From 602cba937366b69550dbef257d79853a9a25ca0d Mon Sep 17 00:00:00 2001 From: agent-exec-transport Date: Fri, 16 Oct 2026 16:21:24 +0000 Subject: [PATCH] agent-exec transport b918a843562d4d29b3efc3da6ee19803/557f53579497 --- Makefile | 14 +- scripts/task.sh | 61 +++-- scripts/task_triage.py | 505 ++++++++++++++++++++++++++++++++++++ tests/test_task.py | 285 +++++++++++++++----- tests/test_task_recovery.py | 9 +- tests/test_task_triage.py | 401 ++++++++++++++++++++++++++++ 6 files changed, 1181 insertions(+), 94 deletions(-) create mode 100644 scripts/task_triage.py create mode 100644 tests/test_task_triage.py diff --git a/Makefile b/Makefile index 3b662624..011ad5b4 100644 --- a/Makefile +++ b/Makefile @@ -13,13 +13,13 @@ PLUGINS := plugins COMBO := core=$(shell git -C $(CORE) rev-parse --short HEAD 2>/dev/null)+backend=$(shell git -C $(BACKEND) rev-parse --short HEAD 2>/dev/null)+frontend=$(shell git -C $(FRONTEND) rev-parse --short HEAD 2>/dev/null)+ui=$(shell git -C $(UI) rev-parse --short HEAD 2>/dev/null)+agentcmd=$(shell git -C deps/agent-command rev-parse --short HEAD 2>/dev/null)+documents=$(shell git -C modules/documents rev-parse --short HEAD 2>/dev/null)+native=$(shell git -C deps/agent-native rev-parse --short HEAD 2>/dev/null)+squad=$(shell git -C deps/agent-squad rev-parse --short HEAD 2>/dev/null)+intent=$(shell git -C deps/agent-intent rev-parse --short HEAD 2>/dev/null) VERSION := $(shell git describe --tags --always --dirty 2>/dev/null)+$(COMBO) -.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-triage task-start task-resume task-sync task-close task-finish facts-verify evidence version agent-exec-validate agent-exec-apply agent-exec-admit +.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-triage task-triage-decide task-start task-resume task-sync task-close task-finish facts-verify evidence version agent-exec-validate agent-exec-apply agent-exec-admit version: @echo "agent-system $(VERSION)" TASK := scripts/task.sh -export PRECHECK_RECEIPT RECEIPT IMPORT_RECEIPT TASK_WRITERS_STOPPED +export PRECHECK_RECEIPT RECEIPT IMPORT_RECEIPT TASK_WRITERS_STOPPED TRIAGE_RECEIPT TASK_WORK_MODE ?= delivery export TASK_WORK_MODE TASK_PARENT_BRANCH ?= $(PARENT) @@ -39,10 +39,13 @@ precheck: @bash $(TASK) precheck task-triage: - @bash $(TASK) triage $(REQ) + @bash $(TASK) triage "$(REQ)" + +task-triage-decide: + @VERDICT="$(VERDICT)" INITIATIVE="$(INITIATIVE)" REASON="$(REASON)" bash $(TASK) triage-decide task-start: - @if [ -z "$(NAME)" ]; then echo "usage: make task-start NAME=feature/"; exit 2; fi + @if [ -z "$(NAME)" ]; then echo "usage: make task-start NAME=feature/ TRIAGE_RECEIPT="; exit 2; fi @bash $(TASK) start "$(NAME)" task-resume: @@ -87,6 +90,9 @@ help: @echo "make agent-exec-validate 校验 AGENT-EXECUTION.md（agent_exec validate）" @echo "make agent-exec-apply RECEIPT= 快进当前分支到成功收据的 transport" @echo "make agent-exec-admit 校验暂存路径具备 transport 来源或属于 host 豁免" + @echo "make task-triage REQ='<需求>' TRIAGE_RECEIPT= 以远端为准发现接纳中的 initiative 并写分流收据（候选存在时 exit 5，待决策）" + @echo "make task-triage-decide TRIAGE_RECEIPT= VERDICT=JOIN|NEW [INITIATIVE=] REASON='<理由>' 记录显式分流决策（不按数量自动 JOIN）" + @echo "make task-start NAME=feature/ TRIAGE_RECEIPT= [KIND=...] [INITIATIVE=...] 建任务前核验分流收据（失败不建分支）" agent-exec-validate: PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py validate --repo . diff --git a/scripts/task.sh b/scripts/task.sh index 3dddb94f..bec000ed 100755 --- a/scripts/task.sh +++ b/scripts/task.sh @@ -74,7 +74,7 @@ cmd_precheck() { [ -n "$done_open" ] && echo "[cards] 已完成但未归档:$(echo $done_open)" local initiatives initiatives=$(active_initiatives | cut -d'|' -f1) - echo "[initiatives] active: $([ -n "$initiatives" ] && echo $initiatives || echo none)" + echo "[initiatives] local cards (not authoritative): $([ -n "$initiatives" ] && echo $initiatives || echo none)" now=$(date +%s) while read -r ref ts; do [ -n "$ref" ] || continue @@ -152,10 +152,28 @@ cmd_start() { echo "$initiative" | grep -qE '^(feature|fix|hotfix|release)/[A-Za-z0-9._-]+$' || die "invalid parent initiative branch" case "${name#*/}" in "${initiative#*/}"-?*) ;; *) die "child slug must start with parent initiative slug plus '-'" ;; esac fi + [ -n "${TRIAGE_RECEIPT:-}" ] || die "TRIAGE_RECEIPT required: make task-triage REQ='<需求>' TRIAGE_RECEIPT= and decide first (no branch created)" fetch_target [ -n "${PRECHECK_RECEIPT:-}" ] || die "PRECHECK_RECEIPT required: capture RECEIPT=/tmp/task-precheck.json CMD='make precheck' make evidence first" PRECHECK_RECEIPT=$(realpath -- "$PRECHECK_RECEIPT") VALIDATE_RECEIPT="$PRECHECK_RECEIPT" VALIDATE_RECEIPT_PURPOSE=start bash "${SCRIPT_DIR}/evidence.sh" + # 分流核验必须先于任何 worktree/分支创建：收据与当前远端候选不一致则无副作用拒绝。 + [ -f "$TRIAGE_RECEIPT" ] || die "triage receipt missing: ${TRIAGE_RECEIPT} (no branch created)" + TRIAGE_RECEIPT=$(realpath -- "$TRIAGE_RECEIPT") + local -a triage_args=(verify --receipt "$TRIAGE_RECEIPT" --kind "$kind") + if [ "$kind" = child ]; then triage_args+=(--initiative "$initiative"); fi + local triage_out triage_line triage_verdict triage_init triage_rsha triage_digest triage_note + if ! triage_out=$(python3 "${SCRIPT_DIR}/task_triage.py" "${triage_args[@]}"); then + printf '%s\n' "$triage_out" >&2 + die "triage receipt rejected; no worktree or branch was created" + fi + triage_line=$(printf '%s\n' "$triage_out" | sed -n 's/^\[triage-verified\] //p') + triage_verdict=$(triage_field verdict "$triage_line") + triage_init=$(triage_field selected_initiative "$triage_line") + triage_rsha=$(triage_field requirement_sha256 "$triage_line") + triage_digest=$(triage_field receipt_sha256 "$triage_line") + [ -n "$triage_verdict" ] && [ -n "$triage_rsha" ] && [ -n "$triage_digest" ] || die "triage verification output unparsable; no branch created" + triage_note="- 需求分流：verdict=${triage_verdict}；selected_initiative=${triage_init:-无}；requirement_sha256=${triage_rsha}；triage_receipt=${TRIAGE_RECEIPT}；receipt_sha256=${triage_digest}" check_worktree before-sync # 已合入的同名分支不得重建（2026-10-05 实测：合入后再 task-start 同名会从新 main 长出 # 第二段历史，空卡覆盖已合入任务卡，两轮证据差点丢失）。复核/追加沿用原分支或换新 slug。 @@ -220,6 +238,7 @@ cmd_start() { - 权威仓：$(git -C "$wt" remote get-url origin 2>/dev/null || echo '(local)')；目标分支：${TARGET_BRANCH}；R0（initial_base_sha）= ${rmain} - 工作区：${wt}（worktree）；分支：${name} ${branch_meta} +${triage_note} ${parent_note} - 授权与交付终点：<本次允许的动作与终点；未授权事项写明> - 交付模式：goal-first（目标、范围和交付终点一次确认；范围内连续推进） @@ -473,28 +492,27 @@ resolve_parent_ref() { # $1=父 initiative 分支 → 打印用于同步/核验 fi } +# 需求分流以 origin 为准（scripts/task_triage.py）：远端 open PR + 远端分支卡片；本地 .task/ 不参与发现。 +# 计数不自动选择 JOIN；候选存在时必须经 triage-decide 记录显式决策。 cmd_triage() { need_repo - local req="${*:-}" list n line b id c - [ -n "$req" ] && echo "[triage] requirement: ${req}" - list=$(active_initiatives) - n=$(printf '%s' "$list" | grep -c '|' || true) - echo "[triage] active initiatives: ${n}" - while IFS= read -r line; do - [ -n "$line" ] || continue - IFS='|' read -r b id c <<<"$line" - echo "[initiative] branch=${b} initiative_id=${id:-无} card=${c}" - done <<<"$list" - if [ "$n" -eq 0 ]; then - echo "[verdict] NEW — no active initiative; start one with make task-start NAME=feature/ KIND=initiative (or KIND=standalone)" - elif [ "$n" -eq 1 ]; then - b=${list%%|*} - echo "[verdict] JOIN — join initiative ${b}: make task-start NAME=${b}- KIND=child INITIATIVE=${b}" - else - echo "[verdict] DECISION_REQUIRED — ${n} active initiatives; the user must choose which one receives this requirement (or approve a new one)" - return 5 - fi + [ -n "${TRIAGE_RECEIPT:-}" ] || die "TRIAGE_RECEIPT required: make task-triage REQ='<需求>' TRIAGE_RECEIPT=/tmp/.json" + local req="${*:-}" + [ -n "$req" ] || die "requirement text required: make task-triage REQ='<需求>' TRIAGE_RECEIPT=" + python3 "${SCRIPT_DIR}/task_triage.py" discover --receipt "$TRIAGE_RECEIPT" --requirement "$req" } + +cmd_triage_decide() { + need_repo + [ -n "${TRIAGE_RECEIPT:-}" ] || die "TRIAGE_RECEIPT required: make task-triage-decide TRIAGE_RECEIPT= VERDICT=JOIN|NEW REASON='<理由>'" + local -a args=(decide --receipt "$TRIAGE_RECEIPT" --verdict "${VERDICT:-}" --reason "${REASON:-}") + if [ -n "${INITIATIVE:-}" ]; then args+=(--initiative "$INITIATIVE"); fi + python3 "${SCRIPT_DIR}/task_triage.py" "${args[@]}" +} + +# triage_field KEY "key=value key2=value2" → value（值不含空格） +triage_field() { printf '%s\n' "$2" | tr ' ' '\n' | sed -n "s/^$1=//p"; } + card_merge() { # $1=branch → "sha|subject|pr"（该分支合入 origin/ 的记录；兼容 merge 提交与 squash 落地） local hit pr hit=$(git log --merges --format='%H|%s' "origin/${TARGET_BRANCH}" 2>/dev/null \ @@ -769,10 +787,11 @@ cmd_finish() { case "${1:-}" in precheck) shift; cmd_precheck "$@" ;; triage) shift; cmd_triage "$@" ;; + triage-decide) shift; cmd_triage_decide "$@" ;; start) shift; cmd_start "$@" ;; resume) shift; cmd_resume "$@" ;; sync) shift; cmd_sync "$@" ;; close) shift; cmd_close "$@" ;; finish) shift; cmd_finish "$@" ;; - *) echo "usage: task.sh [args]"; exit 2 ;; + *) echo "usage: task.sh [args]"; exit 2 ;; esac diff --git a/scripts/task_triage.py b/scripts/task_triage.py new file mode 100644 index 00000000..acf83b24 --- /dev/null +++ b/scripts/task_triage.py @@ -0,0 +1,505 @@ +#!/usr/bin/env python3 +"""远端需求分流收据（task.sh triage / triage-decide / start 共用的唯一分流依据）。 + +事实来源是 origin：`git ls-remote` 取分支头，Gitea open PR 取 base/head 关系，候选分支提交中的 +任务卡取 initiative 元数据。本地 `.task/` 不参与发现。 + +子命令： + discover --receipt PATH --requirement TEXT + 查询远端并写收据；无候选 -> NEW；有候选 -> DECISION_REQUIRED（退出 5，不自动选择 JOIN）。 + decide --receipt PATH --verdict JOIN|NEW [--initiative BRANCH] --reason TEXT + 复查远端；候选集合或任一候选的 HEAD/卡片/PR 字段变化则拒绝（退出 7，收据不改）。 + verify --receipt PATH --kind standalone|initiative|child [--initiative BRANCH] [--max-age-seconds N] + 复查远端并核对收据；供 task.sh start 在创建任何 worktree/分支之前调用（拒绝退出 8）。 + +退出码：0 成功；1 本地错误；2 用法；5 DECISION_REQUIRED；6 REMOTE_UNVERIFIED；7 决策被拒；8 核验被拒。 +仅使用标准库。token 只读自环境变量 GITEA_TOKEN 或 ~/.secrets/gitea 的 ACCESS_TOKEN=，不打印、不写入收据。 +""" +import argparse +import datetime as dt +import hashlib +import http.client +import json +import os +import re +import subprocess +import sys +import urllib.error +import urllib.parse +import urllib.request + +SCHEMA = "agent-system/task-triage/v1" +DEFAULT_API = "https://git-home.u1680832.nyat.app:30319/api/v1" +DEFAULT_TARGET = "main" +DEFAULT_MAX_AGE = 14400 +PAGE_LIMIT = 50 +MAX_PAGES = 100 +SECRETS_FILE = "~/.secrets/gitea" +CARD_RE = re.compile(r"^\.task/[^/]+\.md$") +CANDIDATE_KEYS = ("branch", "head_sha", "card_path", "card_sha256", "pr_number", "pr_state", "pr_base", "pr_head_sha") +RECEIPT_KEYS = ("repo", "root_target", "root_target_sha", "requirement", "discovered_at", + "candidates", "discovery_verdict", "final_verdict", "decision") + +EXIT_LOCAL = 1 +EXIT_USAGE = 2 +EXIT_DECISION_REQUIRED = 5 +EXIT_REMOTE = 6 +EXIT_DECIDE = 7 +EXIT_VERIFY = 8 + + +class Stop(Exception): + """受控失败：退出码 + 不含 token 的说明。""" + + def __init__(self, code, message): + super().__init__(message) + self.code = code + self.message = message + + +def utc_now(): + return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def parse_time(text): + value = dt.datetime.fromisoformat(text) + if value.tzinfo is None: + value = value.replace(tzinfo=dt.timezone.utc) + return value + + +def sha256_bytes(data): + return hashlib.sha256(data).hexdigest() + + +def as_dict(value): + return value if isinstance(value, dict) else {} + + +def target_branch(): + return os.environ.get("TASK_TARGET_BRANCH", "").strip() or DEFAULT_TARGET + + +def git(*args): + return subprocess.run(["git", *args], capture_output=True) + + +def redact_url(url): + """去掉 URL 中的 userinfo（https://user:pw@host 与 user@host:path 两种形式）。""" + if "://" in url: + parts = urllib.parse.urlsplit(url) + host = parts.netloc.rsplit("@", 1)[-1] + return urllib.parse.urlunsplit((parts.scheme, host, parts.path, parts.query, parts.fragment)) + scp = re.match(r"^[^/@]+@([^/]+:.*)$", url) + return scp.group(1) if scp else url + + +def repo_slug(origin_url): + slug = os.environ.get("GITEA_REPO", "").strip() + if not slug: + if "://" in origin_url: + path = urllib.parse.urlsplit(origin_url).path + else: + path = origin_url.split(":", 1)[1] if re.match(r"^[^/]+:", origin_url) else origin_url + segments = [s for s in path.split("/") if s] + if len(segments) < 2: + raise Stop(EXIT_REMOTE, "cannot derive owner/repo from origin; set GITEA_REPO") + name = segments[-1][:-4] if segments[-1].endswith(".git") else segments[-1] + slug = f"{segments[-2]}/{name}" + parts = slug.split("/") + if len(parts) != 2 or not all(parts) or any(p in (".", "..") for p in parts): + raise Stop(EXIT_REMOTE, "invalid owner/repo slug") + return slug + + +def api_base(): + return (os.environ.get("GITEA_API_URL", "").strip() or DEFAULT_API).rstrip("/") + + +def read_token(): + token = os.environ.get("GITEA_TOKEN", "").strip() + if token: + return token + try: + with open(os.path.expanduser(SECRETS_FILE), encoding="utf-8") as fh: + for raw in fh: + line = raw.strip() + if line.startswith("- "): + line = line[2:].strip() + if line.startswith("ACCESS_TOKEN="): + return line.split("=", 1)[1].strip() or None + except (OSError, UnicodeDecodeError): + return None + return None + + +def remote_heads(): + proc = git("ls-remote", "--heads", "origin") + if proc.returncode != 0: + raise Stop(EXIT_REMOTE, "git ls-remote --heads origin failed") + heads = {} + for line in proc.stdout.decode("utf-8", "replace").splitlines(): + sha, _, ref = line.partition("\t") + if ref.startswith("refs/heads/"): + heads[ref[len("refs/heads/"):]] = sha + return heads + + +def http_get_json(url, token): + request = urllib.request.Request( + url, headers={"Accept": "application/json", "Authorization": f"token {token}"}) + try: + with urllib.request.urlopen(request, timeout=20) as response: + body = response.read() + except urllib.error.HTTPError as exc: + raise Stop(EXIT_REMOTE, f"Gitea API returned HTTP {exc.code}") + except (urllib.error.URLError, http.client.HTTPException, OSError) as exc: + raise Stop(EXIT_REMOTE, f"Gitea API unreachable ({type(exc).__name__})") + try: + return json.loads(body.decode("utf-8")) + except ValueError: + raise Stop(EXIT_REMOTE, "Gitea API returned invalid JSON") + + +def open_pulls(api, slug, token): + owner, name = (urllib.parse.quote(part, safe="") for part in slug.split("/")) + pulls = [] + for page in range(1, MAX_PAGES + 1): + url = f"{api}/repos/{owner}/{name}/pulls?state=open&limit={PAGE_LIMIT}&page={page}" + batch = http_get_json(url, token) + if not isinstance(batch, list): + raise Stop(EXIT_REMOTE, "Gitea pulls response is not a list") + if not batch: + return pulls + pulls.extend(batch) + raise Stop(EXIT_REMOTE, f"open pull requests exceed {MAX_PAGES} pages") + + +def fetch_head(branch): + proc = git("fetch", "-q", "origin", f"refs/heads/{branch}") + if proc.returncode != 0: + raise Stop(EXIT_REMOTE, f"cannot fetch origin/{branch}") + rev = git("rev-parse", "--verify", "-q", "FETCH_HEAD^{commit}") + if rev.returncode != 0: + raise Stop(EXIT_REMOTE, f"fetched origin/{branch} has no commit") + return rev.stdout.decode("utf-8", "replace").strip() + + +def task_card_paths(commit): + proc = git("ls-tree", "-r", "-z", "--name-only", commit, "--", ".task") + if proc.returncode != 0: + raise Stop(EXIT_REMOTE, "cannot list task cards at candidate head") + names = proc.stdout.decode("utf-8", "replace").split("\0") + return sorted(name for name in names if CARD_RE.match(name)) + + +def blob_bytes(commit, path): + proc = git("cat-file", "blob", f"{commit}:{path}") + if proc.returncode != 0: + raise Stop(EXIT_REMOTE, f"cannot read {path} at candidate head") + return proc.stdout + + +def card_branch(text): + for line in text.splitlines(): + match = re.match(r"^- 工作区：.*分支：([A-Za-z0-9._/-]+)", line) + if match: + return match.group(1) + return None + + +def card_status(text): + match = re.search(r"；状态：([^；\n]*)", text) + return match.group(1).strip() if match else "" + + +def branch_meta(text, target): + """分支元数据行 '- 分支类型：…；field：value…'；旧卡缺省为 standalone、不接收需求。""" + match = re.search(r"^- 分支类型：([^\n]*)$", text, re.M) + if not match: + return {"kind": "standalone", "initiative_id": "", "root_target": target, "accepting": "false"} + line = match.group(1) + + def field(name, default): + found = re.search("；" + name + r"：([^；]*)", line) + return found.group(1).strip() if found else default + + return { + "kind": line.split("；", 1)[0].strip(), + "initiative_id": field("initiative_id", ""), + "root_target": field("root_target", target), + "accepting": field("accepting_requirements", "false"), + } + + +def goal_excerpt(text): + lines, inside = [], False + for line in text.splitlines(): + if line.startswith("## "): + if inside: + break + inside = line.strip() == "## 目标与非目标" + continue + if inside and line.strip(): + lines.append(line.strip()) + if len(lines) == 5: + break + return " / ".join(lines)[:300] + + +def inspect_branch(target, branch, head_sha, pr): + fetched = fetch_head(branch) + if fetched != head_sha: + raise Stop(EXIT_REMOTE, f"{branch} moved during discovery; retry") + matches = [] + for path in task_card_paths(head_sha): + raw = blob_bytes(head_sha, path) + text = raw.decode("utf-8", "replace") + if card_branch(text) != branch: + continue + meta = branch_meta(text, target) + if meta["kind"] != "initiative" or meta["accepting"] != "true" or meta["root_target"] != target: + continue + if card_status(text).startswith("已完成"): + continue + matches.append((path, raw, text, meta)) + if not matches: + return None + if len(matches) > 1: + raise Stop(EXIT_REMOTE, f"ambiguous initiative cards on {branch}") + path, raw, text, meta = matches[0] + head = as_dict(pr.get("head")) + return { + "branch": branch, + "head_sha": head_sha, + "pr_number": pr["number"], + "pr_state": pr.get("state"), + "pr_base": as_dict(pr.get("base")).get("ref"), + "pr_head_sha": head.get("sha") or "", + "card_path": path, + "card_sha256": sha256_bytes(raw), + "initiative_id": meta["initiative_id"], + "goal_excerpt": goal_excerpt(text), + } + + +def candidates_from(target, api, slug, token, heads): + by_branch = {} + for pr in open_pulls(api, slug, token): + if not isinstance(pr, dict) or pr.get("state") != "open": + continue + base = as_dict(pr.get("base")).get("ref") + head = as_dict(pr.get("head")).get("ref") + if base != target or not isinstance(head, str) or head == target or head not in heads: + continue + if not isinstance(pr.get("number"), int): + raise Stop(EXIT_REMOTE, "malformed pull request number") + by_branch.setdefault(head, []).append(pr) + found = [] + for branch in sorted(by_branch): + if len(by_branch[branch]) != 1: + raise Stop(EXIT_REMOTE, f"ambiguous: several open PRs target {target} from {branch}") + candidate = inspect_branch(target, branch, heads[branch], by_branch[branch][0]) + if candidate: + found.append(candidate) + return found + + +def snapshot(target): + origin = git("remote", "get-url", "origin") + if origin.returncode != 0: + raise Stop(EXIT_REMOTE, "origin remote is not configured") + origin_url = origin.stdout.decode("utf-8", "replace").strip() + heads = remote_heads() + if target not in heads: + raise Stop(EXIT_REMOTE, f"origin has no {target} branch") + slug = repo_slug(origin_url) + api = api_base() + token = read_token() + if not token: + raise Stop(EXIT_REMOTE, "no Gitea token: set GITEA_TOKEN or ~/.secrets/gitea ACCESS_TOKEN") + return { + "repo": {"origin_url": redact_url(origin_url), "api_base": redact_url(api), "repo_slug": slug}, + "root_target": target, + "root_target_sha": heads[target], + "candidates": candidates_from(target, api, slug, token, heads), + } + + +def fingerprint(candidates): + return sorted(tuple(c[key] for key in CANDIDATE_KEYS) for c in candidates) + + +def check_shape(receipt): + missing = [key for key in RECEIPT_KEYS if key not in receipt] + if missing: + raise ValueError(f"missing {missing[0]}") + if not isinstance(receipt["candidates"], list): + raise ValueError("candidates") + for cand in receipt["candidates"]: + if not isinstance(cand, dict) or any(key not in cand for key in CANDIDATE_KEYS): + raise ValueError("candidate") + if not isinstance(as_dict(receipt["requirement"]).get("sha256"), str): + raise ValueError("requirement") + parse_time(receipt["discovered_at"]) + + +def load_receipt(path, code): + try: + with open(path, "rb") as fh: + raw = fh.read() + except OSError: + raise Stop(code, f"receipt missing or unreadable: {path}") + try: + receipt = json.loads(raw.decode("utf-8")) + except ValueError: + raise Stop(code, "receipt is not valid JSON") + if not isinstance(receipt, dict) or receipt.get("schema") != SCHEMA: + raise Stop(code, f"receipt schema is not {SCHEMA}") + try: + check_shape(receipt) + except (ValueError, TypeError): + raise Stop(code, "receipt is malformed") + return receipt, raw + + +def write_receipt(path, receipt): + data = json.dumps(receipt, ensure_ascii=False, indent=2) + "\n" + tmp = f"{path}.tmp-{os.getpid()}" + try: + with open(tmp, "w", encoding="utf-8") as fh: + fh.write(data) + os.replace(tmp, path) + except OSError as exc: + raise Stop(EXIT_LOCAL, f"cannot write receipt {path}: {exc.strerror}") + + +def describe(cand): + return (f"[candidate] branch={cand['branch']} head_sha={cand['head_sha']} pr=#{cand['pr_number']} " + f"base={cand['pr_base']} card={cand['card_path']} card_sha256={cand['card_sha256']} " + f"initiative_id={cand['initiative_id'] or '无'}") + + +def cmd_discover(args): + text = args.requirement.strip() + if not text: + raise Stop(EXIT_USAGE, "requirement text is required") + snap = snapshot(target_branch()) + now = utc_now() + receipt = { + "schema": SCHEMA, + **snap, + "requirement": {"sha256": sha256_bytes(text.encode("utf-8")), + "summary": " ".join(text.split())[:200]}, + "discovered_at": now, + } + path = os.path.realpath(args.receipt) + if not snap["candidates"]: + receipt.update(discovery_verdict="NEW", final_verdict="NEW", decision={ + "verdict": "NEW", "selected_initiative": None, + "rationale": "no accepting initiative", "decided_at": now}) + write_receipt(args.receipt, receipt) + print(f"[verdict] NEW — 远端没有接纳中的 initiative；收据：{path}") + return 0 + receipt.update(discovery_verdict="DECISION_REQUIRED", final_verdict="DECISION_REQUIRED", decision=None) + write_receipt(args.receipt, receipt) + print(f"[triage] 远端接纳中的 initiative 候选: {len(snap['candidates'])}（以 origin 为准，本地 .task 不参与）") + for cand in snap["candidates"]: + print(describe(cand)) + if cand["goal_excerpt"]: + print(f"[goal] branch={cand['branch']} {cand['goal_excerpt']}") + print(f"[verdict] DECISION_REQUIRED — 须人工决策：JOIN 其一，或 NEW 并说明不适配理由；收据：{path}") + for cand in snap["candidates"]: + print(f"[decide] make task-triage-decide TRIAGE_RECEIPT={path} VERDICT=JOIN " + f"INITIATIVE={cand['branch']} REASON='<加入理由>'") + print(f"[decide] make task-triage-decide TRIAGE_RECEIPT={path} VERDICT=NEW REASON='<不适配理由>'") + return EXIT_DECISION_REQUIRED + + +def cmd_decide(args): + if args.verdict not in ("JOIN", "NEW"): + raise Stop(EXIT_DECIDE, "verdict must be JOIN or NEW") + receipt, _ = load_receipt(args.receipt, EXIT_DECIDE) + if receipt["discovery_verdict"] != "DECISION_REQUIRED": + raise Stop(EXIT_DECIDE, f"no decision needed: discovery verdict is {receipt['discovery_verdict']}") + reason = args.reason.strip() + if not reason: + raise Stop(EXIT_DECIDE, "--reason is required") + initiative = args.initiative.strip() + if args.verdict == "JOIN" and not initiative: + raise Stop(EXIT_DECIDE, "JOIN requires --initiative") + if args.verdict == "NEW" and initiative: + raise Stop(EXIT_DECIDE, "NEW must not select an initiative") + current = snapshot(receipt["root_target"]) + if fingerprint(current["candidates"]) != fingerprint(receipt["candidates"]): + raise Stop(EXIT_DECIDE, "candidate set changed since discovery; re-run make task-triage") + if args.verdict == "JOIN" and initiative not in {c["branch"] for c in receipt["candidates"]}: + raise Stop(EXIT_DECIDE, f"{initiative} is not a discovered candidate") + receipt["decision"] = {"verdict": args.verdict, "selected_initiative": initiative or None, + "rationale": reason, "decided_at": utc_now()} + receipt["final_verdict"] = args.verdict + write_receipt(args.receipt, receipt) + print(f"[decided] verdict={args.verdict} selected_initiative={initiative or '无'} " + f"receipt={os.path.realpath(args.receipt)}") + return 0 + + +def cmd_verify(args): + receipt, raw = load_receipt(args.receipt, EXIT_VERIFY) + final = receipt["final_verdict"] + decision = as_dict(receipt["decision"]) + if final not in ("JOIN", "NEW") or decision.get("verdict") != final: + raise Stop(EXIT_VERIFY, "receipt has no final JOIN/NEW decision; run make task-triage-decide") + age = (dt.datetime.now(dt.timezone.utc) - parse_time(receipt["discovered_at"])).total_seconds() + if age > args.max_age_seconds: + raise Stop(EXIT_VERIFY, f"receipt is stale ({int(age)}s old, max {args.max_age_seconds}s); re-run make task-triage") + if age < -300: + raise Stop(EXIT_VERIFY, "receipt discovered_at is in the future") + if receipt["root_target"] != target_branch(): + raise Stop(EXIT_VERIFY, "receipt root target does not match TASK_TARGET_BRANCH") + selected = decision.get("selected_initiative") + if args.kind == "child": + if final != "JOIN" or not args.initiative or args.initiative != selected: + raise Stop(EXIT_VERIFY, f"child requires a JOIN receipt selecting {args.initiative or '(none)'}") + elif final != "NEW": + raise Stop(EXIT_VERIFY, f"{args.kind} requires a NEW receipt, got {final}") + current = snapshot(receipt["root_target"]) + if fingerprint(current["candidates"]) != fingerprint(receipt["candidates"]): + raise Stop(EXIT_VERIFY, "remote initiative candidates changed since triage; re-run make task-triage") + head = next((c["head_sha"] for c in receipt["candidates"] if c["branch"] == selected), None) + print(f"[triage-verified] verdict={final} selected_initiative={selected or '无'} " + f"requirement_sha256={receipt['requirement']['sha256']} receipt_sha256={sha256_bytes(raw)}") + print(f"[triage-head] selected_head_sha={head or '无'}") + return 0 + + +def main(argv=None): + parser = argparse.ArgumentParser(prog="task_triage.py", description="远端需求分流收据") + sub = parser.add_subparsers(dest="cmd", required=True) + p = sub.add_parser("discover") + p.add_argument("--receipt", required=True) + p.add_argument("--requirement", required=True) + p = sub.add_parser("decide") + p.add_argument("--receipt", required=True) + p.add_argument("--verdict", required=True) + p.add_argument("--initiative", default="") + p.add_argument("--reason", default="") + p = sub.add_parser("verify") + p.add_argument("--receipt", required=True) + p.add_argument("--kind", required=True, choices=("standalone", "initiative", "child")) + p.add_argument("--initiative", default="") + p.add_argument("--max-age-seconds", type=int, default=DEFAULT_MAX_AGE) + args = parser.parse_args(argv) + handlers = {"discover": cmd_discover, "decide": cmd_decide, "verify": cmd_verify} + try: + return handlers[args.cmd](args) + except Stop as stop: + if stop.code == EXIT_REMOTE: + print(f"[verdict] REMOTE_UNVERIFIED — {stop.message}") + else: + print(f"[triage] refused ({stop.code}): {stop.message}", file=sys.stderr) + return stop.code + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_task.py b/tests/test_task.py index 148e2e3e..ccf19e1e 100644 --- a/tests/test_task.py +++ b/tests/test_task.py @@ -2,15 +2,65 @@ import os import json import hashlib +import http.server +import threading from pathlib import Path import shutil import shlex import subprocess import tempfile import unittest +import urllib.parse SOURCE = Path(__file__).resolve().parents[1] +TRIAGE_COMMANDS = ("start", "triage", "triage-decide") + + +class FakeGitea: + """Gitea pulls API 桩（进程内 127.0.0.1）：校验 token 请求头，按页返回 self.prs；从不联网。""" + + PAGE = 50 + + def __init__(self, token="dummy"): + self.token = token + self.prs = [] + stub = self + + class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def _send(self, status, obj): + body = json.dumps(obj).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + if self.headers.get("Authorization") != f"token {stub.token}": + return self._send(401, {"message": "unauthorized"}) + parsed = urllib.parse.urlsplit(self.path) + if not parsed.path.startswith("/api/v1/repos/") or not parsed.path.endswith("/pulls"): + return self._send(404, {"message": "not found"}) + page = int(urllib.parse.parse_qs(parsed.query).get("page", ["1"])[0]) + start = (page - 1) * stub.PAGE + return self._send(200, stub.prs[start:start + stub.PAGE]) + + self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.url = f"http://127.0.0.1:{self.server.server_address[1]}/api/v1" + threading.Thread(target=self.server.serve_forever, daemon=True).start() + + def close(self): + self.server.shutdown() + self.server.server_close() + + @staticmethod + def pr(number, branch, sha, base="main", state="open"): + return {"number": number, "state": state, "base": {"ref": base}, + "head": {"ref": branch, "sha": sha}} class TaskLifecycleTests(unittest.TestCase): @@ -20,6 +70,8 @@ class TaskLifecycleTests(unittest.TestCase): self.root = Path(self.tmp.name) self.repo = self.root / "repo" self.remote = self.root / "remote.git" + self.gitea = FakeGitea() + self.addCleanup(self.gitea.close) self.env = os.environ.copy() self.env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0", @@ -27,8 +79,11 @@ class TaskLifecycleTests(unittest.TestCase): "GIT_COMMITTER_NAME": "Lifecycle Test", "GIT_COMMITTER_EMAIL": "test@example.invalid", "GIT_CONFIG_COUNT": "2", "GIT_CONFIG_KEY_0": "gc.auto", "GIT_CONFIG_VALUE_0": "0", "GIT_CONFIG_KEY_1": "maintenance.auto", - "GIT_CONFIG_VALUE_1": "false"}) - for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GITEA_TOKEN", "PRECHECK_RECEIPT", + "GIT_CONFIG_VALUE_1": "false", + # 桩位于 127.0.0.1：禁止经代理转发，保证测试不触达外部网络。 + "NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost"}) + for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GITEA_TOKEN", "GITEA_API_URL", + "GITEA_REPO", "TRIAGE_RECEIPT", "PRECHECK_RECEIPT", "TASK_PARENT_BRANCH", "TASK_PARENT_PR", "FORCE_ABANDON", "TASK_TARGET_BRANCH", "TASK_WORK_MODE", "VALIDATE_RECEIPT_PURPOSE", "TASK_BRANCH_KIND", "TASK_INITIATIVE"): self.env.pop(key, None) @@ -60,8 +115,14 @@ class TaskLifecycleTests(unittest.TestCase): def task(self, cwd, command, *args, **kwargs): if command == "finish" and "env" not in kwargs: kwargs["env"] = self.env | {"TASK_WRITERS_STOPPED": "1"} + if command in TRIAGE_COMMANDS: + # 分流只读远端：注入桩地址与测试专用 token，绝不读取真实 ~/.secrets。 + kwargs["env"] = self.gitea_env() | kwargs.get("env", self.env) return self.run_cmd(cwd, "bash", str(self.repo / "scripts/task.sh"), command, *args, **kwargs) + def gitea_env(self): + return {"GITEA_API_URL": self.gitea.url, "GITEA_REPO": "fixture/repo", "GITEA_TOKEN": self.gitea.token} + def worktree(self, name="feature/test"): wt = self.root / "worktree" self.git(self.repo, "worktree", "add", "-b", name, str(wt), "origin/main") @@ -676,13 +737,41 @@ class TaskLifecycleTests(unittest.TestCase): self.run_cmd(cwd or self.repo, "bash", str(self.repo / "scripts/evidence.sh"), env=env) return receipt + def triage_receipt(self, join=None, reason="fixture decision"): + """经真实 task.sh triage 生成分流收据（对 Gitea 桩）。候选存在时按 join（分支名）或 NEW 显式决策。""" + receipt = self.root / f"triage-{len(list(self.root.glob('triage*.json')))}.json" + result = self.task(self.repo, "triage", "fixture requirement", check=False, + env=self.env | {"TRIAGE_RECEIPT": str(receipt)}) + if result.returncode == 5: + decide_env = self.env | {"TRIAGE_RECEIPT": str(receipt), "REASON": reason, + "VERDICT": "JOIN" if join else "NEW"} + if join: + decide_env["INITIATIVE"] = join + self.task(self.repo, "triage-decide", env=decide_env) + else: + self.assertEqual(result.returncode, 0, result.stdout) + return receipt + + def publish_initiative(self, branch): + """把 initiative 分支推到 origin 并登记 open PR（模拟远端接纳中的大任务）。""" + self.git(self.repo, "push", "origin", f"refs/heads/{branch}:refs/heads/{branch}") + sha = self.git(self.repo, "rev-parse", f"refs/heads/{branch}").stdout.strip() + known = [pr for pr in self.gitea.prs if pr["head"]["ref"] == branch] + if known: + known[0]["head"]["sha"] = sha + else: + self.gitea.prs.append(FakeGitea.pr(len(self.gitea.prs) + 1, branch, sha)) + def test_start_requires_precheck_receipt_before_creating_worktree(self): - result = self.task(self.repo, "start", "feature/new", check=False) + triage = self.triage_receipt() + result = self.task(self.repo, "start", "feature/new", check=False, + env=self.env | {"TRIAGE_RECEIPT": str(triage)}) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertFalse((self.root / "repo-wt-feature-new").exists(), result.stdout) def test_start_preserves_merged_local_and_remote_branch_names(self): receipt = self.capture_precheck() + triage = self.triage_receipt() head = self.git(self.repo, "rev-parse", "HEAD").stdout.strip() for kind in ("local", "remote"): with self.subTest(kind=kind): @@ -695,7 +784,8 @@ class TaskLifecycleTests(unittest.TestCase): ref = "refs/remotes/origin/" + name self.git(self.repo, "fetch", "origin", "refs/heads/" + name + ":" + ref) result = self.task(self.repo, "start", name, check=False, - env=self.env | {"PRECHECK_RECEIPT": str(receipt)}) + env=self.env | {"PRECHECK_RECEIPT": str(receipt), + "TRIAGE_RECEIPT": str(triage)}) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("already merged", result.stdout) self.assertFalse((self.root / ("repo-wt-" + name.replace("/", "-"))).exists()) @@ -703,11 +793,17 @@ class TaskLifecycleTests(unittest.TestCase): def test_start_imports_precheck_and_creates_four_part_commit(self): receipt = self.capture_precheck() - self.task(self.repo, "start", "feature/new", env=self.env | {"PRECHECK_RECEIPT": str(receipt)}) + triage = self.triage_receipt() + self.task(self.repo, "start", "feature/new", + env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage)}) wt = self.root / "repo-wt-feature-new" self.assertTrue((wt / ".task/feature-new.md").is_file()) card = (wt / ".task/feature-new.md").read_text() self.assertIn("make precheck", card) + self.assertIn("- 需求分流：verdict=NEW；selected_initiative=无；requirement_sha256=", card) + self.assertIn(f"triage_receipt={triage.resolve()}；receipt_sha256=", card) + branch_line = card.index("- 分支类型：standalone") + self.assertEqual(card.index("- 需求分流："), card.index("\n", branch_line) + 1) self.assertIn("交付模式：goal-first（目标、范围和交付终点一次确认；范围内连续推进）", card) self.assertIn("实施执行：agent-per-unit（配置：AGENT-EXECUTION.md）", card) self.assertIn("人工决策点：仅限目标/范围/验收变更、重大取舍、高后果操作或无法自动判定的业务验收", card) @@ -974,7 +1070,9 @@ class TaskLifecycleTests(unittest.TestCase): self.git(parent, "commit", "-m", "Parent change") parent_head = self.git(parent, "rev-parse", "HEAD").stdout.strip() receipt = self.capture_precheck(parent) - env = self.env | {"PRECHECK_RECEIPT": str(receipt), "TASK_PARENT_BRANCH": "feature/parent", + triage = self.triage_receipt() + env = self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage), + "TASK_PARENT_BRANCH": "feature/parent", "TASK_PARENT_PR": "https://example.invalid/repo/pulls/42"} self.task(parent, "start", "feature/parent-child", env=env) wt = self.root / "repo-wt-feature-parent-child" @@ -985,17 +1083,20 @@ class TaskLifecycleTests(unittest.TestCase): self.assertEqual((wt / "parent-change").read_text(), "parent feature") def test_start_offline_does_not_create_worktree(self): + triage = self.triage_receipt() self.git(self.repo, "remote", "set-url", "origin", str(self.root / "missing.git")) - result = self.task(self.repo, "start", "feature/offline", check=False) + result = self.task(self.repo, "start", "feature/offline", check=False, + env=self.env | {"TRIAGE_RECEIPT": str(triage)}) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("REMOTE_UNVERIFIED", result.stdout) self.assertFalse((self.root / "repo-wt-feature-offline").exists()) def test_start_rejects_receipt_when_authoritative_base_changed(self): receipt = self.capture_precheck() + triage = self.triage_receipt() self.advance_remote() result = self.task(self.repo, "start", "feature/stale", check=False, - env=self.env | {"PRECHECK_RECEIPT": str(receipt)}) + env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage)}) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertFalse((self.root / "repo-wt-feature-stale").exists()) @@ -1032,80 +1133,77 @@ class TaskLifecycleTests(unittest.TestCase): def start_initiative(self, name="feature/alpha"): receipt = self.capture_precheck() + triage = self.triage_receipt() self.task(self.repo, "start", name, - env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TASK_BRANCH_KIND": "initiative"}) + env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage), + "TASK_BRANCH_KIND": "initiative"}) return self.root / ("repo-wt-" + name.replace("/", "-")) def start_child(self, name="feature/alpha-login", initiative="feature/alpha", check=True): + # 远端须已登记该 initiative（PR+卡片），子任务才能经真实 JOIN 收据启动。 + if initiative: + self.publish_initiative(initiative) + triage = self.triage_receipt(join=initiative) + else: + triage = self.root / "missing-triage.json" receipt = self.capture_precheck() - env = self.env | {"PRECHECK_RECEIPT": str(receipt), "TASK_BRANCH_KIND": "child", - "TASK_INITIATIVE": initiative, + env = self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage), + "TASK_BRANCH_KIND": "child", "TASK_INITIATIVE": initiative, "TASK_PARENT_PR": "https://example.invalid/repo/pulls/7"} result = self.task(self.repo, "start", name, env=env, check=check) return self.root / ("repo-wt-" + name.replace("/", "-")), result - def test_triage_reports_new_without_active_initiatives(self): - result = self.task(self.repo, "triage", "add login") - self.assertIn("[triage] requirement: add login", result.stdout) - self.assertIn("[triage] active initiatives: 0", result.stdout) - self.assertIn("[verdict] NEW", result.stdout) - self.assertNotIn("[initiative]", result.stdout) + def test_triage_without_receipt_is_refused(self): + result = self.task(self.repo, "triage", "add login", check=False) + self.assertNotEqual(result.returncode, 0, result.stdout) + self.assertIn("TRIAGE_RECEIPT required", result.stdout) - def test_triage_joins_single_active_initiative(self): - self.write_initiative_card("feature/alpha") - result = self.task(self.repo, "triage", "add login") - self.assertIn("[triage] active initiatives: 1", result.stdout) - self.assertIn("[initiative] branch=feature/alpha initiative_id=feature-alpha", result.stdout) - self.assertIn("[verdict] JOIN", result.stdout) - self.assertIn("INITIATIVE=feature/alpha", result.stdout) + def test_triage_ignores_local_cards_and_records_new_without_remote_initiative(self): + # main 上的本地卡片不是权威来源：远端没有 PR/分支时必须是 NEW，而不是按本地计数 JOIN。 + self.write_initiative_card("feature/local-only") + receipt = self.triage_receipt() + data = json.loads(receipt.read_text()) + self.assertEqual((data["discovery_verdict"], data["final_verdict"]), ("NEW", "NEW")) + self.assertEqual(data["candidates"], []) + self.assertIsNone(data["decision"]["selected_initiative"]) - def test_triage_requires_decision_for_multiple_active_initiatives(self): - self.write_initiative_card("feature/alpha") - self.write_initiative_card("feature/beta") - result = self.task(self.repo, "triage", check=False) + def test_remote_initiative_needs_explicit_decision_and_never_auto_joins(self): + self.start_initiative("feature/alpha") + self.publish_initiative("feature/alpha") + receipt = self.root / "single-candidate.json" + result = self.task(self.repo, "triage", "add login", check=False, + env=self.env | {"TRIAGE_RECEIPT": str(receipt)}) self.assertEqual(result.returncode, 5, result.stdout) - self.assertIn("[triage] active initiatives: 2", result.stdout) - self.assertIn("[initiative] branch=feature/alpha", result.stdout) - self.assertIn("[initiative] branch=feature/beta", result.stdout) + self.assertIn("[candidate] branch=feature/alpha", result.stdout) self.assertIn("[verdict] DECISION_REQUIRED", result.stdout) + self.assertNotIn("[verdict] JOIN", result.stdout) + data = json.loads(receipt.read_text()) + self.assertEqual(data["final_verdict"], "DECISION_REQUIRED") + self.assertIsNone(data["decision"]) - def test_triage_ignores_closed_merged_pending_and_non_accepting_initiatives(self): - self.write_initiative_card("feature/active") - self.write_initiative_card("feature/done", status="已完成（回填）") - self.write_initiative_card("feature/closed-intake", accepting="false") - self.write_initiative_card("feature/plain", kind="standalone") - self.write_initiative_card("feature/other-root", root="release") - self.write_initiative_card("feature/merged-init") - self.git(self.repo, "switch", "-c", "feature/merged-init") - (self.repo / "merged-work").write_text("landed\n") - self.git(self.repo, "add", "merged-work") - self.git(self.repo, "commit", "-m", "Merged initiative work") - self.git(self.repo, "switch", "main") - self.git(self.repo, "merge", "--no-ff", "feature/merged-init", "-m", - "Merge pull request 'Init' (#9) from feature/merged-init into main") - self.write_initiative_card("feature/pending-init") - self.git(self.repo, "switch", "-c", "feature/card-close-1") - self.git(self.repo, "mv", ".task/feature-pending-init.md", ".task/archive-tmp.md") - (self.repo / ".task/archive").mkdir() - self.git(self.repo, "mv", ".task/archive-tmp.md", ".task/archive/feature-pending-init.md") - self.git(self.repo, "commit", "-m", "Close pending initiative card") - self.git(self.repo, "switch", "main") - self.git(self.repo, "push", "origin", "main", "feature/card-close-1") - result = self.task(self.repo, "triage") - self.assertIn("[triage] active initiatives: 1", result.stdout) - self.assertIn("[initiative] branch=feature/active", result.stdout) - for branch in ("done", "closed-intake", "plain", "other-root", "merged-init", "pending-init"): - self.assertNotIn(f"branch=feature/{branch} ", result.stdout) - self.assertIn("[verdict] JOIN", result.stdout) + def test_multiple_remote_candidates_stop_at_decision_required(self): + self.start_initiative("feature/alpha") + self.publish_initiative("feature/alpha") + self.start_initiative("feature/beta") + self.publish_initiative("feature/beta") + receipt = self.root / "multi-candidate.json" + result = self.task(self.repo, "triage", "add login", check=False, + env=self.env | {"TRIAGE_RECEIPT": str(receipt)}) + self.assertEqual(result.returncode, 5, result.stdout) + self.assertIn("[candidate] branch=feature/alpha", result.stdout) + self.assertIn("[candidate] branch=feature/beta", result.stdout) + self.assertIn("[verdict] DECISION_REQUIRED", result.stdout) + self.assertNotIn("[verdict] JOIN", result.stdout) + self.assertEqual(json.loads(receipt.read_text())["final_verdict"], "DECISION_REQUIRED") def test_precheck_reports_active_initiatives(self): result = self.task(self.repo, "precheck", check=False) - self.assertIn("[initiatives] active: none", result.stdout) + self.assertIn("[initiatives] local cards (not authoritative): none", result.stdout) self.write_initiative_card("feature/alpha") self.write_initiative_card("feature/beta") self.git(self.repo, "push", "origin", "main") result = self.task(self.repo, "precheck", check=False) - self.assertIn("[initiatives] active: feature/alpha feature/beta", result.stdout) + self.assertIn("[initiatives] local cards (not authoritative): feature/alpha feature/beta", result.stdout) def test_start_initiative_records_branch_metadata(self): wt = self.start_initiative() @@ -1120,7 +1218,9 @@ class TaskLifecycleTests(unittest.TestCase): def test_start_standalone_records_default_metadata(self): receipt = self.capture_precheck() - self.task(self.repo, "start", "feature/solo", env=self.env | {"PRECHECK_RECEIPT": str(receipt)}) + triage = self.triage_receipt() + self.task(self.repo, "start", "feature/solo", + env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage)}) text = (self.root / "repo-wt-feature-solo/.task/feature-solo.md").read_text() self.assertIn("- 分支类型：standalone；root_target：main；immediate_target：main；" "accepting_requirements：false", text) @@ -1199,8 +1299,51 @@ class TaskLifecycleTests(unittest.TestCase): self.assertEqual(self.card_helper("card_root_target", card), "main") self.assertEqual(self.card_helper("card_immediate_target", card), "main") self.assertEqual(self.card_helper("card_accepting_requirements", card), "false") - result = self.task(self.repo, "triage") - self.assertIn("[verdict] NEW", result.stdout) + receipt = self.triage_receipt() + self.assertEqual(json.loads(receipt.read_text())["final_verdict"], "NEW") + + def test_start_without_triage_receipt_creates_no_worktree_or_branch(self): + receipt = self.capture_precheck() + result = self.task(self.repo, "start", "feature/untriaged", check=False, + env=self.env | {"PRECHECK_RECEIPT": str(receipt)}) + self.assertNotEqual(result.returncode, 0, result.stdout) + self.assertIn("TRIAGE_RECEIPT required", result.stdout) + self.assertFalse((self.root / "repo-wt-feature-untriaged").exists()) + self.assertEqual(self.git(self.repo, "branch", "--list", "feature/untriaged").stdout.strip(), "") + self.assertEqual(self.git(self.repo, "ls-remote", "--heads", "origin", + "feature/untriaged").stdout.strip(), "") + + def test_start_child_refuses_non_join_receipt_without_side_effects(self): + self.start_initiative("feature/alpha") + self.publish_initiative("feature/alpha") + new_receipt = self.triage_receipt() # 远端存在候选，显式 NEW 决策（非适配） + receipt = self.capture_precheck() + env = self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(new_receipt), + "TASK_BRANCH_KIND": "child", "TASK_INITIATIVE": "feature/alpha", + "TASK_PARENT_PR": "https://example.invalid/repo/pulls/7"} + result = self.task(self.repo, "start", "feature/alpha-login", check=False, env=env) + self.assertNotEqual(result.returncode, 0, result.stdout) + self.assertIn("triage receipt rejected", result.stdout) + self.assertFalse((self.root / "repo-wt-feature-alpha-login").exists()) + self.assertEqual(self.git(self.repo, "branch", "--list", "feature/alpha-login").stdout.strip(), "") + + def test_start_refuses_receipt_when_remote_initiative_moved_after_triage(self): + self.start_initiative("feature/alpha") + self.publish_initiative("feature/alpha") + join = self.triage_receipt(join="feature/alpha") + parent = self.root / "repo-wt-feature-alpha" + (parent / "moved").write_text("later initiative work\n") + self.git(parent, "add", "moved") + self.git(parent, "commit", "-m", "Move initiative after triage") + self.publish_initiative("feature/alpha") + receipt = self.capture_precheck() + env = self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(join), + "TASK_BRANCH_KIND": "child", "TASK_INITIATIVE": "feature/alpha", + "TASK_PARENT_PR": "https://example.invalid/repo/pulls/7"} + result = self.task(self.repo, "start", "feature/alpha-login", check=False, env=env) + self.assertNotEqual(result.returncode, 0, result.stdout) + self.assertIn("remote initiative candidates changed", result.stdout) + self.assertFalse((self.root / "repo-wt-feature-alpha-login").exists()) if __name__ == "__main__": @@ -1222,9 +1365,13 @@ class FinishGuardFixTests(unittest.TestCase): "GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0", "GIT_AUTHOR_NAME": "Guard Test", "GIT_AUTHOR_EMAIL": "guard@example.invalid", "GIT_COMMITTER_NAME": "Guard Test", "GIT_COMMITTER_EMAIL": "guard@example.invalid", - "TASK_WRITERS_STOPPED": "1"}) - for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GITEA_TOKEN", "PRECHECK_RECEIPT"): + "TASK_WRITERS_STOPPED": "1", + "NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost"}) + for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GITEA_TOKEN", "GITEA_API_URL", + "GITEA_REPO", "TRIAGE_RECEIPT", "PRECHECK_RECEIPT"): self.env.pop(key, None) + self.gitea = FakeGitea() + self.addCleanup(self.gitea.close) self.git(self.root, "init", "--bare", "-b", "main", self.root + "/origin.git") self.git(self.root, "init", "-b", "main", str(self.repo)) (self.repo / "tracked").write_text("base\n") @@ -1248,8 +1395,12 @@ class FinishGuardFixTests(unittest.TestCase): ["bash", str(self.repo / "scripts" / "evidence.sh")], cwd=str(self.repo), capture_output=True, text=True, env=self.env | {"RECEIPT": str(receipt), "CMD": "make precheck"}, check=True) + triage = Path(self.root) / "triage.json" + triage_env = self.env | {"GITEA_API_URL": self.gitea.url, "GITEA_REPO": "fixture/repo", + "GITEA_TOKEN": self.gitea.token} + self.task(self.repo, "triage", "guard fixture", env=triage_env | {"TRIAGE_RECEIPT": str(triage)}) self.task(self.repo, "start", "feature/guard", - env=self.env | {"PRECHECK_RECEIPT": str(receipt)}) + env=triage_env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage)}) wt = self.root + "/repo-wt-feature-guard" assert Path(wt).is_dir(), "worktree not created" return Path(wt) diff --git a/tests/test_task_recovery.py b/tests/test_task_recovery.py index 25367c27..f37dc9c5 100644 --- a/tests/test_task_recovery.py +++ b/tests/test_task_recovery.py @@ -1,8 +1,12 @@ """Local recovery is task-scoped and cannot grant delivery readiness.""" import json +import os +import sys import unittest -import test_task as lifecycle +# 允许 `python3 -m unittest tests/test_task_recovery.py` 直接运行（与 discover 模式一致地导入同目录夹具）。 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import test_task as lifecycle # noqa: E402 class TaskRecoveryTests(unittest.TestCase): @@ -178,7 +182,8 @@ class TaskRecoveryTests(unittest.TestCase): self.assertEqual(self.card.read_bytes(), before) start = f.task(self.wt, "start", "feature/bypass", check=False, env=f.env | {"PRECHECK_RECEIPT": str(receipt), "TASK_WORK_MODE": "local", - "VALIDATE_RECEIPT_PURPOSE": "resume-local"}) + "VALIDATE_RECEIPT_PURPOSE": "resume-local", + "TRIAGE_RECEIPT": str(f.triage_receipt())}) self.assertNotEqual(start.returncode, 0, start.stdout) self.assertFalse((f.root / "repo-wt-feature-bypass").exists()) diff --git a/tests/test_task_triage.py b/tests/test_task_triage.py new file mode 100644 index 00000000..3c650243 --- /dev/null +++ b/tests/test_task_triage.py @@ -0,0 +1,401 @@ +"""需求分流收据契约（scripts/task_triage.py）：远端发现、显式决策、核验拒绝。 + +全部对真实临时 Git 仓库（bare origin + clone）与进程内 Gitea 桩运行；不联网，不读取真实 ~/.secrets。 +""" +import hashlib +import json +import os +import socket +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from test_task import FakeGitea # noqa: E402 + +SOURCE = Path(__file__).resolve().parents[1] +SCRIPT = SOURCE / "scripts" / "task_triage.py" +sys.path.insert(0, str(SOURCE / "scripts")) +import task_triage # noqa: E402 + + +def card_text(branch, kind="initiative", accepting="true", root="main", status="进行中"): + slug = branch.replace("/", "-") + return ( + f"# {branch} 任务卡\n" + f"任务主记录：fixture；状态：{status}；owner：Triage Test\n" + f"- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= 0\n" + f"- 工作区：fixture（worktree）；分支：{branch}\n" + f"- 分支类型：{kind}；initiative_id：{slug}；root_target：{root}；" + f"immediate_target：{root}；accepting_requirements：{accepting}\n\n" + "## 目标与非目标\n\n- 交付登录能力\n- 不做第三方登录\n\n## 其它\n\n- 无\n" + ) + + +class TriageTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory(prefix="triage-") + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.remote = self.root / "remote.git" + self.repo = self.root / "repo" + self.receipt = self.root / "receipt.json" + self.env = os.environ.copy() + self.env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null", + "GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0", + "GIT_AUTHOR_NAME": "Triage Test", "GIT_AUTHOR_EMAIL": "triage@example.invalid", + "GIT_COMMITTER_NAME": "Triage Test", "GIT_COMMITTER_EMAIL": "triage@example.invalid", + "GIT_CONFIG_COUNT": "2", "GIT_CONFIG_KEY_0": "gc.auto", + "GIT_CONFIG_VALUE_0": "0", "GIT_CONFIG_KEY_1": "maintenance.auto", + "GIT_CONFIG_VALUE_1": "false", + "NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost", + "GITEA_REPO": "fixture/repo", "GITEA_TOKEN": "dummy"}) + for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GITEA_API_URL", "TASK_TARGET_BRANCH", + "TRIAGE_RECEIPT"): + self.env.pop(key, None) + self.git(self.root, "init", "--bare", "--initial-branch=main", str(self.remote)) + self.git(self.root, "init", "--initial-branch=main", str(self.repo)) + (self.repo / "README").write_text("fixture\n") + self.git(self.repo, "add", "README") + self.git(self.repo, "commit", "-m", "Create fixture") + self.git(self.repo, "remote", "add", "origin", str(self.remote)) + self.git(self.repo, "push", "-u", "origin", "main") + self.gitea = FakeGitea(token="dummy") + self.addCleanup(self.gitea.close) + self.env["GITEA_API_URL"] = self.gitea.url + + # ---- 夹具 ---- + + def git(self, cwd, *args): + return subprocess.run(["git", *args], cwd=cwd, env=self.env, capture_output=True, check=True) + + def make_initiative(self, branch, **card): + self.git(self.repo, "switch", "-c", branch, "main") + path = self.repo / ".task" / (branch.replace("/", "-") + ".md") + path.parent.mkdir(exist_ok=True) + path.write_text(card_text(branch, **card), encoding="utf-8") + self.git(self.repo, "add", ".task") + self.git(self.repo, "commit", "-m", f"Record {branch}") + self.git(self.repo, "push", "origin", branch) + self.git(self.repo, "switch", "main") + return self.head(branch) + + def advance_initiative(self, branch): + self.git(self.repo, "switch", branch) + (self.repo / "more-work.txt").write_text("later initiative work\n") + self.git(self.repo, "add", "more-work.txt") + self.git(self.repo, "commit", "-m", "More initiative work") + self.git(self.repo, "push", "origin", branch) + self.git(self.repo, "switch", "main") + + def head(self, ref): + return subprocess.run(["git", "rev-parse", ref], cwd=self.repo, env=self.env, + capture_output=True, text=True, check=True).stdout.strip() + + def open_pr(self, number, branch, base="main", state="open"): + self.gitea.prs.append(FakeGitea.pr(number, branch, self.head(branch), base=base, state=state)) + + def run_script(self, *args, env=None): + result = subprocess.run([sys.executable, str(SCRIPT), *args], cwd=self.repo, + env=self.env | (env or {}), capture_output=True, text=True, timeout=60) + result.out = result.stdout + result.stderr + return result + + def discover(self, env=None, requirement="add login", receipt=None): + return self.run_script("discover", "--receipt", str(receipt or self.receipt), + "--requirement", requirement, env=env) + + def decide(self, verdict, reason="", initiative=None): + args = ["decide", "--receipt", str(self.receipt), "--verdict", verdict, "--reason", reason] + if initiative: + args += ["--initiative", initiative] + return self.run_script(*args) + + def verify(self, kind, initiative=None, max_age=None, env=None): + args = ["verify", "--receipt", str(self.receipt), "--kind", kind] + if initiative: + args += ["--initiative", initiative] + if max_age is not None: + args += ["--max-age-seconds", str(max_age)] + return self.run_script(*args, env=env) + + def load(self): + return json.loads(self.receipt.read_text(encoding="utf-8")) + + def save(self, data): + self.receipt.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + + def discover_single(self, branch="feature/alpha"): + self.make_initiative(branch) + self.open_pr(len(self.gitea.prs) + 1, branch) + result = self.discover() + self.assertEqual(result.returncode, 5, result.out) + return result + + # ---- 发现 ---- + + def test_remote_initiative_absent_from_main_tree_is_discovered(self): + head = self.make_initiative("feature/alpha") + self.open_pr(1, "feature/alpha") + self.assertNotIn(".task/feature-alpha.md", + subprocess.run(["git", "ls-tree", "-r", "--name-only", "main"], cwd=self.repo, + env=self.env, capture_output=True, text=True).stdout) + result = self.discover() + self.assertEqual(result.returncode, 5, result.out) + self.assertIn("[candidate] branch=feature/alpha", result.out) + self.assertIn("[verdict] DECISION_REQUIRED", result.out) + self.assertIn("make task-triage-decide", result.out) + data = self.load() + cand = data["candidates"][0] + self.assertEqual(cand["head_sha"], head) + self.assertEqual(cand["card_path"], ".task/feature-alpha.md") + self.assertEqual(cand["pr_base"], "main") + self.assertEqual(cand["pr_number"], 1) + self.assertIn("交付登录能力", cand["goal_excerpt"]) + blob = subprocess.run(["git", "cat-file", "blob", f"{head}:.task/feature-alpha.md"], + cwd=self.repo, env=self.env, capture_output=True, check=True).stdout + self.assertEqual(cand["card_sha256"], hashlib.sha256(blob).hexdigest()) + self.assertEqual((data["final_verdict"], data["discovery_verdict"]), + ("DECISION_REQUIRED", "DECISION_REQUIRED")) + self.assertIsNone(data["decision"]) + + def test_zero_candidates_yields_new_without_selecting_an_initiative(self): + result = self.discover() + self.assertEqual(result.returncode, 0, result.out) + self.assertIn("[verdict] NEW", result.out) + data = self.load() + self.assertEqual((data["discovery_verdict"], data["final_verdict"]), ("NEW", "NEW")) + self.assertEqual(data["candidates"], []) + self.assertEqual(data["decision"]["verdict"], "NEW") + self.assertIsNone(data["decision"]["selected_initiative"]) + self.assertEqual(data["decision"]["rationale"], "no accepting initiative") + self.assertEqual(data["requirement"]["sha256"], hashlib.sha256(b"add login").hexdigest()) + self.assertEqual(data["repo"]["repo_slug"], "fixture/repo") + self.assertEqual(data["root_target_sha"], self.head("main")) + + def test_single_candidate_requires_decision_never_auto_join(self): + self.discover_single() + data = self.load() + self.assertEqual(data["final_verdict"], "DECISION_REQUIRED") + self.assertIsNone(data["decision"]) + + def test_multiple_candidates_stop_at_decision_required(self): + self.make_initiative("feature/alpha") + self.make_initiative("feature/beta") + self.open_pr(1, "feature/alpha") + self.open_pr(2, "feature/beta") + result = self.discover() + self.assertEqual(result.returncode, 5, result.out) + self.assertIn("[candidate] branch=feature/alpha", result.out) + self.assertIn("[candidate] branch=feature/beta", result.out) + data = self.load() + self.assertEqual([c["branch"] for c in data["candidates"]], ["feature/alpha", "feature/beta"]) + self.assertEqual(data["final_verdict"], "DECISION_REQUIRED") + + def test_pr_base_not_main_and_closed_pr_are_ignored(self): + self.make_initiative("feature/alpha") + self.make_initiative("feature/beta") + self.open_pr(1, "feature/alpha", base="release") + self.open_pr(2, "feature/beta", state="closed") + result = self.discover() + self.assertEqual(result.returncode, 0, result.out) + self.assertEqual(self.load()["final_verdict"], "NEW") + + def test_card_filters_ignore_closed_intake_done_plain_and_other_root(self): + self.make_initiative("feature/active") + self.make_initiative("feature/closed-intake", accepting="false") + self.make_initiative("feature/done", status="已完成（回填）") + self.make_initiative("feature/plain", kind="standalone") + self.make_initiative("feature/other-root", root="release") + for number, branch in enumerate(("feature/active", "feature/closed-intake", "feature/done", + "feature/plain", "feature/other-root"), start=1): + self.open_pr(number, branch) + result = self.discover() + self.assertEqual(result.returncode, 5, result.out) + self.assertEqual([c["branch"] for c in self.load()["candidates"]], ["feature/active"]) + + def test_remote_discovery_paginates_until_empty(self): + for number in range(1, 51): + self.gitea.prs.append(FakeGitea.pr(number, f"feature/noise-{number}", "0" * 40, base="release")) + self.make_initiative("feature/late") + self.open_pr(51, "feature/late") + result = self.discover() + self.assertEqual(result.returncode, 5, result.out) + self.assertEqual([c["branch"] for c in self.load()["candidates"]], ["feature/late"]) + + def test_api_unreachable_yields_remote_unverified_without_receipt(self): + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + result = self.discover(env={"GITEA_API_URL": f"http://127.0.0.1:{port}/api/v1"}) + self.assertEqual(result.returncode, 6, result.out) + self.assertIn("[verdict] REMOTE_UNVERIFIED", result.out) + self.assertFalse(self.receipt.exists()) + + def test_ls_remote_failure_yields_remote_unverified_without_receipt(self): + self.git(self.repo, "remote", "set-url", "origin", str(self.root / "missing.git")) + result = self.discover() + self.assertEqual(result.returncode, 6, result.out) + self.assertIn("[verdict] REMOTE_UNVERIFIED", result.out) + self.assertFalse(self.receipt.exists()) + + def test_token_from_secrets_file_is_used_and_never_printed(self): + home = self.root / "home" + (home / ".secrets").mkdir(parents=True) + (home / ".secrets" / "gitea").write_text( + "URL_PREFIX=https://example.invalid\n- ACCESS_TOKEN=file-token-7f3a\nACCESS_TOKEN=second-token\n") + self.gitea.token = "file-token-7f3a" + self.make_initiative("feature/alpha") + self.open_pr(1, "feature/alpha") + result = self.discover(env={"HOME": str(home), "GITEA_TOKEN": ""}) + self.assertEqual(result.returncode, 5, result.out) + self.assertNotIn("file-token-7f3a", result.out) + self.assertNotIn("file-token-7f3a", self.receipt.read_text(encoding="utf-8")) + self.gitea.token = "other-token" + rejected = self.discover(env={"HOME": str(home), "GITEA_TOKEN": ""}, receipt=self.root / "other.json") + self.assertEqual(rejected.returncode, 6, rejected.out) + self.assertIn("[verdict] REMOTE_UNVERIFIED", rejected.out) + self.assertNotIn("file-token-7f3a", rejected.out) + + def test_repo_slug_is_derived_from_origin_when_gitea_repo_unset(self): + owner_dir = self.root / "fixture-owner" + owner_dir.mkdir() + named = owner_dir / "fixture-repo.git" + self.git(self.root, "init", "--bare", "--initial-branch=main", str(named)) + self.git(self.repo, "remote", "set-url", "origin", str(named)) + self.git(self.repo, "push", "-u", "origin", "main") + result = self.discover(env={"GITEA_REPO": ""}) + self.assertEqual(result.returncode, 0, result.out) + self.assertEqual(self.load()["repo"]["repo_slug"], "fixture-owner/fixture-repo") + + def test_discover_requires_requirement_text(self): + result = self.discover(requirement=" ") + self.assertEqual(result.returncode, 2, result.out) + + # ---- 决策 ---- + + def test_join_decision_succeeds_and_verifies_child(self): + head = self.make_initiative("feature/alpha") + self.open_pr(1, "feature/alpha") + self.discover() + decided = self.decide("JOIN", "加入登录主线", initiative="feature/alpha") + self.assertEqual(decided.returncode, 0, decided.out) + data = self.load() + self.assertEqual(data["final_verdict"], "JOIN") + self.assertEqual(data["decision"]["selected_initiative"], "feature/alpha") + verified = self.verify("child", initiative="feature/alpha") + self.assertEqual(verified.returncode, 0, verified.out) + self.assertIn("[triage-verified] verdict=JOIN selected_initiative=feature/alpha " + "requirement_sha256=", verified.out) + self.assertIn(f"[triage-head] selected_head_sha={head}", verified.out) + + def test_new_decision_for_single_non_fitting_candidate_requires_reason(self): + self.discover_single() + before = self.receipt.read_bytes() + no_reason = self.decide("NEW", "") + self.assertEqual(no_reason.returncode, 7, no_reason.out) + self.assertEqual(self.receipt.read_bytes(), before) + decided = self.decide("NEW", "不适配：独立的后台巡检任务") + self.assertEqual(decided.returncode, 0, decided.out) + self.assertEqual(self.load()["final_verdict"], "NEW") + self.assertEqual(self.verify("standalone").returncode, 0) + + def test_decide_rejects_unknown_initiative_or_missing_reason_without_changing_receipt(self): + self.make_initiative("feature/alpha") + self.open_pr(1, "feature/alpha") + self.discover() + before = self.receipt.read_bytes() + unknown = self.decide("JOIN", "理由", initiative="feature/beta") + self.assertEqual(unknown.returncode, 7, unknown.out) + no_reason = self.decide("JOIN", "", initiative="feature/alpha") + self.assertEqual(no_reason.returncode, 7, no_reason.out) + self.assertEqual(self.receipt.read_bytes(), before) + + def test_decide_rejects_candidate_head_change_and_keeps_receipt(self): + self.discover_single() + before = self.receipt.read_bytes() + self.advance_initiative("feature/alpha") + result = self.decide("JOIN", "加入", initiative="feature/alpha") + self.assertEqual(result.returncode, 7, result.out) + self.assertIn("candidate set changed", result.out) + self.assertEqual(self.receipt.read_bytes(), before) + + # ---- 核验 ---- + + def test_verify_rejects_stale_receipt(self): + self.assertEqual(self.discover().returncode, 0) + self.assertEqual(self.verify("standalone").returncode, 0) + data = self.load() + data["discovered_at"] = "2001-01-01T00:00:00Z" + self.save(data) + result = self.verify("standalone") + self.assertEqual(result.returncode, 8, result.out) + self.assertIn("stale", result.out) + + def test_verify_rejects_changed_candidate_head(self): + self.discover_single() + self.decide("JOIN", "加入", initiative="feature/alpha") + self.advance_initiative("feature/alpha") + result = self.verify("child", initiative="feature/alpha") + self.assertEqual(result.returncode, 8, result.out) + self.assertIn("candidates changed", result.out) + + def test_verify_rejects_changed_card_digest(self): + self.discover_single() + self.decide("JOIN", "加入", initiative="feature/alpha") + data = self.load() + data["candidates"][0]["card_sha256"] = "0" * 64 + self.save(data) + result = self.verify("child", initiative="feature/alpha") + self.assertEqual(result.returncode, 8, result.out) + + def test_verify_rejects_verdict_and_kind_mismatch(self): + self.discover_single() + self.decide("JOIN", "加入", initiative="feature/alpha") + standalone = self.verify("standalone") + self.assertEqual(standalone.returncode, 8, standalone.out) + self.decide("NEW", "不适配") + child = self.verify("child", initiative="feature/alpha") + self.assertEqual(child.returncode, 8, child.out) + + def test_verify_rejects_child_with_wrong_initiative(self): + self.make_initiative("feature/alpha") + self.make_initiative("feature/beta") + self.open_pr(1, "feature/alpha") + self.open_pr(2, "feature/beta") + self.discover() + self.decide("JOIN", "加入", initiative="feature/alpha") + result = self.verify("child", initiative="feature/beta") + self.assertEqual(result.returncode, 8, result.out) + + def test_verify_rejects_root_target_mismatch(self): + self.discover() + result = self.verify("standalone", env={"TASK_TARGET_BRANCH": "release"}) + self.assertEqual(result.returncode, 8, result.out) + self.assertIn("root target", result.out) + + def test_verify_rejects_wrong_schema_and_missing_receipt(self): + self.discover() + data = self.load() + data["schema"] = "agent-system/other/v9" + self.save(data) + self.assertEqual(self.verify("standalone").returncode, 8) + self.receipt.unlink() + missing = self.verify("standalone") + self.assertEqual(missing.returncode, 8, missing.out) + + +class RedactionTests(unittest.TestCase): + def test_userinfo_is_stripped_from_urls_and_scp_forms(self): + self.assertEqual(task_triage.redact_url("https://user:secret@git.example:30319/o/r.git"), + "https://git.example:30319/o/r.git") + self.assertEqual(task_triage.redact_url("git@git-ssh.example:weichen/Repo.git"), + "git-ssh.example:weichen/Repo.git") + self.assertEqual(task_triage.redact_url("/tmp/plain/remote.git"), "/tmp/plain/remote.git") + + +if __name__ == "__main__": + unittest.main()