#!/usr/bin/env python3 """Print the next manager/worker step for a ship as JSON.""" from __future__ import annotations import argparse import json import re import sys from pathlib import Path from typing import Any from ship_lib import load_yaml SPEC_PLACEHOLDERS = ( "<用户可见的目标", "<明确排除的范围", "<能被浏览器", "<命令、页面路径", ) TASK_PLACEHOLDERS = ( "", ) FAULT_PLACEHOLDER = "<没有故障时删除" DONE_TASK = frozenset({"verified", "merged", "done"}) READY_DEP = frozenset({"verified", "merged", "done"}) STOP_AT_STATE = { "verified": "verified", "review_ready": "pr_opened", "validation_ready": "deployed", "released": "deployed", } DELIVERY_STEPS: tuple[tuple[str, str, str], ...] = ( ("verified", "verify", "tester"), ("pr_opened", "open_pr", "delivery-loop"), ("merged", "merge_pr", "delivery-loop"), ("regression_passed", "regress", "tester"), ("artifact_built", "obtain", "builder"), ("published", "publish", "releaser"), ("deployed", "deploy", "deployer"), ) DELIVERY_ORDER = ("change_ready",) + tuple(s[0] for s in DELIVERY_STEPS) def _out( *, ship_id: str, entry: str, actor: str, step: str, status: str = "ready", skill: str | None = None, task_id: str | None = None, target_state: str | None = None, stop_at: str | None = None, ready_task_ids: list[str] | None = None, summary: str, ) -> dict[str, Any]: return { "shipId": ship_id, "entry": entry, "actor": actor, "step": step, "status": status, "skill": skill, "taskId": task_id, "targetState": target_state, "stopAt": stop_at, "readyTaskIds": ready_task_ids or [], "summary": summary, } def _text(path: Path) -> str: return path.read_text(encoding="utf-8") if path.is_file() else "" def _has_any(text: str, markers: tuple[str, ...]) -> bool: return any(m in text for m in markers) def _fault_section(spec_md: str) -> str: match = re.search(r"^## 故障证据\s*$", spec_md, re.MULTILINE) if not match: return "" rest = spec_md[match.end() :] nxt = re.search(r"^## ", rest, re.MULTILINE) body = rest[: nxt.start()] if nxt else rest return body.strip() def _needs_hunt(spec_md: str) -> bool: body = _fault_section(spec_md) if not body or FAULT_PLACEHOLDER in body: return False return True def _tasks(ship_dir: Path) -> list[dict[str, Any]]: data = load_yaml(ship_dir / "tasks.yaml") if not isinstance(data, dict): return [] tasks = data.get("tasks") return [t for t in tasks if isinstance(t, dict)] if isinstance(tasks, list) else [] def _task_placeholders(tasks: list[dict[str, Any]]) -> bool: blob = json.dumps(tasks, ensure_ascii=False) return _has_any(blob, TASK_PLACEHOLDERS) def _commits(current: dict[str, Any]) -> list[dict[str, Any]]: evidence = current.get("evidence") or {} if not isinstance(evidence, dict): return [] rows = evidence.get("commits") or [] return [r for r in rows if isinstance(r, dict)] def _has_change_commit(current: dict[str, Any]) -> bool: return any(r.get("role") in {"head", "merged"} for r in _commits(current)) def _has_merged_commit(current: dict[str, Any]) -> bool: return any(r.get("role") == "merged" for r in _commits(current)) def _manager(current: dict[str, Any]) -> dict[str, Any]: value = current.get("manager") return value if isinstance(value, dict) else {} def _stop_state(stop_at: str | None) -> str | None: if not stop_at: return None return STOP_AT_STATE.get(stop_at) def _state_index(state: str | None) -> int: if state in DELIVERY_ORDER: return DELIVERY_ORDER.index(state) return 0 def _next_delivery( *, ship_id: str, entry: str, current_state: str, stop_at: str | None, skip_until: str | None = None, ) -> dict[str, Any] | None: target_stop = _stop_state(stop_at) if not stop_at or not target_stop: return _out( ship_id=ship_id, entry=entry, actor="manager", step="ask_user", status="ask_user", stop_at=stop_at, summary="stopAt 未配置或无法映射,主会话向用户确认停止点后再继续", ) start = _state_index(current_state) floor = _state_index(skip_until) if skip_until else start start = max(start, floor) stop_i = _state_index(target_stop) if start >= stop_i: return None nxt = DELIVERY_STEPS[start] state, step, skill = nxt summary = f"派 worker 把 delivery.state 推进到 {state},加载 {skill}" if step == "regress": summary = ( "派 tester worker 做合并后的完整回归:先发现父仓库和子项目的所有 tester " "catalog,逐根执行并为每根写独立快照;再执行项目声明的原生测试套件;" "任一测试面缺证据不得通过" ) return _out( ship_id=ship_id, entry=entry, actor="worker", step=step, skill=skill, target_state=state, stop_at=stop_at, summary=summary, ) def _failed(ship_id: str, entry: str, mgr: dict[str, Any], stop_at: str | None) -> dict[str, Any] | None: if mgr.get("lastResult") not in {"fail", "blocked"}: return None last = str(mgr.get("lastStep") or "unknown") return _out( ship_id=ship_id, entry=entry, actor="manager", step=last, status="blocked", stop_at=stop_at, summary=f"上一步 {last} 失败或阻塞,主会话不要自己做;问用户是否开新 agent 重试", ) def _ready_tasks(tasks: list[dict[str, Any]]) -> list[dict[str, Any]]: by_id = {str(t.get("id")): t for t in tasks if t.get("id")} ready: list[dict[str, Any]] = [] for task in tasks: status = task.get("status") if status not in {"open", "in_progress"}: continue deps = task.get("dependsOn") or [] if not isinstance(deps, list): continue ok = True for dep in deps: other = by_id.get(str(dep)) if other is None or other.get("status") not in READY_DEP: ok = False break if ok: ready.append(task) return ready def decide_next(ship_dir: Path, entry: str) -> dict[str, Any]: current_path = ship_dir / "current.yaml" if not current_path.is_file(): return _out( ship_id=ship_dir.name, entry=entry, actor="manager", step="ensure_record", status="blocked", summary="没有 current.yaml,主会话先跑 create_ship.py", ) current = load_yaml(current_path) if not isinstance(current, dict): return _out( ship_id=ship_dir.name, entry=entry, actor="manager", step="ensure_record", status="blocked", summary="current.yaml 无法解析", ) ship_id = str(current.get("shipId") or ship_dir.name) kind = current.get("kind") stop_at = current.get("stopAt") or (current.get("delivery") or {}).get("stopAt") if isinstance(stop_at, str) and stop_at in {"", "null"}: stop_at = None state = (current.get("delivery") or {}).get("state") or "change_ready" source = current.get("source") if isinstance(current.get("source"), dict) else {} spec = current.get("spec") if isinstance(current.get("spec"), dict) else {} mgr = _manager(current) spec_md = _text(ship_dir / "spec.md") tasks = _tasks(ship_dir) stop = stop_at if isinstance(stop_at, str) else None failed = _failed(ship_id, entry, mgr, stop) if failed: return failed if mgr.get("lastResult") == "ask": last = str(mgr.get("lastStep") or "unknown") return _out( ship_id=ship_id, entry=entry, actor="manager", step="ask_user", status="ask_user", stop_at=stop, summary=f"上一步 {last} 需要用户决定,主会话先问用户,不要自己执行下一步", ) in_flight = mgr.get("inFlight") if isinstance(in_flight, str) and in_flight.strip(): return _out( ship_id=ship_id, entry=entry, actor="manager", step=in_flight, status="blocked", stop_at=stop, summary=f"manager.inFlight={in_flight},先等待该 worker,不要重派或自己做", ) use_pr = entry == "pr" or source.get("kind") == "pr" use_long = (not use_pr) and (entry == "long" or kind == "long") if use_pr: return _decide_pr(ship_id, current, source, mgr, state, stop) if use_long: return _decide_long(ship_dir, ship_id, current, spec, spec_md, tasks, state, stop) return _decide_short(ship_id, spec_md, tasks, current, state, stop) def _decide_short( ship_id: str, spec_md: str, tasks: list[dict[str, Any]], current: dict[str, Any], state: str, stop_at: str | None, ) -> dict[str, Any]: if not spec_md or _has_any(spec_md, SPEC_PLACEHOLDERS) or _task_placeholders(tasks): return _out( ship_id=ship_id, entry="short", actor="manager", step="write_spec", stop_at=stop_at, summary="主会话补齐 spec.md 四段和唯一任务的 expected/verification,不要加载执行类 skill", ) if not stop_at: return _out( ship_id=ship_id, entry="short", actor="manager", step="ask_user", status="ask_user", summary="stopAt 未配置,主会话向用户确认 verified / review_ready / validation_ready / released", ) if state == "change_ready" and _needs_hunt(spec_md) and not _has_change_commit(current): last = _manager(current).get("lastStep") if last != "hunt": return _out( ship_id=ship_id, entry="short", actor="worker", step="hunt", skill="hunt", stop_at=stop_at, summary="派 hunt worker 确认根因、复现和验收信号", ) if state == "change_ready" and not _has_change_commit(current): return _out( ship_id=ship_id, entry="short", actor="worker", step="implement", skill=None, task_id=str(tasks[0]["id"]) if tasks and tasks[0].get("id") else "T-01", stop_at=stop_at, summary="派 worker 按 spec.md 改代码,不要加载 delivery-loop", ) nxt = _next_delivery( ship_id=ship_id, entry="short", current_state=state, stop_at=stop_at ) if nxt: return nxt return _out( ship_id=ship_id, entry="short", actor="manager", step="report", status="done", stop_at=stop_at, summary="已到停止点,主会话按账本写完成报告", ) def _decide_long( ship_dir: Path, ship_id: str, current: dict[str, Any], spec: dict[str, Any], spec_md: str, tasks: list[dict[str, Any]], state: str, stop_at: str | None, ) -> dict[str, Any]: spec_status = spec.get("specStatus") handoff = spec.get("handoff") prd = spec.get("prd") handoff_path = ship_dir / str(handoff) if handoff else ship_dir / "spec" / "handoff.yaml" if spec_status not in {"approved"}: if spec_status in {None, "", "draft"} and not prd: return _out( ship_id=ship_id, entry="long", actor="worker", step="specifier_new", skill="specifier", stop_at=stop_at, summary="派 specifier worker 写 spec/ 下的 PRD,主会话不要读 specifier", ) if spec_status in {"changes_requested"}: return _out( ship_id=ship_id, entry="long", actor="worker", step="specifier_review", skill="specifier", stop_at=stop_at, summary="派 specifier worker 按用户意见出新 PRD 版本", ) return _out( ship_id=ship_id, entry="long", actor="manager", step="ask_user", status="ask_user", stop_at=stop_at, summary="PRD 待用户批准或提出修改;主会话问用户,不要自己改 PRD", ) if not handoff_path.is_file(): return _out( ship_id=ship_id, entry="long", actor="worker", step="specifier_handoff", skill="specifier", stop_at=stop_at, summary="已批准但缺少 handoff,派 specifier worker 生成 spec/handoff.yaml", ) if _task_placeholders(tasks): return _out( ship_id=ship_id, entry="long", actor="manager", step="materialize_tasks", stop_at=stop_at, summary="主会话从 handoff.yaml 写入 tasks.yaml,不要加载 specifier", ) unfinished = [t for t in tasks if t.get("status") not in DONE_TASK] if unfinished: ready = _ready_tasks(tasks) if not ready: return _out( ship_id=ship_id, entry="long", actor="manager", step="ask_user", status="blocked", stop_at=stop_at, summary="有未完成任务但依赖未就绪,主会话报告阻塞,不要自己开发", ) open_ready = [t for t in ready if t.get("status") == "open"] if open_ready: ids = [str(t["id"]) for t in open_ready if t.get("id")] first = ids[0] return _out( ship_id=ship_id, entry="long", actor="worker", step="implement_task", skill=None, task_id=first, ready_task_ids=ids, stop_at=stop_at, summary="派 worker 实现无依赖冲突的任务;多个 id 可并行,仍全部等完", ) inprog = [t for t in ready if t.get("status") == "in_progress"] task = inprog[0] tid = str(task.get("id") or "") last = _manager(current).get("lastStep") if last == "implement_task": return _out( ship_id=ship_id, entry="long", actor="worker", step="verify_task", skill="tester", task_id=tid, target_state="verified", stop_at=stop_at, summary="派 tester worker 验证该任务,主会话不要自己点浏览器", ) task_stop = task.get("stopAt") or "merged" if last == "verify_task" and task_stop in {"pr_opened", "merged", "review_ready"}: return _out( ship_id=ship_id, entry="long", actor="worker", step="open_pr", skill="delivery-loop", task_id=tid, target_state="pr_opened", stop_at=stop_at, summary="派 worker 为该子任务开 PR,默认不部署生产", ) if last == "open_pr" and task_stop in {"merged"}: return _out( ship_id=ship_id, entry="long", actor="worker", step="merge_pr", skill="delivery-loop", task_id=tid, target_state="merged", stop_at=stop_at, summary="派 worker 合并该子任务 PR,不要升级生产", ) return _out( ship_id=ship_id, entry="long", actor="manager", step="ask_user", status="blocked", task_id=tid, stop_at=stop_at, summary=f"任务 {tid} 处于 in_progress,但 lastStep={last!r},主会话先核对账本", ) parent_state = state if _state_index(state) < _state_index("merged"): parent_state = "merged" nxt = _next_delivery( ship_id=ship_id, entry="long", current_state=parent_state, stop_at=stop_at, skip_until="merged", ) if nxt: return nxt return _out( ship_id=ship_id, entry="long", actor="manager", step="report", status="done", stop_at=stop_at, summary="已到停止点,主会话按账本写完成报告", ) def _decide_pr( ship_id: str, current: dict[str, Any], source: dict[str, Any], mgr: dict[str, Any], state: str, stop_at: str | None, ) -> dict[str, Any]: if not source.get("ref"): return _out( ship_id=ship_id, entry="pr", actor="manager", step="bind_pr", stop_at=stop_at, summary="主会话把 PR/MR 写入 source.ref 和 evidence.prs,不要加载 check", ) if not stop_at: return _out( ship_id=ship_id, entry="pr", actor="manager", step="ask_user", status="ask_user", summary="stopAt 未配置,主会话向用户确认合并后停在哪一档", ) if not _has_merged_commit(current) and state not in { "merged", "regression_passed", "artifact_built", "published", "deployed", }: if mgr.get("lastStep") == "check_pr" and mgr.get("lastResult") == "pass": return _out( ship_id=ship_id, entry="pr", actor="worker", step="merge_pr", skill="delivery-loop", target_state="merged", stop_at=stop_at, summary="派 worker 合并 PR,并以合并 commit 作为后续来源", ) if mgr.get("lastStep") == "check_pr" and mgr.get("lastResult") == "ask": return _out( ship_id=ship_id, entry="pr", actor="worker", step="fix_pr", skill=None, stop_at=stop_at, summary="检查要求修复且当前请求含修复时,派 worker 在 PR 分支上改;不要扩范围", ) return _out( ship_id=ship_id, entry="pr", actor="worker", step="check_pr", skill="check", stop_at=stop_at, summary="派 check worker 看差异、审核和 CI,主会话不要自己审代码", ) nxt = _next_delivery( ship_id=ship_id, entry="pr", current_state=state if state != "change_ready" else "merged", stop_at=stop_at, skip_until="merged", ) if nxt: return nxt return _out( ship_id=ship_id, entry="pr", actor="manager", step="report", status="done", stop_at=stop_at, summary="已到停止点,主会话按账本写完成报告", ) 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("--entry", choices=("short", "long", "pr"), required=True) args = parser.parse_args() ship_dir = Path(args.project_root).resolve() / ".pouch" / "ship" / args.ship_id if not ship_dir.is_dir(): doc = _out( ship_id=args.ship_id, entry=args.entry, actor="manager", step="ensure_record", status="blocked", summary="没有 ship 目录,主会话先跑 create_ship.py", ) json.dump(doc, sys.stdout, ensure_ascii=False, indent=2) sys.stdout.write("\n") return 0 doc = decide_next(ship_dir, args.entry) json.dump(doc, sys.stdout, ensure_ascii=False, indent=2) sys.stdout.write("\n") return 0 if __name__ == "__main__": raise SystemExit(main())