From 1b65375b46fc1645594fd949c36385552972f318 Mon Sep 17 00:00:00 2001 From: rctruta <17259677+rctruta@users.noreply.github.com> Date: Mon, 6 Jul 2026 09:04:27 -0400 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20stale=20running=20markers=20self-hea?= =?UTF-8?q?l=20=E2=80=94=20crashed=20executor=20no=20longer=20blocks=20res?= =?UTF-8?q?ubmission=20(TODO=20#12)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Observed live 2026-07-06: session teardown killed the API mid-execution of 209fc5df, leaving an orphaned running.json. check_registry would have refused every resubmission of that config forever ("already running — poll instead"); the marker had to be removed by hand. has_running_marker is now liveness-aware (single choke point; check_registry and the API's is_running inherit): - unreadable/corrupt marker -> stale - older than MAX_MARKER_AGE_SECONDS (6h; also caps PID-reuse window) -> stale - same host + recorded PID dead -> stale - different host within age -> conservatively alive (can't probe) Stale markers are removed on detection (self-heal) with a loud [WARN]. 6 regression tests incl. the exact 209fc5df scenario. --- TODO.md | 6 ++ sql_benchmarks/running_marker.py | 63 ++++++++++++++++++- sql_benchmarks_tests/test_running_marker.py | 67 +++++++++++++++++++++ 3 files changed, 134 insertions(+), 2 deletions(-) diff --git a/TODO.md b/TODO.md index b713f3e..1c2be9a 100644 --- a/TODO.md +++ b/TODO.md @@ -190,3 +190,9 @@ Session-start and session-end records with model, goal, outcome. Written to `--l ## 11. agent_runs/ is a flat pile — restructure as / directories **STATUS: OPEN** — surfaced 2026-07-06 (Ramona). All traces live flat in `sql_benchmarks/experiments/agent_runs/*.jsonl`. Target shape: `agent_runs//` holding everything pertaining to that run (trace, and later: sealed analysis, per-run manifest), mirroring the capsule convention in `results//`. For multi-agent runs the orchestrator's id is the natural directory; specialist traces live inside it. Deferred by choice — corpus collection takes priority; do the reorg before the corpus gets big enough for the move to be painful (analyzer + committed-trace paths need a migration in the same PR). + +## 12. Stale running markers — a crashed executor blocked resubmission forever + +**STATUS: SHIPPED 2026-07-06** — `has_running_marker` is now liveness-aware. Observed live: a session teardown killed the API mid-execution of `209fc5df`, leaving an orphaned `running.json`; `check_registry` would have answered "already running — poll instead" to every resubmission of that config, permanently, and the marker had to be removed by hand. + +Fix (single choke point — `running_marker.has_running_marker`, inherited by `check_registry` and the API's `is_running`): a marker only counts as running if it is readable, younger than `MAX_MARKER_AGE_SECONDS` (6h — also caps the PID-reuse window), and (same-host) its recorded PID is alive. Stale markers are removed on detection (self-heal) with a loud `[WARN]`. Different-host markers within the age window are conservatively treated as alive (can't probe a remote PID). 6 regression tests pin each rule, including the exact 209fc5df scenario (dead PID → not running → marker gone). diff --git a/sql_benchmarks/running_marker.py b/sql_benchmarks/running_marker.py index 25ae338..98997ca 100644 --- a/sql_benchmarks/running_marker.py +++ b/sql_benchmarks/running_marker.py @@ -69,8 +69,67 @@ def read_running_marker(results_dir: str, exp_id: str) -> Optional[dict]: return None -def has_running_marker(results_dir: str, exp_id: str) -> bool: - return os.path.exists(marker_path(results_dir, exp_id)) +# A run older than this is presumed dead regardless of PID state — also +# bounds the PID-reuse false-alive window. Generous: real experiments run +# minutes to ~an hour. +MAX_MARKER_AGE_SECONDS = 6 * 3600 + + +def _pid_alive(pid) -> bool: + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + except PermissionError: + return True # exists, owned by someone else + except (OverflowError, ValueError, TypeError): + return False # garbage pid in the marker + + +def has_running_marker(results_dir: str, exp_id: str, + max_age_seconds: float = MAX_MARKER_AGE_SECONDS) -> bool: + """True only if the marker exists AND the run it describes is plausibly + alive. A crashed executor (killed API process, torn-down session) leaves + an orphaned marker that would otherwise block resubmission of the same + config forever — observed live 2026-07-06 (capsule 209fc5df: session + teardown killed the API mid-execution; the stale marker had to be + removed by hand). See TODO.md #12. + + Staleness rules, in order: + - unreadable/corrupt marker -> stale + - older than max_age_seconds -> stale (any host; also caps + the PID-reuse window) + - same host and recorded PID not alive -> stale + - different host, within age -> assumed alive (can't probe) + + Stale markers are REMOVED (self-heal) with a loud warning, so the + status endpoint and check_registry recover without operator surgery.""" + payload = read_running_marker(results_dir, exp_id) + if payload is None: + # Missing entirely, or unreadable. If the file exists but can't be + # parsed, it can't testify that anything is running — remove it. + if os.path.exists(marker_path(results_dir, exp_id)): + print(f"[WARN] corrupt running marker for {exp_id} — removing (stale)") + clear_running_marker(results_dir, exp_id) + return False + + age = time.time() - float(payload.get("started_at") or 0) + if age > max_age_seconds: + print(f"[WARN] running marker for {exp_id} is {age/3600:.1f}h old " + f"(max {max_age_seconds/3600:.1f}h) — presumed dead, removing") + clear_running_marker(results_dir, exp_id) + return False + + if payload.get("hostname") == socket.gethostname(): + pid = payload.get("pid") + if not _pid_alive(pid): + print(f"[WARN] running marker for {exp_id} names dead pid {pid} — " + "executor crashed; removing stale marker") + clear_running_marker(results_dir, exp_id) + return False + + return True def clear_running_marker(results_dir: str, exp_id: str) -> None: diff --git a/sql_benchmarks_tests/test_running_marker.py b/sql_benchmarks_tests/test_running_marker.py index 3d4d8fb..3c78a7d 100644 --- a/sql_benchmarks_tests/test_running_marker.py +++ b/sql_benchmarks_tests/test_running_marker.py @@ -89,3 +89,70 @@ def test_write_overwrites_existing(tmp_path): write_running_marker(str(tmp_path), "abc12345") second = read_running_marker(str(tmp_path), "abc12345") assert second["started_at"] >= first["started_at"] + + +# --------------------------------------------------------------------------- +# Staleness detection (TODO #12) — a crashed executor must not block +# resubmission forever. Observed live 2026-07-06: session teardown killed +# the API mid-execution of 209fc5df; the orphaned marker required manual +# removal. +# --------------------------------------------------------------------------- + +import json +import time as _time + +from sql_benchmarks.running_marker import has_running_marker, marker_path + + +def _write_marker(tmp_path, exp_id, **overrides): + import socket + exp_dir = os.path.join(str(tmp_path), exp_id) + os.makedirs(exp_dir, exist_ok=True) + payload = {"experiment_id": exp_id, "started_at": _time.time(), + "pid": os.getpid(), "hostname": socket.gethostname()} + payload.update(overrides) + with open(os.path.join(exp_dir, "running.json"), "w") as f: + json.dump(payload, f) + + +def test_live_marker_is_running(tmp_path): + """Fresh marker, this process's PID, this host -> running.""" + _write_marker(tmp_path, "aaaa0001") + assert has_running_marker(str(tmp_path), "aaaa0001") is True + assert os.path.exists(marker_path(str(tmp_path), "aaaa0001")) # untouched + + +def test_dead_pid_marker_is_stale_and_self_heals(tmp_path): + """The 209fc5df case: executor crashed, PID gone. Marker must read as + not-running AND be removed so resubmission proceeds.""" + _write_marker(tmp_path, "aaaa0002", pid=99999999) # no such pid + assert has_running_marker(str(tmp_path), "aaaa0002") is False + assert not os.path.exists(marker_path(str(tmp_path), "aaaa0002")) + + +def test_overage_marker_is_stale_even_with_live_pid(tmp_path): + """Age cap bounds the PID-reuse window: a marker older than max age is + dead even if its PID happens to be alive.""" + _write_marker(tmp_path, "aaaa0003", started_at=_time.time() - 7 * 3600) + assert has_running_marker(str(tmp_path), "aaaa0003") is False + assert not os.path.exists(marker_path(str(tmp_path), "aaaa0003")) + + +def test_other_host_within_age_assumed_alive(tmp_path): + """Can't probe a PID on another machine — within the age window we + stay conservative and treat it as running.""" + _write_marker(tmp_path, "aaaa0004", hostname="some-other-box", pid=1) + assert has_running_marker(str(tmp_path), "aaaa0004") is True + + +def test_corrupt_marker_is_stale_and_removed(tmp_path): + exp_dir = os.path.join(str(tmp_path), "aaaa0005") + os.makedirs(exp_dir) + with open(os.path.join(exp_dir, "running.json"), "w") as f: + f.write("{broken") + assert has_running_marker(str(tmp_path), "aaaa0005") is False + assert not os.path.exists(marker_path(str(tmp_path), "aaaa0005")) + + +def test_missing_marker_is_not_running(tmp_path): + assert has_running_marker(str(tmp_path), "aaaa0006") is False From cbda19a6eaffaeef1b90d1c68790a9694d8938f9 Mon Sep 17 00:00:00 2001 From: rctruta <17259677+rctruta@users.noreply.github.com> Date: Mon, 6 Jul 2026 09:31:14 -0400 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20SUSPENDED=20state=20+=20resume=20?= =?UTF-8?q?=E2=80=94=20the=20capsule=20is=20the=20durable=20hand-off?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Edge-4's rerun proved no poll budget is "right" for open-ended workloads: fable-5 correctly chose a 16GB out-of-memory sort that outlived even a 30-minute budget while executing legitimately. Synchronous waiting conflates "slow" with "dead". - Poll timeout while the experiment still runs -> outcome "suspended" (non-error, experiment_id preserved). "poll_failed" now means the experiment actually failed or polling errored. - Orchestrator.resume(exp_id): skip config_builder, poll, run analyzer. Callable minutes later, next session, or after a crash — the content-addressed capsule is the durable hand-off point. - CLI: multi_agent.py --resume + --poll-budget-seconds. 4 new tests (31 total): suspend-not-fail on timeout, resume completes when ready, resume re-suspends while running, resume surfaces real execution failures. Ships with (same branch): structured REFUSAL state, contract-declared poll_budget_seconds, liveness-aware self-healing running markers (TODO #12). Decisions-log entry added. --- docs/decisions_log.md | 14 +++ scripts/multi_agent.py | 21 +++- scripts/run_study.py | 4 +- sql_benchmarks/agent_orchestrator.py | 106 ++++++++++++++-- .../studies/edge4_rerun_budget.yaml | 28 +++++ sql_benchmarks_tests/test_orchestrator.py | 118 ++++++++++++++++++ 6 files changed, 279 insertions(+), 12 deletions(-) create mode 100644 sql_benchmarks/experiments/studies/edge4_rerun_budget.yaml diff --git a/docs/decisions_log.md b/docs/decisions_log.md index 83894a8..347fa9d 100644 --- a/docs/decisions_log.md +++ b/docs/decisions_log.md @@ -336,3 +336,17 @@ Human-and-agent parity: `sqlbench project means ` (CLI), **Decision (own repo?).** The harness (`agent_tools/specialist/orchestrator/trace`, `run_study`, analyzer, grader) stays IN this repo until a second consumer exists. Extraction trigger: ai-security-testbed importing it, or the article requiring a citable standalone artifact. Extracting now would freeze the API exactly while the edge-case studies are telling us which states/tools are missing (e.g., the refusal state that edge-3 is expected to surface). **Cross-refs.** Edge-case contracts `edge1_novel_goal`, `edge3_adversarial_goal`, `edge4_ambiguous_goal`; run_study schema v2 (`models:` list — weak→strong ladder in one contract). + +--- + +## 2026-07-06 — SUSPENDED state + resume: the capsule is the durable hand-off + +**Decision.** Poll timeout while the experiment is still executing is now `suspended`, not `poll_failed` — a first-class non-error outcome carrying the experiment_id. `Orchestrator.resume(exp_id)` (CLI: `multi_agent.py --resume `) picks up at the analyzer once the capsule completes, for a few thousand tokens. Also this session: structured REFUSAL state (`HANDOFF: impossible reason=…`), contract-declared `poll_budget_seconds`, and liveness-aware running markers (TODO #12 — self-healing on dead PID / over-age / corrupt). + +**Fork closed.** Tuning poll budgets per suite. Edge-4's rerun proved no budget is "right": fable-5 correctly chose a 16GB out-of-memory sort that outlived even a 30-minute budget *while executing legitimately*. Synchronous waiting conflates "slow" with "dead". + +**Fork opened.** Long-running experiments become first-class: submit → suspend → resume across sessions or crashes. The capsule (content-addressed, server-side, durable) is the hand-off point — the same object that already anchors reproducibility now anchors continuity. Pairs with the liveness-aware markers: "dead" is now detected by evidence (PID, age), never inferred from silence. + +**Live validation of the recovery story (same day).** A session teardown killed the runner mid-poll and the API mid-execution. Traces survived (append-per-event JSONL); the stale running marker was the only manual surgery — and that class is now self-healing. + +**Cross-refs.** TODO #12, edge-3/edge-4 studies, Finding 18/19, `[[work-before-the-work]]` (states uncovered by pre-deployment probing — exactly the deliverable of the role Ramona is naming). diff --git a/scripts/multi_agent.py b/scripts/multi_agent.py index 0da5404..87b9307 100644 --- a/scripts/multi_agent.py +++ b/scripts/multi_agent.py @@ -42,16 +42,24 @@ def main(): help="litellm model identifier") parser.add_argument("--goal", type=str, required=True, help="Natural-language goal for the orchestrator") + parser.add_argument("--resume", metavar="EXP_ID", + help="Resume a SUSPENDED run: skip config_builder, poll " + "this experiment_id, then run the analyzer. The " + "goal is still required (the analyzer needs it).") + parser.add_argument("--poll-budget-seconds", type=float, default=180.0, + help="How long the poll stage waits before suspending.") args = parser.parse_args() console.print(Panel(f"[bold cyan]GOAL:[/bold cyan] {args.goal}", title="🤖 Multi-Agent Orchestrator")) - console.print(f"[dim]Model: {args.model}[/dim]") + console.print(f"[dim]Model: {args.model}" + + (f" | resuming {args.resume}" if args.resume else "") + "[/dim]") - orch = Orchestrator(goal=args.goal, model=args.model) + orch = Orchestrator(goal=args.goal, model=args.model, + poll_budget_seconds=args.poll_budget_seconds) console.print(f"[dim]Orchestrator trace: {orch.trace.path}[/dim]\n") - result = orch.run() + result = orch.resume(args.resume) if args.resume else orch.run() console.print(f"\n[bold]Outcome:[/bold] {result.outcome}") console.print(f"[dim]experiment_id: {result.experiment_id}[/dim]") @@ -61,6 +69,13 @@ def main(): if result.outcome == "complete" and result.analysis: console.print(Panel(Markdown(result.analysis), title="📊 Analysis", border_style="green")) + elif result.outcome == "suspended": + console.print( + f"\n[yellow]SUSPENDED[/yellow] — experiment {result.experiment_id} is still " + f"executing server-side. Resume any time with:\n" + f" python scripts/multi_agent.py --resume {result.experiment_id} " + f"--goal --model {args.model}" + ) elif result.error: console.print(f"\n[bold red]Error:[/bold red] {result.error}") sys.exit(1) diff --git a/scripts/run_study.py b/scripts/run_study.py index 89e7247..866ff25 100644 --- a/scripts/run_study.py +++ b/scripts/run_study.py @@ -83,7 +83,9 @@ def run_cell_rep(contract: dict, study_id: str, cell_name: str, rep: int, return "ran" # run_agent prints its own outcome; trace has run_end else: # multi_agent from sql_benchmarks.agent_orchestrator import Orchestrator - orch = Orchestrator(goal=contract["goal"], model=model) + orch = Orchestrator(goal=contract["goal"], model=model, + poll_budget_seconds=float( + contract.get("poll_budget_seconds", 180))) orch.trace.prompt_provenance(components={}, ablation_flags={ "architecture": "orchestrator", **study_stamp}) result = orch.run() diff --git a/sql_benchmarks/agent_orchestrator.py b/sql_benchmarks/agent_orchestrator.py index 54cc017..827ed6b 100644 --- a/sql_benchmarks/agent_orchestrator.py +++ b/sql_benchmarks/agent_orchestrator.py @@ -52,6 +52,16 @@ HANDOFF: experiment_id=<8-hex> +If the goal CANNOT be satisfied with the engines and suites this lab +actually has (e.g. it names a database that is not in the catalog, or +asks for a workload no suite covers), DO NOT substitute something else +and do not force a submission. Produce instead: + + HANDOFF: impossible reason= + +That is a successful outcome — an honest refusal is worth more than a +silently substituted experiment. + That signals you're done. Do not analyze results — that's a different specialist. Do not produce a narrative — just the HANDOFF line. """ @@ -121,13 +131,20 @@ _HANDOFF_RE = re.compile(r"HANDOFF:\s*experiment_id=([0-9a-f]{8})", re.IGNORECASE) +_REFUSAL_RE = re.compile(r"HANDOFF:\s*impossible\s+reason=(.+)", re.IGNORECASE) def _parse_handoff(content: str) -> Optional[dict]: - """The config_builder signals completion with `HANDOFF: experiment_id=<8hex>`.""" + """config_builder completion signals: `HANDOFF: experiment_id=<8hex>` + (built) or `HANDOFF: impossible reason=` (structured refusal — + the state edge-3 showed was missing: without it, honest models thrash + through mechanical failures to say 'can't').""" m = _HANDOFF_RE.search(content or "") if m: return {"experiment_id": m.group(1)} + r = _REFUSAL_RE.search(content or "") + if r: + return {"impossible": r.group(1).strip()} return None @@ -167,7 +184,7 @@ def poll_until_terminal(experiment_id: str, @dataclass class OrchestratorResult: - outcome: str # "complete" | "config_builder_failed" | "poll_failed" | "analyzer_failed" + outcome: str # "complete" | "refused" | "suspended" | "config_builder_failed" | "poll_failed" | "analyzer_failed" experiment_id: Optional[str] analysis: Optional[str] orchestrator_run_id: str @@ -176,16 +193,24 @@ class OrchestratorResult: class Orchestrator: - """Runs the state machine: config_builder → poll → analyzer. + """Runs the state machine: config_builder → poll → analyzer, with a + REFUSED terminal state (config_builder judged the goal unsatisfiable + with the lab's actual engines/suites — a first-class honest outcome, + not a failure). Each stage's specialist gets its own JSONL trace. The orchestrator's own trace records `delegate` events pointing at each specialist's `sub_run_id`, so a reader can walk the multi-agent tree by opening - the orchestrator trace and following the sub_run_ids.""" + the orchestrator trace and following the sub_run_ids. + + `poll_budget_seconds` is contract-declarable: edge-4 showed the fixed + 180s default silently kills legitimately slow suites (sort_spill) — + uniform cross-model 'failure' that was actually a harness defect.""" - def __init__(self, goal: str, model: str): + def __init__(self, goal: str, model: str, poll_budget_seconds: float = 180.0): self.goal = goal self.model = model + self.poll_budget_seconds = poll_budget_seconds self.trace = AgentTrace( goal=f"[orchestrator] {goal}", model=model, agents_md_loaded=False, max_turns=0, @@ -202,6 +227,15 @@ def run(self) -> OrchestratorResult: input_summary=self.goal[:200], outcome=cb.outcome, output_summary=(json.dumps(cb.output) if cb.output else None), ) + if cb.outcome == "ok" and cb.output and "impossible" in cb.output: + # Structured refusal — terminal, honest, cheap. Not a failure. + reason = cb.output["impossible"] + self.trace.run_end("refused", turns_used=0, error=None) + return OrchestratorResult( + outcome="refused", experiment_id=None, + analysis=f"REFUSED: {reason}", + orchestrator_run_id=self.trace.run_id, sub_run_ids=sub_ids, + ) if cb.outcome != "ok" or not cb.output or "experiment_id" not in cb.output: self.trace.run_end("config_builder_failed", turns_used=0, error=cb.error) return OrchestratorResult( @@ -211,14 +245,69 @@ def run(self) -> OrchestratorResult: ) exp_id = cb.output["experiment_id"] - # Stage 2: pure-Python poller - poll_result = poll_until_terminal(exp_id) + # Stage 2: pure-Python poller (budget contract-declarable) + interval = 3.0 + poll_result = poll_until_terminal( + exp_id, + max_polls=max(1, int(self.poll_budget_seconds / interval)), + interval_seconds=interval, + ) self.trace.delegate( stage="poll", sub_run_id=None, input_summary=f"experiment_id={exp_id}", outcome=poll_result["status"], output_summary=json.dumps(poll_result), ) + if poll_result["status"] == "timeout": + # SUSPENDED, not failed: the experiment is still executing + # server-side and will finish regardless of who's watching. + # No poll budget is "right" for open-ended workloads (edge-4: + # a legitimate 16GB out-of-memory sort outlived a 30-minute + # budget). The run is resumable: `Orchestrator.resume(exp_id)` + # picks up at the analyzer for a few thousand tokens once the + # capsule completes. + self.trace.run_end("suspended", turns_used=0, error=None) + return OrchestratorResult( + outcome="suspended", experiment_id=exp_id, analysis=None, + orchestrator_run_id=self.trace.run_id, sub_run_ids=sub_ids, + error=None, + ) + if poll_result["status"] != "complete": + self.trace.run_end("poll_failed", turns_used=0, error=str(poll_result)) + return OrchestratorResult( + outcome="poll_failed", experiment_id=exp_id, analysis=None, + orchestrator_run_id=self.trace.run_id, sub_run_ids=sub_ids, + error=f"poll returned {poll_result['status']}: {poll_result.get('detail') or poll_result.get('error')}", + ) + + return self._analyzer_stage(exp_id, sub_ids) + + def resume(self, exp_id: str) -> OrchestratorResult: + """Pick up a SUSPENDED run: skip config_builder (the experiment + already exists), check/poll the capsule, then run the analyzer. + Cheap — the analyzer is a few thousand tokens; polling is free. + + Callable any time after suspension: minutes later, next session, + or after a crash — the capsule is the durable hand-off point.""" + sub_ids: dict = {"config_builder": None} # skipped on resume + interval = 3.0 + poll_result = poll_until_terminal( + exp_id, + max_polls=max(1, int(self.poll_budget_seconds / interval)), + interval_seconds=interval, + ) + self.trace.delegate( + stage="poll", sub_run_id=None, + input_summary=f"experiment_id={exp_id} (resume)", + outcome=poll_result["status"], + output_summary=json.dumps(poll_result), + ) + if poll_result["status"] == "timeout": + self.trace.run_end("suspended", turns_used=0, error=None) + return OrchestratorResult( + outcome="suspended", experiment_id=exp_id, analysis=None, + orchestrator_run_id=self.trace.run_id, sub_run_ids=sub_ids, + ) if poll_result["status"] != "complete": self.trace.run_end("poll_failed", turns_used=0, error=str(poll_result)) return OrchestratorResult( @@ -226,8 +315,9 @@ def run(self) -> OrchestratorResult: orchestrator_run_id=self.trace.run_id, sub_run_ids=sub_ids, error=f"poll returned {poll_result['status']}: {poll_result.get('detail') or poll_result.get('error')}", ) + return self._analyzer_stage(exp_id, sub_ids) - # Stage 3: analyzer + def _analyzer_stage(self, exp_id: str, sub_ids: dict) -> OrchestratorResult: # Fetch the config's `definitions` block server-side (no LLM tokens) # so the analyzer knows what the partition labels MEAN (row counts). # Run-4 lesson: without this the analyzer had to assume "large is diff --git a/sql_benchmarks/experiments/studies/edge4_rerun_budget.yaml b/sql_benchmarks/experiments/studies/edge4_rerun_budget.yaml new file mode 100644 index 0000000..e659e92 --- /dev/null +++ b/sql_benchmarks/experiments/studies/edge4_rerun_budget.yaml @@ -0,0 +1,28 @@ +# Edge 4 RERUN with the two states shipped after the first execution: +# - poll_budget_seconds (contract-declared; first run died at the fixed +# 180s default while sort_spill was still executing — Finding 19) +# - refusal state exists but should NOT trigger here (goal is valid). +# +# Cost-conscious design (Ramona, 2026-07-06): n=1 per model, two models +# only — this verifies a GATE, not a distribution. Replication buys +# nothing for defect-fix verification. + +meta: + name: "Edge 4 rerun — suite-aware poll budget verification" + description: > + Same ambiguous sort_spill goal; poll budget raised to 30 min. + Success = poll survives to terminal state and analyzer produces a + graded answer. + +driver: multi_agent +models: [anthropic/claude-fable-5, gemini/gemini-2.5-flash] +replications: 1 +poll_budget_seconds: 1800 + +goal: > + I have a feeling DuckDB gets much slower when it has to sort data that + doesn't fit in memory. Can you check whether that's true? + +cells: + orchestrated: + flags: {} diff --git a/sql_benchmarks_tests/test_orchestrator.py b/sql_benchmarks_tests/test_orchestrator.py index 03020cc..601dd18 100644 --- a/sql_benchmarks_tests/test_orchestrator.py +++ b/sql_benchmarks_tests/test_orchestrator.py @@ -426,3 +426,121 @@ def fake_completion(model, messages, tools, tool_choice): stop_msgs = [m for m in captured_messages if m.get("role") == "user" and "STOP" in (m.get("content") or "")] assert stop_msgs, "expected escalating STOP coaching after repeated identical failures" + + +# --------------------------------------------------------------------------- +# Refusal state + contract-declared poll budget (from edge-case studies 3+4) +# --------------------------------------------------------------------------- + +def test_parse_handoff_impossible(): + out = _parse_handoff("HANDOFF: impossible reason=MongoDB is not an available engine") + assert out == {"impossible": "MongoDB is not an available engine"} + # experiment_id form still wins when present + assert _parse_handoff("HANDOFF: experiment_id=abcd1234") == {"experiment_id": "abcd1234"} + + +def test_orchestrator_refusal_is_terminal_honest_and_cheap(): + """config_builder's structured refusal ends the run: no poll, no + analyzer, outcome='refused' (not a failure), reason surfaced.""" + with patch.object(agent_orchestrator, "run_specialist") as mock_run_spec, \ + patch.object(agent_orchestrator, "poll_until_terminal") as mock_poll: + mock_run_spec.return_value = SpecialistResult( + role="config_builder", outcome="ok", + output={"impossible": "no MongoDB engine in the catalog"}, + sub_run_id="cb-run", turns_used=3, + ) + result = Orchestrator(goal="benchmark MongoDB", model="test/model").run() + + assert result.outcome == "refused" + assert result.experiment_id is None + assert "no MongoDB engine" in result.analysis + assert result.error is None # refusal is not an error + assert mock_poll.call_count == 0 # never polled + assert mock_run_spec.call_count == 1 # analyzer never invoked + + +def test_poll_budget_is_contract_declarable(): + """Edge-4 defect: fixed 180s budget killed slow suites. The budget + must flow Orchestrator(poll_budget_seconds=...) -> poll max_polls.""" + with patch.object(agent_orchestrator, "run_specialist") as mock_run_spec, \ + patch.object(agent_orchestrator, "poll_until_terminal") as mock_poll: + mock_run_spec.side_effect = [ + SpecialistResult(role="config_builder", outcome="ok", + output={"experiment_id": "aabb0007"}, + sub_run_id="cb", turns_used=1), + SpecialistResult(role="analyzer", outcome="ok", + output={"analysis": "FINAL ANSWER: x" * 20}, + sub_run_id="an", turns_used=1), + ] + mock_poll.return_value = {"status": "complete", "polls": 1} + Orchestrator(goal="g", model="m", poll_budget_seconds=1800).run() + + _, kwargs = mock_poll.call_args + assert kwargs["max_polls"] == 600 # 1800s / 3s interval + + +# --------------------------------------------------------------------------- +# Suspended state + resume (edge-4 follow-through: no poll budget is "right" +# for open-ended workloads — a 16GB out-of-memory sort outlived a 30-min +# budget while executing legitimately; the capsule is the durable hand-off) +# --------------------------------------------------------------------------- + +def test_poll_timeout_suspends_not_fails(): + """Timeout while the experiment is still executing -> outcome + 'suspended' (not an error), experiment_id preserved for resume, + analyzer NOT invoked (its tokens are spent later, on resume).""" + with patch.object(agent_orchestrator, "run_specialist") as mock_run_spec, \ + patch.object(agent_orchestrator, "poll_until_terminal") as mock_poll: + mock_run_spec.return_value = SpecialistResult( + role="config_builder", outcome="ok", + output={"experiment_id": "aabb0008"}, + sub_run_id="cb", turns_used=2) + mock_poll.return_value = {"status": "timeout", "polls": 600} + result = Orchestrator(goal="g", model="m", poll_budget_seconds=1800).run() + + assert result.outcome == "suspended" + assert result.experiment_id == "aabb0008" + assert result.error is None # suspension is not an error + assert mock_run_spec.call_count == 1 # analyzer never ran + + +def test_resume_completes_when_capsule_ready(): + """resume(exp_id) skips config_builder entirely: poll -> analyzer.""" + with patch.object(agent_orchestrator, "run_specialist") as mock_run_spec, \ + patch.object(agent_orchestrator, "poll_until_terminal") as mock_poll: + mock_poll.return_value = {"status": "complete", "polls": 1} + mock_run_spec.return_value = SpecialistResult( + role="analyzer", outcome="ok", + output={"analysis": "FINAL ANSWER: spill confirmed"}, + sub_run_id="an", turns_used=3) + with patch.object(agent_orchestrator.httpx, "get") as mock_get: + mock_get.return_value.json.return_value = {"config": {}} + result = Orchestrator(goal="g", model="m").resume("aabb0009") + + assert result.outcome == "complete" + assert result.analysis == "FINAL ANSWER: spill confirmed" + assert result.sub_run_ids["config_builder"] is None # skipped on resume + assert mock_run_spec.call_count == 1 # analyzer only + + +def test_resume_resuspends_if_still_running(): + with patch.object(agent_orchestrator, "run_specialist") as mock_run_spec, \ + patch.object(agent_orchestrator, "poll_until_terminal") as mock_poll: + mock_poll.return_value = {"status": "timeout", "polls": 60} + result = Orchestrator(goal="g", model="m").resume("aabb000a") + assert result.outcome == "suspended" + assert result.experiment_id == "aabb000a" + assert mock_run_spec.call_count == 0 + + +def test_resume_surfaces_execution_failure(): + """A capsule that FAILED while suspended must come back as poll_failed + with the failure detail, not as suspended.""" + with patch.object(agent_orchestrator, "run_specialist") as mock_run_spec, \ + patch.object(agent_orchestrator, "poll_until_terminal") as mock_poll: + mock_poll.return_value = {"status": "failed", "polls": 1, + "detail": "[execute] out of disk"} + result = Orchestrator(goal="g", model="m").resume("aabb000b") + assert result.outcome == "poll_failed" + assert "out of disk" in result.error + assert mock_run_spec.call_count == 0