feat(ship): add stage audit trail and sortable ids

skills/ship/scripts/ship_lib.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 
 import re
 import sys
+from datetime import date, datetime, timezone
 from pathlib import Path
 from typing import Any
 
@@ -14,7 +15,10 @@ if str(_REPO_ROOT) not in sys.path:
 
 from pouch.yaml_io import safe_dump, safe_load  # noqa: E402
 
-SHIP_ID_RE = re.compile(r"^[a-z][a-z0-9]*(?:-[a-z0-9]+)*$")
+SHIP_ID_RE = re.compile(
+    r"^(?:[a-z][a-z0-9]*(?:-[a-z0-9]+)*|\d{8}-[a-z][a-z0-9]*(?:-[a-z0-9]+)*)$"
+)
+DATED_SHIP_ID_RE = re.compile(r"^\d{8}-[a-z][a-z0-9]*(?:-[a-z0-9]+)*$")
 DATE_ID_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$|^\d{8}$")
 KIND = frozenset({"short", "long"})
 SHIP_STATUS = frozenset(
@@ -47,6 +51,24 @@ DELIVERY_STATE = frozenset(
 SOURCE_KIND = frozenset({"user", "hunt", "ack", "pr", "feishu"})
 COMMIT_ROLE = frozenset({"baseline", "merged", "head"})
 
+# These directories are the human-readable audit trail for the orchestration
+# phases. The step list is intentionally kept here so create_ship and
+# next_step cannot drift apart.
+STAGE_DEFINITIONS = (
+    ("1-spec", "规格与审批", ("write_spec", "specifier_new", "specifier_review", "specifier_handoff", "materialize_tasks", "hunt")),
+    ("2-develop", "开发", ("implement", "implement_task",)),
+    ("3-test", "验证与回归", ("verify", "verify_task", "regress")),
+    ("4-review", "PR 与合并", ("open_pr", "check_pr", "fix_pr", "merge_pr")),
+    ("5-build", "构建", ("obtain",)),
+    ("6-publish", "发布", ("publish",)),
+    ("7-deploy", "部署", ("deploy",)),
+)
+STAGE_BY_STEP = {
+    step: (directory, title)
+    for directory, title, steps in STAGE_DEFINITIONS
+    for step in steps
+}
+
 
 def skill_dir() -> Path:
     return Path(__file__).resolve().parents[1]
@@ -72,3 +94,115 @@ def check_ship_id(ship_id: str) -> str | None:
             f"shipId 无效: {ship_id!r}（小写字母开头，仅小写字母、数字、连字符）"
         )
     return None
+
+
+def check_dated_ship_id(ship_id: str) -> str | None:
+    """Validate the convention for newly-created records.
+
+    check_ship_id remains backward compatible so existing ship records can be
+    inspected. New entry points should use this stricter check.
+    """
+    err = check_ship_id(ship_id)
+    if err:
+        return err
+    if not DATED_SHIP_ID_RE.fullmatch(ship_id):
+        return (
+            f"新 ship-id 必须是 YYYYMMDD-短语格式: {ship_id!r}"
+        )
+    return None
+
+
+def dated_ship_id(phrase: str, *, today: date | None = None) -> str:
+    """Build a sortable ship id from a short lowercase phrase."""
+    phrase = re.sub(r"[^a-z0-9]+", "-", phrase.strip().lower()).strip("-")
+    if not phrase:
+        raise ValueError("ship-id 短语不能为空")
+    value = f"{(today or date.today()):%Y%m%d}-{phrase}"
+    err = check_dated_ship_id(value)
+    if err:
+        raise ValueError(err)
+    return value
+
+
+def stage_for_step(step: str | None) -> tuple[str, str] | None:
+    return STAGE_BY_STEP.get(str(step)) if step else None
+
+
+def ensure_stage_layout(ship_dir: Path) -> None:
+    """Create the visible phase folders and their small status manifests."""
+    for order, (directory, title, steps) in enumerate(STAGE_DEFINITIONS, start=1):
+        stage_dir = ship_dir / directory
+        stage_dir.mkdir(parents=True, exist_ok=True)
+        manifest = stage_dir / "stage.yaml"
+        if not manifest.exists():
+            manifest.write_text(
+                safe_dump({
+                    "order": order,
+                    "stage": directory,
+                    "title": title,
+                    "status": "pending",
+                    "steps": list(steps),
+                    "startedAt": None,
+                    "finishedAt": None,
+                }),
+                encoding="utf-8",
+            )
+        activity = stage_dir / "activity.md"
+        if not activity.exists():
+            activity.write_text(
+                f"# {directory} {title}\n\n状态：pending\n\n"
+                "## 活动记录\n\n",
+                encoding="utf-8",
+            )
+
+
+def _stage_manifest(path: Path) -> dict[str, Any]:
+    try:
+        value = safe_load(path.read_text(encoding="utf-8"))
+    except Exception:  # noqa: BLE001
+        value = None
+    return value if isinstance(value, dict) else {}
+
+
+def record_stage_event(
+    ship_dir: Path,
+    *,
+    step: str | None,
+    status: str,
+    summary: str = "",
+    task_id: str | None = None,
+) -> None:
+    """Append one dispatch/result event to the visible stage audit trail."""
+    resolved = stage_for_step(step)
+    if resolved is None:
+        return
+    ensure_stage_layout(ship_dir)
+    directory, title = resolved
+    stage_dir = ship_dir / directory
+    manifest_path = stage_dir / "stage.yaml"
+    manifest = _stage_manifest(manifest_path)
+    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    event_key = f"{step}|{status}|{task_id or ''}|{summary.strip()}"
+    if manifest.get("lastEventKey") == event_key:
+        return
+    manifest["lastEventKey"] = event_key
+    if status in {"ready", "in_progress"}:
+        if manifest.get("status") == "pending":
+            manifest["startedAt"] = now
+        manifest["status"] = "in_progress"
+    elif status in {"pass", "fail", "blocked", "ask"}:
+        manifest["lastResult"] = status
+        manifest["lastStep"] = step
+        if status == "pass":
+            manifest["status"] = "completed"
+            manifest["finishedAt"] = now
+        elif status in {"fail", "blocked"}:
+            manifest["status"] = "blocked"
+    manifest_path.write_text(safe_dump(manifest), encoding="utf-8")
+    line = f"- `{now}` `{step}` status=`{status}`"
+    if task_id:
+        line += f" task=`{task_id}`"
+    if summary:
+        line += f"：{summary.strip()}"
+    with (stage_dir / "activity.md").open("a", encoding="utf-8") as handle:
+        handle.write(line + "\n")