feat: rename bugfix-ship/requirement-flow to short-ship/long-ship with shared ship records

short-ship and long-ship share one ship-id record under .pouch/ship/<id>/.
specifier writes PRDs into spec/; delivery-loop appends evidence pointers.
Old names stay as deprecation stubs so existing installs keep working.
This commit is contained in:
2026-10-13 23:11:58 +08:00
parent 93341b3c69
commit af6e139d30
47 changed files with 1497 additions and 222 deletions
+106
View File
@@ -0,0 +1,106 @@
#!/usr/bin/env python3
"""Append one evidence pointer to current.yaml and re-validate."""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
from typing import Any
from ship_lib import dump_yaml, load_yaml
from validate_ship import validate_ship_dir
KINDS = ("testRun", "artifact", "deploy", "pr", "commit")
KEY = {
"testRun": "testRuns",
"artifact": "artifacts",
"deploy": "deploys",
"pr": "prs",
"commit": "commits",
}
def _entry(kind: str, args: argparse.Namespace) -> dict[str, Any]:
if kind == "testRun":
if not args.path or not args.result:
raise SystemExit("testRun 需要 --path 和 --result")
row: dict[str, Any] = {"path": args.path, "result": args.result}
if args.suite:
row["suite"] = args.suite
return row
if kind == "artifact":
if not args.sha:
raise SystemExit("artifact 需要 --sha")
row = {"sha": args.sha}
if args.digest:
row["digest"] = args.digest
if args.ref:
row["ref"] = args.ref
if args.venue:
row["venue"] = args.venue
return row
if kind == "deploy":
if not args.env:
raise SystemExit("deploy 需要 --env")
row = {"env": args.env}
if args.sha:
row["revision"] = args.sha
if args.health:
row["health"] = args.health
return row
if kind == "pr":
if not args.url:
raise SystemExit("pr 需要 --url")
row = {"url": args.url, "status": args.pr_status or "opened"}
return row
if kind == "commit":
if not args.sha or not args.role:
raise SystemExit("commit 需要 --sha 和 --role")
return {"sha": args.sha, "role": args.role}
raise SystemExit(f"未知 kind: {kind}")
def main() -> int:
parser = argparse.ArgumentParser(description="向 ship current.yaml 追加证据指针")
parser.add_argument("ship_dir")
parser.add_argument("--kind", required=True, choices=KINDS)
parser.add_argument("--path")
parser.add_argument("--result")
parser.add_argument("--suite")
parser.add_argument("--sha")
parser.add_argument("--digest")
parser.add_argument("--ref")
parser.add_argument("--venue")
parser.add_argument("--env")
parser.add_argument("--health")
parser.add_argument("--url")
parser.add_argument("--pr-status")
parser.add_argument("--role")
args = parser.parse_args()
ship_dir = Path(args.ship_dir).resolve()
current_path = ship_dir / "current.yaml"
data = load_yaml(current_path)
if not isinstance(data, dict):
raise SystemExit("current.yaml 不是 mapping")
evidence = data.setdefault("evidence", {})
if not isinstance(evidence, dict):
raise SystemExit("evidence 不是 mapping")
key = KEY[args.kind]
bucket = evidence.setdefault(key, [])
if not isinstance(bucket, list):
raise SystemExit(f"evidence.{key} 不是列表")
bucket.append(_entry(args.kind, args))
dump_yaml(current_path, data)
issues = validate_ship_dir(ship_dir)
if issues:
for item in issues:
print(f"✗ {item}", file=sys.stderr)
return 1
print(f"✓ 已追加 evidence.{key}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+130
View File
@@ -0,0 +1,130 @@
#!/usr/bin/env python3
"""Create a .pouch/ship/<id>/ record from templates. Refuse to overwrite."""
from __future__ import annotations
import argparse
import shutil
import sys
from pathlib import Path
from ship_lib import check_ship_id, skill_dir
from validate_ship import validate_ship_dir
def _replace(text: str, mapping: dict[str, str]) -> str:
for key, value in mapping.items():
text = text.replace(f"<{key}>", value)
text = text.replace(f"{{{{ {key} }}}}", value)
return text
def create_ship(
project_root: Path,
ship_id: str,
kind: str,
title: str,
*,
stop_at: str | None = None,
source_kind: str = "user",
source_ref: str | None = None,
) -> Path:
err = check_ship_id(ship_id)
if err:
raise SystemExit(err)
if kind not in {"short", "long"}:
raise SystemExit(f"kind 无效: {kind}")
title = title.strip()
if not title:
raise SystemExit("title 不能为空")
dest = project_root / ".pouch" / "ship" / ship_id
if dest.exists():
raise SystemExit(f"拒绝覆盖已有路径: {dest}")
templates = skill_dir() / "templates"
dest.mkdir(parents=True)
mapping = {
"ship-id": ship_id,
"title": title,
"kind": kind,
"short title": title,
}
current = _replace(
(templates / "current.template.yaml").read_text(encoding="utf-8"),
mapping,
)
current = current.replace("kind: short", f"kind: {kind}", 1)
current = current.replace("status: draft", "status: draft", 1)
if stop_at:
current = current.replace("stopAt: null", f"stopAt: {stop_at}")
if source_kind != "user":
current = current.replace("kind: user", f"kind: {source_kind}", 1)
if source_ref:
current = current.replace("ref: null", f"ref: {source_ref}", 1)
if kind == "long":
current = current.replace("status: draft", "status: spec_review", 1)
(dest / "current.yaml").write_text(current, encoding="utf-8")
(dest / "spec.md").write_text(
_replace((templates / "spec.template.md").read_text(encoding="utf-8"), mapping),
encoding="utf-8",
)
(dest / "tasks.yaml").write_text(
_replace(
(templates / "tasks.template.yaml").read_text(encoding="utf-8"), mapping
),
encoding="utf-8",
)
(dest / "status.md").write_text(
_replace(
(templates / "status.template.md").read_text(encoding="utf-8"),
{**mapping, "status": "spec_review" if kind == "long" else "draft",
"stopAt": stop_at or "null", "delivery.state": "change_ready"},
),
encoding="utf-8",
)
if kind == "long":
spec_dir = dest / "spec"
spec_dir.mkdir()
(spec_dir / "brief.md").write_text(f"# {title}\n\n", encoding="utf-8")
(spec_dir / "review.yaml").write_text("[]\n", encoding="utf-8")
(spec_dir / "decisions.yaml").write_text("[]\n", encoding="utf-8")
(spec_dir / "risks.yaml").write_text("[]\n", encoding="utf-8")
issues = validate_ship_dir(dest)
if issues:
shutil.rmtree(dest)
for item in issues:
print(f"✗ {item}", file=sys.stderr)
raise SystemExit("新建后校验失败,已删除不完整目录")
return dest
def main() -> int:
parser = argparse.ArgumentParser(description="新建 ship 记录")
parser.add_argument("--project-root", required=True)
parser.add_argument("--ship-id", required=True)
parser.add_argument("--kind", choices=("short", "long"), required=True)
parser.add_argument("--title", required=True)
parser.add_argument("--stop-at", default=None)
parser.add_argument("--source-kind", default="user")
parser.add_argument("--source-ref", default=None)
args = parser.parse_args()
dest = create_ship(
Path(args.project_root).resolve(),
args.ship_id,
args.kind,
args.title,
stop_at=args.stop_at,
source_kind=args.source_kind,
source_ref=args.source_ref,
)
print(dest)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+74
View File
@@ -0,0 +1,74 @@
#!/usr/bin/env python3
"""Shared ship-id and YAML helpers. Zero third-party deps."""
from __future__ import annotations
import re
import sys
from pathlib import Path
from typing import Any
_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
sys.path.insert(0, str(_REPO_ROOT))
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]+)*$")
DATE_ID_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$|^\d{8}$")
KIND = frozenset({"short", "long"})
SHIP_STATUS = frozenset(
{
"draft",
"spec_review",
"approved",
"in_progress",
"blocked",
"done",
"parked",
"superseded",
}
)
TASK_STATUS = frozenset(
{"open", "in_progress", "blocked", "verified", "merged", "done"}
)
DELIVERY_STATE = frozenset(
{
"change_ready",
"verified",
"pr_opened",
"merged",
"regression_passed",
"artifact_built",
"published",
"deployed",
}
)
SOURCE_KIND = frozenset({"user", "hunt", "ack", "pr", "feishu"})
COMMIT_ROLE = frozenset({"baseline", "merged", "head"})
def skill_dir() -> Path:
return Path(__file__).resolve().parents[1]
def load_yaml(path: Path) -> Any:
return safe_load(path.read_text(encoding="utf-8"))
def dump_yaml(path: Path, data: Any) -> None:
path.write_text(safe_dump(data), encoding="utf-8")
def check_ship_id(ship_id: str) -> str | None:
if not ship_id:
return "shipId 为空"
if len(ship_id) > 64:
return f"shipId 超过 64 字符: {ship_id}"
if DATE_ID_RE.fullmatch(ship_id):
return f"shipId 不能是日期: {ship_id}"
if not SHIP_ID_RE.fullmatch(ship_id):
return (
f"shipId 无效: {ship_id!r}(小写字母开头,仅小写字母、数字、连字符)"
)
return None
+279
View File
@@ -0,0 +1,279 @@
#!/usr/bin/env python3
"""Validate a .pouch/ship/<id>/ record. Exit 0 on success, 1 on issues."""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
from typing import Any
from ship_lib import (
COMMIT_ROLE,
DELIVERY_STATE,
KIND,
SHIP_STATUS,
SOURCE_KIND,
TASK_STATUS,
check_ship_id,
load_yaml,
)
def _issue(issues: list[str], msg: str) -> None:
issues.append(msg)
def _mapping(value: Any, path: str, issues: list[str]) -> dict[str, Any] | None:
if value is None:
_issue(issues, f"{path} 缺失")
return None
if not isinstance(value, dict):
_issue(issues, f"{path} 必须是 mapping")
return None
return value
def _project_root(ship_dir: Path) -> Path:
# .pouch/ship/<id>
return ship_dir.parent.parent.parent
def validate_ship_dir(ship_dir: Path) -> list[str]:
issues: list[str] = []
ship_dir = ship_dir.resolve()
current_path = ship_dir / "current.yaml"
if not current_path.is_file():
return [f"缺少 {current_path}"]
try:
current = load_yaml(current_path)
except Exception as exc: # noqa: BLE001
return [f"无法解析 current.yaml: {exc}"]
data = _mapping(current, "current.yaml", issues)
if data is None:
return issues
ship_id = data.get("shipId")
if not isinstance(ship_id, str):
_issue(issues, "current.yaml.shipId 必须是字符串")
else:
err = check_ship_id(ship_id)
if err:
_issue(issues, err)
if ship_id != ship_dir.name:
_issue(
issues,
f"目录名与 shipId 不一致: {ship_dir.name} != {ship_id}",
)
kind = data.get("kind")
if kind not in KIND:
_issue(issues, f"kind 无效: {kind!r}(short|long)")
status = data.get("status")
if status not in SHIP_STATUS:
_issue(issues, f"status 无效: {status!r}")
if not isinstance(data.get("title"), str) or not str(data.get("title")).strip():
_issue(issues, "title 不能为空")
source = data.get("source")
if source is not None:
src = _mapping(source, "source", issues)
if src is not None and src.get("kind") not in SOURCE_KIND:
_issue(issues, f"source.kind 无效: {src.get('kind')!r}")
spec = _mapping(data.get("spec"), "spec", issues)
spec_path = ship_dir / "spec.md"
if not spec_path.is_file():
_issue(issues, "缺少 spec.md")
if spec is not None:
rel = spec.get("path") or "spec.md"
if rel != "spec.md":
_issue(issues, f"spec.path 必须是 spec.md,实际: {rel}")
if kind == "long":
spec_dir = ship_dir / "spec"
if not spec_dir.is_dir():
_issue(issues, "kind=long 缺少 spec/ 目录")
revision = spec.get("revision")
prd = spec.get("prd")
if revision and not prd:
_issue(issues, "long spec.revision 已设但 spec.prd 为空")
if prd:
prd_file = ship_dir / str(prd)
if not prd_file.is_file():
_issue(issues, f"缺少 PRD 文件: {prd}")
handoff = spec.get("handoff")
spec_status = spec.get("specStatus")
if spec_status == "approved" and not handoff:
_issue(issues, "specStatus=approved 但缺少 spec.handoff")
if handoff:
handoff_file = ship_dir / str(handoff)
if not handoff_file.is_file():
_issue(issues, f"缺少 handoff 文件: {handoff}")
elif kind == "short":
for field in (
"revision",
"specStatus",
"prd",
"approvedRevision",
"reviews",
"handoff",
):
if spec.get(field) not in (None, "", []):
_issue(issues, f"kind=short 时 spec.{field} 必须为空")
tasks_meta = _mapping(data.get("tasks"), "tasks", issues)
tasks_rel = "tasks.yaml"
if tasks_meta is not None:
tasks_rel = str(tasks_meta.get("path") or "tasks.yaml")
tasks_path = ship_dir / tasks_rel
if not tasks_path.is_file():
_issue(issues, f"缺少 {tasks_rel}")
else:
issues.extend(_validate_tasks(tasks_path, ship_id, kind))
delivery = data.get("delivery")
if delivery is not None:
deliv = _mapping(delivery, "delivery", issues)
if deliv is not None:
state = deliv.get("state")
if state not in DELIVERY_STATE:
_issue(issues, f"delivery.state 无效: {state!r}")
evidence = _mapping(data.get("evidence"), "evidence", issues)
if evidence is not None:
issues.extend(_validate_evidence(evidence, _project_root(ship_dir)))
status_md = ship_dir / "status.md"
if not status_md.is_file():
_issue(issues, "缺少 status.md")
return issues
def _validate_tasks(path: Path, ship_id: Any, kind: Any) -> list[str]:
issues: list[str] = []
try:
data = load_yaml(path)
except Exception as exc: # noqa: BLE001
return [f"无法解析 tasks.yaml: {exc}"]
mapping = _mapping(data, "tasks.yaml", issues)
if mapping is None:
return issues
if mapping.get("shipId") != ship_id:
_issue(
issues,
f"tasks.yaml.shipId 与 current.yaml 不一致: "
f"{mapping.get('shipId')!r} != {ship_id!r}",
)
tasks = mapping.get("tasks")
if not isinstance(tasks, list) or not tasks:
_issue(issues, "tasks.yaml.tasks 必须是非空列表")
return issues
if kind == "short" and len(tasks) != 1:
_issue(issues, f"kind=short 必须恰好 1 条任务,实际 {len(tasks)}")
ids: list[str] = []
for i, item in enumerate(tasks):
prefix = f"tasks[{i}]"
row = _mapping(item, prefix, issues)
if row is None:
continue
tid = row.get("id")
if not isinstance(tid, str) or not tid.strip():
_issue(issues, f"{prefix}.id 不能为空")
else:
if tid in ids:
_issue(issues, f"重复任务 id: {tid}")
ids.append(tid)
if not isinstance(row.get("title"), str) or not str(row.get("title")).strip():
_issue(issues, f"{prefix}.title 不能为空")
depends = row.get("dependsOn") or []
if not isinstance(depends, list):
_issue(issues, f"{prefix}.dependsOn 必须是列表")
status = row.get("status")
if status not in TASK_STATUS:
_issue(issues, f"{prefix}.status 无效: {status!r}")
expected = row.get("expected")
if not isinstance(expected, list) or not expected:
_issue(issues, f"{prefix}.expected 必须是非空列表")
verification = row.get("verification")
if not isinstance(verification, list) or not verification:
_issue(issues, f"{prefix}.verification 必须是非空列表")
for i, item in enumerate(tasks):
if not isinstance(item, dict):
continue
for dep in item.get("dependsOn") or []:
if dep not in ids:
_issue(issues, f"tasks[{i}].dependsOn 引用未知 id: {dep}")
if dep == item.get("id"):
_issue(issues, f"tasks[{i}] 不能依赖自己")
return issues
def _validate_evidence(evidence: dict[str, Any], project_root: Path) -> list[str]:
issues: list[str] = []
for key in ("testRuns", "artifacts", "deploys", "prs", "commits"):
value = evidence.get(key)
if value is None:
_issue(issues, f"evidence.{key} 缺失")
continue
if not isinstance(value, list):
_issue(issues, f"evidence.{key} 必须是列表")
continue
for i, item in enumerate(value):
prefix = f"evidence.{key}[{i}]"
row = _mapping(item, prefix, issues)
if row is None:
continue
if key == "testRuns":
rel = row.get("path")
if not isinstance(rel, str) or not rel.strip():
_issue(issues, f"{prefix}.path 不能为空")
else:
target = project_root / rel
if not target.is_file():
_issue(issues, f"{prefix}.path 不存在: {rel}")
if row.get("result") not in {
"pass",
"fail",
"environment-fail",
"skipped",
"not-run",
"partial",
}:
_issue(issues, f"{prefix}.result 无效: {row.get('result')!r}")
elif key == "artifacts":
if not row.get("sha"):
_issue(issues, f"{prefix}.sha 不能为空")
elif key == "deploys":
if not row.get("env"):
_issue(issues, f"{prefix}.env 不能为空")
elif key == "prs":
if not row.get("url"):
_issue(issues, f"{prefix}.url 不能为空")
elif key == "commits":
if not row.get("sha"):
_issue(issues, f"{prefix}.sha 不能为空")
if row.get("role") not in COMMIT_ROLE:
_issue(issues, f"{prefix}.role 无效: {row.get('role')!r}")
return issues
def main() -> int:
parser = argparse.ArgumentParser(description="校验 ship 记录目录")
parser.add_argument("path", help=".pouch/ship/<id> 或其中的 current.yaml")
args = parser.parse_args()
path = Path(args.path)
ship_dir = path.parent if path.name == "current.yaml" else path
issues = validate_ship_dir(ship_dir)
if issues:
for item in issues:
print(f"✗ {item}", file=sys.stderr)
return 1
print(f"✓ ship 记录通过: {ship_dir}")
return 0
if __name__ == "__main__":
raise SystemExit(main())