# /// script # requires-python = ">=3.10" # dependencies = [] # /// """Course helper: check, push and test the n8n workflows that a CLI coding agent wrote. uv run tools/flow.py doctor # is everything ready? uv run tools/flow.py check my-work/helpdesk.json # find mistakes before n8n sees the file uv run tools/flow.py push my-work/log-report.json # import + publish (one n8n restart) uv run tools/flow.py off my-log-report # unpublish by file or webhook path uv run tools/flow.py diff reference/baseline-report.json my-work/log-report.json [--full] # what changed, node by node uv run tools/flow.py call my-log-report requests/week2.json [-v] uv run tools/flow.py test cases/lab01.json [-v] [--webhook base-report] uv run tools/flow.py logs [--from 2026-10-26 --to 2026-10-26] [--tag before --source live] [--last 10] # the run logs uv run tools/flow.py ops [--reset] # live logs · added FAQ · DLQ · alerts · fault switch (day 77) uv run tools/flow.py bench my-cost requests/faq-vpn.json -n 10 # p50 · p95 · max · failures uv run tools/flow.py usage [--clear] # recorded calls, tokens and cost by model / user / route uv run tools/flow.py budget [--daily 0.001] [--per-min 2] [--reset] # the daily budget and per-user limit uv run tools/flow.py cache [--clear] # the answer cache (Lab 02 task 2) uv run tools/flow.py fault usage slow|error|ok # make the data API slow or failing (drills) uv run tools/flow.py llm down|up # outage drill: a broken OpenAI key in n8n, then restore Every command takes -v (--verbose) for details: the request and response JSON of each call (call, test, bench, fault, doctor), the nodes and links of a workflow (check, push), and the docker / n8n CLI commands that ran. Run it from the day folder (Last3Mon/77). n8n is the day 65 install (~/day65-n8n, Docker project day65-n8n), started from Last3Mon/65/n8n. Environment overrides: N8N_CONTAINER (default: the running n8n service of project day65-n8n, else day65-n8n-n8n-1) N8N_HOME (default ~/day65-n8n), N8N_URL (default http://localhost:5678) """ from __future__ import annotations import argparse import datetime as dt import hashlib import json import math import os import re import shutil import subprocess import sys import time import urllib.error import urllib.request import uuid from pathlib import Path ROOT = Path(__file__).resolve().parent.parent CACHE = Path(__file__).resolve().parent / ".cache" def n8n_home() -> Path: """Where the official n8n install lives: N8N_HOME, else Last3Mon/65/n8n/.env, else ~/day65-n8n.""" if os.getenv("N8N_HOME"): return Path(os.environ["N8N_HOME"]).expanduser() env_file = ROOT.parent / "65" / "n8n" / ".env" if env_file.exists(): for line in env_file.read_text(encoding="utf-8").splitlines(): if line.startswith("N8N_HOME="): value = line.split("=", 1)[1].strip().strip("'\"") if value: return (env_file.parent / Path(value).expanduser()).resolve() return Path.home() / "day65-n8n" N8N_HOME = n8n_home() START_HINT = "cd ~/aio-01d4-stu/Last3Mon/65/n8n && docker compose start (n8n install: ~/day65-n8n)" def find_container() -> str: """The n8n container of the day 65 project, whichever folder it was started from.""" if os.getenv("N8N_CONTAINER"): return os.environ["N8N_CONTAINER"] if shutil.which("docker"): result = subprocess.run(["docker", "ps", "-a", "--filter", "label=com.docker.compose.project=day65-n8n", "--filter", "label=com.docker.compose.service=n8n", "--format", "{{.Names}}"], capture_output=True, text=True) names = result.stdout.split() if names: return names[0] return "day65-n8n-n8n-1" CONTAINER = find_container() N8N_URL = os.getenv("N8N_URL", "http://localhost:5678").rstrip("/") # Versions written in AGENTS.md. Other versions still work if n8n knows them (warning only). RECOMMENDED = { "n8n-nodes-base.webhook": 2, "n8n-nodes-base.respondToWebhook": 1.1, "n8n-nodes-base.if": 2.2, "n8n-nodes-base.switch": 3.2, "n8n-nodes-base.set": 3.4, "n8n-nodes-base.code": 2, "n8n-nodes-base.httpRequest": 4.2, "n8n-nodes-base.httpRequestTool": 4.2, "@n8n/n8n-nodes-langchain.agent": 3.1, "@n8n/n8n-nodes-langchain.lmChatOpenAi": 1.2, "@n8n/n8n-nodes-langchain.vectorStoreInMemory": 1.3, "@n8n/n8n-nodes-langchain.embeddingsOpenAi": 1.2, "@n8n/n8n-nodes-langchain.documentDefaultDataLoader": 1.1, } FAULT_TARGETS = ("faq", "tickets", "prompts", "usage", "dlq") NODE_SETTINGS = ("retryOnFail", "maxTries", "waitBetweenTries", "onError", "alwaysOutputData", "executeOnce") USES_DATA_API = True # day 77: run logs, FAQ (and its additions), alerts (and day 76 tickets · DLQ) come from the data API AGENT_HOOKS: tuple[str, ...] = () # no ready-made agents on day 77 (base-report and my-helpdesk are pushed by hand) COURSE_CREDENTIALS = {"openAiCourse0001": "OpenAI (.env)"} # OpenAI only from day 66 (no Ollama) CONNECTION_TYPES = {"main", "ai_tool", "ai_languageModel", "ai_memory", "ai_outputParser", "ai_embedding", "ai_document"} SECRET = re.compile(r"\bsk-[A-Za-z0-9_-]{16,}") OK, BAD, WARN = "✔", "✘", "!" VERBOSE = False # -v on any command (set in main) # ---------------------------------------------------------------- helpers def die(message: str) -> None: print(f"{BAD} {message}", file=sys.stderr) sys.exit(1) def docker(*args: str, stdin: str | None = None, check: bool = True) -> subprocess.CompletedProcess: if not shutil.which("docker"): die("docker command not found. Start Docker first.") result = subprocess.run(["docker", *args], input=stdin, capture_output=True, text=True) if VERBOSE: show_command(args, result) if check and result.returncode != 0: die(f"docker {' '.join(args[:2])} failed:\n{(result.stderr or result.stdout).strip()}") return result def show_json(label: str, value: object) -> None: """Print one JSON value under a label, indented below the result line (-v). API keys are masked.""" text = json.dumps(value, ensure_ascii=False, indent=2) if not isinstance(value, str) else value print(f" {label}") for row in SECRET.sub("sk-***", text).splitlines(): print(f" {row}") def detail(label: str, value: object) -> None: """show_json, only with -v.""" if VERBOSE: show_json(label, value) def show_command(args: tuple[str, ...], result: subprocess.CompletedProcess) -> None: """-v: one docker command and the end of its output. Data dumps (cat of an export) are not printed.""" command = " ".join(arg if len(arg) <= 90 else arg[:87] + "..." for arg in args) print(f" $ docker {command} → exit {result.returncode}") if "cat /tmp" in " ".join(args): return rows = [row for row in (result.stdout + result.stderr).splitlines() if row.strip()] for row in rows[-6:]: print(f" {SECRET.sub('sk-***', row)[:160]}") def n8n(*args: str, stdin: str | None = None, check: bool = True) -> subprocess.CompletedProcess: return docker("exec", "-i", CONTAINER, "n8n", *args, stdin=stdin, check=check) def container_running() -> bool: if not shutil.which("docker"): return False result = subprocess.run(["docker", "inspect", "-f", "{{.State.Running}}", CONTAINER], capture_output=True, text=True) return result.stdout.strip() == "true" def http(method: str, url: str, body: object | None = None, timeout: float = 300) -> tuple[int, object, float]: data = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8") request = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) start = time.monotonic() try: with urllib.request.urlopen(request, timeout=timeout) as response: status, raw = response.status, response.read() except urllib.error.HTTPError as error: status, raw = error.code, error.read() except (urllib.error.URLError, TimeoutError, ConnectionError) as error: return 0, {"error": str(getattr(error, "reason", error))}, time.monotonic() - start text = raw.decode("utf-8", errors="replace") try: payload: object = json.loads(text) if text.strip() else "" except json.JSONDecodeError: payload = text return status, payload, time.monotonic() - start def load_json(path: Path) -> object: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: die(f"file not found: {path}") except json.JSONDecodeError as error: die(f"{path} is not valid JSON (line {error.lineno}, column {error.colno}): {error.msg}") def resolve(path: str) -> Path: candidate = Path(path) if candidate.exists() or candidate.is_absolute(): return candidate return ROOT / path def node_versions() -> dict[str, list[float]]: """Node types and versions that this n8n really has (exported once, then cached).""" cache = CACHE / "node-versions.json" if cache.exists(): return json.loads(cache.read_text()) if not container_running(): return {} n8n("export:nodes", "--output=/tmp/course-nodes.json") raw = docker("exec", CONTAINER, "sh", "-c", "cat /tmp/course-nodes.json && rm -f /tmp/course-nodes.json").stdout versions: dict[str, set[float]] = {} for node in json.loads(raw): value = node.get("version") versions.setdefault(node["name"], set()).update(value if isinstance(value, list) else [value]) result = {name: sorted(values) for name, values in versions.items()} CACHE.mkdir(exist_ok=True) cache.write_text(json.dumps(result)) return result def walk_strings(value: object, where: str = ""): if isinstance(value, str): yield where, value elif isinstance(value, dict): for key, item in value.items(): yield from walk_strings(item, f"{where}.{key}" if where else key) elif isinstance(value, list): for index, item in enumerate(value): yield from walk_strings(item, f"{where}[{index}]") def webhooks(workflow: dict) -> list[tuple[str, str]]: found = [] for node in workflow.get("nodes", []): if node.get("type") == "n8n-nodes-base.webhook": parameters = node.get("parameters", {}) found.append((str(parameters.get("httpMethod", "GET")).upper(), str(parameters.get("path", "")).strip("/"))) return found # ---------------------------------------------------------------- check def check_workflow(workflow: object) -> tuple[list[str], list[str]]: errors: list[str] = [] warnings: list[str] = [] if not isinstance(workflow, dict): return ["the file must be one workflow object { name, nodes, connections }"], warnings if not str(workflow.get("name", "")).strip(): errors.append("missing workflow 'name'") nodes = workflow.get("nodes") connections = workflow.get("connections") if not isinstance(nodes, list) or not nodes: return errors + ["'nodes' must be a non-empty list"], warnings if not isinstance(connections, dict): errors.append("'connections' must be an object") connections = {} known = node_versions() names: dict[str, dict] = {} for index, node in enumerate(nodes): label = f"node #{index + 1}" if not isinstance(node, dict): errors.append(f"{label} is not an object") continue name = node.get("name") if not name: errors.append(f"{label} has no 'name'") continue label = f"node '{name}'" if name in names: errors.append(f"{label} appears twice (node names must be unique)") names[name] = node node_type, version = node.get("type"), node.get("typeVersion") if not node_type or version is None: errors.append(f"{label} needs 'type' and 'typeVersion'") elif known and node_type not in known: errors.append(f"{label}: type '{node_type}' does not exist in this n8n") elif known and version not in known[node_type]: errors.append(f"{label}: {node_type} has no typeVersion {version} (available: {known[node_type]})") elif node_type in RECOMMENDED and version != RECOMMENDED[node_type]: warnings.append(f"{label}: typeVersion {version} (AGENTS.md uses {RECOMMENDED[node_type]})") if not isinstance(node.get("parameters"), dict): errors.append(f"{label} needs 'parameters' (use {{}} when empty)") position = node.get("position") if not (isinstance(position, list) and len(position) == 2): errors.append(f"{label} needs 'position': [x, y]") for kind, credential in (node.get("credentials") or {}).items(): credential_id = (credential or {}).get("id") if credential_id not in COURSE_CREDENTIALS: errors.append(f"{label}: credential '{kind}' must use a course id {list(COURSE_CREDENTIALS)}, got {credential_id!r}") for where, text in walk_strings(node.get("parameters", {})): if SECRET.search(text): errors.append(f"{label}: parameters.{where} looks like an API key. Use a credential instead.") if "{{" in text and "}}" in text and not text.startswith("="): warnings.append(f"{label}: parameters.{where} has {{{{ }}}} but does not start with '=' (it stays plain text)") # day 75: OpenAI no longer serves gpt-4.1-nano, and gpt-5.4-nano refuses max_tokens over HTTP if "gpt-4.1-nano" in text: errors.append(f"{label}: parameters.{where} uses gpt-4.1-nano (OpenAI no longer serves it). Use gpt-5.4-nano.") if "chat/completions" in str(node.get("parameters", {}).get("url", "")) and re.search(r"\bmax_tokens\b", text): errors.append(f"{label}: parameters.{where} has max_tokens. Use max_completion_tokens (gpt-5.4-nano refuses max_tokens).") # day 76: node settings live next to "parameters", not inside it (n8n ignores them there) misplaced = [key for key in NODE_SETTINGS if key in (node.get("parameters") or {})] if misplaced: errors.append(f"{label}: {', '.join(misplaced)} must be at the node level next to 'parameters', not inside it") if node.get("retryOnFail") and not 1 < int(node.get("maxTries") or 3) <= 5: errors.append(f"{label}: maxTries must be 2..5 with retryOnFail (n8n allows up to 5)") if "onError" in node and node["onError"] not in ("stopWorkflow", "continueRegularOutput", "continueErrorOutput"): errors.append(f"{label}: onError must be stopWorkflow, continueRegularOutput or continueErrorOutput") targets_of: dict[str, set[str]] = {} for source, outputs in connections.items(): if source not in names: errors.append(f"connections: source '{source}' is not a node name") continue if not isinstance(outputs, dict): errors.append(f"connections['{source}'] must be an object like {{\"main\": [[...]]}}") continue for kind, branches in outputs.items(): if kind not in CONNECTION_TYPES: errors.append(f"connections['{source}']: unknown connection type '{kind}'") for branch in branches or []: for link in branch or []: target = (link or {}).get("node") if target not in names: errors.append(f"connections['{source}'] points to '{target}', which is not a node name") elif link.get("type") != kind: errors.append(f"connections['{source}'] -> '{target}': link type must be '{kind}'") else: targets_of.setdefault(target, set()).add(kind) if kind != "main": targets_of.setdefault(f"{source}::from", set()).add(kind) has_respond = any(node.get("type") == "n8n-nodes-base.respondToWebhook" for node in names.values()) for name, node in names.items(): node_type = node.get("type", "") parameters = node.get("parameters", {}) if isinstance(node.get("parameters"), dict) else {} if node_type == "n8n-nodes-base.webhook": if not str(parameters.get("path", "")).strip("/"): errors.append(f"node '{name}': webhook needs a 'path'") if parameters.get("responseMode") == "responseNode" and not has_respond: errors.append(f"node '{name}': responseMode is 'responseNode' but there is no Respond to Webhook node") if node_type == "@n8n/n8n-nodes-langchain.agent" and "ai_languageModel" not in targets_of.get(name, set()): errors.append(f"node '{name}': AI Agent needs a chat model connected with 'ai_languageModel'") if node_type.endswith("Tool") or node_type.startswith("@n8n/n8n-nodes-langchain.tool"): if "ai_tool" not in targets_of.get(f"{name}::from", set()): errors.append(f"node '{name}': a tool must be connected to an AI Agent with 'ai_tool'") if node_type.startswith("@n8n/n8n-nodes-langchain.lmChat"): if "ai_languageModel" not in targets_of.get(f"{name}::from", set()): errors.append(f"node '{name}': a chat model must be connected to an AI Agent with 'ai_languageModel'") is_sub_node = bool(targets_of.get(f"{name}::from")) # model, tool, embeddings, loader: connected by ai_* links if node_type != "n8n-nodes-base.webhook" and not node_type.startswith("@n8n/n8n-nodes-langchain.lm") and not is_sub_node \ and not (node_type.endswith("Tool") or node_type.startswith("@n8n/n8n-nodes-langchain.tool")) \ and "main" not in targets_of.get(name, set()) and "trigger" not in node_type.lower(): warnings.append(f"node '{name}': nothing is connected into it (it never runs)") error_workflow = (workflow.get("settings") or {}).get("errorWorkflow") if error_workflow and not re.fullmatch(r"[A-Za-z0-9]{10,21}", str(error_workflow)): errors.append(f"settings.errorWorkflow must be the id of the error workflow (e.g. errHandler76001), got {error_workflow!r}") if any(node.get("type") == "n8n-nodes-base.errorTrigger" for node in names.values()) and not workflow.get("id"): warnings.append("an error workflow needs a fixed top-level \"id\" (other workflows name it in settings.errorWorkflow)") if not webhooks(workflow) and not any(node.get("type") == "n8n-nodes-base.errorTrigger" for node in names.values()): warnings.append("no Webhook node: 'push' works, but 'call' and 'test' need a webhook path") return errors, warnings def show_workflow(workflow: object) -> None: """-v: the nodes, links and webhooks of a workflow file.""" if not VERBOSE or not isinstance(workflow, dict): return print(f" workflow '{workflow.get('name')}' · {len(workflow.get('nodes') or [])} node(s)") for node in workflow.get("nodes") or []: if isinstance(node, dict): extra = [key for key in ("onError", "retryOnFail") if key in node] print(f" {node.get('name')} · {node.get('type')} v{node.get('typeVersion')}" + (f" · {', '.join(extra)}" if extra else "")) print(" links") for source, outputs in (workflow.get("connections") or {}).items(): for kind, branches in (outputs or {}).items(): for index, branch in enumerate(branches or []): targets = ", ".join(str((link or {}).get("node")) for link in branch or []) if targets: print(f" {source} --{kind}[{index}]--> {targets}") for method, hook in webhooks(workflow): print(f" webhook {method} /webhook/{hook}") def cmd_check(args: argparse.Namespace) -> int: path = resolve(args.file) workflow = load_json(path) show_workflow(workflow) errors, warnings = check_workflow(workflow) for warning in warnings: print(f"{WARN} {warning}") for error in errors: print(f"{BAD} {error}") if errors: print(f"{BAD} {path.name}: {len(errors)} error(s). Fix them and run check again.") return 1 print(f"{OK} {path.name}: no errors" + (f" ({len(warnings)} warning(s))" if warnings else "")) return 0 # ---------------------------------------------------------------- push def stable_id(path: Path) -> str: return "w" + hashlib.sha1(path.stem.encode("utf-8")).hexdigest()[:15] def wait_until_ready(paths: list[tuple[str, str]], seconds: int = 90) -> bool: deadline = time.monotonic() + seconds while time.monotonic() < deadline: status, _, _ = http("GET", f"{N8N_URL}/healthz", timeout=3) if status == 200: break time.sleep(1) else: return False for method, path in paths: while time.monotonic() < deadline: # A webhook that is registered for POST answers a GET with a hint, not "is not registered". status, payload, _ = http("GET" if method != "GET" else "HEAD", f"{N8N_URL}/webhook/{path}", timeout=5) text = json.dumps(payload, ensure_ascii=False) if "not registered for" in text or status not in (0, 404, 500, 502, 503): break time.sleep(1) time.sleep(2) # webhooks of the other published workflows finish registering right after return time.monotonic() < deadline def exported_workflows() -> list[dict]: exported = docker("exec", CONTAINER, "sh", "-c", "n8n export:workflow --all --output=/tmp/course-all.json >/dev/null 2>&1; " "cat /tmp/course-all.json 2>/dev/null; rm -f /tmp/course-all.json", check=False).stdout try: return json.loads(exported) if exported.strip() else [] except json.JSONDecodeError: return [] def prepare(path: Path) -> dict | None: workflow = load_json(path) show_workflow(workflow) errors, warnings = check_workflow(workflow) for warning in warnings: print(f"{WARN} {path.name}: {warning}") if errors: for error in errors: print(f"{BAD} {path.name}: {error}") return None workflow = dict(workflow) if not re.fullmatch(r"[A-Za-z0-9]{10,}", str(workflow.get("id", ""))): workflow["id"] = stable_id(path) workflow["active"] = False workflow.setdefault("settings", {}).setdefault("executionOrder", "v1") for node in workflow["nodes"]: node.setdefault("id", str(uuid.uuid5(uuid.NAMESPACE_URL, f"{workflow['id']}/{node['name']}"))) if node["type"] == "n8n-nodes-base.webhook": node.setdefault("webhookId", str(uuid.uuid5(uuid.NAMESPACE_URL, f"{workflow['id']}/webhook/{node['name']}"))) return workflow def cmd_push(args: argparse.Namespace) -> int: paths = [resolve(file) for file in args.files] workflows = [prepare(path) for path in paths] if None in workflows: print(f"{BAD} not pushed. Fix the errors first (uv run tools/flow.py check ).") return 1 if not container_running(): die(f"n8n container '{CONTAINER}' is not running. {START_HINT}") mine_ids = {workflow["id"] for workflow in workflows} hooks = {hook for workflow in workflows for hook in webhooks(workflow)} # Another published workflow with the same webhook path would block this one. for other in exported_workflows(): if other.get("id") not in mine_ids and other.get("active") and hooks & set(webhooks(other)): n8n("unpublish:workflow", f"--id={other['id']}") print(f"{WARN} unpublished '{other.get('name')}' ({other['id']}): it used the same webhook path") for workflow in workflows: docker("exec", "-i", CONTAINER, "sh", "-c", "cat > /tmp/course-workflow.json && n8n import:workflow --input=/tmp/course-workflow.json; " "status=$?; rm -f /tmp/course-workflow.json; exit $status", stdin=json.dumps(workflow, ensure_ascii=False)) n8n("publish:workflow", f"--id={workflow['id']}") # The CLI changes the database only; one restart makes the running n8n load the new versions (about 5 s). docker("restart", CONTAINER) ready = wait_until_ready(sorted(hooks)) for workflow in workflows: print(f"{OK} pushed '{workflow['name']}' (id {workflow['id']})") for method, hook in sorted(webhooks(workflow)): print(f" {method} {N8N_URL}/webhook/{hook}") if not ready: print(f"{WARN} n8n is still starting. Wait a few seconds before 'call' or 'test'.") return 0 def cmd_off(args: argparse.Namespace) -> int: """Unpublish workflows, given as files or webhook paths, to see how the others behave when one role is down.""" if not container_running(): die(f"n8n container '{CONTAINER}' is not running. {START_HINT}") exported = exported_workflows() detail("published now", [f"{item.get('id')} · {item.get('name')} · " + ", ".join(f"/{hook}" for _, hook in webhooks(item)) for item in exported if item.get("active")]) wanted: dict[str, str] = {} for item in args.files: path = resolve(item) if (ROOT / item).exists() or Path(item).exists() else None if path: workflow = load_json(path) workflow_id = workflow.get("id") if re.fullmatch(r"[A-Za-z0-9]{10,}", str(workflow.get("id", ""))) else stable_id(path) wanted[workflow_id] = item hooks = {hook for _, hook in webhooks(workflow)} else: hooks = {item.strip("/")} for other in exported: if other.get("active") and hooks & {hook for _, hook in webhooks(other)}: wanted[other["id"]] = item active = {item.get("id") for item in exported if item.get("active")} for workflow_id, item in wanted.items(): if workflow_id in active: n8n("unpublish:workflow", f"--id={workflow_id}") print(f"{OK} unpublished {item} (id {workflow_id})") docker("restart", CONTAINER) wait_until_ready([]) print(f"{OK} n8n restarted. Turn it back on with: uv run tools/flow.py push ") return 0 # ---------------------------------------------------------------- call / test def request_body(spec: str | dict, base: Path = ROOT) -> object: if isinstance(spec, dict): return spec text = spec.strip() if text.startswith("{"): try: return json.loads(text) except json.JSONDecodeError as error: die(f"request is not valid JSON: {error.msg}") path = Path(text) if not path.exists(): path = base / text return load_json(path) def cmd_call(args: argparse.Namespace) -> int: body = request_body(args.request) if args.request else {} hook = args.path.strip('/') status, payload, seconds = http("POST", f"{N8N_URL}/webhook/{hook}", body) print(f"HTTP {status} · {seconds:.1f}s") if args.verbose: show_json(f"→ request POST /webhook/{hook}", body) show_json(f"← response HTTP {status}", payload) return 0 if 200 <= status < 300 else 1 print(json.dumps(payload, ensure_ascii=False, indent=2) if not isinstance(payload, str) else payload) return 0 if 200 <= status < 300 else 1 def field(payload: object, dotted: str) -> object: value = payload for part in dotted.split("."): if isinstance(value, dict): value = value.get(part) elif isinstance(value, list) and part.isdigit() and int(part) < len(value): value = value[int(part)] else: return None return value def as_text(value: object) -> str: return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) def judge(expect: dict, status: int, payload: object, seconds: float = 0.0) -> list[str]: problems = [] if "max_seconds" in expect and seconds > float(expect["max_seconds"]): problems.append(f"took {seconds:.1f}s (expected at most {expect['max_seconds']}s)") if status != expect.get("status", 200): problems.append(f"status {status} (expected {expect.get('status', 200)})") for name, wanted in (expect.get("equals") or {}).items(): actual = field(payload, name) if actual != wanted: problems.append(f"{name} = {as_text(actual)[:60]} (expected {as_text(wanted)})") for name, bounds in (expect.get("range") or {}).items(): actual = field(payload, name) low, high = bounds if not isinstance(actual, (int, float)) or isinstance(actual, bool) or not low <= actual <= high: problems.append(f"{name} = {as_text(actual)[:40]} (expected {low}..{high})") for name, words in (expect.get("contains") or {}).items(): words = [words] if isinstance(words, str) else words text = as_text(field(payload, name) or "") if not any(word in text for word in words): problems.append(f"{name} has none of {words}") for name, words in (expect.get("not_contains") or {}).items(): words = [words] if isinstance(words, str) else words text = as_text(field(payload, name) or "") leaked = [word for word in words if word in text] if leaked: problems.append(f"{name} must not have {leaked}") return problems def cmd_test(args: argparse.Namespace) -> int: path = resolve(args.cases) suite = load_json(path) hook = (args.webhook or suite.get("webhook", "")).strip("/") cases = suite.get("cases", []) if not hook or not cases: die(f"{path.name} needs \"webhook\" and a non-empty \"cases\" list") # "before"/"after": calls made once around the cases, e.g. turn a fault on and off again. if not args.no_before: for step in suite.get("before", []): run_step("before", step) try: return run_cases(args, path, suite, hook, cases) finally: for step in suite.get("after", []): run_step("after", step) def run_step(label: str, step: dict) -> None: """A step is {"webhook": path} (POST to n8n), {"url": ...} (POST to the data API, e.g. the fault switch) or {"llm": "down"|"up"} (the outage drill) or {"sleep": 3} (wait, e.g. for an error workflow that runs late). "status": 409 means that status is expected (e.g. a refused deploy).""" if "sleep" in step: print(f"… {label}: wait {step['sleep']}s") time.sleep(float(step["sleep"])) return if step.get("llm") in ("down", "up"): code = cmd_llm(argparse.Namespace(state=step["llm"])) if code and label == "before": die("could not change the OpenAI credential (llm drill)") if step["llm"] == "down": time.sleep(1) return if step.get("url"): target = str(step["url"]) else: target = f"{N8N_URL}/webhook/{str(step.get('webhook', '')).strip('/')}" body = request_body(step.get("body") or step.get("request", "{}")) status, payload, seconds = http("POST", target, body) shown = json.dumps(payload, ensure_ascii=False)[:120] if not isinstance(payload, str) else payload[:120] good = status == step["status"] if "status" in step else 200 <= status < 300 print(f"{OK if good else BAD} {label}: POST {target.replace(N8N_URL, '')} · HTTP {status} · {seconds:.1f}s · {shown}") detail(f"→ request POST {target.replace(N8N_URL, '')}", body) detail(f"← response HTTP {status}", payload) if not good and label == "before": die(f"the 'before' step failed ({target}). Is it pushed / is the data API the day 77 version?") def parse_sets(items: list[str] | None) -> dict: """--set release=1.1.0 --set no_cache=true → {"release": "1.1.0", "no_cache": true}""" out: dict = {} for item in items or []: key, _, value = item.partition("=") try: out[key] = json.loads(value) except json.JSONDecodeError: out[key] = value return out def run_cases(args: argparse.Namespace, path: Path, suite: dict, hook: str, cases: list) -> int: print(f"POST {N8N_URL}/webhook/{hook} · {len(cases)} case(s) × {args.repeat}") results, passed = [], 0 for case in cases: for step in case.get("before", []): run_step("before", step) try: passed += run_case(args, case, hook, results) finally: for step in case.get("after", []): run_step("after", step) return finish(path, hook, results, passed) def run_case(args: argparse.Namespace, case: dict, hook: str, results: list) -> int: passed = 0 body = request_body(case.get("body") or case.get("request", "{}")) if getattr(args, "set", None) and isinstance(body, dict): body = {**body, **parse_sets(args.set)} case_hook = (case.get("webhook") or hook).strip("/") if not args.webhook else hook # "url": POST to that address instead of the webhook (e.g. check /usage/summary after the questions) target = str(case["url"]) if case.get("url") else f"{N8N_URL}/webhook/{case_hook}" shown = target.replace(N8N_URL, "") for run in range(1, args.repeat + 1): status, payload, seconds = http("POST", target, body) problems = judge(case.get("expect", {}), status, payload, seconds) passed += not problems mark = OK if not problems else BAD suffix = f" #{run}" if args.repeat > 1 else "" print(f"{mark} {case.get('name', '?')}{suffix} · {seconds:.1f}s{cost_note(payload)}" + ("" if not problems else " · " + "; ".join(problems))) if args.verbose: show_json(f"→ request POST {shown}", body) show_json(f"← response HTTP {status}", payload) results.append({"case": case.get("name"), "webhook": case_hook if not case.get("url") else shown, "run": run, "passed": not problems, "problems": problems, "seconds": round(seconds, 1), "status": status, "request": body, "response": payload}) return passed def cost_note(payload: object) -> str: """' · 자동 답변 · 1 call · $0.000123' when the answer says how many LLM calls it made and what they cost (day 75).""" if not isinstance(payload, dict) or not isinstance(payload.get("cost_usd"), (int, float)): return "" calls = payload.get("llm_calls") route = f" · {payload['route']}" if isinstance(payload.get("route"), str) else "" return route + (f" · {calls} call{'s' if calls != 1 else ''}" if isinstance(calls, int) else "") + f" · ${payload['cost_usd']:.6f}" LAST_RESULTS: list = [] def finish(path: Path, hook: str, results: list, passed: int) -> int: LAST_RESULTS[:] = results total = len(results) print(f"Passed {passed}/{total}") runs = ROOT / "runs" runs.mkdir(exist_ok=True) stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S") report = runs / f"{path.stem}-{stamp}.json" report.write_text(json.dumps({"cases": str(path.name), "webhook": hook, "passed": passed, "total": total, "results": results}, ensure_ascii=False, indent=2), encoding="utf-8") print(f"report: {report.relative_to(ROOT)}") return 0 if passed == total else 1 # ---------------------------------------------------------------- gate / diff / release def cmd_gate(args: argparse.Namespace) -> int: """A release gate: run a test file and decide PASS/FAIL from the pass rate and p95 of the case times.""" cmd_test(args) results = LAST_RESULTS total = len(results) passed = sum(1 for item in results if item["passed"]) times = [item["seconds"] for item in results if item.get("status") == 200] p95 = percentile(times, 0.95) if times else 0.0 rate = passed / total if total else 0.0 reasons = [] if rate < args.min_pass: reasons.append(f"pass rate {rate:.0%} < {args.min_pass:.0%}") if args.max_p95 is not None and p95 > args.max_p95: reasons.append(f"p95 {p95:.1f}s > {args.max_p95}s") sets = parse_sets(args.set) label = ", ".join(f"{k}={v}" for k, v in sets.items()) or "as is" print(f"gate [{label}]: pass {passed}/{total} ({rate:.0%}) · p95 {p95:.1f}s → " + ("PASS" if not reasons else "FAIL · " + "; ".join(reasons))) if sets.get("release") and not args.no_record: record = {"release": str(sets["release"]), "verdict": "rollback" if reasons else "promote", "passed": passed, "total": total, "p95": round(p95, 2), "cases": Path(args.cases).name, "source": "flow.py gate"} status, payload, _ = http("POST", f"http://localhost:8800/gates/{args.name}", record, timeout=10) print(f"{OK if status == 200 else WARN} gate record → /gates/{args.name}: {record['release']} {record['verdict']}" + ("" if status == 200 else f" (not stored: HTTP {status})")) detail(f"← response HTTP {status}", payload) return 0 if not reasons else 1 def flat(value: object, prefix: str = "") -> dict: if isinstance(value, dict): out: dict = {} for key, item in value.items(): out.update(flat(item, f"{prefix}.{key}" if prefix else str(key))) return out if isinstance(value, list): out = {} for index, item in enumerate(value): out.update(flat(item, f"{prefix}[{index}]")) return out or {prefix: []} return {prefix: value} def cmd_diff(args: argparse.Namespace) -> int: """Node-by-node difference of two workflow files: added/removed nodes, changed settings, links.""" old, new = load_json(resolve(args.old)), load_json(resolve(args.new)) old_nodes = {node["name"]: node for node in old.get("nodes", [])} new_nodes = {node["name"]: node for node in new.get("nodes", [])} print(f"{old.get('name', args.old)} → {new.get('name', args.new)}") changed = 0 for name in sorted(set(new_nodes) - set(old_nodes)): changed += 1 print(f"+ node '{name}' ({new_nodes[name].get('type', '?').split('.')[-1]})") for name in sorted(set(old_nodes) - set(new_nodes)): changed += 1 print(f"- node '{name}' ({old_nodes[name].get('type', '?').split('.')[-1]})") skip = ("position", "id", "webhookId") for name in sorted(set(old_nodes) & set(new_nodes)): a = flat({k: v for k, v in old_nodes[name].items() if k not in skip}) b = flat({k: v for k, v in new_nodes[name].items() if k not in skip}) for key in sorted(set(a) | set(b)): if a.get(key) == b.get(key): continue changed += 1 before, after = as_text(a.get(key, "(none)")), as_text(b.get(key, "(none)")) if len(before) > 70 or len(after) > 70: print(f"~ '{name}' {key}: changed ({len(before)} → {len(after)} chars)") if args.full or args.verbose: print(f" before: {before}\n after: {after}") else: print(f"~ '{name}' {key}: {before} → {after}") def links(workflow: dict) -> set: return {(src, kind, dst["node"]) for src, kinds in workflow.get("connections", {}).items() for kind, outs in kinds.items() for out in outs for dst in out} for src, kind, dst in sorted(links(new) - links(old)): changed += 1 print(f"+ link {src} → {dst} ({kind})") for src, kind, dst in sorted(links(old) - links(new)): changed += 1 print(f"- link {src} → {dst} ({kind})") print(f"{changed} change(s)") return 0 def bucket(user: str) -> int: """Same rule as AGENTS.md: sum of the character codes of the user name, modulo 100.""" return sum(ord(ch) for ch in user) % 100 def cmd_release(args: argparse.Namespace) -> int: url = f"http://localhost:8800/releases/{args.name}" if args.reset: status, payload, _ = http("POST", url + "/reset", {}, timeout=10) elif args.require_gate: status, payload, _ = http("POST", url + "/policy", {"require_gate": args.require_gate == "on"}, timeout=10) elif args.version or args.canary or args.no_canary: body: dict = {} if args.version: body["active"] = args.version if args.canary: body["canary"] = {"release": args.canary, "percent": args.percent} if args.no_canary: body["canary"] = None status, payload, _ = http("POST", url, body, timeout=10) else: status, payload, _ = http("GET", url, timeout=10) detail(f"← response HTTP {status}", payload) if status == 409 and isinstance(payload, dict): gate = payload.get("last_gate") print(f"{BAD} refused: {payload.get('release')} has no passing gate record" + (f" (last gate: {gate.get('verdict')} {gate.get('passed')}/{gate.get('total')} · {gate.get('source')})" if gate else " (never gated)") + "\n → run the gate (flow.py gate ... --set release=) or my-release-check first") return 1 if status != 200 or not isinstance(payload, dict): print(f"{BAD} {as_text(payload)}") return 1 _, gates, _ = http("GET", f"http://localhost:8800/gates/{args.name}", timeout=10) latest = {} for record in (gates.get("gates", []) if isinstance(gates, dict) else []): latest[record.get("release")] = record canary = payload.get("canary") print(f"{OK} {args.name}: active {payload['active']}" + (f" · canary {canary['release']} {canary['percent']}%" if canary else " · no canary") + (" · deploy needs a passing gate" if payload.get("require_gate") else "")) for version, info in sorted(payload.get("releases", {}).items()): mark = "*" if version == payload["active"] else ("c" if canary and version == canary["release"] else " ") gate = latest.get(version) seen = f" gate {gate['verdict']} {gate.get('passed')}/{gate.get('total')}" if gate else "" print(f" {mark} {version} prompt {info.get('prompt_version')} model {info.get('model')} {info.get('note', '')}{seen}") if canary: names = ["이준호", "박서연", "최유진", "한지훈", "김민지", "정우성"] print(" canary users: " + (", ".join(f"{n}({bucket(n)})" for n in names if bucket(n) < canary["percent"]) or "none")) return 0 # ---------------------------------------------------------------- bench / fault / llm def percentile(values: list[float], share: float) -> float: ordered = sorted(values) if not ordered: return 0.0 # nearest rank, as in the lecture: position = ceil(share × n), counted from 1 (10 runs: p50 = 5th, p95 = 10th) index = min(len(ordered) - 1, max(0, math.ceil(share * len(ordered) - 1e-9) - 1)) return ordered[index] # ---------------------------------------------------------------- bench / fault / llm def percentile(values: list[float], share: float) -> float: ordered = sorted(values) if not ordered: return 0.0 # nearest rank, as in the lecture: position = ceil(share × n), counted from 1 (10 runs: p50 = 5th, p95 = 10th) index = min(len(ordered) - 1, max(0, math.ceil(share * len(ordered) - 1e-9) - 1)) return ordered[index] def cmd_bench(args: argparse.Namespace) -> int: body = request_body(args.request) if args.request else {} if args.set and isinstance(body, dict): body = {**body, **parse_sets(args.set)} hook = args.path.strip("/") print(f"POST {N8N_URL}/webhook/{hook} × {args.n}") detail(f"→ request POST /webhook/{hook} (the same body every time)", body) seconds_list, failures, rows = [], 0, [] for run in range(1, args.n + 1): status, payload, seconds = http("POST", f"{N8N_URL}/webhook/{hook}", body) good = 200 <= status < 300 failures += not good seconds_list.append(seconds) extra = "" if isinstance(payload, dict): extra = " · " + ", ".join(f"{key}={as_text(payload[key])[:40]}" for key in ("route", "llm_calls", "cached") if key in payload) if isinstance(payload.get("version"), dict): extra += f", release={payload['version'].get('release')}" print(f"{OK if good else BAD} #{run} · HTTP {status} · {seconds:.2f}s{extra}") detail(f"← response HTTP {status}", payload) rows.append({"run": run, "status": status, "seconds": round(seconds, 3), "response": payload}) p50, p95 = percentile(seconds_list, 0.50), percentile(seconds_list, 0.95) print(f"p50 {p50:.2f}s · p95 {p95:.2f}s · max {max(seconds_list):.2f}s · failures {failures}/{args.n}") if VERBOSE: print(" sorted " + " ".join(f"{value:.2f}" for value in sorted(seconds_list))) steps: dict[str, list[float]] = {} for row in rows: timings = row["response"].get("timings") if isinstance(row["response"], dict) else None for key, value in (timings or {}).items(): if isinstance(value, (int, float)): steps.setdefault(key, []).append(value) if steps: print(" timings (average ms, runs) " + " · ".join(f"{key} {sum(values) / len(values):.0f} ({len(values)})" for key, values in steps.items())) runs = ROOT / "runs" runs.mkdir(exist_ok=True) report = runs / f"bench-{hook}-{dt.datetime.now().strftime('%Y%m%d-%H%M%S')}.json" report.write_text(json.dumps({"webhook": hook, "request": body, "n": args.n, "p50": round(p50, 3), "p95": round(p95, 3), "max": round(max(seconds_list), 3), "failures": failures, "runs": rows}, ensure_ascii=False, indent=2), encoding="utf-8") print(f"report: {report.relative_to(ROOT)}") return 0 if not failures else 1 # ---------------------------------------------------------------- logs (day 77) def cmd_logs(args: argparse.Namespace) -> int: body = {"from": args.date_from or "", "to": args.date_to or "", "tag": args.tag or "", "source": args.source} status, payload, seconds = http("POST", "http://localhost:8800/logs/query", body, timeout=20) detail(f"→ POST http://localhost:8800/logs/query · HTTP {status} · {seconds:.2f}s", body) if status != 200 or not isinstance(payload, dict): print(f"{BAD} {as_text(payload)} (is the data API the day 77 version? uv run tools/flow.py doctor)") return 1 rows = payload["logs"] errors = sum(1 for r in rows if r.get("status") == "error") print(f"{len(rows)} run(s) · errors {errors}" + (f" ({errors / len(rows):.1%})" if rows else "")) routes: dict[str, int] = {} for row in rows: key = str(row.get("route") or "오류") routes[key] = routes.get(key, 0) + 1 if routes: print(" route: " + " · ".join(f"{k} {v}" for k, v in sorted(routes.items(), key=lambda kv: -kv[1]))) for row in rows[-args.last:] if args.last else []: print(f" {row.get('started_at')} {str(row.get('status')):<7} {str(row.get('route')):<6} {row.get('duration_ms'):>5}ms " f"{row.get('topic')} {str(row.get('error_node') or '')} {str(row.get('question'))[:24]}") if rows: detail("← last run (all fields)", rows[-1]) return 0 # ---------------------------------------------------------------- ops (day 76 · 77) def cmd_ops(args: argparse.Namespace) -> int: base = "http://localhost:8800" if args.reset: for path in ("/tickets-reset", "/dlq-clear", "/alerts-clear", "/logs-reset", "/faq-reset"): status, payload, seconds = http("POST", base + path, {}, timeout=10) detail(f"→ POST {base + path} · HTTP {status} · {seconds:.2f}s", payload) body = {target: "ok" for target in FAULT_TARGETS} status, payload, _ = http("POST", base + "/admin/fault", body, timeout=10) detail(f"→ POST {base}/admin/fault · HTTP {status}", payload) print(f"{OK if status == 200 else BAD} live run logs, added FAQ, created tickets, DLQ and alerts emptied, fault switch all ok") return 0 if status == 200 else 1 _, created, _ = http("GET", base + "/tickets-created", timeout=10) _, dlq, _ = http("GET", base + "/dlq", timeout=10) _, alerts, _ = http("GET", base + "/alerts", timeout=10) _, fault, _ = http("GET", base + "/admin/fault", timeout=10) if not all(isinstance(x, dict) for x in (created, dlq, alerts, fault)): print(f"{BAD} data API did not answer: is it the day 77 version? (uv run tools/flow.py doctor)") return 1 on = {key: value for key, value in fault.items() if key in FAULT_TARGETS and value != "ok"} if "slow" in on.values(): on["slow_seconds"] = fault.get("slow_seconds") print(f"fault switch: {'all ok' if not on else as_text(on)}") _, live, _ = http("GET", base + "/logs?source=live", timeout=10) _, extra, _ = http("GET", base + "/faq-extra", timeout=10) if isinstance(live, dict): tags: dict[str, int] = {} for row in live.get("logs", []): tags[str(row.get("tag") or "-")] = tags.get(str(row.get("tag") or "-"), 0) + 1 print(f"live run logs: {live.get('count')}" + (" (" + " · ".join(f"tag {k} {v}" for k, v in tags.items()) + ")" if tags else "")) if isinstance(extra, dict): print(f"added FAQ: {extra.get('count')}" + (" (" + ", ".join(str(f.get("id")) for f in extra.get("faq", [])) + ")" if extra.get("count") else "")) print(f"created tickets: {created.get('count')}") for item in created.get("tickets", [])[-5:]: print(f" {item.get('ticket_id')} key {item.get('key')} {item.get('category')}/{item.get('priority')} {str(item.get('title'))[:30]}") print(f"DLQ: {dlq.get('size')} (parked {dlq.get('parked', 0)})") for item in dlq.get("items", [])[-5:]: print(f" {item.get('dlq_id')} {(item.get('item') or {}).get('id')} attempts {item.get('attempts')} " f"parked {item.get('parked')} {str(item.get('reason'))[:50]}") print(f"alerts: {alerts.get('count')}") for item in alerts.get("alerts", [])[-3:]: extra = f" · count {item['count']}" if "count" in item else "" print(f" [{item.get('severity')}] {item.get('workflow')} · {item.get('reason')} · {item.get('node', '-')}{extra} · {str(item.get('message'))[:60]}") detail("← GET /dlq", dlq) detail("← GET /alerts", alerts) return 0 # ---------------------------------------------------------------- usage / budget (day 75) def cmd_usage(args: argparse.Namespace) -> int: if args.clear: status, payload, _ = http("POST", "http://localhost:8800/usage-clear", {}, timeout=10) print(f"{OK if status == 200 else BAD} usage cleared: {as_text(payload)}") return 0 if status == 200 else 1 status, payload, seconds = http("GET", "http://localhost:8800/usage/summary", timeout=10) detail(f"← GET /usage/summary · HTTP {status} · {seconds:.2f}s", payload) if status != 200 or not isinstance(payload, dict): print(f"{BAD} {as_text(payload)}") return 1 print(f"requests {payload['requests']} · calls {payload['calls']} · tokens in {payload['prompt_tokens']}" f" / out {payload['completion_tokens']} · ${payload['cost_usd']:.6f}") for title in ("by_model", "by_user", "by_route"): for key, row in sorted(payload.get(title, {}).items(), key=lambda kv: -kv[1]["cost_usd"]): print(f" {title[3:]:<6} {key:<14} {row['calls']:>4} calls · in {row['prompt_tokens']:>7}" f" · out {row['completion_tokens']:>6} · ${row['cost_usd']:.6f}") show_budget() return 0 def show_budget() -> dict: status, b, seconds = http("GET", "http://localhost:8800/budget", timeout=10) detail(f"← GET /budget · HTTP {status} · {seconds:.2f}s", b) if isinstance(b, dict) and "daily_usd" in b: print(f"budget ${b['daily_usd']} / used ${b['used_usd']:.6f} · remaining ${b['remaining_usd']:.6f} · over {b['over']}" f" · per user {b.get('user_limit_per_min')}/min · question {b.get('max_question_chars')} chars") return b print(f"{BAD} budget: {as_text(b)}") return {} def cmd_budget(args: argparse.Namespace) -> int: """Show or change the daily budget, the per-user limit and the question length limit (people decide these).""" body = {key: value for key, value in (("daily_usd", args.daily), ("user_limit_per_min", args.per_min), ("max_question_chars", args.chars)) if value is not None} if args.reset or body: target = "http://localhost:8800/budget/reset" if args.reset else "http://localhost:8800/budget" status, payload, seconds = http("POST", target, body, timeout=10) detail(f"→ request POST {target}", body) detail(f"← response HTTP {status} · {seconds:.2f}s", payload) if status != 200: print(f"{BAD} {as_text(payload)}") return 1 print(f"{OK} " + ("budget back to data/budget.json" if args.reset else f"budget changed: {as_text(body)}")) return 0 if show_budget() else 1 def cmd_fault(args: argparse.Namespace) -> int: targets = list(FAULT_TARGETS) if args.target == "all" else [args.target] body: dict = {target: args.mode for target in targets} if args.seconds is not None: body["slow_seconds"] = args.seconds status, payload, seconds = http("POST", "http://localhost:8800/admin/fault", body, timeout=10) print(f"{OK if status == 200 else BAD} fault switch: {as_text(payload)}") detail("→ request POST http://localhost:8800/admin/fault", body) detail(f"← response HTTP {status} · {seconds:.2f}s", payload) return 0 if status == 200 else 1 def cmd_llm(args: argparse.Namespace) -> int: """Outage drill: register the course OpenAI credential with a broken key (down) or the real key from .env (up).""" script = ROOT.parent / "65" / "n8n" / "register-credentials.sh" if not script.exists(): die(f"{script} not found") env = dict(os.environ, N8N_CONTAINER=CONTAINER) if args.state == "down": env["OPENAI_API_KEY"] = "sk-invalid-key-for-drill-000000" else: env.pop("OPENAI_API_KEY", None) result = subprocess.run(["bash", str(script)], env=env, capture_output=True, text=True) detail(f"$ bash {script.relative_to(ROOT.parent)} → exit {result.returncode}", (result.stdout + result.stderr).strip()) if result.returncode != 0: print(result.stderr.strip() or result.stdout.strip()) return 1 print(f"{OK} OpenAI credential: " + ("broken key (drill). Restore with: uv run tools/flow.py llm up" if args.state == "down" else "real key from .env restored")) return 0 # ---------------------------------------------------------------- doctor def cmd_doctor(args: argparse.Namespace) -> int: failures = 0 def line(good: bool, text: str, hint: str = "", seen: object = None) -> None: nonlocal failures failures += not good print(f"{OK if good else BAD} {text}" + ("" if good or not hint else f"\n → {hint}")) if seen is not None: detail("seen", seen) installed = (N8N_HOME / "compose.yml").exists() print(f"{OK if installed else WARN} n8n install folder {N8N_HOME}" + ("" if installed else "\n → day 65 SETUP-N8N 5단계 (get.n8n.io 설치). 다른 곳에 설치했다면 Last3Mon/65/n8n/.env 에 N8N_HOME=<폴더>")) running = container_running() line(running, f"n8n container {CONTAINER}", START_HINT) status, payload, _ = http("GET", f"{N8N_URL}/healthz", timeout=5) line(status == 200, f"n8n {N8N_URL}", "wait 10 seconds after starting n8n, then run doctor again", {"GET": "/healthz", "status": status, "response": payload}) status, payload, _ = http("GET", "http://localhost:8800/health", timeout=5) if USES_DATA_API: line(status == 200, "data API http://localhost:8800", "cd ~/aio-01d4-stu/Last3Mon/77/data-api && docker compose up -d --force-recreate (start n8n first)", {"GET": "/health", "status": status, "response": payload}) fault_status, fault, _ = http("GET", "http://localhost:8800/admin/fault", timeout=5) price_status, prices, _ = http("GET", "http://localhost:8800/prices", timeout=5) budget_status, b, _ = http("GET", "http://localhost:8800/budget", timeout=5) models = prices.get("models", {}) if isinstance(prices, dict) else {} dlq_status, dlq, _ = http("GET", "http://localhost:8800/dlq", timeout=5) line(price_status == 200 and budget_status == 200 and "gpt-5.4-nano" in models and isinstance(b, dict) and "max_question_chars" in b and dlq_status == 200 and isinstance(dlq, dict) and "parked" in dlq and isinstance(fault, dict) and "dlq" in fault, "dead letter queue /dlq · alerts · fault switch dlq (day 76 data API)", {"GET": "/prices", "status": price_status, "response": prices, "budget": b, "dlq": dlq, "fault": fault}) logs_status, logs, _ = http("GET", "http://localhost:8800/logs?source=seed&limit=1", timeout=5) extra_status, extra, _ = http("GET", "http://localhost:8800/faq-extra", timeout=5) line(logs_status == 200 and isinstance(logs, dict) and logs.get("count") == 666 and extra_status == 200, "run logs /logs (666 seeded runs) · FAQ additions /faq-extra (day 77 data API)", "the data API is an older day's version: cd ~/aio-01d4-stu/Last3Mon/77/data-api && docker compose up -d --force-recreate && sleep 3", {"GET": "/logs", "status": logs_status, "count": logs.get("count") if isinstance(logs, dict) else None, "faq_extra": extra}) live_status, live, _ = http("GET", "http://localhost:8800/logs?source=live", timeout=5) if isinstance(live, dict) and live.get("count"): print(f"{WARN} {live.get('count')} live run log(s) left from an earlier run\n → uv run tools/flow.py ops --reset") if isinstance(extra, dict) and extra.get("count"): print(f"{WARN} {extra.get('count')} FAQ entr(ies) added earlier (Lab 03)\n → uv run tools/flow.py ops --reset") if isinstance(dlq, dict) and dlq.get("size"): print(f"{WARN} DLQ is not empty ({dlq.get('size')} item(s), left from an earlier drill)\n → uv run tools/flow.py ops --reset") if isinstance(b, dict) and "daily_usd" in b: start = load_json(ROOT / "data-api" / "data" / "budget.json") if any(b.get(key) != value for key, value in start.items()): print(f"{WARN} budget is not the start state: {as_text({key: b.get(key) for key in start})}" f"\n → uv run tools/flow.py budget --reset") if fault_status == 200 and isinstance(fault, dict) and any(fault.get(key, "ok") != "ok" for key in FAULT_TARGETS): print(f"{WARN} fault switch is on: {fault}\n → uv run tools/flow.py fault all ok") else: print(f"{OK if status == 200 else WARN} data API http://localhost:8800 (not used today)") if running: probe = docker("exec", CONTAINER, "node", "-e", "fetch('http://data-api:8800/health').then(r=>console.log(r.status)).catch(()=>console.log(0))", check=False).stdout.strip() if USES_DATA_API: line(probe == "200", "n8n → http://data-api:8800", "data API must join the day65-n8n_default network (compose.yml)", {"from n8n": "GET http://data-api:8800/health", "status": probe}) exported = docker("exec", CONTAINER, "sh", "-c", "n8n export:credentials --all --output=/tmp/course-cred.json >/dev/null 2>&1; " "cat /tmp/course-cred.json 2>/dev/null; rm -f /tmp/course-cred.json", check=False).stdout try: found = json.loads(exported) if exported.strip() else [] except json.JSONDecodeError: found = [] ids = {item.get("id") for item in found} detail("credentials in n8n (id · name · type, no secrets)", [f"{item.get('id')} · {item.get('name')} · {item.get('type')}" for item in found]) for credential_id, name in COURSE_CREDENTIALS.items(): line(credential_id in ids, f"credential '{name}' ({credential_id})", "cd ~/aio-01d4-stu/Last3Mon/65/n8n && bash register-credentials.sh") catalog = node_versions() line(bool(catalog), "node catalog (for check)", seen={"node types": len(catalog), "cache": str(CACHE.relative_to(ROOT))}) for hook in AGENT_HOOKS: _, payload, _ = http("GET", f"{N8N_URL}/webhook/{hook}", timeout=5) line("not registered for" in json.dumps(payload, ensure_ascii=False), f"agent webhook /{hook}", "uv run tools/flow.py push ", {"GET": f"/webhook/{hook}", "response": payload}) agents = [name for name in ("codex", "claude", "agy") if shutil.which(name)] line(bool(agents), "CLI coding agent: " + (", ".join(agents) if agents else "none"), "install one of codex / claude / agy (see Lecture/77/00-DAY77-PREP.md)", {name: shutil.which(name) for name in agents}) print("ready" if not failures else f"{failures} item(s) to fix") return 0 if not failures else 1 def main() -> int: global VERBOSE common = argparse.ArgumentParser(add_help=False) common.add_argument("-v", "--verbose", action="store_true", help="details: request and response JSON, workflow nodes and links, docker commands") parser = argparse.ArgumentParser(description="Check, push and test course n8n workflows.") commands = parser.add_subparsers(dest="command", required=True) commands.add_parser("doctor", parents=[common], help="check n8n, data API, credentials and CLI agents").set_defaults(run=cmd_doctor) item = commands.add_parser("check", parents=[common], help="find mistakes in a workflow file") item.add_argument("file") item.set_defaults(run=cmd_check) item = commands.add_parser("push", parents=[common], help="import and publish workflow files (one n8n restart)") item.add_argument("files", nargs="+") item.set_defaults(run=cmd_push) item = commands.add_parser("off", parents=[common], help="unpublish workflow files or webhook paths (a role that is down)") item.add_argument("files", nargs="+") item.set_defaults(run=cmd_off) item = commands.add_parser("call", parents=[common], help="POST one request to a webhook path") item.add_argument("path", help="webhook path, e.g. my-notice") item.add_argument("request", nargs="?", help="request file or inline JSON") item.set_defaults(run=cmd_call) item = commands.add_parser("test", parents=[common], help="run a test case file") item.add_argument("cases") item.add_argument("--repeat", type=int, default=1, help="run every case N times") item.add_argument("--webhook", help="override the webhook path in the case file") item.add_argument("--no-before", action="store_true", help="skip the 'before' steps") item.add_argument("--set", action="append", help="add key=value to every request body (value is JSON when it parses)") item.set_defaults(run=cmd_test) item = commands.add_parser("gate", parents=[common], help="release gate: run a test file, PASS/FAIL from pass rate and p95") item.add_argument("cases") item.add_argument("--repeat", type=int, default=1) item.add_argument("--webhook") item.add_argument("--no-before", action="store_true") item.add_argument("--set", action="append", help="add key=value to every request body, e.g. release=1.1.0") item.add_argument("--min-pass", type=float, default=1.0, help="share of cases that must pass (default 1.0)") item.add_argument("--max-p95", type=float, help="seconds") item.add_argument("--name", default="helpdesk", help="release name for the gate record (default helpdesk)") item.add_argument("--no-record", action="store_true", help="do not store the verdict in /gates") item.set_defaults(run=cmd_gate) item = commands.add_parser("diff", parents=[common], help="compare two workflow files node by node") item.add_argument("old") item.add_argument("new") item.add_argument("--full", action="store_true", help="print long values (prompts) before and after (-v does too)") item.set_defaults(run=cmd_diff) item = commands.add_parser("release", parents=[common], help="show or change the active release / canary in the data API") item.add_argument("name") item.add_argument("version", nargs="?") item.add_argument("--canary") item.add_argument("--percent", type=int, default=30) item.add_argument("--no-canary", action="store_true") item.add_argument("--reset", action="store_true", help="active 1.0.0, no canary, deploy policy off") item.add_argument("--require-gate", choices=["on", "off"], help="deploy only releases whose last gate record is promote") item.set_defaults(run=cmd_release) item = commands.add_parser("bench", parents=[common], help="POST the same request N times: p50, p95, max, failures") item.add_argument("path", help="webhook path, e.g. my-helpdesk") item.add_argument("request", nargs="?", help="request file or inline JSON") item.add_argument("-n", type=int, default=10, help="how many calls (default 10)") item.add_argument("--set", action="append", help="add key=value to the request body, e.g. release=2.0.0") item.set_defaults(run=cmd_bench) item = commands.add_parser("usage", parents=[common], help="recorded LLM calls, tokens and cost by model / user / route, and the budget") item.add_argument("--clear", action="store_true", help="empty the usage records (the budget counts from 0 again)") item.set_defaults(run=cmd_usage) item = commands.add_parser("budget", parents=[common], help="show or change the daily budget and the limits") item.add_argument("--daily", type=float, help="daily budget in USD, e.g. 0.001") item.add_argument("--per-min", type=int, help="requests per user per minute, e.g. 2") item.add_argument("--chars", type=int, help="longest question sent as it is, e.g. 500") item.add_argument("--reset", action="store_true", help="back to data/budget.json") item.set_defaults(run=cmd_budget) item = commands.add_parser("logs", parents=[common], help="quick look at the run logs in the data API (day 77)") item.add_argument("--from", dest="date_from", help="first day, e.g. 2026-10-26 (inclusive)") item.add_argument("--to", dest="date_to", help="last day (inclusive)") item.add_argument("--tag") item.add_argument("--source", default="all", choices=["all", "seed", "live"]) item.add_argument("--last", type=int, default=8, help="show the last N runs (0: counts only)") item.set_defaults(run=cmd_logs) item = commands.add_parser("ops", parents=[common], help="live logs, added FAQ, DLQ, alerts and the fault switch at a glance (day 77)") item.add_argument("--reset", action="store_true", help="empty live logs, added FAQ, created tickets, DLQ and alerts, fault switch all ok") item.set_defaults(run=cmd_ops) item = commands.add_parser("fault", parents=[common], help="data API fault switch for drills") item.add_argument("target", choices=[*FAULT_TARGETS, "all"]) item.add_argument("mode", choices=["ok", "slow", "error", "flaky"]) item.add_argument("--seconds", type=float, help="how long 'slow' waits (default 8)") item.set_defaults(run=cmd_fault) item = commands.add_parser("llm", parents=[common], help="outage drill: break (down) or restore (up) the OpenAI credential in n8n") item.add_argument("state", choices=["down", "up"]) item.set_defaults(run=cmd_llm) args = parser.parse_args() VERBOSE = args.verbose return args.run(args) if __name__ == "__main__": sys.exit(main())