chore: sync pouch skill workflows

tests/test_ship_record.py

@@ -1,15 +1,44 @@
 from __future__ import annotations
 
+import json
 import subprocess
 import sys
 import tempfile
 import unittest
 from pathlib import Path
 
+from pouch.yaml_io import safe_dump, safe_load
+
 REPO_ROOT = Path(__file__).resolve().parents[1]
 CREATE = REPO_ROOT / "skills" / "ship" / "scripts" / "create_ship.py"
 VALIDATE = REPO_ROOT / "skills" / "ship" / "scripts" / "validate_ship.py"
 APPEND = REPO_ROOT / "skills" / "ship" / "scripts" / "append_evidence.py"
+NEXT = REPO_ROOT / "skills" / "ship" / "scripts" / "next_step.py"
+
+FILLED_SPEC = """# 导出按钮
+
+## 要做什么
+
+列表页给登录用户一个导出按钮。
+
+## 不做什么
+
+不改权限模型。
+
+## 可观察验收
+
+- 登录后列表页可见导出按钮
+
+## 怎么验
+
+- 浏览器打开 /list 看到按钮
+"""
+
+FAULT_SPEC = FILLED_SPEC + """
+## 故障证据
+
+线上列表页没有导出按钮。复现：登录后打开 /list。
+"""
 
 
 def _run(script: Path, *args: str, cwd: Path | None = None) -> subprocess.CompletedProcess[str]:
@@ -155,5 +184,173 @@ class CreateAndValidateTests(unittest.TestCase):
             self.assertIn("不存在", appended.stderr)
 
 
