1056 lines
53 KiB
Python
1056 lines
53 KiB
Python
|
# /// 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/helpdesk.json # import + publish (one n8n restart)
|
|||
|
|
uv run tools/flow.py off my-helpdesk # unpublish by file or webhook path
|
|||
|
|
uv run tools/flow.py diff reference/helpdesk-v1.json reference/helpdesk-v2.json [--full] # what changed, node by node
|
|||
|
|
uv run tools/flow.py call my-helpdesk requests/faq-vpn.json [-v]
|
|||
|
|
uv run tools/flow.py test cases/lab02.json [-v] [--repeat 3] [--set release=1.1.0] # --set adds a field to every body
|
|||
|
|
uv run tools/flow.py gate cases/golden.json --set release=1.1.0 --min-pass 1.0 --max-p95 3 # PASS/FAIL, recorded in /gates
|
|||
|
|
uv run tools/flow.py bench my-helpdesk requests/faq-vpn.json -n 10 [--set release=2.0.0] # p50 · p95 · max · failures
|
|||
|
|
uv run tools/flow.py release helpdesk [1.1.0] [--canary 1.2.0 --percent 50 | --no-canary] [--reset] [--require-gate on|off]
|
|||
|
|
uv run tools/flow.py fault faq 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/74). 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,
|
|||
|
|
}
|
|||
|
|
USES_DATA_API = True # day 74: FAQ, tickets, prompt registry, releases and gate records come from the data API
|
|||
|
|
AGENT_HOOKS: tuple[str, ...] = () # no ready-made agents on day 74 (rel-v1 · rel-v2 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)")
|
|||
|
|
|
|||
|
|
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)")
|
|||
|
|
if not webhooks(workflow):
|
|||
|
|
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 <file>).")
|
|||
|
|
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 <the workflow file>")
|
|||
|
|
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, 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). "status": 409 means that status is expected (e.g. a refused deploy)."""
|
|||
|
|
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 74 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
|
|||
|
|
for run in range(1, args.repeat + 1):
|
|||
|
|
status, payload, seconds = http("POST", f"{N8N_URL}/webhook/{case_hook}", 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" + ("" if not problems else " · " + "; ".join(problems)))
|
|||
|
|
if args.verbose:
|
|||
|
|
show_json(f"→ request POST /webhook/{case_hook}", body)
|
|||
|
|
show_json(f"← response HTTP {status}", payload)
|
|||
|
|
results.append({"case": case.get("name"), "webhook": case_hook, "run": run, "passed": not problems, "problems": problems,
|
|||
|
|
"seconds": round(seconds, 1), "status": status, "request": body, "response": payload})
|
|||
|
|
return passed
|
|||
|
|
|
|||
|
|
|
|||
|
|
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=<version>) 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
|
|||
|
|
|
|||
|
|
|
|||
|
|
def cmd_fault(args: argparse.Namespace) -> int:
|
|||
|
|
targets = ["faq", "tickets", "prompts"] 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/74/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)
|
|||
|
|
rel_status, rel, _ = http("GET", "http://localhost:8800/releases/helpdesk", timeout=5)
|
|||
|
|
gate_status, gates, _ = http("GET", "http://localhost:8800/gates/helpdesk", timeout=5)
|
|||
|
|
line(rel_status == 200 and gate_status == 200 and isinstance(rel, dict) and "2.1.0" in rel.get("releases", {}),
|
|||
|
|
"releases /releases/helpdesk · gates /gates/helpdesk (day 74 data API)",
|
|||
|
|
"the data API is an older day's version: cd ~/aio-01d4-stu/Last3Mon/74/data-api && docker compose up -d --force-recreate",
|
|||
|
|
{"GET": "/releases/helpdesk", "status": rel_status, "response": rel, "gates": gates})
|
|||
|
|
if isinstance(rel, dict) and (rel.get("active") != "1.0.0" or rel.get("canary") or rel.get("require_gate")):
|
|||
|
|
print(f"{WARN} release state is not the start state: active {rel.get('active')} · canary {rel.get('canary')}"
|
|||
|
|
f" · require_gate {rel.get('require_gate')}\n → uv run tools/flow.py release helpdesk --reset")
|
|||
|
|
stats_status, stats, _ = http("GET", "http://localhost:8800/cache-stats", timeout=5)
|
|||
|
|
detail("cache GET /cache-stats", {"status": stats_status, "response": stats})
|
|||
|
|
if fault_status == 200 and isinstance(fault, dict) and any(fault.get(key, "ok") != "ok" for key in ("faq", "tickets", "prompts")):
|
|||
|
|
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 <agent workflow file>", {"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/74/00-DAY74-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("fault", parents=[common], help="data API fault switch for drills")
|
|||
|
|
item.add_argument("target", choices=["faq", "tickets", "prompts", "all"])
|
|||
|
|
item.add_argument("mode", choices=["ok", "slow", "error"])
|
|||
|
|
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())
|