feat(ack): shard task records for merge-safe workspaces

tests/test_ack_task_store.py

@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = REPO_ROOT / "skills" / "ack" / "scripts"
+VALIDATOR = SCRIPTS / "validate_tasks.py"
+SELECTOR = SCRIPTS / "select_tasks.py"
+MIGRATOR = SCRIPTS / "migrate_tasks.py"
+RECORDER = SCRIPTS / "record_task.py"
+sys.path.insert(0, str(SCRIPTS))
+
+from task_store import materialize_board  # noqa: E402
+
+
+def manifest() -> dict:
+    return {
+        "version": 1,
+        "ackVersion": "0.20.0",
+        "taskStore": {"version": 1, "directory": "task-records"},
+        "project": {"name": "demo", "orchestration": {
+            "profileVersion": 1,
+            "mode": "manual",
+            "allowedWorktrees": [],
+            "modelAllowlist": {},
+            "profiles": {},
+            "defaults": {},
+        }},
+        "tasks": [],
+        "workerReceipts": [],
+    }
+
+
+def record(record_id: str, task_id: str, task: dict, parents: list[str] | None = None) -> dict:
+    return {
+        "recordVersion": 1,
+        "recordId": record_id,
+        "taskId": task_id,
+        "parents": parents or [],
+        "kind": "task.snapshot",
+        "task": task,
+    }
+
+
+class AckTaskStoreTests(unittest.TestCase):
+    def write_board(self, root: Path, records: list[dict]) -> Path:
+        board = root / "tasks.yaml"
+        board.write_text(json.dumps(manifest()), encoding="utf-8")
+        record_dir = root / "task-records"
+        record_dir.mkdir()
+        for item in records:
+            (record_dir / f"{item['recordId']}.yaml").write_text(
+                json.dumps(item), encoding="utf-8"
+            )
+        return board
+
+    def test_materializes_current_task_from_one_head(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            path = self.write_board(
+                root,
+                [record("EV-1", "T-1", {"id": "T-1", "title": "one", "status": "open"})],
+            )
+            from validate_tasks import load_document
+
+            board = load_document(path)
+            self.assertEqual([task["id"] for task in board["tasks"]], ["T-1"])
+            self.assertEqual(board["summary"]["open"], ["T-1"])
+            self.assertNotIn("taskStoreErrors", board)
+
+    def test_two_heads_are_reported_instead_of_last_write_wins(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            path = self.write_board(
+                root,
+                [
+                    record("EV-1", "T-1", {"id": "T-1", "title": "one", "status": "open"}),
+                    record("EV-2", "T-1", {"id": "T-1", "title": "two", "status": "verified"}, ["EV-1"]),
+                    record("EV-3", "T-1", {"id": "T-1", "title": "three", "status": "blocked"}, ["EV-1"]),
+                ],
+            )
+            result = subprocess.run(
+                [sys.executable, str(VALIDATOR), str(path)],
+                cwd=REPO_ROOT,
+                text=True,
+                capture_output=True,
+                check=False,
+            )
+            self.assertEqual(result.returncode, 1)
+            self.assertIn("未合并 head", result.stderr)
+
+    def test_selector_reads_records_without_changing_cli_shape(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            path = self.write_board(
+                root,
+                [record("EV-1", "T-1", {"id": "T-1", "title": "one", "status": "open"})],
+            )
+            result = subprocess.run(
+                [sys.executable, str(SELECTOR), str(path)],
+                cwd=REPO_ROOT,
+                text=True,
+                capture_output=True,
+                check=False,
+            )
+            self.assertEqual(result.returncode, 0, result.stderr)
+            payload = json.loads(result.stdout)
+            self.assertEqual(payload["selection"]["selectedTaskIds"], ["T-1"])
+
+    def test_migrator_dry_run_does_not_modify_legacy_board(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            path = Path(temp) / "tasks.yaml"
+            original = {
+                "version": 1,
+                "project": {"name": "demo"},
+                "tasks": [{"id": "T-1", "title": "one", "status": "open"}],
+                "workerReceipts": [],
+            }
+            path.write_text(json.dumps(original), encoding="utf-8")
+            result = subprocess.run(
+                [sys.executable, str(MIGRATOR), str(path)],
+                cwd=REPO_ROOT,
+                text=True,
+                capture_output=True,
+                check=False,
+            )
+            self.assertEqual(result.returncode, 0, result.stderr)
+            self.assertEqual(json.loads(path.read_text()), original)
+            self.assertFalse((path.parent / "task-records").exists())
+
+    def test_migrator_apply_keeps_backup_and_materializes_same_board(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            path = Path(temp) / "tasks.yaml"
+            path.write_text(json.dumps({
+                "version": 1,
+                "project": {"name": "demo"},
+                "tasks": [{"id": "T-1", "title": "one", "status": "open"}],
+                "workerReceipts": [],
+            }), encoding="utf-8")
+            result = subprocess.run(
+                [sys.executable, str(MIGRATOR), str(path), "--apply"],
+                cwd=REPO_ROOT,
+                text=True,
+                capture_output=True,
+                check=False,
+            )
+            self.assertEqual(result.returncode, 0, result.stderr)
+            self.assertTrue((path.parent / "tasks.legacy.yaml").is_file())
+            self.assertTrue((path.parent / "task-records" / "EV-T-1-initial.yaml").is_file())
+            from validate_tasks import load_document
+
+            board = load_document(path)
+            self.assertEqual(board["tasks"][0]["title"], "one")
+
+    def test_recorder_appends_exclusive_snapshot(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            path = self.write_board(root, [])
+            snapshot = root / "task.json"
+            snapshot.write_text(
+                json.dumps({"id": "T-1", "title": "one", "status": "open"}),
+                encoding="utf-8",
+            )
+            result = subprocess.run(
+                [
+                    sys.executable, str(RECORDER), str(path), "--task-id", "T-1",
+                    "--task-json", str(snapshot), "--record-id", "EV-1",
+                ],
+                cwd=REPO_ROOT,
+                text=True,
+                capture_output=True,
+                check=False,
+            )
+            self.assertEqual(result.returncode, 0, result.stderr)
+            self.assertTrue((root / "task-records" / "EV-1.yaml").is_file())
+            duplicate = subprocess.run(
+                [
+                    sys.executable, str(RECORDER), str(path), "--task-id", "T-1",
+                    "--task-json", str(snapshot), "--record-id", "EV-1",
+                ],
+                cwd=REPO_ROOT,
+                text=True,
+                capture_output=True,
+                check=False,
+            )
+            self.assertEqual(duplicate.returncode, 1)
+            self.assertIn("拒绝覆盖", duplicate.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()