agent-exec transport 9c7c1417e62f511b5f90669f58d69da9/682c27506bb4

policy/governance.md

@@ -54,6 +54,11 @@ TASK_WORK_MODE=local make task-resume PRECHECK_RECEIPT=/tmp/my-local-resume.json
 
 LOCAL_READY 只用于局部恢复；新任务入口始终校验严格 READY，正式验收与合入另行同步。
 恢复核验实际 worktree/卡片/分支、R0 祖先关系及源码外原始收据；脏现场、Git 操作、组件漂移和未知远端仍拒绝。
+`resume-local` 只接受 LOCAL_READY 收据，且须同时满足：卡片位于本 worktree 的 `.task/` 且状态未以“已”开头
+（已完成/已放弃等闭合任务拒绝）；收据内 `[task-state]` 的登记路径与 R0 等于当前登记任务；收据前后与当前 HEAD
+一致、工作区干净且组件状态未变；收据 `root_target/immediate_target` 等于卡片范围；目标只允许快进落后，回退或分叉拒绝。
+`resume-delivery` 仍要求 READY 且收据目标等于当前权威目标。LOCAL_READY 始终是 SELF 本地工作记录，
+不能用于建任务、正式交付、Review、合入、CI、GATE 或 REPLAY。
 旧收据允许作为历史证据原样导入，不自动成为当前恢复或交付准入；格式继续为 schema 1。
 
 堆叠（standalone）任务使用 `PARENT=feature/parent PARENT_PR=<父PR链接>`；子任务改用 `KIND=child`。父分支须包含本批主干 R0；

scripts/evidence.py

@@ -27,6 +27,7 @@ BRANCH = re.compile(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,199}\Z")
 # 可选的分支目标字段：两者须同时出现或同时缺省（旧收据没有）。
 TARGET_FIELDS = ("root_target", "immediate_target")
 PRE_MERGE_CLOSED = "已完成（合入前闭合）"
+TASK_STATE = re.compile(r"^\[task-state\] registered=(.+) initial_base=([0-9a-f]+)$", re.M)
 BASE_FIELDS = {"schema", "source", "timestamp", "command", "command_sha256", "cwd", "common_dir",
                "before", "after", "after_error", "exit_code", "output", "output_sha256", "receipt_sha256"}
 
@@ -314,6 +315,25 @@ def validate_precheck(receipt, purpose="start"):
                 git("merge-base", "--is-ancestor", after["target_head"], current["target_head"])
             except EvidenceError:
                 raise EvidenceError("Precheck target history has rewound or diverged") from None
+            validate_local_scope(receipt, current, context)
+
+
+def validate_local_scope(receipt, current, context):
+    """LOCAL_READY continues one active registered task; it remains SELF local-work evidence only."""
+    card = find_card()
+    status = re.search(r"；状态：\s*([^；\s][^；\n]*)", card.read_text(encoding="utf-8"))
+    if card.resolve().parent != context["root"] / ".task" or not status or status.group(1).startswith("已"):
+        raise EvidenceError("Local continuation requires an active registered task")
+    states = TASK_STATE.findall(receipt["output"])
+    if len(states) != 1 or Path(states[0][0]) != context["root"] or states[0][1] != context["r0"]:
+        raise EvidenceError("Precheck belongs to a different task identity or initial baseline")
+    for obj in (receipt["before"], receipt["after"]):
+        if obj["status"] or obj["untracked"] or obj["tracked_diff_sha256"] != digest(b""):
+            raise EvidenceError("Precheck workspace was not clean")
+        if obj["submodules"] != current["submodules"]:
+            raise EvidenceError("Precheck component state has changed")
+    if tuple(receipt.get(key) for key in TARGET_FIELDS) != task_targets():
+        raise EvidenceError("Precheck authorization scope differs from the registered task")
 
 
 def find_card():
@@ -457,8 +477,10 @@ def main():
         if imported or validated:
             receipt = read_receipt(imported or validated)
             if validated:
-                validate_precheck(receipt, os.environ.get("VALIDATE_RECEIPT_PURPOSE", "start"))
-                report(receipt, "precheck validated")
+                purpose = os.environ.get("VALIDATE_RECEIPT_PURPOSE", "start")
+                validate_precheck(receipt, purpose)
+                report(receipt, "local continuation validated (local work only; not delivery evidence)"
+                       if purpose == "resume-local" else "precheck validated")
             else:
                 card = find_card()
                 append_card(card, receipt)

tests/test_evidence.py

@@ -477,6 +477,119 @@ class EvidenceTests(unittest.TestCase):
         self.assertNotEqual(self.import_receipt().returncode, 0)
         self.assertEqual(self.card.read_text(), "# Test task\n")
 
+    def local_task(self, status="进行中"):
+        """Register a clean linked task worktree whose precheck prints the real local task context."""
+        wt = self.base.resolve() / "task-wt"
+        self.git("worktree", "add", "-qb", "feature/local", str(wt), self.head)
+        (wt / ".task").mkdir()
+        (wt / ".task/feature-local.md").write_text(
+            f"# Local task\n任务主记录：fixture；状态：{status}；owner：Evidence Test\n"
+            f"- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= {self.head}\n"
+            f"- 工作区：{wt}（worktree）；分支：feature/local\n")
+        (wt / "Makefile").write_text(
+            f"VERDICT ?= LOCAL_READY\nprecheck:\n\t@python3 {SCRIPTS / 'evidence.py'} --local-context\n"
+            "\t@echo '[verdict] $(VERDICT) — fixture'\n")
+        self.git("add", ".task", "Makefile", cwd=wt)
+        self.git("commit", "-qm", "Register local task", cwd=wt)
+        return wt
+
+    def validate(self, wt, purpose="resume-local", **values):
+        return self.run_evidence(cwd=wt, VALIDATE_RECEIPT=self.receipt,
+                                 VALIDATE_RECEIPT_PURPOSE=purpose, **values)
+
+    def advance_target(self):
+        head = self.git("commit-tree", "-p", self.head, "-m", "Advance target", self.head + "^{tree}")
+        self.git("update-ref", "refs/remotes/origin/main", head)
+
+    def test_local_continuation_accepts_registered_task_with_known_lag(self):
+        wt = self.local_task()
+        self.capture("make precheck", cwd=wt)
+        result = self.validate(wt)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("local continuation validated", result.stdout)
+        self.assertIn("src=SELF", result.stdout)
+        self.advance_target()
+        lagging = self.validate(wt)
+        self.assertEqual(lagging.returncode, 0, lagging.stderr)
+
+    def test_local_ready_never_authorizes_start_delivery_or_promoted_source(self):
+        wt = self.local_task()
+        self.capture("make precheck", cwd=wt)
+        for purpose in ("start", "resume-delivery"):
+            with self.subTest(purpose=purpose):
+                result = self.validate(wt, purpose)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn("Precheck receipt is not READY", result.stderr)
+        for source in ("GATE", "REPLAY"):
+            with self.subTest(source=source):
+                result = self.validate(wt, SRC=source)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn("Local evidence is SELF", result.stderr)
+
+    def test_delivery_resume_still_requires_ready_and_current_target(self):
+        wt = self.local_task()
+        self.capture("make precheck", cwd=wt, VERDICT="READY")
+        local = self.validate(wt)
+        self.assertNotEqual(local.returncode, 0)
+        self.assertIn("Precheck receipt is not LOCAL_READY", local.stderr)
+        current = self.validate(wt, "resume-delivery")
+        self.assertEqual(current.returncode, 0, current.stderr)
+        self.assertIn("precheck validated", current.stdout)
+        self.advance_target()
+        stale = self.validate(wt, "resume-delivery")
+        self.assertNotEqual(stale.returncode, 0)
+        self.assertIn("authoritative target has changed", stale.stderr)
+
+    def test_local_continuation_rejects_closed_task(self):
+        wt = self.local_task(status="已完成（合入前闭合）")
+        self.capture("make precheck", cwd=wt)
+        result = self.validate(wt)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("active registered task", result.stderr)
+
+    def test_local_continuation_rejects_dirty_workspace_or_changed_head(self):
+        wt = self.local_task()
+        self.capture("make precheck", cwd=wt)
+        (wt / "extra").write_text("unsaved\n")
+        dirty = self.validate(wt)
+        self.assertNotEqual(dirty.returncode, 0)
+        self.assertIn("Task workspace is not ready", dirty.stderr)
+        (wt / "extra").unlink()
+        self.git("commit", "--allow-empty", "-qm", "Advance task", cwd=wt)
+        moved = self.validate(wt)
+        self.assertNotEqual(moved.returncode, 0)
+        self.assertIn("source HEAD has changed", moved.stderr)
+
+    def test_local_continuation_rejects_mismatched_identity_baseline_workspace_or_scope(self):
+        wt = self.local_task()
+        original = self.capture("make precheck", cwd=wt)
+        identity = "different task identity or initial baseline"
+        cases = (
+            ("task identity", identity,
+             lambda d: d.update(output=d["output"].replace(str(wt), str(self.base / "other")))),
+            ("initial baseline", identity,
+             lambda d: d.update(output=d["output"].replace(self.head, "0" * 40))),
+            ("missing identity", identity,
+             lambda d: d.update(output="[verdict] LOCAL_READY — fixture\n")),
+            ("dirty receipt", "workspace was not clean",
+             lambda d: d["after"].update(status="?? extra\n", untracked={"extra": None})),
+            ("tracked diff", "workspace was not clean",
+             lambda d: d["before"].update(tracked_diff_sha256="1" * 64)),
+            ("target scope", "authorization scope",
+             lambda d: d.update(immediate_target="feature/other")),
+            ("missing scope", "authorization scope",
+             lambda d: [d.pop(key) for key in ("root_target", "immediate_target")]),
+        )
+        for name, message, change in cases:
+            with self.subTest(case=name):
+                data = json.loads(json.dumps(original))
+                change(data)
+                data["output_sha256"] = hashlib.sha256(data["output"].encode()).hexdigest()
+                self.resign(data)
+                result = self.validate(wt)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn(message, result.stderr)
+
 
 if __name__ == "__main__":
     unittest.main()