diff --git a/mcp/court-mcp/orchestrator.py b/mcp/court-mcp/orchestrator.py index a4f97ce..7cdbf2c 100644 --- a/mcp/court-mcp/orchestrator.py +++ b/mcp/court-mcp/orchestrator.py @@ -9,9 +9,6 @@ | webhook 队列 | ``pending-webhook/.json`` (PR-14) | | 跑着的 court window | tmux ``agent-court-dashboard`` session windows (PR-13) | -加 SY-4 后多一处: -- ``retry-queue.json`` (重试) - bug 类型 (来自 issue #18 的诊断): - pending 已 approve 但 tmux window 没起来 → 状态机断了 - tmux window 残留 (claude crash 没退干净) → seen 标 EXECUTING, 新轮询不再 dispatch @@ -20,7 +17,7 @@ ## MVP v1 范围 (本模块) - **不重写写路径**: watcher / router / approval / resolver 现状不动 -- 提供**单一只读统一视图** ``snapshot()``: 把 4+1 处状态 join 成 ``list[Run]`` +- 提供**单一只读统一视图** ``snapshot()``: 把 4 处状态 join 成 ``list[Run]`` - ``reconcile()``: 检测 4 类不一致, 返 ``list[Inconsistency]`` (不主动修, 让 caller 或 dashboard UI 决定) - ``get_metrics()`` 统计供监控用 @@ -45,7 +42,7 @@ class RunState(str, Enum): """跟 seen_state.last_action 一一对应 + 几个派生状态.""" - QUEUED = "queued" # 在 retry queue / pending-webhook 等 + QUEUED = "queued" # 在 pending-webhook / pending-approval 等 BOOTSTRAP = "bootstrap" # 首次启动 hydrate, 历史 issue PENDING_APPROVAL = "pending_approval" # pending-approval/*.json 在 (INTAKE/PLAN) DISPATCHED = "dispatched" # last_action=DISPATCHED_DASHBOARD, tmux window 应该在 @@ -53,8 +50,6 @@ class RunState(str, Enum): DONE = "done" # last_action=DONE_DASHBOARD FAILED = "failed" # last_action=SPAWN_FAILED REJECTED = "rejected" # last_action=REJECTED_DASHBOARD - DEFERRED_CAPACITY = "deferred_capacity" - TIMEOUT_KILLED = "timeout_killed" UNKNOWN = "unknown" @@ -65,14 +60,12 @@ class RunState(str, Enum): "DONE_DASHBOARD": RunState.DONE, "SPAWN_FAILED": RunState.FAILED, "REJECTED_DASHBOARD": RunState.REJECTED, - "DEFERRED_CAPACITY": RunState.DEFERRED_CAPACITY, - "TIMEOUT_KILLED": RunState.TIMEOUT_KILLED, } @dataclass(frozen=True) class Run: - """单个 issue 的统一视图 (4+1 处状态 join 后的派生).""" + """单个 issue 的统一视图 (4 处状态 join 后的派生).""" issue_key: str # "#" repo: str number: int @@ -83,8 +76,6 @@ class Run: tmux_window: str = "" # 名字; "" = 没 window tmux_window_alive: bool = False has_pending_approval: bool = False - in_retry_queue: bool = False - retry_attempt: int = 0 dispatched_at: str = "" def to_dict(self) -> dict[str, Any]: @@ -146,7 +137,6 @@ def __init__( def snapshot(self) -> Snapshot: seen = self._load_seen() pending_keys = self._collect_pending_approval_keys() - retry_map = self._collect_retry_queue() tmux_windows = self._collect_tmux_windows() tmux_window_set = set(tmux_windows) @@ -157,16 +147,14 @@ def snapshot(self) -> Snapshot: issue_key=key, raw=raw, has_pending_approval=key in pending_keys, - in_retry_queue=key in retry_map, - retry_attempt=retry_map.get(key, 0), tmux_windows=tmux_window_set, ) if run is not None: runs.append(run) if run.tmux_window: seen_window_names.add(run.tmux_window) - # 把只在 retry queue / pending 里, 但 seen 还没记的 issue 也补成 Run - for key in (set(pending_keys) | set(retry_map)) - set(seen.keys()): + # 把只在 pending 里, 但 seen 还没记的 issue 也补成 Run + for key in set(pending_keys) - set(seen.keys()): repo, num = _split_key(key) if not repo or num is None: continue @@ -176,8 +164,6 @@ def snapshot(self) -> Snapshot: number=num, state=RunState.QUEUED, has_pending_approval=(key in pending_keys), - in_retry_queue=(key in retry_map), - retry_attempt=retry_map.get(key, 0), stage="INTAKE" if key in pending_keys else "", )) orphan_windows = [ @@ -199,7 +185,7 @@ def snapshot(self) -> Snapshot: ) def reconcile(self) -> list[Inconsistency]: - """返当前 4+1 处状态间的不一致清单. 不主动修.""" + """返当前 4 处状态间的不一致清单. 不主动修.""" return self.snapshot().inconsistencies def get_run(self, issue_key: str) -> Run | None: @@ -212,7 +198,7 @@ def get_metrics(self) -> dict[str, int]: return self.snapshot().metrics # ------------------------------------------------------------------ - # 状态收集 (4+1 处, 全部容错) + # 状态收集 (4 处, 全部容错) # ------------------------------------------------------------------ def _load_seen(self) -> dict[str, Any]: @@ -247,27 +233,6 @@ def _collect_pending_approval_keys(self) -> set[str]: keys.add(f"{repo}#{num}") return keys - def _collect_retry_queue(self) -> dict[str, int]: - """issue_key → 当前 attempt 次数. SY-4 retry_queue 文件.""" - path = self.state_dir / "retry-queue.json" - if not path.is_file(): - return {} - try: - raw = json.loads(path.read_text()) - except (OSError, json.JSONDecodeError): - return {} - if not isinstance(raw, dict): - return {} - out: dict[str, int] = {} - for key, entry in raw.items(): - if not isinstance(entry, dict): - continue - try: - out[key] = int(entry.get("attempt", 0)) - except (TypeError, ValueError): - out[key] = 0 - return out - def _collect_tmux_windows(self) -> list[str]: """``tmux list-windows -t -F '#{window_name}'`` → list. tmux 不可用返 [].""" try: @@ -304,7 +269,7 @@ def _reconcile_internal( kind="dispatched_window_gone", severity=SEVERITY_ERROR, detail=f"seen.last_action=DISPATCHED_DASHBOARD 但 tmux window {r.tmux_window!r} 已不存在", - suggested_fix="orchestrator 标 FAILED + push retry queue, 或 caller 手动 spawn", + suggested_fix="orchestrator 标 FAILED, 或 caller 手动重新 spawn", )) # I-2: seen=EXECUTING 但 tmux window 不在 → claude crash 在中途 elif r.state == RunState.EXECUTING and r.tmux_window and not r.tmux_window_alive: @@ -313,18 +278,9 @@ def _reconcile_internal( kind="executing_window_gone", severity=SEVERITY_ERROR, detail=f"seen.last_action=EXECUTING 但 tmux window {r.tmux_window!r} 已不存在", - suggested_fix="标 FAILED + push retry queue", - )) - # I-3: seen 已 DONE 但 retry queue 里还有这条 → stale entry - if r.state in {RunState.DONE, RunState.REJECTED} and r.in_retry_queue: - out.append(Inconsistency( - issue_key=r.issue_key, - kind="retry_stale_after_done", - severity=SEVERITY_WARN, - detail=f"seen 已 {r.state.value} 但 retry queue 里仍有条目 (attempt={r.retry_attempt})", - suggested_fix="orchestrator 调 retry_queue.remove(issue_key)", + suggested_fix="标 FAILED, 或 caller 手动重新 spawn", )) - # I-4: pending-approval 还在但 seen 已 DONE → result 写后没清 pending + # I-3: pending-approval 还在但 seen 已 DONE → result 写后没清 pending if r.state in {RunState.DONE, RunState.REJECTED} and r.has_pending_approval: out.append(Inconsistency( issue_key=r.issue_key, @@ -333,7 +289,7 @@ def _reconcile_internal( detail=f"seen 已 {r.state.value} 但 pending-approval/*.json 还在", suggested_fix="清掉 pending-approval/.json + .result + .lock", )) - # I-5: tmux window 在但 seen 完全没记录 (没经过 router 走 spawn) → 手动起的孤儿 + # I-4: tmux window 在但 seen 完全没记录 (没经过 router 走 spawn) → 手动起的孤儿 for win in orphan_windows: out.append(Inconsistency( issue_key="", # 没对应 issue @@ -356,7 +312,6 @@ def _compute_metrics( out["total"] = len(runs) out["active"] = sum(1 for r in runs if r.state in {RunState.DISPATCHED, RunState.EXECUTING}) out["pending_approval_count"] = sum(1 for r in runs if r.has_pending_approval) - out["in_retry_queue"] = sum(1 for r in runs if r.in_retry_queue) out["orphan_tmux_windows"] = len(orphan_windows) out["inconsistencies"] = len(inconsistencies) out["inconsistencies_error"] = sum(1 for i in inconsistencies if i.severity == SEVERITY_ERROR) @@ -383,8 +338,6 @@ def _build_run_from_seen( issue_key: str, raw: Any, has_pending_approval: bool, - in_retry_queue: bool, - retry_attempt: int, tmux_windows: set[str], ) -> Run | None: if not isinstance(raw, dict): @@ -409,7 +362,5 @@ def _build_run_from_seen( tmux_window=tmux_window, tmux_window_alive=(bool(tmux_window) and tmux_window in tmux_windows), has_pending_approval=has_pending_approval, - in_retry_queue=in_retry_queue, - retry_attempt=retry_attempt, dispatched_at=str(raw.get("dispatched_at", "")), ) diff --git a/mcp/court-mcp/tests/test_court_reconcile_cli.py b/mcp/court-mcp/tests/test_court_reconcile_cli.py index ba1875e..4be9fd8 100644 --- a/mcp/court-mcp/tests/test_court_reconcile_cli.py +++ b/mcp/court-mcp/tests/test_court_reconcile_cli.py @@ -41,17 +41,19 @@ def test_clean_state_exits_zero(tmp_path, capsys): def test_warn_inconsistency_exits_one(tmp_path, capsys, monkeypatch): - """retry queue 残留 DONE 条目 → warn-only → exit 1.""" + """seen 已 DONE 但 pending-approval 还在 → warn-only → exit 1.""" _seed_seen(tmp_path, { "foo/bar#1": {"last_action": "DONE_DASHBOARD"}, }) - (tmp_path / "gitea-watcher" / "retry-queue.json").write_text(json.dumps({ - "foo/bar#1": {"attempt": 1, "next_at": 0, "last_error": "old", "last_failed_at": 0}, + pending_dir = tmp_path / "gitea-watcher" / "pending-approval" + pending_dir.mkdir(parents=True, exist_ok=True) + (pending_dir / "foo-bar-1-intake.json").write_text(json.dumps({ + "repo": "foo/bar", "number": 1, "stage": "INTAKE", })) rc = cli.main(["--court-root", str(tmp_path)]) out = capsys.readouterr().out assert rc == 1 - assert "retry_stale_after_done" in out + assert "pending_after_done" in out assert "[WARN]" in out diff --git a/mcp/court-mcp/tests/test_orchestrator.py b/mcp/court-mcp/tests/test_orchestrator.py index 7e022b8..e50916a 100644 --- a/mcp/court-mcp/tests/test_orchestrator.py +++ b/mcp/court-mcp/tests/test_orchestrator.py @@ -39,11 +39,6 @@ def _write_pending_approval(tmp_path: Path, slug: str, meta: dict, *, has_result (sd / f"{slug}.result").write_text("{}") -def _write_retry_queue(tmp_path: Path, data: dict): - sd = _state_dir(tmp_path) - (sd / "retry-queue.json").write_text(json.dumps(data)) - - class _Orchestrator(orch.Orchestrator): """注入 tmux_windows, 避免依赖真 tmux.""" @@ -103,21 +98,6 @@ def test_snapshot_pending_already_resulted_not_counted_as_pending(tmp_path): assert r.state == orch.RunState.DISPATCHED -def test_snapshot_includes_retry_queue_attempt(tmp_path): - _write_seen(tmp_path, { - "demo/x#1": {"last_action": "SPAWN_FAILED", "stage": "INTAKE"}, - }) - _write_retry_queue(tmp_path, { - "demo/x#1": {"attempt": 2, "next_at": 0, "last_error": "boom", "last_failed_at": 0}, - }) - o = _Orchestrator(tmp_path, tmux_windows=[]) - r = o.get_run("demo/x#1") - assert r is not None - assert r.state == orch.RunState.FAILED - assert r.in_retry_queue is True - assert r.retry_attempt == 2 - - # --------------------------------------------------------------------------- # reconcile: 4 类不一致 # --------------------------------------------------------------------------- @@ -146,21 +126,8 @@ def test_reconcile_detects_executing_window_gone(tmp_path): assert "executing_window_gone" in kinds -def test_reconcile_detects_retry_stale_after_done(tmp_path): - """I-3: seen 已 DONE 但 retry queue 还有条 → 警告 (应该 remove).""" - _write_seen(tmp_path, { - "foo/bar#1": {"last_action": "DONE_DASHBOARD"}, - }) - _write_retry_queue(tmp_path, { - "foo/bar#1": {"attempt": 1, "next_at": 0, "last_error": "old", "last_failed_at": 0}, - }) - o = _Orchestrator(tmp_path, tmux_windows=[]) - inc = o.reconcile() - assert any(i.kind == "retry_stale_after_done" for i in inc) - - def test_reconcile_detects_pending_after_done(tmp_path): - """I-4: seen 已 DONE 但 pending-approval/*.json 还在 → 警告.""" + """I-3: seen 已 DONE 但 pending-approval/*.json 还在 → 警告.""" _write_seen(tmp_path, { "demo/x#1": {"last_action": "DONE_DASHBOARD"}, }) @@ -173,7 +140,7 @@ def test_reconcile_detects_pending_after_done(tmp_path): def test_reconcile_detects_tmux_window_orphan(tmp_path): - """I-5: tmux window 在但 seen 没对应 → 警告.""" + """I-4: tmux window 在但 seen 没对应 → 警告.""" o = _Orchestrator(tmp_path, tmux_windows=[orch.WATCHER_WINDOW, "manual-test-1"]) inc = o.reconcile() assert len(inc) == 1 @@ -241,16 +208,6 @@ def test_corrupt_seen_json_treated_as_empty(tmp_path): assert snap.runs == [] -def test_corrupt_retry_queue_json_ignored(tmp_path): - _write_seen(tmp_path, {"a/b#1": {"last_action": "SPAWN_FAILED"}}) - sd = _state_dir(tmp_path) - (sd / "retry-queue.json").write_text("garbage") - o = _Orchestrator(tmp_path, tmux_windows=[]) - r = o.get_run("a/b#1") - assert r is not None - assert r.in_retry_queue is False - - def test_missing_state_dir_returns_empty_snapshot(tmp_path): o = _Orchestrator(tmp_path / "nope", tmux_windows=[]) snap = o.snapshot() diff --git a/mcp/court-mcp/tests/test_workflow_loader.py b/mcp/court-mcp/tests/test_workflow_loader.py index 71a7a67..22a89e1 100644 --- a/mcp/court-mcp/tests/test_workflow_loader.py +++ b/mcp/court-mcp/tests/test_workflow_loader.py @@ -15,8 +15,6 @@ def test_parse_text_with_frontmatter_returns_typed_config(): text = """--- schema_version: 1 branch_prefix: "feat/auto-" -max_concurrent_runs: 5 -run_timeout_seconds: 900 allowed_labels: ["agent-ok", "auto"] require_approval_stages: ["INTAKE"] working_dir_strategy: worktree @@ -32,8 +30,6 @@ def test_parse_text_with_frontmatter_returns_typed_config(): config, prompt = wl.parse_workflow_text(text) assert config.schema_version == 1 assert config.branch_prefix == "feat/auto-" - assert config.max_concurrent_runs == 5 - assert config.run_timeout_seconds == 900 assert config.allowed_labels == ("agent-ok", "auto") assert config.require_approval_stages == ("INTAKE",) assert config.working_dir_strategy == "worktree" @@ -47,7 +43,6 @@ def test_parse_text_without_frontmatter_uses_all_defaults(): config, prompt = wl.parse_workflow_text(text) assert config.schema_version == 1 assert config.branch_prefix == "feat/auto-" - assert config.max_concurrent_runs == 3 assert config.allowed_labels == () assert config.tracker.provider == "gitea" assert prompt == "# just a prompt\n\nno frontmatter at all" @@ -65,18 +60,6 @@ def test_invalid_working_dir_strategy_raises(): wl.parse_workflow_text(text) -def test_max_concurrent_zero_raises(): - text = "---\nmax_concurrent_runs: 0\n---\nbody\n" - with pytest.raises(wl.WorkflowParseError, match="max_concurrent_runs"): - wl.parse_workflow_text(text) - - -def test_negative_timeout_raises(): - text = "---\nrun_timeout_seconds: -1\n---\nbody\n" - with pytest.raises(wl.WorkflowParseError, match="run_timeout_seconds"): - wl.parse_workflow_text(text) - - def test_invalid_tracker_provider_raises(): text = "---\ntracker:\n provider: jira\n---\nbody\n" with pytest.raises(wl.WorkflowParseError, match="tracker.provider"): @@ -94,6 +77,22 @@ def test_unknown_frontmatter_fields_are_ignored_forward_compat(): assert config.branch_prefix == "feat/auto-" # defaults preserved +def test_removed_sy4_concurrency_keys_are_tolerated(): + """已废弃的 SY-4 并发/重试 key 仍可能出现在 WORKFLOW.md frontmatter; loader 应忽略而非报错.""" + text = """--- +schema_version: 1 +max_concurrent_runs: 3 +run_timeout_seconds: 1800 +retry_max: 3 +retry_backoff_base_seconds: 60 +--- +body +""" + config, _ = wl.parse_workflow_text(text) + assert config.branch_prefix == "feat/auto-" # defaults preserved, no error raised + assert not hasattr(config, "max_concurrent_runs") + + def test_bad_yaml_frontmatter_raises_parse_error(): text = "---\n:: not valid yaml ::\n---\nbody\n" with pytest.raises(wl.WorkflowParseError, match="YAML"): diff --git a/mcp/court-mcp/workflow_loader.py b/mcp/court-mcp/workflow_loader.py index 8b739bb..f8078d5 100644 --- a/mcp/court-mcp/workflow_loader.py +++ b/mcp/court-mcp/workflow_loader.py @@ -10,7 +10,6 @@ --- schema_version: 1 branch_prefix: "feat/auto-" - max_concurrent_runs: 3 ... --- @@ -67,10 +66,6 @@ class WorkflowConfig: schema_version: int = 1 branch_prefix: str = "feat/auto-" - max_concurrent_runs: int = 3 - run_timeout_seconds: int = 1800 - retry_max: int = 3 - retry_backoff_base_seconds: int = 60 working_dir_strategy: str = "inplace" # inplace | worktree (SY-2 启用) allowed_labels: tuple[str, ...] = () # 空 = 不过滤; 非空 = issue 必须命中一个 require_approval_stages: tuple[str, ...] = ("INTAKE", "PLAN") @@ -101,13 +96,6 @@ def from_dict(cls, raw: dict[str, Any]) -> "WorkflowConfig": data[k] = tuple(v) else: raise WorkflowParseError(f"{k} 必须是 list, got {type(v).__name__}") - for k in ("max_concurrent_runs", "run_timeout_seconds", "retry_max", "retry_backoff_base_seconds"): - if k in data and not isinstance(data[k], int): - raise WorkflowParseError(f"{k} 必须是 int, got {type(data[k]).__name__}") - if k in data and data[k] < 0: - raise WorkflowParseError(f"{k} 不能为负, got {data[k]!r}") - if data.get("max_concurrent_runs") == 0: - raise WorkflowParseError("max_concurrent_runs 必须 > 0") # tracker 嵌套 tracker_raw = data.pop("tracker", None) if tracker_raw is None: