536 lines
21 KiB
Python
536 lines
21 KiB
Python
#!/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"
|
||
SECTION_RE = re.compile(r"^([A-Za-z][A-Za-z0-9_-]*):$")
|
||
TOKEN_RE = re.compile(r"[A-Za-z0-9._~+/=-]+")
|
||
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, status=None):
|
||
super().__init__(message)
|
||
self.code = code
|
||
self.message = message
|
||
self.status = status
|
||
|
||
|
||
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 secrets_file_token():
|
||
"""HOME/IMC 分段时只取 HOME 段的 ACCESS_TOKEN（同 policy/facts.md F2）；无分段时取第一条。"""
|
||
try:
|
||
with open(os.path.expanduser(SECRETS_FILE), encoding="utf-8") as fh:
|
||
lines = fh.read().splitlines()
|
||
except (OSError, UnicodeDecodeError):
|
||
return None
|
||
first = home = None
|
||
sectioned = False
|
||
section = None
|
||
for raw in lines:
|
||
line = raw.strip()
|
||
header = SECTION_RE.match(line)
|
||
if header:
|
||
sectioned = True
|
||
section = header.group(1)
|
||
continue
|
||
if line.startswith("- "):
|
||
line = line[2:].strip()
|
||
if not line.startswith("ACCESS_TOKEN="):
|
||
continue
|
||
value = line.split("=", 1)[1].strip()
|
||
if first is None:
|
||
first = value
|
||
if section == "HOME" and home is None:
|
||
home = value
|
||
return (home if sectioned else first) or None
|
||
|
||
|
||
def read_token():
|
||
token = os.environ.get("GITEA_TOKEN", "").strip() or secrets_file_token()
|
||
if not token:
|
||
return None
|
||
if not TOKEN_RE.fullmatch(token):
|
||
raise Stop(EXIT_REMOTE, "Gitea token is malformed (expected one value without whitespace); "
|
||
"check GITEA_TOKEN or ~/.secrets/gitea")
|
||
return token
|
||
|
||
|
||
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):
|
||
try:
|
||
request = urllib.request.Request(
|
||
url, headers={"Accept": "application/json", "Authorization": f"token {token}"})
|
||
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}", status=exc.code) from None
|
||
except (urllib.error.URLError, http.client.HTTPException, OSError) as exc:
|
||
raise Stop(EXIT_REMOTE, f"Gitea API unreachable ({type(exc).__name__})") from None
|
||
except Stop:
|
||
raise
|
||
except Exception as exc:
|
||
# 异常文本可能含 Authorization 头（如 http.client 的 Invalid header value），只报类型名。
|
||
raise Stop(EXIT_REMOTE, f"Gitea API request failed ({type(exc).__name__})") from None
|
||
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())
|