+def _create(root: Path, ship_id: str, kind: str, *, stop_at: str | None = None) -> Path:
+    args = [
+        "--project-root",
+        str(root),
+        "--ship-id",
+        ship_id,
+        "--kind",
+        kind,
+        "--title",
+        "列表页增加导出按钮",
+    ]
+    if stop_at:
+        args.extend(["--stop-at", stop_at])
+    result = _run(CREATE, *args)
+    if result.returncode != 0:
+        raise AssertionError(result.stderr)
+    return root / ".pouch" / "ship" / ship_id
+
+
+def _next(root: Path, ship_id: str, entry: str) -> dict:
+    result = _run(
+        NEXT,
+        "--project-root",
+        str(root),
+        "--ship-id",
+        ship_id,
+        "--entry",
+        entry,
+    )
+    if result.returncode != 0:
+        raise AssertionError(result.stderr or result.stdout)
+    return json.loads(result.stdout)
+
+
+def _patch_current(ship_dir: Path, **fields) -> None:
+    path = ship_dir / "current.yaml"
+    data = safe_load(path.read_text(encoding="utf-8"))
+    for key, value in fields.items():
+        if key == "delivery_state":
+            data.setdefault("delivery", {})["state"] = value
+        elif key == "manager":
+            data["manager"] = value
+        elif key == "source":
+            data["source"] = value
+        elif key == "spec":
+            data["spec"].update(value)
+        else:
+            data[key] = value
+    path.write_text(safe_dump(data), encoding="utf-8")
+
+
+def _fill_task(ship_dir: Path) -> None:
+    path = ship_dir / "tasks.yaml"
+    data = safe_load(path.read_text(encoding="utf-8"))
+    data["tasks"][0]["expected"] = ["登录用户看到导出"]
+    data["tasks"][0]["verification"] = ["打开 /list"]
+    path.write_text(safe_dump(data), encoding="utf-8")
+
+
+class NextStepTests(unittest.TestCase):
+    def test_new_short_ship_writes_spec(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            _create(root, "add-export-button", "short")
+            doc = _next(root, "add-export-button", "short")
+            self.assertEqual(doc["step"], "write_spec")
+            self.assertEqual(doc["actor"], "manager")
+
+    def test_filled_short_without_stop_asks_user(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            ship_dir = _create(root, "add-export-button", "short")
+            (ship_dir / "spec.md").write_text(FILLED_SPEC, encoding="utf-8")
+            _fill_task(ship_dir)
+            doc = _next(root, "add-export-button", "short")
+            self.assertEqual(doc["status"], "ask_user")
+
+    def test_short_implement_then_verify(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            ship_dir = _create(root, "add-export-button", "short", stop_at="verified")
+            (ship_dir / "spec.md").write_text(FILLED_SPEC, encoding="utf-8")
+            _fill_task(ship_dir)
+            doc = _next(root, "add-export-button", "short")
+            self.assertEqual(doc["step"], "implement")
+            self.assertEqual(doc["actor"], "worker")
+            self.assertIsNone(doc["skill"])
+
+            _patch_current(
+                ship_dir,
+                manager={"lastStep": "implement", "lastResult": "pass", "inFlight": None},
+            )
+            data = safe_load((ship_dir / "current.yaml").read_text(encoding="utf-8"))
+            data["evidence"]["commits"] = [{"sha": "abc1234deadbeef", "role": "head"}]
+            (ship_dir / "current.yaml").write_text(safe_dump(data), encoding="utf-8")
+            doc = _next(root, "add-export-button", "short")
+            self.assertEqual(doc["step"], "verify")
+            self.assertEqual(doc["skill"], "tester")
+
+            _patch_current(ship_dir, delivery_state="verified")
+            doc = _next(root, "add-export-button", "short")
+            self.assertEqual(doc["status"], "done")
+
+    def test_fault_spec_dispatches_hunt(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            ship_dir = _create(root, "add-export-button", "short", stop_at="verified")
+            (ship_dir / "spec.md").write_text(FAULT_SPEC, encoding="utf-8")
+            _fill_task(ship_dir)
+            doc = _next(root, "add-export-button", "short")
+            self.assertEqual(doc["step"], "hunt")
+            self.assertEqual(doc["skill"], "hunt")
+
+    def test_long_starts_with_specifier_worker(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            _create(root, "tenant-permissions", "long")
+            doc = _next(root, "tenant-permissions", "long")
+            self.assertEqual(doc["step"], "specifier_new")
+            self.assertEqual(doc["skill"], "specifier")
+            self.assertEqual(doc["actor"], "worker")
+
+    def test_pr_check_then_merge(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            ship_dir = _create(root, "pr-export", "short", stop_at="released")
+            _patch_current(
+                ship_dir,
+                source={"kind": "pr", "ref": "https://git.example.com/app/-/merge_requests/12"},
+            )
+            data = safe_load((ship_dir / "current.yaml").read_text(encoding="utf-8"))
+            data["evidence"]["prs"] = [
+                {"url": "https://git.example.com/app/-/merge_requests/12", "status": "opened"}
+            ]
+            (ship_dir / "current.yaml").write_text(safe_dump(data), encoding="utf-8")
+            doc = _next(root, "pr-export", "pr")
+            self.assertEqual(doc["step"], "check_pr")
+            self.assertEqual(doc["skill"], "check")
+
+            _patch_current(
+                ship_dir,
+                manager={"lastStep": "check_pr", "lastResult": "pass", "inFlight": None},
+            )
+            doc = _next(root, "pr-export", "pr")
+            self.assertEqual(doc["step"], "merge_pr")
+            self.assertEqual(doc["skill"], "delivery-loop")
+
+    def test_failed_worker_blocks_manager(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            ship_dir = _create(root, "add-export-button", "short", stop_at="verified")
+            (ship_dir / "spec.md").write_text(FILLED_SPEC, encoding="utf-8")
+            _fill_task(ship_dir)
+            _patch_current(
+                ship_dir,
+                manager={"lastStep": "implement", "lastResult": "fail", "inFlight": None},
+            )
+            doc = _next(root, "add-export-button", "short")
+            self.assertEqual(doc["status"], "blocked")
+            self.assertIn("不要自己做", doc["summary"])
+
+    def test_entry_skills_are_managers(self) -> None:
+        for name in ("short-ship", "long-ship", "pr-ship"):
+            text = (REPO_ROOT / "skills" / name / "SKILL.md").read_text(encoding="utf-8")
+            self.assertIn("开新 agent", text)
+            self.assertNotIn("读取并执行", text)
+
+
 if __name__ == "__main__":
     unittest.main()