"""Lifecycle regressions against real disposable Git repositories and remotes.""" import os import json import hashlib from pathlib import Path import shutil import shlex import subprocess import tempfile import unittest SOURCE = Path(__file__).resolve().parents[1] class TaskLifecycleTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix="task-lifecycle-") self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.repo = self.root / "repo" self.remote = self.root / "remote.git" 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": "Lifecycle Test", "GIT_AUTHOR_EMAIL": "test@example.invalid", "GIT_COMMITTER_NAME": "Lifecycle Test", "GIT_COMMITTER_EMAIL": "test@example.invalid"}) for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GITEA_TOKEN", "PRECHECK_RECEIPT", "TASK_PARENT_BRANCH", "TASK_PARENT_PR", "FORCE_ABANDON", "TASK_TARGET_BRANCH"): self.env.pop(key, None) self.git(self.root, "init", "--bare", "--initial-branch=main", str(self.remote)) # receive-pack 默认在后台跑 gc --auto，与 TemporaryDirectory 清理竞态 # （CI 实测 OSError 39 remote.git not empty）；夹具远端无留存价值，直接关。 self.git(self.remote, "config", "gc.auto", "0") self.git(self.remote, "config", "receive.autogc", "false") self.git(self.root, "init", "--initial-branch=main", str(self.repo)) shutil.copytree(SOURCE / "scripts", self.repo / "scripts") (self.repo / "Makefile").write_text("precheck:\n\t@bash scripts/task.sh precheck\n") (self.repo / "tracked").write_text("saved\n") (self.repo / ".gitignore").write_text("cache/\n") self.git(self.repo, "add", ".") 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") def run_cmd(self, cwd, *args, check=True, env=None): result = subprocess.run(args, cwd=cwd, env=env or self.env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=45) if check: self.assertEqual(result.returncode, 0, result.stdout) return result def git(self, cwd, *args, **kwargs): return self.run_cmd(cwd, "git", *args, **kwargs) def task(self, cwd, command, *args, **kwargs): if command == "finish" and "env" not in kwargs: kwargs["env"] = self.env | {"TASK_WRITERS_STOPPED": "1"} return self.run_cmd(cwd, "bash", str(self.repo / "scripts/task.sh"), command, *args, **kwargs) def worktree(self, name="feature/test"): wt = self.root / "worktree" self.git(self.repo, "worktree", "add", "-b", name, str(wt), "origin/main") return wt def advance_remote(self): (self.repo / "new-main").write_text("new main\n") self.git(self.repo, "add", "new-main") self.git(self.repo, "commit", "-m", "Advance main") self.git(self.repo, "push", "origin", "main") def add_submodule(self): child = self.root / "component" self.git(self.root, "init", "--initial-branch=main", str(child)) (child / "content").write_text("v1\n") self.git(child, "add", ".") self.git(child, "commit", "-m", "Component v1") self.git(self.repo, "submodule", "add", str(child), "component") self.git(self.repo, "commit", "-m", "Pin component") self.git(self.repo, "push", "origin", "main") return child def test_precheck_rejects_unreachable_authority(self): self.git(self.repo, "remote", "set-url", "origin", str(self.root / "missing.git")) result = self.task(self.repo, "precheck", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("REMOTE_UNVERIFIED", result.stdout) def test_precheck_feature_behind_requires_sync(self): wt = self.worktree() self.advance_remote() result = self.task(wt, "precheck", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("SYNC_REQUIRED", result.stdout) def test_precheck_routes_existing_project_delivery_policy(self): result = self.task(self.repo, "precheck") self.assertNotIn("policy/delivery.md", result.stdout) policy = self.repo / "policy/delivery.md" policy.parent.mkdir() policy.write_text("# Project delivery policy\n") self.git(self.repo, "add", "policy") self.git(self.repo, "commit", "-m", "Adopt project delivery policy") result = self.task(self.repo, "precheck") reading = next(line for line in result.stdout.splitlines() if line.startswith("[required-reading]")) self.assertIn("policy/delivery.md", reading) def test_sync_materializes_updated_component(self): child = self.add_submodule() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") (child / "content").write_text("v2\n") self.git(child, "commit", "-am", "Component v2") latest = self.git(child, "rev-parse", "HEAD").stdout.strip() self.git(self.repo / "component", "fetch", "origin") self.git(self.repo / "component", "checkout", "--detach", latest) self.git(self.repo, "commit", "-am", "Update component") self.git(self.repo, "push", "origin", "main") result = self.task(wt, "sync") self.assertIn("[synced]", result.stdout) self.assertEqual((wt / "component/content").read_text(), "v2\n") self.assertEqual(self.git(wt / "component", "rev-parse", "HEAD").stdout.strip(), latest) def test_sync_preserves_component_feature_branch(self): child = self.add_submodule() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") self.git(wt / "component", "switch", "-c", "feature/owned") before = self.git(wt / "component", "rev-parse", "HEAD").stdout (child / "content").write_text("v2\n") self.git(child, "commit", "-am", "Component v2") latest = self.git(child, "rev-parse", "HEAD").stdout.strip() self.git(self.repo / "component", "fetch", "origin") self.git(self.repo / "component", "checkout", "--detach", latest) self.git(self.repo, "commit", "-am", "Update component") self.git(self.repo, "push", "origin", "main") result = self.task(wt, "sync", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertNotIn("[synced]", result.stdout) self.assertEqual(self.git(wt / "component", "rev-parse", "HEAD").stdout, before) self.assertEqual(self.git(wt / "component", "branch", "--show-current").stdout.strip(), "feature/owned") def test_sync_keeps_unchanged_component_branch(self): self.add_submodule() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") self.git(wt / "component", "switch", "-c", "feature/owned") self.task(wt, "sync") self.assertEqual(self.git(wt / "component", "branch", "--show-current").stdout.strip(), "feature/owned") def test_sync_preserves_dirty_child(self): self.add_submodule() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") (wt / "component/content").write_text("unsaved\n") self.advance_remote() before = self.git(wt, "rev-parse", "HEAD").stdout result = self.task(wt, "sync", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout, before) self.assertEqual((wt / "component/content").read_text(), "unsaved\n") def test_finish_preserves_modified_untracked_and_ignored_files(self): for kind, rel in (("tracked", "tracked"), ("untracked", "unique"), ("ignored", "cache/unique")): with self.subTest(kind=kind): wt = self.root / f"worktree-{kind}" self.git(self.repo, "worktree", "add", "-b", f"feature/{kind}", str(wt), "origin/main") path = wt / rel path.parent.mkdir(exist_ok=True) path.write_text("unique user content\n") result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue(path.exists(), result.stdout) self.assertEqual(path.read_text(), "unique user content\n") def test_finish_rejects_subdirectory_without_deleting_it(self): wt = self.worktree() result = self.task(wt / "scripts", "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue((wt / "scripts/task.sh").is_file(), result.stdout) def test_finish_preserves_locked_worktree(self): wt = self.worktree() self.git(self.repo, "worktree", "lock", "--reason", "active owner", str(wt)) result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue(wt.exists(), result.stdout) def test_finish_preserves_component_untracked_file(self): self.add_submodule() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") unique = wt / "component/unique" unique.write_text("cannot regenerate\n") result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue(unique.is_file(), result.stdout) def test_finish_preserves_component_commits_off_the_pinned_head(self): self.add_submodule() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") child = wt / "component" pinned = self.git(child, "rev-parse", "HEAD").stdout.strip() self.git(child, "switch", "-c", "feature/unique-history") (child / "content").write_text("unique committed work\n") self.git(child, "commit", "-am", "Preserve this commit") unique = self.git(child, "rev-parse", "HEAD").stdout.strip() self.git(child, "checkout", "--detach", pinned) result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue(wt.exists(), result.stdout) self.assertEqual(self.git(child, "rev-parse", "feature/unique-history").stdout.strip(), unique) def test_finish_preserves_linked_component_worktrees_at_the_pinned_head(self): self.add_submodule() for state in ("clean", "staged", "unstaged"): with self.subTest(state=state): wt = self.root / f"worktree-{state}" self.git(self.repo, "worktree", "add", "-b", f"feature/{state}", str(wt), "origin/main") self.git(wt, "submodule", "update", "--init", "--recursive") child = wt / "component" sibling = self.root / f"component-{state}" self.git(child, "worktree", "add", "--detach", str(sibling), "HEAD") head = self.git(sibling, "rev-parse", "HEAD").stdout if state != "clean": (sibling / "content").write_text("unique sibling work\n") if state == "staged": self.git(sibling, "add", "content") status = self.git(sibling, "status", "--porcelain").stdout result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue(wt.exists(), result.stdout) self.assertEqual(self.git(child, "rev-parse", "HEAD").stdout, head) self.assertEqual(self.git(child, "status", "--porcelain").stdout, "") self.assertEqual(self.git(sibling, "rev-parse", "HEAD").stdout, head) self.assertEqual(self.git(sibling, "status", "--porcelain").stdout, status) self.assertEqual((sibling / "content").read_text(), "v1\n" if state == "clean" else "unique sibling work\n") def nested_component_update(self, attached): child = self.add_submodule() inner = self.root / "inner" self.git(self.root, "init", "--initial-branch=main", str(inner)) (inner / "inner-content").write_text("inner v1\n") self.git(inner, "add", ".") self.git(inner, "commit", "-m", "Inner v1") self.git(child, "submodule", "add", str(inner), "inner") self.git(child, "commit", "-m", "Pin inner") self.git(self.repo / "component", "fetch", "origin") self.git(self.repo / "component", "checkout", "--detach", "origin/main") self.git(self.repo, "commit", "-am", "Pin nested component") self.git(self.repo, "push", "origin", "main") wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") nested = wt / "component/inner" if attached: self.git(nested, "switch", "-c", "feature/owned-inner") before = self.git(nested, "rev-parse", "HEAD").stdout.strip() (inner / "inner-content").write_text("inner v2\n") self.git(inner, "commit", "-am", "Inner v2") latest = self.git(inner, "rev-parse", "HEAD").stdout.strip() self.git(child / "inner", "fetch", "origin") self.git(child / "inner", "checkout", "--detach", latest) self.git(child, "commit", "-am", "Update inner pin") self.git(self.repo / "component", "fetch", "origin") self.git(self.repo / "component", "checkout", "--detach", "origin/main") self.git(self.repo, "commit", "-am", "Update nested component") self.git(self.repo, "push", "origin", "main") return wt, nested, before, latest def test_sync_preserves_nested_feature_branch_when_parent_pin_changes(self): wt, nested, before, _ = self.nested_component_update(attached=True) result = self.task(wt, "sync", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertNotIn("[synced]", result.stdout) self.assertEqual(self.git(nested, "rev-parse", "HEAD").stdout.strip(), before) self.assertEqual(self.git(nested, "branch", "--show-current").stdout.strip(), "feature/owned-inner") def test_sync_materializes_nested_detached_components(self): wt, nested, _, latest = self.nested_component_update(attached=False) self.task(wt, "sync") self.assertEqual(self.git(nested, "rev-parse", "HEAD").stdout.strip(), latest) self.assertEqual((nested / "inner-content").read_text(), "inner v2\n") def test_finish_preserves_linked_nested_component_worktree(self): wt, nested, before, _ = self.nested_component_update(attached=False) sibling = self.root / "nested-sibling" self.git(nested, "worktree", "add", "--detach", str(sibling), "HEAD") result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue(wt.exists(), result.stdout) self.assertEqual(self.git(nested, "rev-parse", "HEAD").stdout.strip(), before) self.assertEqual(self.git(sibling, "rev-parse", "HEAD").stdout.strip(), before) self.assertEqual(self.git(sibling, "status", "--porcelain").stdout, "") self.assertEqual((nested / "inner-content").read_text(), "inner v1\n") self.assertEqual((sibling / "inner-content").read_text(), "inner v1\n") def test_finish_clean_worktree_has_accurate_remote_result(self): wt = self.worktree() # A missing remote branch is not a successful remote deletion. result = self.task(wt, "finish") self.assertFalse(wt.exists(), result.stdout) self.assertNotIn("本地+远端", result.stdout) def test_finish_recycles_clean_submodule_without_shared_config_changes(self): self.add_submodule() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") result = self.task(wt, "finish") self.assertFalse(wt.exists(), result.stdout) self.assertTrue((self.repo / "component/content").is_file(), result.stdout) self.assertEqual(self.git(self.repo, "config", "--get", "submodule.component.url").stdout.strip(), str(self.root / "component")) def test_finish_requires_writer_exit_confirmation(self): wt = self.worktree() result = self.task(wt, "finish", check=False, env=self.env) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue(wt.exists(), result.stdout) def unmerged_task(self): wt = self.worktree() (wt / "tracked").write_text("unmerged task work\n") self.git(wt, "commit", "-am", "Keep unmerged work") self.git(wt, "push", "-u", "origin", "HEAD") return wt, self.git(wt, "rev-parse", "HEAD").stdout.strip() def test_finish_abandon_retains_local_remote_tips_and_reason(self): wt, head = self.unmerged_task() reason = "superseded by a smaller task" result = self.task(wt, "finish", "--abandon", reason) self.assertFalse(wt.exists(), result.stdout) self.assertEqual(self.git(self.repo, "rev-parse", "refs/heads/feature/test").stdout.strip(), head) self.assertEqual(self.git(self.remote, "rev-parse", "refs/heads/feature/test").stdout.strip(), head) note = self.git(self.repo, "notes", "--ref=refs/notes/task-abandon", "show", head).stdout self.assertIn(reason, note) self.assertIn("feature/test", note) self.assertEqual(self.git(self.repo, "show", f"{head}:tracked").stdout, "unmerged task work\n") def test_finish_abandon_without_reason_preserves_checkout_and_refs(self): wt, head = self.unmerged_task() result = self.task(wt, "finish", "--abandon", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue(wt.exists(), result.stdout) self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout.strip(), head) self.assertEqual(self.git(self.repo, "rev-parse", "refs/heads/feature/test").stdout.strip(), head) self.assertEqual(self.git(self.remote, "rev-parse", "refs/heads/feature/test").stdout.strip(), head) self.assertEqual(self.git(wt, "status", "--porcelain").stdout, "") self.assertNotEqual(self.git(self.repo, "notes", "--ref=refs/notes/task-abandon", "show", head, check=False).returncode, 0) def test_finish_deletes_only_merged_remote_branch(self): wt = self.worktree() self.git(wt, "push", "-u", "origin", "HEAD") result = self.task(wt, "finish") self.assertIn("remote=DELETED", result.stdout) self.assertEqual(self.git(self.repo, "ls-remote", "--heads", "origin", "refs/heads/feature/test").stdout, "") def test_finish_preserves_remote_unmerged_tip(self): wt = self.worktree() self.git(self.repo, "switch", "-c", "feature/remote-update") (self.repo / "remote-work").write_text("another writer") self.git(self.repo, "add", "remote-work") self.git(self.repo, "commit", "-m", "Unmerged work") self.git(self.repo, "push", "origin", "HEAD:refs/heads/feature/test") self.git(self.repo, "switch", "main") result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue(wt.exists(), result.stdout) self.assertTrue(self.git(self.repo, "ls-remote", "--heads", "origin", "refs/heads/feature/test").stdout) def test_finish_remote_deletion_lease_preserves_concurrent_update(self): wt = self.worktree() old = self.git(wt, "rev-parse", "HEAD").stdout.strip() self.git(wt, "push", "-u", "origin", "HEAD") self.git(self.repo, "switch", "-c", "feature/racing") (self.repo / "race-work").write_text("concurrent remote work") self.git(self.repo, "add", "race-work") self.git(self.repo, "commit", "-m", "Concurrent remote work") latest = self.git(self.repo, "rev-parse", "HEAD").stdout.strip() self.git(self.repo, "push", "origin", "feature/racing") self.git(self.repo, "switch", "main") hook = self.repo / ".git/hooks/pre-push" hook.write_text("#!/bin/sh\n" + shlex.join(["git", "--git-dir", str(self.remote), "update-ref", "refs/heads/feature/test", latest, old]) + "\n") hook.chmod(0o755) result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("[partial]", result.stdout) current = self.git(self.repo, "ls-remote", "--heads", "origin", "refs/heads/feature/test").stdout self.assertTrue(current.startswith(latest), current) def test_sync_commit_failure_never_reports_synced(self): wt = self.worktree() (wt / ".task").mkdir() (wt / ".task/feature-test.md").write_text("# Task\n") self.git(wt, "add", ".task") self.git(wt, "commit", "-m", "Create task card") hook = self.repo / ".git/hooks/pre-commit" hook.write_text("#!/bin/sh\nexit 1\n") hook.chmod(0o755) result = self.task(wt, "sync", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertNotIn("[synced]", result.stdout) self.assertIn(".task/feature-test.md", self.git(wt, "status", "--porcelain").stdout) def task_card(self, wt): card = wt / ".task/feature-test.md" card.parent.mkdir(exist_ok=True) card.write_text("# Task\n") self.git(wt, "add", ".task") self.git(wt, "commit", "-m", "Create task card") return card def capture_evidence(self, wt, command="printf 'checked\\n'"): card = self.task_card(wt) self.run_cmd(wt, "bash", "scripts/evidence.sh", env=self.env | {"CMD": command}) return card def commit_evidence(self, wt): self.git(wt, "add", ".task") self.git(wt, "commit", "-m", "Record evidence") def test_sync_warns_when_json_evidence_predates_noncard_changes(self): wt = self.worktree() self.capture_evidence(wt) self.commit_evidence(wt) self.advance_remote() result = self.task(wt, "sync") self.assertIn("[evidence] WARN", result.stdout) self.assertIn("重跑", result.stdout) def test_sync_keeps_json_evidence_fresh_after_metadata_only_changes(self): wt = self.worktree() self.capture_evidence(wt) # A second run sees only the first run's uncommitted card receipt. self.run_cmd(wt, "bash", "scripts/evidence.sh", env=self.env | {"CMD": "true"}) self.commit_evidence(wt) (self.repo / ".task").mkdir() (self.repo / ".task/other.md").write_text("Other task metadata\n") self.git(self.repo, "add", ".task") self.git(self.repo, "commit", "-m", "Record other task") self.git(self.repo, "push", "origin", "main") result = self.task(wt, "sync") self.assertNotIn("[evidence] WARN", result.stdout) def test_sync_warns_when_json_before_state_contains_uncommitted_code(self): wt = self.worktree() (wt / "tracked").write_text("uncommitted code\n") self.capture_evidence(wt, command="git restore tracked") self.commit_evidence(wt) result = self.task(wt, "sync") self.assertIn("[evidence] WARN", result.stdout) self.assertIn("unverified", result.stdout) def test_sync_warns_when_json_after_state_contains_uncommitted_code(self): wt = self.worktree() self.capture_evidence(wt, command="printf 'uncommitted code\\n' > tracked") self.git(wt, "restore", "tracked") self.commit_evidence(wt) result = self.task(wt, "sync") self.assertIn("[evidence] WARN", result.stdout) self.assertIn("unverified", result.stdout) def test_sync_does_not_execute_json_receipt_command(self): wt = self.worktree() sentinel = self.root / "command-was-executed" self.capture_evidence(wt, command="printf ran > " + shlex.quote(str(sentinel))) self.assertTrue(sentinel.is_file()) sentinel.unlink() self.commit_evidence(wt) self.task(wt, "sync") self.assertFalse(sentinel.exists()) def rewrite_receipt(self, card, change): prefix, block = card.read_text().split("```json\n", 1) record, suffix = block.split("\n```", 1) receipt = json.loads(record) change(receipt) receipt.pop("receipt_sha256") serialized = json.dumps(receipt, sort_keys=True, ensure_ascii=False, separators=(",", ":")) receipt["receipt_sha256"] = hashlib.sha256(serialized.encode()).hexdigest() card.write_text(prefix + "```json\n" + json.dumps(receipt) + "\n```" + suffix) def test_sync_warns_for_unresolved_json_evidence_head(self): wt = self.worktree() card = self.capture_evidence(wt) self.rewrite_receipt(card, lambda receipt: receipt["before"].update(head="0" * 40)) self.commit_evidence(wt) result = self.task(wt, "sync") self.assertIn("[evidence] WARN", result.stdout) self.assertIn("unverified", result.stdout) def test_sync_warns_for_missing_json_receipt_fields_even_with_valid_checksum(self): wt = self.worktree() card = self.capture_evidence(wt) self.rewrite_receipt(card, lambda receipt: receipt.pop("command")) self.commit_evidence(wt) result = self.task(wt, "sync") self.assertIn("[evidence] WARN", result.stdout) self.assertIn("unverified", result.stdout) def test_sync_warns_for_malformed_json_evidence(self): wt = self.worktree() card = self.task_card(wt) card.write_text(card.read_text() + '\n```json\n{"schema": 1, invalid}\n```\n') self.commit_evidence(wt) result = self.task(wt, "sync") self.assertIn("[evidence] WARN", result.stdout) self.assertIn("unverified", result.stdout) def test_sync_warns_for_unresolved_legacy_evidence_head(self): wt = self.worktree() card = self.task_card(wt) card.write_text(card.read_text() + "\n| 2026-10-05T00:00:00Z | check | `0000000000` | 0 | SELF | abc |\n") self.commit_evidence(wt) result = self.task(wt, "sync") self.assertIn("[evidence] WARN", result.stdout) self.assertIn("unverified", result.stdout) def test_sync_still_warns_for_stale_legacy_table_evidence(self): wt = self.worktree() card = self.task_card(wt) head = self.git(wt, "rev-parse", "--short=10", "HEAD").stdout.strip() card.write_text(card.read_text() + f"\n| 2026-10-05T00:00:00Z | check | `{head}` | 0 | SELF | abc |\n") self.commit_evidence(wt) self.advance_remote() result = self.task(wt, "sync") self.assertIn("[evidence] WARN", result.stdout) def test_sync_reads_legacy_commands_with_escaped_pipes(self): wt = self.worktree() card = self.task_card(wt) head = self.git(wt, "rev-parse", "--short=10", "HEAD").stdout.strip() card.write_text(card.read_text() + f"\n| 2026-10-05T00:00:00Z | `false \\|\\| true` | `{head}` | 0 | SELF | abc |\n") self.commit_evidence(wt) self.advance_remote() result = self.task(wt, "sync") self.assertIn("[synced]", result.stdout) self.assertIn("[evidence] WARN", result.stdout) self.assertNotIn("unverified", result.stdout) def test_finish_preserves_main_and_active_index_lock(self): result = self.task(self.repo, "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue((self.repo / "tracked").is_file()) wt = self.worktree() index_lock = Path(self.git(wt, "rev-parse", "--path-format=absolute", "--git-path", "index.lock").stdout.strip()) index_lock.write_text("another writer") result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertTrue(wt.exists(), result.stdout) self.assertEqual(index_lock.read_text(), "another writer") def test_verify_rejects_recursive_drift_even_when_root_has_local_edits(self): child = self.add_submodule() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") (wt / "tracked").write_text("root implementation edit\n") self.run_cmd(wt, "python3", "scripts/check-worktree.py", "verify") (child / "content").write_text("v2\n") self.git(child, "commit", "-am", "Component v2") self.git(wt / "component", "fetch", "origin") self.git(wt / "component", "checkout", "--detach", "origin/main") result = self.run_cmd(wt, "python3", "scripts/check-worktree.py", "verify", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) def capture_precheck(self, cwd=None): receipt = self.root / "precheck.json" env = self.env | {"RECEIPT": str(receipt), "CMD": "make precheck"} self.run_cmd(cwd or self.repo, "bash", str(self.repo / "scripts/evidence.sh"), env=env) return receipt def test_start_requires_precheck_receipt_before_creating_worktree(self): result = self.task(self.repo, "start", "feature/new", check=False) 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() head = self.git(self.repo, "rev-parse", "HEAD").stdout.strip() for kind in ("local", "remote"): with self.subTest(kind=kind): name = "feature/merged-" + kind if kind == "local": self.git(self.repo, "branch", name, head) ref = "refs/heads/" + name else: self.git(self.repo, "push", "origin", head + ":refs/heads/" + name) 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)}) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("already merged", result.stdout) self.assertFalse((self.root / ("repo-wt-" + name.replace("/", "-"))).exists()) self.assertEqual(self.git(self.repo, "rev-parse", ref).stdout.strip(), head) 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)}) 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("交付模式：manual（前序人工决策，智能体 Review/合入）", card) self.assertIn("authorized-continuous", card) self.assertIn("流程依据：仓库根目录 policy/delivery.md", card) for field in ("授权范围", "授权来源", "授权终点", "停止条件"): self.assertIn(f"{field}：无", card) for field in ("当前阶段", "当前增量", "方案引用", "实施计划引用", "测试计划引用"): self.assertIn(field + "：", card) self.assertIn("## 关联方向与规划\n", card) planning = card.split("## 关联方向与规划\n", 1)[1].split("\n## ", 1)[0] for field in ("适用状态", "主记录路径/版本", "对应里程碑"): self.assertIn(field + "：待核对", planning) self.assertIn("无适用规划时说明原因", planning) self.assertIn("路径按仓库根目录定位，归档后不变", planning) self.assertIn("只引用唯一正文", planning) self.assertIn("| ID | 描述 | 项目符合度及依据 | 优先级 | 验收条件 | 依赖 | 结论 |", card) reviews = card.split("## 阶段评审\n", 1)[1].split("\n## ", 1)[0] self.assertIn("| 阶段 | 对象版本 | 结论 | 评审人 | 时间 | 依据 |", reviews) for stage in ("需求拆解", "需求评估", "需求表确认", "方案设计", "实施计划", "实施与自检", "正式测试与验收", "代码 Review", "合入", "合入核验与收口"): row = next(line for line in reviews.splitlines() if line.startswith(f"| {stage} |")) self.assertEqual(row.split("|")[3].strip(), "未评审") self.assertNotIn("通过", reviews) self.assertIn("## 探索试验与迭代", card) for field in ("假设", "范围", "判断和停止条件", "批准依据", "结果", "变更影响", "需重审补测"): self.assertIn(field, card.split("## 探索试验与迭代\n", 1)[1].split("\n## ", 1)[0]) self.assertIn("## 规划进展与下一步\n", card) next_step = card.split("## 规划进展与下一步\n", 1)[1].split("\n## ", 1)[0] for field in ("本次推进/剩余", "首选下一任务", "前置条件/待确认", "重复任务核对"): self.assertIn(field + "：待核对", next_step) self.assertIn("推荐依据：待填写", next_step) self.assertIn("预期产物/验收条件：待定义", next_step) self.assertIn("无适用规划、已完成或受阻时如实说明，不虚构任务", next_step) self.assertIn("已有/在途任务、PR 与规划记录", next_step) self.assertIn("启动授权：未授权", next_step) self.assertIn("建议不自动启动新任务，仍须明确授权", next_step) self.assertIn("## 合入与清理", card) for field in ("合入记录", "清理记录"): self.assertIn(field + "：NOT_RUN", card) body = self.git(wt, "log", "-1", "--format=%B").stdout for field in ("Why:", "Evidence:", "Impact:", "Refs:"): self.assertIn(field, body) def install_post_merge_hook(self): hook = self.repo / ".git/hooks/post-merge" shutil.copyfile(SOURCE / "scripts/hooks/post-merge", hook) hook.chmod(0o755) def publish_merged_task(self): publisher = self.root / "publisher" self.git(self.root, "clone", str(self.remote), str(publisher)) self.git(publisher, "switch", "-c", "feature/merged") card = publisher / ".task/feature-merged.md" card.parent.mkdir() card.write_text("# Merged task\n任务主记录：待回填；状态：进行中；owner：test\n" "- 工作区：fixture；分支：feature/merged\n") self.git(publisher, "add", ".task") self.git(publisher, "commit", "-m", "Record merged task") self.git(publisher, "switch", "main") self.git(publisher, "merge", "--no-ff", "feature/merged", "-m", "Merge pull request 'Task' (#7) from feature/merged into main") self.git(publisher, "push", "origin", "main") return self.git(publisher, "rev-parse", "HEAD").stdout.strip(), card.read_text() def assert_post_merge_only_reminds_owner(self, operation): expected_head, expected_card = self.publish_merged_task() self.git(self.repo, "fetch", "origin") old_head = self.git(self.repo, "rev-parse", "HEAD").stdout.strip() local_heads = self.git(self.repo, "show-ref", "--heads").stdout worktrees = self.git(self.repo, "worktree", "list", "--porcelain").stdout remote_refs = self.git(self.remote, "show-ref").stdout self.install_post_merge_hook() result = self.git(self.repo, *operation) self.assertIn("make task-close", result.stdout) self.assertIn("授权", result.stdout) self.assertNotIn("[close]", result.stdout) self.assertEqual(self.git(self.repo, "branch", "--show-current").stdout.strip(), "main") self.assertEqual(self.git(self.repo, "rev-parse", "HEAD").stdout.strip(), expected_head) self.assertEqual(self.git(self.repo, "show-ref", "--heads").stdout, local_heads.replace(old_head, expected_head)) self.assertEqual(self.git(self.repo, "worktree", "list", "--porcelain").stdout, worktrees.replace(old_head, expected_head)) self.assertEqual(self.git(self.remote, "show-ref").stdout, remote_refs) self.assertEqual(self.git(self.repo, "status", "--porcelain").stdout, "") self.assertEqual((self.repo / ".task/feature-merged.md").read_text(), expected_card) self.assertFalse((self.repo / ".task/archive").exists()) def test_post_merge_main_merge_only_reminds_owner(self): self.assert_post_merge_only_reminds_owner(("merge", "--ff-only", "origin/main")) def test_post_merge_main_pull_only_reminds_owner(self): self.assert_post_merge_only_reminds_owner(("pull", "--ff-only")) def test_post_merge_is_silent_outside_main_checkout(self): wt = self.worktree() self.install_post_merge_hook() self.advance_remote() result = self.git(wt, "merge", "--ff-only", "origin/main") self.assertNotIn("make task-close", result.stdout) self.assertNotIn("[close]", result.stdout) self.git(self.repo, "switch", "-c", "feature/primary") self.git(wt, "switch", "-c", "feature/advance") (wt / "feature-work").write_text("another task\n") self.git(wt, "add", "feature-work") self.git(wt, "commit", "-m", "Advance feature") result = self.git(self.repo, "merge", "--ff-only", "feature/advance") self.assertNotIn("make task-close", result.stdout) self.assertNotIn("[close]", result.stdout) def test_start_stacked_branch_records_parent_and_main_baseline(self): base = self.git(self.repo, "rev-parse", "origin/main").stdout.strip() parent = self.worktree("feature/parent") (parent / "parent-change").write_text("parent feature") self.git(parent, "add", "parent-change") 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", "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" card = (wt / ".task/feature-parent-child.md").read_text() self.assertIn(f"initial_base_sha）= {base}", card) self.assertIn(f"parent_head_sha={parent_head}", card) self.assertIn("https://example.invalid/repo/pulls/42", card) self.assertEqual((wt / "parent-change").read_text(), "parent feature") def test_start_offline_does_not_create_worktree(self): self.git(self.repo, "remote", "set-url", "origin", str(self.root / "missing.git")) result = self.task(self.repo, "start", "feature/offline", check=False) 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() self.advance_remote() result = self.task(self.repo, "start", "feature/stale", check=False, env=self.env | {"PRECHECK_RECEIPT": str(receipt)}) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertFalse((self.root / "repo-wt-feature-stale").exists()) def test_verify_rejects_uninitialized_component(self): self.add_submodule() wt = self.worktree() result = self.run_cmd(wt, "python3", "scripts/check-worktree.py", "verify", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) if __name__ == "__main__": unittest.main() class FinishGuardFixTests(unittest.TestCase): """2026-10-05 两处 finish 守卫误报的回归：IDE 产物锁不再误报、真实 Git 锁仍拦； 子仓远端跟踪引用不再卡 pinned-history，本地独有工作仍保全。""" def setUp(self): self.root = tempfile.mkdtemp(prefix="finish-guard-") self.addCleanup(shutil.rmtree, self.root, ignore_errors=True) self.repo = Path(self.root) / "repo" shutil.copytree(SOURCE / "scripts", self.repo / "scripts") (self.repo / "Makefile").write_text("precheck:\n\t@bash scripts/task.sh precheck\n") 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": "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"): self.env.pop(key, None) 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") self.git(self.repo, "add", ".") self.git(self.repo, "commit", "-q", "-m", "base") self.git(self.repo, "remote", "add", "origin", self.root + "/origin.git") self.git(self.repo, "push", "-q", "origin", "main") def git(self, cwd, *args): return subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True, text=True, env=self.env) def task(self, cwd, *args, check=True, env=None): return subprocess.run(["bash", str(self.repo / "scripts" / "task.sh"), *args], cwd=str(cwd), capture_output=True, text=True, env=env or self.env, check=check) def worktree(self): receipt = Path(self.root) / "precheck.json" subprocess.run( ["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) self.task(self.repo, "start", "feature/guard", env=self.env | {"PRECHECK_RECEIPT": str(receipt)}) wt = self.root + "/repo-wt-feature-guard" assert Path(wt).is_dir(), "worktree not created" return Path(wt) def add_submodule_pinned_off_main(self): """pin 在 feature 顶端（origin/main 不在 pin 历史内）——修复前的永久卡死场景。""" child = Path(self.root) / "component" self.git(self.root, "init", "-b", "main", str(child)) (child / "content").write_text("v1\n") self.git(child, "add", ".") self.git(child, "-c", "user.name=T", "-c", "user.email=t@t", "commit", "-q", "-m", "v1") self.git(self.root, "init", "--bare", "-b", "main", self.root + "/component-origin.git") self.git(child, "remote", "add", "origin", self.root + "/component-origin.git") self.git(child, "push", "-q", "origin", "main") self.git(child, "switch", "-c", "feature/pin-tip") (child / "content").write_text("v2 pin tip\n") self.git(child, "commit", "-am", "v2") pin_tip = subprocess.run(["git", "-C", str(child), "rev-parse", "HEAD"], capture_output=True, text=True, check=True, env=self.env).stdout.strip() # 把 pin-tip 推上远端，子仓 add 时按该分支检出 self.git(child, "push", "-q", "origin", "feature/pin-tip") self.git(self.repo, "-c", "protocol.file.allow=always", "submodule", "add", "-b", "feature/pin-tip", self.root + "/component-origin.git", "component") self.git(self.repo, "commit", "-q", "-am", "Pin component at feature tip") self.git(self.repo, "push", "-q", "origin", "main") # 让子仓本地带上 origin/main 的远端跟踪引用（不在 pin 历史内） self.git(child, "fetch", "origin") return child, pin_tip def test_ide_lock_in_git_dir_does_not_block_finish(self): wt = self.worktree() (self.repo / "tracked").write_text("implementation done\n") self.git(self.repo, "add", "-A") self.git(self.repo, "commit", "-q", "-m", "done") self.git(self.repo, "push", "-q", "origin", "feature/guard") self.git(self.repo, "merge", "--no-ff", "-q", "-m", "Merge feature/guard", "feature/guard") self.git(self.repo, "push", "-q", "origin", "main") ide = Path(subprocess.run(["git", "-C", str(self.repo), "rev-parse", "--path-format=absolute", "--git-common-dir"], capture_output=True, text=True, check=True).stdout.strip()) / "cursor" / "crepe" ide.mkdir(parents=True, exist_ok=True) stale = ide / "build.lock" stale.write_text("stale IDE artifact") result = self.task(wt, "finish", check=False) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertFalse(wt.exists()) self.assertTrue(stale.is_file(), "IDE artifact must be preserved untouched") def test_real_git_lock_still_blocks_finish(self): wt = self.worktree() (self.repo / "tracked").write_text("implementation done\n") self.git(self.repo, "add", "-A") self.git(self.repo, "commit", "-q", "-m", "done") self.git(self.repo, "push", "-q", "origin", "feature/guard") self.git(self.repo, "merge", "--no-ff", "-q", "-m", "Merge feature/guard", "feature/guard") self.git(self.repo, "push", "-q", "origin", "main") common = Path(subprocess.run(["git", "-C", str(self.repo), "rev-parse", "--path-format=absolute", "--git-common-dir"], capture_output=True, text=True, check=True).stdout.strip()) (common / "config.lock").write_text("live writer") result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0) self.assertTrue(wt.exists()) def test_remote_tracking_refs_do_not_block_finish(self): self.add_submodule_pinned_off_main() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") (self.repo / "tracked").write_text("implementation done\n") self.git(self.repo, "add", "-A") self.git(self.repo, "commit", "-q", "-m", "done") self.git(self.repo, "push", "-q", "origin", "feature/guard") self.git(self.repo, "merge", "--no-ff", "-q", "-m", "Merge feature/guard", "feature/guard") self.git(self.repo, "push", "-q", "origin", "main") # 模拟实际场景：子仓检出于 pin 顶端且带有远端跟踪引用与克隆 reflog result = self.task(wt, "finish", check=False) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertFalse(wt.exists()) def test_local_default_branch_at_origin_main_does_not_block(self): """克隆默认分支停在 origin/main 顶端（pin 为 feature 顶端时不在其历史内） 且 HEAD reflog 含克隆/检出轨迹——均为远端有副本的状态，不得阻断 finish。""" self.add_submodule_pinned_off_main() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") (self.repo / "tracked").write_text("implementation done\n") self.git(self.repo, "add", "-A") self.git(self.repo, "commit", "-q", "-m", "done") self.git(self.repo, "push", "-q", "origin", "feature/guard") self.git(self.repo, "merge", "--no-ff", "-q", "-m", "Merge feature/guard", "feature/guard") self.git(self.repo, "push", "-q", "origin", "main") result = self.task(wt, "finish", check=False) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertFalse(wt.exists()) def test_local_unique_commit_still_preserved(self): self.add_submodule_pinned_off_main() wt = self.worktree() self.git(wt, "submodule", "update", "--init", "--recursive") child = wt / "component" pinned = subprocess.run(["git", "-C", str(child), "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip() self.git(child, "switch", "-c", "feature/unique") (child / "content").write_text("unique local work\n") self.git(child, "commit", "-am", "unique") unique = subprocess.run(["git", "-C", str(child), "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip() self.git(child, "checkout", "--detach", pinned) (self.repo / "tracked").write_text("done\n") self.git(self.repo, "add", "-A") self.git(self.repo, "commit", "-q", "-m", "done") self.git(self.repo, "push", "-q", "origin", "feature/guard") self.git(self.repo, "merge", "--no-ff", "-q", "-m", "Merge feature/guard", "feature/guard") self.git(self.repo, "push", "-q", "origin", "main") result = self.task(wt, "finish", check=False) self.assertNotEqual(result.returncode, 0) self.assertTrue(wt.exists()) kept = subprocess.run(["git", "-C", str(child), "rev-parse", "feature/unique"], capture_output=True, text=True, check=True).stdout.strip() self.assertEqual(kept, unique)