diff --git a/policy/governance.md b/policy/governance.md index 41904cbe..faf4b235 100644 --- a/policy/governance.md +++ b/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； diff --git a/scripts/evidence.py b/scripts/evidence.py index a40a5e84..afbbd4ca 100644 --- a/scripts/evidence.py +++ b/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) diff --git a/tests/test_evidence.py b/tests/test_evidence.py index 01fc018b..08810a5f 100644 --- a/tests/test_evidence.py +++ b/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()