From 383dfed94d746f550123eaf21cff71e0a8415d22 Mon Sep 17 00:00:00 2001 From: lisihao Date: Sun, 31 May 2026 19:13:34 -0400 Subject: [PATCH] Fix dispatch pool reconciliation and PM closeout guards --- harness/lib/graph_node_dispatcher.py | 197 +++++++++++++++++- .../tests/graph/test_graph_dispatch_submit.py | 33 ++- .../test-graph-dispatch-stale-scrollback.py | 125 +++++++++++ harness/tests/test_pm_dispatch.py | 64 ++++++ harness/tools/pm_dispatch.py | 77 ++++++- 5 files changed, 475 insertions(+), 21 deletions(-) diff --git a/harness/lib/graph_node_dispatcher.py b/harness/lib/graph_node_dispatcher.py index b186e3be4..5ff618017 100644 --- a/harness/lib/graph_node_dispatcher.py +++ b/harness/lib/graph_node_dispatcher.py @@ -29,6 +29,8 @@ def _harness_dir() -> Path: HARNESS_DIR = _harness_dir() +if str(HARNESS_DIR / "lib") not in sys.path: + sys.path.insert(0, str(HARNESS_DIR / "lib")) SPRINTS_DIR = HARNESS_DIR / "sprints" MULTI_TASK_RUN_DIR = HARNESS_DIR / "run" / "multi-task" SESSION = os.environ.get("SOLAR_HARNESS_SESSION", "solar-harness") @@ -59,6 +61,7 @@ def _harness_dir() -> Path: re.I, ) PANE_RATE_LIMIT_FALLBACK_SEC = int(os.environ.get("SOLAR_PANE_RATE_LIMIT_FALLBACK_SEC", "900")) +OPERATOR_CONTRACT_CLOSEOUT_COOLDOWN_SEC = int(os.environ.get("SOLAR_GRAPH_OPERATOR_CONTRACT_CLOSEOUT_COOLDOWN_SEC", "900")) def _effective_graph_max_parallel(default: int = 8) -> int: @@ -141,6 +144,14 @@ def _prune_expired_operator_blocks() -> None: "dispatch prompt not dismissed", "late_settle_blocked", } + +try: + from pane_overlay_state import pane_overlay_detail, pane_overlay_blocked, prompt_match_is_stale, tail_has_idle_prompt_footer +except Exception: # pragma: no cover - keep dispatcher usable in partial installs + pane_overlay_detail = None # type: ignore + pane_overlay_blocked = None # type: ignore + prompt_match_is_stale = None # type: ignore + tail_has_idle_prompt_footer = None # type: ignore STATE_READ_PREFLIGHT = """ ## 必须先读状态 (防写入 hook 卡死) @@ -1185,6 +1196,51 @@ def _scope_lines(values: Any) -> str: return "\n".join(f"- `{v}`" for v in values) +def _write_scope_preflight_block(sid: str, node: dict[str, Any]) -> str: + """Warn builders when write-scope artifacts already exist from another sprint. + + Several early S01 graphs use generic files such as + `sprints/s01-req-N5-handoff.md`. Those paths can survive from a different + sprint and must not be treated as current evidence. + """ + scopes = node.get("write_scope") or [] + if isinstance(scopes, str): + scopes = [scopes] + rows: list[str] = [] + sprint_re = re.compile(r"sprint-[A-Za-z0-9_.\-\u4e00-\u9fff]+") + for raw in scopes: + scope = str(raw or "").strip() + if not scope or any(ch in scope for ch in "*?[]"): + continue + path = (HARNESS_DIR / scope).expanduser() if not scope.startswith("/") else Path(scope).expanduser() + if not path.exists() or not path.is_file(): + continue + try: + stat = path.stat() + sample = path.read_text(encoding="utf-8", errors="replace")[:12000] + except Exception: + continue + refs = sorted(set(sprint_re.findall(sample))) + foreign_refs = [ref for ref in refs if ref != sid] + contains_current = sid in sample + if foreign_refs or not contains_current: + mtime = datetime.datetime.fromtimestamp(stat.st_mtime, tz=datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + rows.append( + f"- `{scope}` exists already (mtime={mtime}, size={stat.st_size}); " + f"contains_current_sprint={str(contains_current).lower()}; " + f"foreign_sprint_refs={', '.join(foreign_refs[:3]) if foreign_refs else 'N/A'}" + ) + if not rows: + return "## Write Scope Preflight\n\n- No pre-existing stale write-scope artifacts detected." + return ( + "## Write Scope Preflight\n\n" + "The following declared output paths already exist but do not clearly belong to this sprint. " + "Treat them as stale inputs, not as completion evidence. Overwrite with current-sprint content " + "or explain why a different scoped artifact is required.\n\n" + + "\n".join(rows) + ) + + def _acceptance_lines(values: Any) -> str: if not values: return "- N/A" @@ -1398,6 +1454,42 @@ def _operator_terminal_result_closeout( } +def _cooldown_operator_after_contract_closeout(operator_id: str, closeout: dict[str, Any]) -> dict[str, Any]: + operator_id = str(operator_id or "").strip() + if not operator_id or OPERATOR_CONTRACT_CLOSEOUT_COOLDOWN_SEC <= 0: + return {"ok": False, "reason": "operator_cooldown_disabled_or_missing"} + try: + if str(HARNESS_DIR / "lib") not in sys.path: + sys.path.insert(0, str(HARNESS_DIR / "lib")) + import operator_flow_control as ofc # type: ignore + + expires_at = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta( + seconds=OPERATOR_CONTRACT_CLOSEOUT_COOLDOWN_SEC + ) + persisted = ofc.persist_operator_block( + operator_id, + "cooldown", + expires_at=expires_at, + reason="contract_closeout_failed", + source="graph_node_dispatcher", + evidence_text=json.dumps(closeout, ensure_ascii=False)[-4000:], + ) + runtime = ofc.set_operator_state( + operator_id, + "cooldown", + ttl_seconds=OPERATOR_CONTRACT_CLOSEOUT_COOLDOWN_SEC, + ) + return { + "ok": bool(runtime.get("runtime_state") == "cooldown" or persisted.get("ok")), + "operator_id": operator_id, + "cooldown_sec": OPERATOR_CONTRACT_CLOSEOUT_COOLDOWN_SEC, + "persisted": persisted, + "runtime": runtime, + } + except Exception as exc: + return {"ok": False, "reason": f"{type(exc).__name__}: {exc}", "operator_id": operator_id} + + def _reconcile_existing_dispatches(graph: dict[str, Any], graph_path: str | Path) -> list[dict[str, Any]]: sid = str(graph.get("sprint_id") or Path(graph_path).stem.replace(".task_graph", "")) repaired: list[dict[str, Any]] = [] @@ -1420,6 +1512,31 @@ def _reconcile_existing_dispatches(graph: dict[str, Any], graph_path: str | Path eval_verdict = "FAIL" else: eval_verdict = "" + if handoff_file and eval_verdict in {"PASS", "FAIL"} and status in {"passed", "failed"}: + stale_eval_keys = [ + "eval_assigned_to", + "eval_dispatch_id", + "eval_retry_reason", + "last_eval_closeout_failure", + "last_eval_operator_cooldown_after_closeout", + ] + cleared = [key for key in stale_eval_keys if key in node] + if cleared: + for key in cleared: + node.pop(key, None) + node["eval_json"] = eval_json_path + node["updated_at"] = _utc_now() + repaired.append( + { + "node": node_id, + "status": status, + "reason": "canonical_eval_verdict_cleared_stale_eval_state", + "cleared": cleared, + "eval_json": eval_json_path, + "verdict": eval_verdict, + } + ) + continue if handoff_file and eval_verdict in {"PASS", "FAIL"} and status in {"pending", "queued", "blocked", "assigned", "dispatched", "in_progress", "running", "reviewing", "ready_for_review", "needs_human_review", "failed_review", ""}: pane = str(node.get("assigned_to") or "").strip() dispatch_id = str(node.get("dispatch_id") or "").strip() @@ -1500,12 +1617,20 @@ def _reconcile_existing_dispatches(graph: dict[str, Any], graph_path: str | Path if closeout: pane = str(node.get("assigned_to") or "").strip() dispatch_id = str(node.get("dispatch_id") or "").strip() + operator_cooldown = {} + if closeout.get("reason") == "failed_contract_closeout": + operator_cooldown = _cooldown_operator_after_contract_closeout( + str(closeout.get("operator_id") or ""), + closeout, + ) if pane and dispatch_id: release_lease(pane, dispatch_id, f"graph_dispatch_reconcile_{closeout['reason']}") node.pop("assigned_to", None) node.pop("dispatch_id", None) node["dispatch_retry_reason"] = closeout["reason"] node["last_operator_closeout_failure"] = closeout + if operator_cooldown: + node["last_operator_cooldown_after_closeout"] = operator_cooldown node["updated_at"] = _utc_now() node["status"] = "pending" graph.setdefault("node_results", {}).pop(node_id, None) @@ -1514,7 +1639,7 @@ def _reconcile_existing_dispatches(graph: dict[str, Any], graph_path: str | Path sid, pane, dispatch_id, - {"node": node_id, **closeout}, + {"node": node_id, **closeout, "operator_cooldown": operator_cooldown}, ) repaired.append( { @@ -1525,6 +1650,7 @@ def _reconcile_existing_dispatches(graph: dict[str, Any], graph_path: str | Path "reason": closeout["reason"], "operator_status": closeout.get("operator_status"), "result_json": closeout.get("result_json"), + "operator_cooldown": operator_cooldown, } ) continue @@ -1617,6 +1743,27 @@ def _reconcile_existing_dispatches(graph: dict[str, Any], graph_path: str | Path # because no live lease exists. set_node_status(graph, node_id, "dispatched", pane=pane, dispatch_id=dispatch_id) continue + if lease_live and not unavailable_reason and not _pane_tui_busy(pane): + acquired_at = _parse_utc(str((lease or {}).get("acquired_at") or "")) + now = datetime.datetime.now(datetime.timezone.utc) + if acquired_at and (now - acquired_at).total_seconds() > 120: + release_lease(pane, dispatch_id, "graph_dispatch_reconcile_live_lease_idle_without_submit_ack") + node.pop("assigned_to", None) + node.pop("dispatch_id", None) + node["dispatch_retry_reason"] = "live_lease_idle_without_submit_ack" + node["updated_at"] = _utc_now() + node["status"] = "pending" + graph.setdefault("node_results", {}).pop(node_id, None) + repaired.append( + { + "node": node_id, + "pane": pane, + "dispatch_id": dispatch_id, + "status": "pending", + "reason": "live_lease_idle_without_submit_ack", + } + ) + continue if not lease_live: release_lease( pane, @@ -1701,6 +1848,15 @@ def _reconcile_existing_dispatches(graph: dict[str, Any], graph_path: str | Path } break if terminal_operator_assignment: + operator_cooldown = {} + failed_operator = "" + pane_value = str(terminal_operator_assignment.get("pane") or "") + if pane_value.startswith("operator:"): + failed_operator = pane_value.split(":", 1)[1].strip() + operator_cooldown = _cooldown_operator_after_contract_closeout( + failed_operator, + terminal_operator_assignment, + ) if terminal_operator_assignment["dispatch_id"]: release_lease( terminal_operator_assignment["pane"], @@ -1710,6 +1866,8 @@ def _reconcile_existing_dispatches(graph: dict[str, Any], graph_path: str | Path _clear_eval_assignments(node) node["eval_retry_reason"] = terminal_operator_assignment["reason"] node["last_eval_closeout_failure"] = terminal_operator_assignment + if operator_cooldown: + node["last_eval_operator_cooldown_after_closeout"] = operator_cooldown node["updated_at"] = _utc_now() repaired.append( { @@ -1720,6 +1878,7 @@ def _reconcile_existing_dispatches(graph: dict[str, Any], graph_path: str | Path "reason": terminal_operator_assignment["reason"], "operator_status": terminal_operator_assignment.get("operator_status"), "result_json": terminal_operator_assignment.get("result_json"), + "operator_cooldown": operator_cooldown, } ) continue @@ -2349,6 +2508,7 @@ def build_dispatch_text(payload: dict[str, Any], pane: str) -> str: plan_artifacts.get("physical_plan_ir_path", ""), ] ) + write_scope_preflight = _write_scope_preflight_block(str(sid), node) return f"""{STATE_READ_PREFLIGHT} {DEFINITION_OF_DONE_POLICY} @@ -2395,6 +2555,8 @@ def build_dispatch_text(payload: dict[str, Any], pane: str) -> str: {_scope_lines(node.get("write_scope"))} +{write_scope_preflight} + {architecture_block} ## Acceptance @@ -2805,6 +2967,8 @@ def _prompt_match_followed_by_idle_default_prompt(text: str, match: re.Match[str """ if match is None: return False + if prompt_match_is_stale: + return bool(prompt_match_is_stale(text, match)) after = text[match.end():] return bool(re.search(r"❯[\s\u00a0]+Try\s+\"", after)) or _tail_has_idle_prompt_footer(after) @@ -2816,6 +2980,8 @@ def _tail_has_idle_prompt_footer(text: str) -> bool: even after Claude returns to a clean prompt/footer. Treating that history as live state strands otherwise idle panes. """ + if tail_has_idle_prompt_footer: + return bool(tail_has_idle_prompt_footer(text)) lines = [line.rstrip() for line in text.splitlines()] footer_prefixes = ( "⏵", @@ -2878,6 +3044,7 @@ def _pane_prompt_residue_is_stale_scrollback(pane: str, text: str) -> bool: def _pane_tui_busy(pane: str) -> bool: tail = _pane_tail(pane) bottom = "\n".join(tail.splitlines()[-40:]) + overlay = pane_overlay_detail(tail) if pane_overlay_detail else {"state": "none", "type": ""} prompt_is_empty = "❯" in bottom and not _pane_current_prompt_has_residue(bottom) if PANE_RATE_LIMIT_OPTIONS_MODAL_RE.search(bottom): if _dismiss_rate_limit_options_modal(pane): @@ -2908,7 +3075,7 @@ def _pane_tui_busy(pane: str) -> bool: return False return True if PANE_SURVEY_PROMPT_RE.search(bottom): - if prompt_is_empty: + if overlay.get("state") == "stale_scrollback_ignored" or prompt_is_empty: return False return True confirmation_match = PANE_CONFIRMATION_PROMPT_RE.search(bottom) @@ -2930,6 +3097,8 @@ def _pane_tui_busy(pane: str) -> bool: # to return busy here before `_pane_unavailable_reason()` could clear it, # which left panes permanently stranded. if PANE_QUEUED_PROMPT_RE.search(bottom): + if overlay.get("state") == "stale_scrollback_ignored": + return False if _clear_stale_prompt_residue(pane): time.sleep(0.3) tail = _pane_tail(pane) @@ -3058,6 +3227,7 @@ def _pane_unavailable_reason(pane: str) -> str: return str(health.get("reason") or "provider_health_unavailable") tail = _pane_tail(pane) bottom = "\n".join(tail.splitlines()[-40:]) + overlay = pane_overlay_detail(tail) if pane_overlay_detail else {"state": "none", "type": ""} if PANE_RATE_LIMIT_OPTIONS_MODAL_RE.search(bottom): if _dismiss_rate_limit_options_modal(pane): tail = _pane_tail(pane) @@ -3083,10 +3253,12 @@ def _pane_unavailable_reason(pane: str) -> str: if PANE_TUI_UNAVAILABLE_RE.search(bottom): return "rate_limit_or_api_error" if PANE_SURVEY_PROMPT_RE.search(bottom): - if "❯" in bottom and not _pane_current_prompt_has_residue(bottom): + if overlay.get("state") == "stale_scrollback_ignored" or ("❯" in bottom and not _pane_current_prompt_has_residue(bottom)): return "" return "survey_prompt_blocked" if PANE_QUEUED_PROMPT_RE.search(bottom): + if overlay.get("state") == "stale_scrollback_ignored": + return "" if _clear_stale_prompt_residue(pane): tail = _pane_tail(pane) bottom = "\n".join(tail.splitlines()[-40:]) @@ -3116,7 +3288,18 @@ def _pane_hygiene_entries() -> dict[str, Any]: except Exception: return {} panes = payload.get("panes") - return panes if isinstance(panes, dict) else {} + if isinstance(panes, dict): + return panes + # Live registries may be stored as a flat map: + # {"session:win.pane": {"state": "needs_respawn", ...}}. + # Honor that shape so bad panes do not re-enter evaluator capacity. + if isinstance(payload, dict): + return { + str(key): value + for key, value in payload.items() + if isinstance(value, dict) and "state" in value + } + return {} def _recover_pane_hygiene_if_idle(pane: str, state: str) -> bool: @@ -3200,6 +3383,9 @@ def _pane_has_matching_queued_prompt(pane: str, instruction_file: Path) -> bool: def _pane_dispatch_prompt_reason(tail: str) -> str: bottom = "\n".join((tail or "").splitlines()[-40:]) + overlay = pane_overlay_detail(tail) if pane_overlay_detail else {"state": "none", "type": ""} + if overlay.get("state") == "stale_scrollback_ignored": + return "" edit_match = re.search(r"Do you want to make this edit|Do you want to overwrite|allow all edits during this session", bottom, re.I) if edit_match and not _prompt_match_followed_by_idle_default_prompt(bottom, edit_match): return "edit_confirmation_prompt" @@ -4990,7 +5176,8 @@ def _discover_evaluators(dry_run: bool = False) -> list[dict[str, Any]]: rate_limit_blocks = _persist_pane_rate_limit_block(pane, title, tail, quota_exhausted) if quota_exhausted else [] runtime_unavailable_reason = "" if cooldown_reason else _pane_runtime_unavailable_reason(pane, title) unavailable_reason = ( - cooldown_reason + _pane_hygiene_unavailable_reason(pane) + or cooldown_reason or _multi_task_direct_dispatch_unavailable_reason(pane, current_command=current_command) or runtime_unavailable_reason or _pane_unavailable_reason(pane) diff --git a/harness/tests/graph/test_graph_dispatch_submit.py b/harness/tests/graph/test_graph_dispatch_submit.py index b57bc9e6d..0aad334ce 100644 --- a/harness/tests/graph/test_graph_dispatch_submit.py +++ b/harness/tests/graph/test_graph_dispatch_submit.py @@ -273,6 +273,11 @@ def test_operator_completed_without_handoff_requeues_pending(self, tmp_harness, release_calls = [] monkeypatch.setattr(gnd, "release_lease", lambda *a, **k: release_calls.append(a) or {"released": True}) + monkeypatch.setattr( + gnd, + "_cooldown_operator_after_contract_closeout", + lambda operator_id, closeout: {"ok": True, "operator_id": operator_id, "cooldown_sec": 900}, + ) repaired = gnd._reconcile_existing_dispatches(graph, sprints / f"{sid}.task_graph.json") @@ -298,6 +303,11 @@ def test_operator_completed_without_handoff_requeues_pending(self, tmp_harness, "reason": "failed_contract_closeout", "operator_status": "completed", "result_json": str(result_json), + "operator_cooldown": { + "ok": True, + "operator_id": "mini-reasonix-deepseek-v4-flash-builder-1", + "cooldown_sec": 900, + }, } ] @@ -358,17 +368,18 @@ def test_eval_operator_completed_without_eval_json_clears_assignment(self, tmp_h "graph_eval_reconcile_failed_contract_closeout", ) ] - assert repaired == [ - { - "node": "N1", - "pane": "operator:mini-reasonix-deepseek-v4-builder", - "dispatch_id": f"graph-eval-{sid}-N1-q1", - "status": "reviewing", - "reason": "eval_failed_contract_closeout", - "operator_status": "completed", - "result_json": str(result_json), - } - ] + assert len(repaired) == 1 + item = repaired[0] + assert item["node"] == "N1" + assert item["pane"] == "operator:mini-reasonix-deepseek-v4-builder" + assert item["dispatch_id"] == f"graph-eval-{sid}-N1-q1" + assert item["status"] == "reviewing" + assert item["reason"] == "eval_failed_contract_closeout" + assert item["operator_status"] == "completed" + assert item["result_json"] == str(result_json) + assert item["operator_cooldown"]["ok"] is True + assert item["operator_cooldown"]["operator_id"] == "mini-reasonix-deepseek-v4-builder" + assert item["operator_cooldown"]["cooldown_sec"] == 900 def test_reconcile_accepts_lowercase_passed_eval_sidecar(self, tmp_harness, monkeypatch): """Evaluator sidecars may write verdict=passed; reconcile must still close the node.""" diff --git a/harness/tests/test-graph-dispatch-stale-scrollback.py b/harness/tests/test-graph-dispatch-stale-scrollback.py index 797434dfd..3471b15bb 100644 --- a/harness/tests/test-graph-dispatch-stale-scrollback.py +++ b/harness/tests/test-graph-dispatch-stale-scrollback.py @@ -2,7 +2,9 @@ from __future__ import annotations import importlib.util +import json import sys +import tempfile from pathlib import Path @@ -103,6 +105,26 @@ def main() -> int: mod._pane_tail = lambda pane, lines=80: confirmation_prompt assert mod._pane_tui_busy("solar-harness-lab:0.1") is True + stale_confirmation_prompt = """ + Bash command + + python3 scripts/check.py + + Do you want to proceed? + ❯ 1. Yes + 2. No + + Esc to cancel · Tab to amend + +──────────────────────────────────────────────────────────────── +❯ +──────────────────────────────────────────────────────────────── + ⏵⏵ bypass permissions on (shift+tab to cycle) · esc to interrupt +""" + mod._pane_tail = lambda pane, lines=80: stale_confirmation_prompt + assert mod._pane_unavailable_reason("solar-harness-lab:0.1") == "" + assert mod._pane_tui_busy("solar-harness-lab:0.1") is False + stale_busy_marker_with_empty_prompt = """ ⎿ ~/.solar/harness/lib/benchmark/schemas.py @@ -119,6 +141,109 @@ def main() -> int: mod._pane_tail = lambda pane, lines=80: stale_busy_marker_with_empty_prompt assert mod._pane_tui_busy("solar-harness:0.3") is False + with tempfile.TemporaryDirectory() as tmp: + original_harness = mod.HARNESS_DIR + original_sprints = mod.SPRINTS_DIR + tmp_path = Path(tmp) + mod.HARNESS_DIR = tmp_path + mod.SPRINTS_DIR = tmp_path / "sprints" + (mod.SPRINTS_DIR / "graph-acks").mkdir(parents=True) + released: list[tuple[str, str, str]] = [] + mod.read_lease = lambda pane: { + "dispatch_id": "dispatch-live-no-ack", + "expires_at": "2099-01-01T00:00:00Z", + "acquired_at": "2026-01-01T00:00:00Z", + } + mod.release_lease = lambda pane, dispatch_id, reason: released.append((pane, dispatch_id, reason)) or {"released": True} + mod._pane_title = lambda pane: "Builder | 状态:working/graph_node_idle_assigned" + mod._pane_tail = lambda pane, lines=80: live_prompt.replace("finalize sprint", "") + mod._pane_cooldown_reason = lambda pane: "" + mod._pane_runtime_unavailable_reason = lambda pane, title="": "" + mod._pane_unavailable_reason = lambda pane: "" + mod._pane_tui_busy = lambda pane: False + graph = { + "sprint_id": "sprint-live-lease-no-ack", + "nodes": [ + { + "id": "N1", + "status": "assigned", + "assigned_to": "solar-harness-lab:0.2", + "dispatch_id": "dispatch-live-no-ack", + } + ], + } + repaired = mod._reconcile_existing_dispatches(graph, tmp_path / "sprint-live-lease-no-ack.task_graph.json") + assert repaired and repaired[0]["reason"] == "live_lease_idle_without_submit_ack" + assert graph["nodes"][0]["status"] == "pending" + assert "assigned_to" not in graph["nodes"][0] + assert released[0] == ( + "solar-harness-lab:0.2", + "dispatch-live-no-ack", + "graph_dispatch_reconcile_live_lease_idle_without_submit_ack", + ) + mod.HARNESS_DIR = original_harness + mod.SPRINTS_DIR = original_sprints + + with tempfile.TemporaryDirectory() as tmp: + original_harness = mod.HARNESS_DIR + original_sprints = mod.SPRINTS_DIR + tmp_path = Path(tmp) + sid = "sprint-canonical-eval-cleanup" + mod.HARNESS_DIR = tmp_path + mod.SPRINTS_DIR = tmp_path / "sprints" + mod.SPRINTS_DIR.mkdir(parents=True) + (mod.SPRINTS_DIR / f"{sid}.N1-handoff.md").write_text("# handoff\n", encoding="utf-8") + (mod.SPRINTS_DIR / f"{sid}.N1-eval.json").write_text( + json.dumps({"verdict": "PASS", "status": "passed"}), + encoding="utf-8", + ) + graph = { + "sprint_id": sid, + "nodes": [ + { + "id": "N1", + "status": "passed", + "eval_retry_reason": "eval_failed_contract_closeout", + "last_eval_closeout_failure": {"reason": "eval_failed_contract_closeout"}, + "last_eval_operator_cooldown_after_closeout": {"ok": True}, + "eval_assigned_to": "operator:bad-evaluator", + "eval_dispatch_id": "eval-dispatch-stale", + } + ], + } + + repaired = mod._reconcile_existing_dispatches(graph, tmp_path / f"{sid}.task_graph.json") + node = graph["nodes"][0] + assert repaired and repaired[0]["reason"] == "canonical_eval_verdict_cleared_stale_eval_state" + assert node["status"] == "passed" + assert node["eval_json"].endswith(f"{sid}.N1-eval.json") + assert "eval_retry_reason" not in node + assert "last_eval_closeout_failure" not in node + assert "eval_assigned_to" not in node + assert "eval_dispatch_id" not in node + mod.HARNESS_DIR = original_harness + mod.SPRINTS_DIR = original_sprints + + with tempfile.TemporaryDirectory() as tmp: + original_harness = mod.HARNESS_DIR + tmp_path = Path(tmp) + mod.HARNESS_DIR = tmp_path + stale = tmp_path / "sprints" / "s01-req-N5-handoff.md" + stale.parent.mkdir(parents=True) + stale.write_text( + "sprint: `sprint-old-runtime`\n# old handoff\n", + encoding="utf-8", + ) + block = mod._write_scope_preflight_block( + "sprint-current-runtime", + {"write_scope": ["sprints/s01-req-N5-handoff.md"]}, + ) + assert "Write Scope Preflight" in block + assert "Treat them as stale inputs" in block + assert "sprint-old-runtime" in block + assert "contains_current_sprint=false" in block + mod.HARNESS_DIR = original_harness + print("PASS graph dispatcher ignores stale completed prompt scrollback") return 0 diff --git a/harness/tests/test_pm_dispatch.py b/harness/tests/test_pm_dispatch.py index c5199555c..293f78fc0 100644 --- a/harness/tests/test_pm_dispatch.py +++ b/harness/tests/test_pm_dispatch.py @@ -24,6 +24,12 @@ def _load_pm_dispatch(): return module +def test_normalize_role_maps_builder_main_to_builder(): + pm_dispatch = _load_pm_dispatch() + assert pm_dispatch.normalize_role("builder_main") == "builder" + assert pm_dispatch.normalize_role("builder-main") == "builder" + + def test_select_operator_by_role_prefers_capsule_operator_constraints(monkeypatch): pm_dispatch = _load_pm_dispatch() monkeypatch.setattr( @@ -211,3 +217,61 @@ def _unexpected_status(*args, **kwargs): rc = pm_dispatch.cmd_compile_request(args) assert rc == 2 assert touched["status"] is False + + +def test_reconcile_marks_planner_result_without_required_artifacts_failed(monkeypatch, tmp_path, capsys): + pm_dispatch = _load_pm_dispatch() + monkeypatch.setattr(pm_dispatch, "HARNESS_DIR", tmp_path) + monkeypatch.setattr(pm_dispatch, "SPRINTS_DIR", tmp_path / "sprints") + monkeypatch.setattr(pm_dispatch, "PM_INBOX_DIR", tmp_path / "run" / "pm-inbox") + monkeypatch.setattr(pm_dispatch, "_active_pm_task_ids", lambda: set()) + pm_dispatch.SPRINTS_DIR.mkdir(parents=True) + pm_dispatch.PM_INBOX_DIR.mkdir(parents=True) + result_path = pm_dispatch.SPRINTS_DIR / "sprint-demo.N0.pm-result.md" + result_path.write_text("# PM Task Result\n\n## 已完成\n- only summary\n", encoding="utf-8") + pm_dispatch.write_pm_task_record("pm-demo", { + "task_id": "pm-demo", + "sprint_id": "sprint-demo", + "node_id": "N0", + "requested_role": "planner", + "status": "submitted", + "result_path": str(result_path), + }) + + rc = pm_dispatch.cmd_reconcile(argparse.Namespace(max_age_minutes=30, apply=True, json=True, limit=40)) + + assert rc == 0 + updated = pm_dispatch.read_pm_task_record("pm-demo") + assert updated["status"] == "failed_contract_closeout" + assert updated["closeout_status"]["ok"] is False + assert str(pm_dispatch.SPRINTS_DIR / "sprint-demo.plan.md") in updated["closeout_status"]["missing_artifacts"] + assert "fail_contract_closeout" in capsys.readouterr().out + + +def test_reconcile_completes_planner_when_required_artifacts_exist(monkeypatch, tmp_path): + pm_dispatch = _load_pm_dispatch() + monkeypatch.setattr(pm_dispatch, "HARNESS_DIR", tmp_path) + monkeypatch.setattr(pm_dispatch, "SPRINTS_DIR", tmp_path / "sprints") + monkeypatch.setattr(pm_dispatch, "PM_INBOX_DIR", tmp_path / "run" / "pm-inbox") + monkeypatch.setattr(pm_dispatch, "_active_pm_task_ids", lambda: set()) + pm_dispatch.SPRINTS_DIR.mkdir(parents=True) + pm_dispatch.PM_INBOX_DIR.mkdir(parents=True) + result_path = pm_dispatch.SPRINTS_DIR / "sprint-demo.N0.pm-result.md" + result_path.write_text("# PM Task Result\n", encoding="utf-8") + (pm_dispatch.SPRINTS_DIR / "sprint-demo.plan.md").write_text("# Plan\n", encoding="utf-8") + (pm_dispatch.SPRINTS_DIR / "sprint-demo.task_graph.json").write_text('{"nodes":[]}\n', encoding="utf-8") + pm_dispatch.write_pm_task_record("pm-demo", { + "task_id": "pm-demo", + "sprint_id": "sprint-demo", + "node_id": "N0", + "requested_role": "planner", + "status": "submitted", + "result_path": str(result_path), + }) + + rc = pm_dispatch.cmd_reconcile(argparse.Namespace(max_age_minutes=30, apply=True, json=True, limit=40)) + + assert rc == 0 + updated = pm_dispatch.read_pm_task_record("pm-demo") + assert updated["status"] == "completed" + assert updated["closeout_status"]["ok"] is True diff --git a/harness/tools/pm_dispatch.py b/harness/tools/pm_dispatch.py index 250e0d9cf..9d1844b61 100755 --- a/harness/tools/pm_dispatch.py +++ b/harness/tools/pm_dispatch.py @@ -24,6 +24,7 @@ import shutil import subprocess import sys +import tempfile import textwrap import time import uuid @@ -47,6 +48,7 @@ # ── 角色别名映射 ─────────────────────────────────────────────────────────────── ROLE_ALIASES: dict[str, str] = { "build": "builder", + "builder-main": "builder", "implementation": "builder", "implementer": "builder", "coder": "builder", @@ -424,8 +426,8 @@ def _write_health_cache(operator_id: str, ok: bool, reason: str) -> None: "checked_at": _now(), "checked_at_epoch": time.time(), } - tmp = str(path) + ".tmp" - with open(tmp, "w", encoding="utf-8") as f: + fd, tmp = tempfile.mkstemp(prefix=f"{operator_id}.", suffix=".tmp", dir=str(path.parent)) + with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(payload, f, indent=2, ensure_ascii=False) os.replace(tmp, str(path)) @@ -1091,6 +1093,30 @@ def _active_pm_task_ids() -> set[str]: return active +def _pm_expected_artifacts(record: dict[str, Any]) -> list[Path]: + """Artifacts that prove a PM role task actually satisfied its contract.""" + role = normalize_role(str(record.get("requested_role") or "")) + sprint_id = str(record.get("sprint_id") or "").strip() + if not sprint_id: + return [] + if role == "planner": + return [ + SPRINTS_DIR / f"{sprint_id}.plan.md", + SPRINTS_DIR / f"{sprint_id}.task_graph.json", + ] + return [] + + +def _pm_closeout_status(record: dict[str, Any]) -> dict[str, Any]: + expected = _pm_expected_artifacts(record) + missing = [str(path) for path in expected if not path.exists() or path.stat().st_size <= 0] + return { + "ok": not missing, + "expected_artifacts": [str(path) for path in expected], + "missing_artifacts": missing, + } + + def _record_age_minutes(record: dict[str, Any], path: Path) -> float: for key in ("submitted_at", "created_at", "updated_at", "ts"): parsed = _parse_utc(str(record.get(key) or "")) @@ -1909,19 +1935,60 @@ def cmd_reconcile(args: argparse.Namespace) -> int: task_id = str(record.get("task_id") or path.stem) status = str(record.get("status") or "").strip() - if status in {"completed", "cancelled", "failed", "failed_missing_pm_result"}: + if status == "completed": + closeout = _pm_closeout_status(record) + if closeout.get("ok"): + continue + actions.append({ + "task_id": task_id, + "action": "fail_contract_closeout", + "reason": "completed_without_required_artifacts", + **closeout, + }) + if apply_changes: + record["task_id"] = task_id + record["status"] = "failed_contract_closeout" + record["failed_at"] = now + record["failure_reason"] = "completed_without_required_artifacts" + record["closeout_status"] = closeout + record.setdefault("reconcile_history", []).append( + {"ts": now, "action": "fail_contract_closeout", "reason": "completed_without_required_artifacts", **closeout} + ) + write_pm_task_record(task_id, record) + continue + if status in {"cancelled", "failed", "failed_missing_pm_result", "failed_contract_closeout"}: continue result_path = Path(str(record.get("result_path") or "")) result_exists = bool(str(result_path) and result_path.exists()) if result_exists: - actions.append({"task_id": task_id, "action": "complete", "reason": "result_path_exists"}) + closeout = _pm_closeout_status(record) + if not closeout.get("ok"): + actions.append({ + "task_id": task_id, + "action": "fail_contract_closeout", + "reason": "result_path_exists_but_required_artifacts_missing", + **closeout, + }) + if apply_changes: + record["task_id"] = task_id + record["status"] = "failed_contract_closeout" + record["failed_at"] = now + record["failure_reason"] = "result_path_exists_but_required_artifacts_missing" + record["closeout_status"] = closeout + record.setdefault("reconcile_history", []).append( + {"ts": now, "action": "fail_contract_closeout", "reason": "result_path_exists_but_required_artifacts_missing", **closeout} + ) + write_pm_task_record(task_id, record) + continue + actions.append({"task_id": task_id, "action": "complete", "reason": "result_path_exists", **closeout}) if apply_changes: record["task_id"] = task_id record["status"] = "completed" record["completed_at"] = now + record["closeout_status"] = closeout record.setdefault("reconcile_history", []).append( - {"ts": now, "action": "complete", "reason": "result_path_exists"} + {"ts": now, "action": "complete", "reason": "result_path_exists", **closeout} ) write_pm_task_record(task_id, record) continue