Merge pull request #595: Add remote triage receipts and the PR base gate

.gitea/workflows/ci.yaml

@@ -1,7 +1,6 @@
 name: ci
 on:
   pull_request:
-    branches: [main]
   push:
     branches: [main]
 
@@ -156,6 +155,8 @@ jobs:
           print("Checkout complete; credentials were not persisted in Git configuration")
           PY_CHECKOUT
           unset TOKEN SUBMODULE_TOKEN
+      - name: PR base gate
+        run: python3 scripts/task_pr_check.py ci
       - name: Install Go 1.27.1
         run: |
           set -euo pipefail

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 task-pr-check 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/<slug>"; exit 2; fi
+	@if [ -z "$(NAME)" ]; then echo "usage: make task-start NAME=feature/<slug> TRIAGE_RECEIPT=<path>"; exit 2; fi
 	@bash $(TASK) start "$(NAME)"
 
 task-resume:
@@ -57,6 +60,10 @@ task-close:
 task-finish:
 	@bash $(TASK) finish $(ABANDON)
 
+# PR base/head 门禁（两级分支）：本地分支卡片期望目标 vs Gitea PR 与 origin；FINAL=1 用于 initiative 收口。
+task-pr-check:
+	python3 scripts/task_pr_check.py local $(if $(PR),--pr $(PR)) $(if $(FINAL),--final)
+
 # 证据写入：执行命令并把结果行追加进当前任务卡（AGENTS.md「证据与验证门禁」）。
 # 本地执行只记 SELF；RECEIPT 捕获启动收据，IMPORT_RECEIPT 导入原始收据而不重跑。
 # 用法：CMD="<单行命令>" make evidence [TAIL=15]；退出码随被执行命令。
@@ -87,6 +94,10 @@ help:
 	@echo "make agent-exec-validate  校验 AGENT-EXECUTION.md（agent_exec validate）"
 	@echo "make agent-exec-apply RECEIPT=<path>  快进当前分支到成功收据的 transport"
 	@echo "make agent-exec-admit  校验暂存路径具备 transport 来源或属于 host 豁免"
+	@echo "make task-triage REQ='<需求>' TRIAGE_RECEIPT=<path>  以远端为准发现接纳中的 initiative 并写分流收据（候选存在时 exit 5，待决策）"
+	@echo "make task-triage-decide TRIAGE_RECEIPT=<path> VERDICT=JOIN|NEW [INITIATIVE=<branch>] REASON='<理由>'  记录显式分流决策（不按数量自动 JOIN）"
+	@echo "make task-start NAME=feature/<slug> TRIAGE_RECEIPT=<path> [KIND=...] [INITIATIVE=...]  建任务前核验分流收据（失败不建分支）"
+	@echo "make task-pr-check [PR=N] [FINAL=1]  核验当前任务分支 PR 的 base/head 门禁（远端为准；FINAL=1 用于 initiative 收口）"
 
 agent-exec-validate:
 	PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py validate --repo .

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=<path> 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/<slug> KIND=initiative (or KIND=standalone)"
-  elif [ "$n" -eq 1 ]; then
-    b=${list%%|*}
-    echo "[verdict] JOIN — join initiative ${b}: make task-start NAME=${b}-<topic> 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/<name>.json"
+  local req="${*:-}"
+  [ -n "$req" ] || die "requirement text required: make task-triage REQ='<需求>' TRIAGE_RECEIPT=<path>"
+  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=<path> 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/<target> 的记录；兼容 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 <precheck|triage|start|resume|sync|close|finish> [args]"; exit 2 ;;
+  *) echo "usage: task.sh <precheck|triage|triage-decide|start|resume|sync|close|finish> [args]"; exit 2 ;;
 esac

scripts/task_pr_check.py

@@ -0,0 +1,291 @@
+#!/usr/bin/env python3
+"""PR base/head 门禁（两级分支模型：child → initiative → main）。
+
+子命令：
+  local [--pr N] [--final]
+      核验当前检出的任务分支及其 PR（事实来源是 Gitea 与 origin）：PR 处于开放状态、base 为卡片期望目标、
+      PR head 与本地 HEAD 一致、origin 分支头与本地 HEAD 一致、期望目标分支存在；child 的父 initiative
+      须仍有开放 PR 指向其 root_target。--final（仅 initiative）：accepting_requirements=false，且
+      `## 子任务登记` 中「必需=是」的子任务为「已合入」（PR 已合并入本 initiative）或「已延期」。
+  ci
+      仅 pull_request 事件：按 GITHUB_HEAD_REF 在检出的 .task/ 顶层取卡片，核对 GITHUB_BASE_REF。
+      不联网、不需要 token；非 pull_request 事件跳过（退出 0）。
+
+期望目标：child → immediate_target；其余（initiative / standalone / 无卡）→ root_target（无卡即 main）。
+
+退出码：0 通过；1 门禁拒绝（PR_CHECK_FAILED）；2 用法；6 远端无法核验（REMOTE_UNVERIFIED，fail closed）。
+仅使用标准库，复用 task_triage 的 Gitea 客户端与 token 读取。token 不打印、不写入任何产物。
+"""
+import argparse
+import os
+import re
+import subprocess
+import sys
+import urllib.parse
+from pathlib import Path
+
+from task_triage import (EXIT_REMOTE, Stop, api_base, as_dict, card_branch, http_get_json,
+                         open_pulls, read_token, repo_slug, target_branch)
+
+EXIT_FAILED = 1
+EXIT_USAGE = 2
+CHILD_COLUMNS = ("子任务分支", "必需", "状态", "PR")
+DONE_STATES = ("已合入", "已延期")
+
+
+def sh(*args):
+    return subprocess.run(["git", *args], capture_output=True, text=True,
+                          encoding="utf-8", errors="replace")
+
+
+def standalone(target):
+    return {"kind": "standalone", "initiative_id": "", "root_target": target,
+            "immediate_target": target, "accepting": "false"}
+
+
+def card_meta(text, target):
+    """两级分支元数据行；无该行即旧卡，按 standalone 处理（与 task.sh 的 card_branch_kind 一致）。"""
+    match = re.search(r"^- 分支类型：([^\n]*)$", text, re.M)
+    if not match:
+        return standalone(target)
+    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),
+            "immediate_target": field("immediate_target", target),
+            "accepting": field("accepting_requirements", "false")}
+
+
+def expected_base(meta):
+    return meta["immediate_target"] if meta["kind"] == "child" else meta["root_target"]
+
+
+def find_card(branch):
+    """只看检出树的 .task/ 顶层（archive 不参与）；返回 (路径, 文本)，无卡返回 (None, None)。"""
+    found = []
+    for path in sorted(Path(".task").glob("*.md")):
+        text = path.read_text(encoding="utf-8", errors="replace")
+        if card_branch(text) == branch:
+            found.append((path, text))
+    if len(found) > 1:
+        raise Stop(EXIT_FAILED, f"ambiguous task cards for {branch}: {', '.join(str(p) for p, _ in found)}")
+    return found[0] if found else (None, None)
+
+
+def branch_meta(branch, target):
+    _, text = find_card(branch)
+    return (standalone(target) if text is None else card_meta(text, target)), text
+
+
+def current_branch():
+    proc = sh("symbolic-ref", "--quiet", "--short", "HEAD")
+    if proc.returncode != 0 or not proc.stdout.strip():
+        raise Stop(EXIT_USAGE, "detached HEAD: check out the task branch first")
+    return proc.stdout.strip()
+
+
+def local_head():
+    proc = sh("rev-parse", "--verify", "-q", "HEAD^{commit}")
+    if proc.returncode != 0:
+        raise Stop(EXIT_USAGE, "cannot resolve local HEAD")
+    return proc.stdout.strip()
+
+
+def remote_head(branch):
+    proc = sh("ls-remote", "origin", f"refs/heads/{branch}")
+    if proc.returncode != 0:
+        raise Stop(EXIT_REMOTE, "git ls-remote origin failed")
+    for line in proc.stdout.splitlines():
+        sha, _, ref = line.partition("\t")
+        if ref == f"refs/heads/{branch}":
+            return sha
+    return None
+
+
+def remote_context():
+    origin = sh("remote", "get-url", "origin")
+    if origin.returncode != 0:
+        raise Stop(EXIT_REMOTE, "origin remote is not configured")
+    slug = repo_slug(origin.stdout.strip())
+    token = read_token()
+    if not token:
+        raise Stop(EXIT_REMOTE, "no Gitea token: set GITEA_TOKEN or ~/.secrets/gitea ACCESS_TOKEN")
+    return api_base(), slug, token
+
+
+def pull_request(api, slug, token, number):
+    owner, name = (urllib.parse.quote(part, safe="") for part in slug.split("/"))
+    pr = http_get_json(f"{api}/repos/{owner}/{name}/pulls/{number}", token)
+    if not isinstance(pr, dict):
+        raise Stop(EXIT_REMOTE, f"Gitea PR #{number} response is malformed")
+    return pr
+
+
+def base_ref(pr):
+    return as_dict(pr.get("base")).get("ref")
+
+
+def head_ref(pr):
+    return as_dict(pr.get("head")).get("ref")
+
+
+def head_sha(pr):
+    return as_dict(pr.get("head")).get("sha")
+
+
+def parent_pr_open(api, slug, token, parent, root):
+    return any(isinstance(pr, dict) and pr.get("state") == "open"
+               and head_ref(pr) == parent and base_ref(pr) == root
+               for pr in open_pulls(api, slug, token))
+
+
+def pr_number(cell):
+    for pattern in (r"pulls/(\d+)", r"#(\d+)", r"^(\d+)$"):
+        found = re.search(pattern, cell.strip())
+        if found:
+            return int(found.group(1))
+    return None
+
+
+def child_registry(text):
+    """解析 `## 子任务登记` 表（列：子任务分支 | 必需 | 状态 | PR）；缺失即拒绝（fail closed）。"""
+    lines = text.splitlines()
+    starts = [i for i, line in enumerate(lines) if line.strip() == "## 子任务登记"]
+    if not starts:
+        raise Stop(EXIT_FAILED, "initiative card has no ## 子任务登记 section")
+    rows, columns = [], None
+    for line in lines[starts[0] + 1:]:
+        if line.startswith("## "):
+            break
+        if not line.strip().startswith("|"):
+            continue
+        cells = [cell.strip() for cell in line.strip().strip("|").split("|")]
+        if columns is None:
+            if not all(name in cells for name in CHILD_COLUMNS):
+                raise Stop(EXIT_FAILED, "子任务登记 table must have columns 子任务分支 | 必需 | 状态 | PR")
+            columns = {name: cells.index(name) for name in CHILD_COLUMNS}
+            continue
+        if all(re.fullmatch(r":?-{3,}:?", cell) for cell in cells):
+            continue
+        if len(cells) <= max(columns.values()):
+            raise Stop(EXIT_FAILED, f"malformed 子任务登记 row: {line.strip()}")
+        rows.append({name: cells[index] for name, index in columns.items()})
+    if columns is None:
+        raise Stop(EXIT_FAILED, "子任务登记 section has no table")
+    return rows
+
+
+def final_problems(api, slug, token, branch, meta, text):
+    problems = []
+    if meta["accepting"] != "false":
+        problems.append(f"initiative {branch} still accepts requirements "
+                        f"(accepting_requirements={meta['accepting']}); set false before --final")
+    for row in child_registry(text):
+        name, required, status, pr_cell = row["子任务分支"], row["必需"], row["状态"], row["PR"]
+        if required == "否":
+            continue
+        if required != "是":
+            problems.append(f"child {name} has invalid 必需 value {required!r}")
+        elif status == "已合入":
+            number = pr_number(pr_cell)
+            if number is None:
+                problems.append(f"child {name} is 已合入 but PR cell {pr_cell!r} names no PR")
+                continue
+            pr = pull_request(api, slug, token, number)
+            if pr.get("merged") is not True or base_ref(pr) != branch:
+                problems.append(f"child {name} PR #{number} is not merged into {branch}")
+        elif status not in DONE_STATES:
+            problems.append(f"required child {name} is {status!r}; must be 已合入 or 已延期")
+    return problems
+
+
+def verdict(problems, kind, base, head):
+    if problems:
+        for problem in problems:
+            print(f"[pr-check] FAIL {problem}")
+        print("[verdict] PR_CHECK_FAILED")
+        return EXIT_FAILED
+    print(f"[verdict] PR_CHECK_OK base={base} head={head} kind={kind}")
+    return 0
+
+
+def cmd_local(args):
+    if args.pr is not None and args.pr < 1:
+        raise Stop(EXIT_USAGE, "--pr must be a positive PR number")
+    branch = current_branch()
+    head = local_head()
+    meta, text = branch_meta(branch, target_branch())
+    kind, expected = meta["kind"], expected_base(meta)
+    api, slug, token = remote_context()
+    if args.pr is not None:
+        pr = pull_request(api, slug, token, args.pr)
+    else:
+        pulls = [pr for pr in open_pulls(api, slug, token) if isinstance(pr, dict) and head_ref(pr) == branch]
+        if not pulls:
+            raise Stop(EXIT_FAILED, f"no open PR has head {branch}")
+        if len(pulls) > 1:
+            raise Stop(EXIT_FAILED, f"{len(pulls)} open PRs have head {branch}; pass --pr N")
+        pr = pulls[0]
+    number = pr.get("number")
+    problems = []
+    if pr.get("merged") is True:
+        problems.append(f"PR #{number} is already merged")
+    elif pr.get("state") != "open":
+        problems.append(f"PR #{number} is not open (state={pr.get('state')})")
+    if base_ref(pr) != expected:
+        problems.append(f"PR #{number} base is {base_ref(pr)!r}; expected {expected!r} for {kind} branch {branch}")
+    if head_sha(pr) != head:
+        problems.append(f"PR #{number} head {head_sha(pr) or '(none)'} != local HEAD {head}")
+    remote = remote_head(branch)
+    if remote is None:
+        problems.append(f"origin has no branch {branch}")
+    elif remote != head:
+        problems.append(f"origin/{branch} is {remote}, local HEAD is {head}")
+    if remote_head(expected) is None:
+        problems.append(f"base branch {expected} is absent on origin")
+    if kind == "child" and not parent_pr_open(api, slug, token, expected, meta["root_target"]):
+        problems.append(f"parent initiative {expected} has no open PR into {meta['root_target']}")
+    if args.final and kind == "initiative":
+        problems.extend(final_problems(api, slug, token, branch, meta, text))
+    return verdict(problems, kind, expected, head)
+
+
+def cmd_ci(_args):
+    event = os.environ.get("GITHUB_EVENT_NAME", "")
+    if event != "pull_request":
+        print(f"[pr-check] skip: event {event or '(none)'} is not pull_request")
+        return 0
+    base = os.environ.get("GITHUB_BASE_REF", "").strip()
+    head = os.environ.get("GITHUB_HEAD_REF", "").strip()
+    if not base or not head:
+        raise Stop(EXIT_FAILED, "pull_request event lacks GITHUB_BASE_REF or GITHUB_HEAD_REF")
+    meta, _ = branch_meta(head, target_branch())
+    expected = expected_base(meta)
+    if base != expected:
+        raise Stop(EXIT_FAILED, f"PR base {base!r} differs from expected {expected!r} "
+                                f"for {meta['kind']} branch {head}")
+    return verdict([], meta["kind"], base, head)
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(prog="task_pr_check.py", description="PR base/head 门禁（两级分支模型）")
+    sub = parser.add_subparsers(dest="cmd", required=True)
+    local = sub.add_parser("local")
+    local.add_argument("--pr", type=int)
+    local.add_argument("--final", action="store_true")
+    sub.add_parser("ci")
+    args = parser.parse_args(argv)
+    try:
+        return cmd_local(args) if args.cmd == "local" else cmd_ci(args)
+    except Stop as stop:
+        print(f"[pr-check] FAIL {stop.message}")
+        print("[verdict] REMOTE_UNVERIFIED" if stop.code == EXIT_REMOTE else "[verdict] PR_CHECK_FAILED")
+        return stop.code
+
+
+if __name__ == "__main__":
+    sys.exit(main())

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())

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)

tests/test_task_pr_check.py

@@ -0,0 +1,401 @@
+"""PR base/head 门禁（scripts/task_pr_check.py）：临时真实 Git 仓库（bare origin + clone）+ 进程内 Gitea 桩。
+
+全部离线运行：不联网、不读取真实 ~/.secrets（HOME 指向临时目录），token 为合成值。
+"""
+import http.server
+import json
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+import urllib.parse
+from pathlib import Path
+
+SOURCE = Path(__file__).resolve().parents[1]
+SCRIPT = SOURCE / "scripts" / "task_pr_check.py"
+
+TOKEN = "synthetic-gitea-token-not-real"
+
+
+def card_text(branch, kind, root="main", immediate="main", accepting="false", registry=None):
+    slug = branch.replace("/", "-")
+    lines = [
+        f"# {branch} 任务卡",
+        "任务主记录：fixture；状态：进行中；owner：PR Check Test",
+        "- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= 0",
+        f"- 工作区：fixture（worktree）；分支：{branch}",
+        f"- 分支类型：{kind}；initiative_id：{slug}；root_target：{root}；"
+        f"immediate_target：{immediate}；accepting_requirements：{accepting}",
+        "",
+        "## 目标与非目标",
+        "",
+        "- 交付 PR 门禁",
+        "",
+    ]
+    if registry is not None:
+        lines += ["## 子任务登记", "", "| 子任务分支 | 必需 | 状态 | PR |", "| --- | --- | --- | --- |"]
+        lines += [f"| {branch_name} | {required} | {status} | {pr} |"
+                  for branch_name, required, status, pr in registry]
+        lines.append("")
+    return "\n".join(lines) + "\n"
+
+
+def pr(number, branch, base, sha, state="open", merged=False):
+    return {"number": number, "state": state, "merged": merged,
+            "base": {"ref": base}, "head": {"ref": branch, "sha": sha}}
+
+
+class FakeGitea:
+    """Gitea pulls API 桩（127.0.0.1）：校验 token，列表（分页+state）与单 PR 查询；从不联网。"""
+
+    PAGE = 50
+
+    def __init__(self, token=TOKEN):
+        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)
+                query = urllib.parse.parse_qs(parsed.query)
+                parts = parsed.path.split("/")
+                if parts[1:4] != ["api", "v1", "repos"] or len(parts) not in (7, 8) or parts[6] != "pulls":
+                    return self._send(404, {"message": "not found"})
+                if len(parts) == 8:
+                    found = [item for item in stub.prs if str(item["number"]) == parts[7]]
+                    return self._send(200, found[0]) if found else self._send(404, {"message": "not found"})
+                state = query.get("state", [""])[0]
+                items = [item for item in stub.prs if not state or item["state"] == state]
+                page = int(query.get("page", ["1"])[0])
+                start = (page - 1) * stub.PAGE
+                return self._send(200, items[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()
+
+
+def closed_port_url():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        port = sock.getsockname()[1]
+    return f"http://127.0.0.1:{port}/api/v1"
+
+
+class PrCheckTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="pr-check-")
+        self.addCleanup(self.tmp.cleanup)
+        root = Path(self.tmp.name)
+        self.remote = root / "remote.git"
+        self.repo = root / "repo"
+        self.env = {k: v for k, v in os.environ.items()
+                    if not k.startswith(("GITHUB_", "GIT_")) and k not in {
+                        "GITEA_API_URL", "GITEA_REPO", "GITEA_TOKEN", "TASK_TARGET_BRANCH"}}
+        self.env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null",
+                         "GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0",
+                         "GIT_AUTHOR_NAME": "PR Check Test", "GIT_AUTHOR_EMAIL": "pr-check@example.invalid",
+                         "GIT_COMMITTER_NAME": "PR Check Test", "GIT_COMMITTER_EMAIL": "pr-check@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",
+                         "HOME": str(root / "no-home"),
+                         "NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost",
+                         "GITEA_REPO": "fixture/repo", "GITEA_TOKEN": TOKEN})
+        self.git(root, "init", "--bare", "--initial-branch=main", str(self.remote))
+        self.git(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()
+        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,
+                              text=True, check=True).stdout.strip()
+
+    def head(self, ref="HEAD"):
+        return self.git(self.repo, "rev-parse", ref)
+
+    def make_branch(self, branch, base="main"):
+        self.git(self.repo, "switch", "-c", branch, base)
+        self.git(self.repo, "push", "-u", "origin", branch)
+        return self.head()
+
+    def commit_card(self, branch, text, base="main"):
+        self.make_branch(branch, base)
+        path = self.repo / ".task" / (branch.replace("/", "-") + ".md")
+        path.parent.mkdir(exist_ok=True)
+        path.write_text(text, 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)
+        return self.head()
+
+    def set_card(self, branch, text):
+        self.git(self.repo, "switch", branch)
+        path = self.repo / ".task" / (branch.replace("/", "-") + ".md")
+        path.write_text(text, encoding="utf-8")
+        self.git(self.repo, "add", ".task")
+        self.git(self.repo, "commit", "-m", f"Update {branch}")
+        self.git(self.repo, "push", "origin", branch)
+        return self.head()
+
+    def commit_work(self, branch, push=True):
+        self.git(self.repo, "switch", branch)
+        count = len(list(self.repo.glob("work-*.txt")))
+        (self.repo / f"work-{count}.txt").write_text("more work\n")
+        self.git(self.repo, "add", ".")
+        self.git(self.repo, "commit", "-m", "More work")
+        if push:
+            self.git(self.repo, "push", "origin", "HEAD")
+        return self.head()
+
+    def build_child(self):
+        self.commit_card("feature/init", card_text("feature/init", "initiative", accepting="true"))
+        self.commit_card("feature/init-child",
+                         card_text("feature/init-child", "child", root="main", immediate="feature/init"),
+                         base="feature/init")
+        return self.head()
+
+    def invoke(self, *args, env=None, drop=()):
+        merged = {k: v for k, v in (self.env | (env or {})).items() if k not in drop}
+        result = subprocess.run([sys.executable, str(SCRIPT), *args], cwd=self.repo, env=merged,
+                                capture_output=True, text=True, timeout=60)
+        result.out = result.stdout + result.stderr
+        self.assertNotIn(TOKEN, result.out)
+        return result
+
+    def local(self, *args, env=None, drop=()):
+        return self.invoke("local", *args, env=env, drop=drop)
+
+    @staticmethod
+    def fails(result):
+        return [line for line in result.stdout.splitlines() if line.startswith("[pr-check] FAIL")]
+
+    # ---- local：基址与分支类型 ----
+
+    def test_child_pr_into_main_is_rejected(self):
+        self.build_child()
+        self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init")),
+                          pr(2, "feature/init-child", "main", self.head())]
+        result = self.local()
+        self.assertEqual(result.returncode, 1, result.out)
+        self.assertEqual(len(self.fails(result)), 1, result.out)
+        self.assertIn("expected 'feature/init' for child branch feature/init-child", result.out)
+        self.assertIn("[verdict] PR_CHECK_FAILED", result.out)
+
+    def test_child_pr_into_initiative_passes(self):
+        self.build_child()
+        head = self.head()
+        self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init")),
+                          pr(2, "feature/init-child", "feature/init", head)]
+        result = self.local()
+        self.assertEqual(result.returncode, 0, result.out)
+        self.assertIn(f"[verdict] PR_CHECK_OK base=feature/init head={head} kind=child", result.out)
+
+    def test_child_is_rejected_when_parent_initiative_pr_is_closed(self):
+        self.build_child()
+        self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init"), state="closed"),
+                          pr(2, "feature/init-child", "feature/init", self.head())]
+        result = self.local()
+        self.assertEqual(result.returncode, 1, result.out)
+        self.assertEqual(len(self.fails(result)), 1, result.out)
+        self.assertIn("parent initiative feature/init has no open PR into main", result.out)
+
+    def test_standalone_and_initiative_base_gate(self):
+        self.make_branch("feature/other")
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        self.commit_card("feature/init-x", card_text("feature/init-x", "initiative", accepting="true"))
+        for branch, kind in (("feature/solo", "standalone"), ("feature/init-x", "initiative")):
+            self.git(self.repo, "switch", branch)
+            head = self.head()
+            for base, expected_ok in (("main", True), ("feature/other", False)):
+                with self.subTest(kind=kind, base=base):
+                    self.gitea.prs = [pr(1, branch, base, head)]
+                    result = self.local()
+                    if expected_ok:
+                        self.assertEqual(result.returncode, 0, result.out)
+                        self.assertIn(f"[verdict] PR_CHECK_OK base=main head={head} kind={kind}", result.out)
+                    else:
+                        self.assertEqual(result.returncode, 1, result.out)
+                        self.assertIn(f"expected 'main' for {kind} branch {branch}", result.out)
+
+    # ---- local：PR 状态与 head ----
+
+    def test_stale_pr_head_is_rejected(self):
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        stale = self.head()
+        self.gitea.prs = [pr(1, "feature/solo", "main", stale)]
+        fresh = self.commit_work("feature/solo")
+        result = self.local()
+        self.assertEqual(result.returncode, 1, result.out)
+        self.assertEqual(len(self.fails(result)), 1, result.out)
+        self.assertIn(f"PR #1 head {stale} != local HEAD {fresh}", result.out)
+
+    def test_remote_branch_head_differing_from_local_is_rejected(self):
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        remote = self.head()
+        local = self.commit_work("feature/solo", push=False)
+        self.gitea.prs = [pr(1, "feature/solo", "main", local)]
+        result = self.local()
+        self.assertEqual(result.returncode, 1, result.out)
+        self.assertEqual(len(self.fails(result)), 1, result.out)
+        self.assertIn(f"origin/feature/solo is {remote}, local HEAD is {local}", result.out)
+
+    def test_closed_and_merged_prs_are_rejected(self):
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        head = self.head()
+        for kwargs, reason in (({"state": "closed"}, "PR #1 is not open (state=closed)"),
+                               ({"state": "closed", "merged": True}, "PR #1 is already merged")):
+            with self.subTest(reason=reason):
+                self.gitea.prs = [pr(1, "feature/solo", "main", head, **kwargs)]
+                result = self.local("--pr", "1")
+                self.assertEqual(result.returncode, 1, result.out)
+                self.assertIn(reason, result.out)
+
+    def test_pr_lookup_requires_one_open_pr_or_explicit_number(self):
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        head = self.head()
+        self.make_branch("feature/other")
+        self.git(self.repo, "switch", "feature/solo")
+        self.gitea.prs = [pr(1, "feature/solo", "feature/other", head), pr(2, "feature/solo", "main", head)]
+        ambiguous = self.local()
+        self.assertEqual(ambiguous.returncode, 1, ambiguous.out)
+        self.assertIn("2 open PRs have head feature/solo; pass --pr N", ambiguous.out)
+        explicit = self.local("--pr", "2")
+        self.assertEqual(explicit.returncode, 0, explicit.out)
+        self.gitea.prs = []
+        missing = self.local()
+        self.assertEqual(missing.returncode, 1, missing.out)
+        self.assertIn("no open PR has head feature/solo", missing.out)
+
+    # ---- 远端不可核验：fail closed ----
+
+    def test_unreachable_or_unauthorized_api_fails_closed_with_exit_6(self):
+        self.build_child()
+        self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init")),
+                          pr(2, "feature/init-child", "feature/init", self.head())]
+        for label, env in (("unreachable", {"GITEA_API_URL": closed_port_url()}),
+                           ("unauthorized", {"GITEA_TOKEN": "wrong-token"})):
+            with self.subTest(label):
+                result = self.local(env=env)
+                self.assertEqual(result.returncode, 6, result.out)
+                self.assertIn("[verdict] REMOTE_UNVERIFIED", result.out)
+
+    def test_token_is_first_access_token_line_of_secrets_file(self):
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        home = Path(self.tmp.name) / "secrets-home"
+        (home / ".secrets").mkdir(parents=True)
+        (home / ".secrets" / "gitea").write_text(
+            f"URL_PREFIX=https://example.invalid\n- ACCESS_TOKEN={TOKEN}\nACCESS_TOKEN=other-token\n")
+        self.gitea.prs = [pr(1, "feature/solo", "main", self.head())]
+        result = self.local(env={"HOME": str(home)}, drop=("GITEA_TOKEN",))
+        self.assertEqual(result.returncode, 0, result.out)
+
+    # ---- local --final ----
+
+    def test_final_rejects_initiative_still_accepting_requirements(self):
+        self.commit_card("feature/init", card_text("feature/init", "initiative", accepting="true", registry=[]))
+        self.gitea.prs = [pr(1, "feature/init", "main", self.head())]
+        result = self.local("--final")
+        self.assertEqual(result.returncode, 1, result.out)
+        self.assertIn("still accepts requirements", result.out)
+
+    def test_final_requires_required_children_merged_or_deferred(self):
+        self.commit_card("feature/init", card_text("feature/init", "initiative", accepting="true"))
+        scenarios = [
+            ("merged and deferred pass",
+             [("feature/init-a", "是", "已合入", "#2"), ("feature/init-b", "否", "进行中", "-"),
+              ("feature/init-c", "是", "已延期", "-")],
+             [pr(2, "feature/init-a", "feature/init", "a" * 40, state="closed", merged=True)],
+             None),
+            ("required child still in progress",
+             [("feature/init-d", "是", "进行中", "-")], [], "required child feature/init-d is '进行中'"),
+            ("open PR is not merged",
+             [("feature/init-e", "是", "已合入", "#3")],
+             [pr(3, "feature/init-e", "feature/init", "b" * 40, state="open")],
+             "child feature/init-e PR #3 is not merged into feature/init"),
+            ("merged into main instead",
+             [("feature/init-f", "是", "已合入", "#4")],
+             [pr(4, "feature/init-f", "main", "c" * 40, state="closed", merged=True)],
+             "child feature/init-f PR #4 is not merged into feature/init"),
+        ]
+        for label, registry, children, reason in scenarios:
+            with self.subTest(label):
+                head = self.set_card("feature/init", card_text("feature/init", "initiative",
+                                                               accepting="false", registry=registry))
+                self.gitea.prs = [pr(1, "feature/init", "main", head)] + children
+                result = self.local("--final")
+                if reason is None:
+                    self.assertEqual(result.returncode, 0, result.out)
+                    self.assertIn("[verdict] PR_CHECK_OK base=main", result.out)
+                else:
+                    self.assertEqual(result.returncode, 1, result.out)
+                    self.assertIn(reason, result.out)
+
+    # ---- ci 模式 ----
+
+    def ci(self, base=None, head=None, event="pull_request"):
+        env = {"GITHUB_EVENT_NAME": event}
+        if base is not None:
+            env["GITHUB_BASE_REF"] = base
+        if head is not None:
+            env["GITHUB_HEAD_REF"] = head
+        # ci 模式不得需要网络或 token：指向关闭端口并移除 token，证明零依赖。
+        return self.invoke("ci", env=env | {"GITEA_API_URL": closed_port_url()}, drop=("GITEA_TOKEN",))
+
+    def test_ci_mode_gates_pull_request_base_against_head_card(self):
+        self.build_child()
+        result = self.ci(base="feature/init", head="feature/no-card")
+        self.assertEqual(result.returncode, 1, result.out)  # 无卡 → standalone → 期望 main
+        self.assertIn("expected 'main' for standalone branch feature/no-card", result.out)
+        cases = [
+            ("skips non pull_request events", dict(event="push", base="main", head="feature/init-child"), 0,
+             "[pr-check] skip: event push is not pull_request"),
+            ("child into main is rejected", dict(base="main", head="feature/init-child"), 1,
+             "expected 'feature/init' for child branch feature/init-child"),
+            ("child into initiative passes", dict(base="feature/init", head="feature/init-child"), 0,
+             "[verdict] PR_CHECK_OK base=feature/init head=feature/init-child kind=child"),
+            ("card-less head defaults to main", dict(base="main", head="feature/no-card"), 0,
+             "[verdict] PR_CHECK_OK base=main head=feature/no-card kind=standalone"),
+            ("card-less head into other base is rejected", dict(base="feature/init", head="feature/no-card"), 1,
+             "expected 'main' for standalone branch feature/no-card"),
+            ("missing base ref is rejected", dict(base=None, head="feature/init-child"), 1,
+             "lacks GITHUB_BASE_REF or GITHUB_HEAD_REF"),
+        ]
+        for label, kwargs, code, text in cases:
+            with self.subTest(label):
+                result = self.ci(**kwargs)
+                self.assertEqual(result.returncode, code, result.out)
+                self.assertIn(text, result.out)
+
+
+if __name__ == "__main__":
+    unittest.main()

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())
 

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()