diff --git a/docs/deep-dive.md b/docs/deep-dive.md index b87d190..b2a5be6 100644 --- a/docs/deep-dive.md +++ b/docs/deep-dive.md @@ -254,7 +254,7 @@ A stable system is not one that claims to have no edges — it is one whose edge - **`.env` and `grapharc.toml` follow the same discovery rule: the working directory, and nowhere else.** Neither searches parent directories — a run must not be governed by a file you did not know about, and must not be *billed* to one either. **This is a behaviour change:** the credential loader used to walk up to `/`, so a `.env` in an ancestor directory (a `$HOME` one on a shared box, a client project one above a demo checkout) was picked up silently. If you relied on that, move the file into the directory you run from, `export` the variable, or pass `env_file=` to name it explicitly. A real environment variable still beats any file. - **`grapharc run` has no budget unless you give it one.** Set any of `--max-tokens`, `--max-iterations`, `--max-seconds`, or `--max-concurrency`; without them each dimension is unlimited and the gate admits a topology of any worst-case cost. -**Verified this pass:** `pytest` → green, 2,156 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.7` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. +**Verified this pass:** `pytest` → green, 2,180 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.7` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. [ROADMAP.md](../ROADMAP.md) tracks what is built and what is not, item by item. diff --git a/grapharc/cli/plan.py b/grapharc/cli/plan.py index 57484fb..b9b2f75 100644 --- a/grapharc/cli/plan.py +++ b/grapharc/cli/plan.py @@ -233,6 +233,38 @@ def _executed_run_ids(record: dict[str, Any]) -> list[str]: return [str(scalar)] if scalar else [] + +def _unfinished_execution(trace_path: Path, record: dict[str, Any]) -> str | None: + """A run whose nodes ran but which `plan.json` never recorded finishing. + + `executed_run_id` is stamped *after* `loop.run()` returns, so a run that + is killed partway through never reaches the stamp: the MCP `execute` + work budget expiring, a SIGKILL, an OOM-kill. The record then says the + plan was never executed while the tree may already have been changed, + and the #100 reissue guard waves a second run through on the strength of + one human approval -- which is the property #100 was closed to protect. + + The trace is written *as the run proceeds*, so it is the only place that + evidence survives a kill. Returns the offending run id, or None. + """ + if not trace_path.is_file(): + return None + from grapharc.observe.trace import TraceReadError, TraceRecorder, began_execution + + known = {str(r) for r in _executed_run_ids(record)} + try: + events = TraceRecorder(trace_path).read_events() + except (OSError, TraceReadError): + # A torn or unreadable trace is not evidence of a half-run, and + # refusing here would wedge every plan sitting beside a damaged file. + # `grapharc trace` is what reports a trace that cannot be read. + return None + for event in events: + if event.run_id not in known and began_execution(event.phase): + return event.run_id + return None + + def find_unexecuted_plan(runs_root: Path | None = None) -> Path | None: """The newest saved plan `go` has not executed yet, or None.""" import json @@ -395,6 +427,25 @@ def execute_plan( run_dir = plan_file.parent trace_path = run_dir / "trace.jsonl" + # A previous attempt that began and never recorded finishing. Checked + # after the `executed_run_id` guard above and separately from it, because + # the two are different facts: that one is "this ran, cleanly, once", this + # one is "something ran and we do not know how far it got". + unfinished = _unfinished_execution(trace_path, record) + if unfinished and not again: + return fail( + f"{plan_file} has an attempt, run {unfinished}, that began executing " + "and never recorded finishing — it may have changed the tree partway " + f"through. What it did reach is in the trace: `grapharc trace {trace_path}`. " + "Pass --again to run the plan a second time anyway; an approval binds " + "to the plan's fingerprint, which does not change between runs, so a " + "re-run of a mutating plan spends an earlier yes.", + as_json=as_json, + command="go", + plan=str(plan_file), + unfinished_run_id=str(unfinished), + ) + try: settings = load_settings(config_path) model_spec = settings.resolve("model", model_spec, record.get("model")) diff --git a/grapharc/mcp/driver.py b/grapharc/mcp/driver.py index 8bd8f97..d022558 100644 --- a/grapharc/mcp/driver.py +++ b/grapharc/mcp/driver.py @@ -21,18 +21,62 @@ from __future__ import annotations import asyncio +import contextlib import json +import os +import signal import sys from pathlib import Path from typing import Any PLAN_FILENAME = "plan.json" -#: A parked `execute` lives inside one MCP call, and hosts time tool calls -#: out. The default stays under typical host limits; a timeout leaves the -#: plan unexecuted and the call safe to reissue. +#: How long a parked `execute` waits for a human to answer. A parked call +#: lives inside one MCP call and hosts time tool calls out, so the default +#: stays under typical host limits. This bounds *the wait*, nothing else: +#: a park that expires without an answer leaves the plan unexecuted, and +#: only that case is safe to reissue. DEFAULT_APPROVAL_TIMEOUT = 240.0 +#: How long the *work* gets, once a plan is admitted and (if mutating) a +#: human has said yes. Separate from the park on purpose. These used to be +#: one budget of `approval_timeout + 120s` covering both, which meant a human +#: approving near the end of the park left roughly two minutes for the run +#: itself — and a governed run's agent phases delegate to Claude Code, which +#: reads files, edits them and verifies. Killing an approved run at 120s does +#: not protect anything; it severs an approved run partway through its work, +#: which for a mutating plan means partway through mutating the tree. +#: +#: 1800s matches `GRAPHARC_SLACK_WORK_TIMEOUT`, which the Slack path carved +#: out of one shared timeout for this exact reason. Overridable because the +#: right ceiling is a property of the operator's machine, not of this file. +DEFAULT_WORK_TIMEOUT = 1800.0 + + +def work_timeout(env: dict[str, str] | None = None) -> float: + """The work budget, from `GRAPHARC_MCP_WORK_TIMEOUT` or the default. + + Refuses a non-numeric or non-positive override rather than falling back + to the default: an operator who sets this has a ceiling in mind, and + silently substituting a different one is how a run gets killed at a + limit nobody chose. + """ + source = os.environ if env is None else env + raw = source.get("GRAPHARC_MCP_WORK_TIMEOUT") + if raw is None or raw == "": + return DEFAULT_WORK_TIMEOUT + try: + seconds = float(raw) + except ValueError: + raise DriverError( + f"GRAPHARC_MCP_WORK_TIMEOUT must be a number of seconds, not {raw!r}" + ) from None + if seconds <= 0: + raise DriverError( + f"GRAPHARC_MCP_WORK_TIMEOUT must be positive, not {seconds!r}" + ) + return seconds + class DriverError(Exception): """A tool call that cannot proceed, with the reason as the message.""" @@ -56,6 +100,23 @@ def confine_run_dir(root: Path, run_dir: str) -> Path: return resolved +def _kill_tree(process: asyncio.subprocess.Process) -> None: + """SIGKILL the child's whole process group, falling back to the child. + + The group is the point: see `start_new_session` in `run_cli`. A group + that has already exited raises `ProcessLookupError`, which is the + success case arriving early, and a platform without `killpg` still gets + the old single-process behaviour rather than an exception. + """ + # `ProcessLookupError` and `PermissionError` are `OSError`; `AttributeError` + # is a platform with no `killpg` at all. + with contextlib.suppress(AttributeError, OSError): + os.killpg(os.getpgid(process.pid), signal.SIGKILL) + return + with contextlib.suppress(ProcessLookupError): + process.kill() + + async def run_cli( argv: list[str], *, cwd: Path, timeout: float | None = None ) -> tuple[int, str, str]: @@ -68,14 +129,23 @@ async def run_cli( cwd=cwd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + # Its own process group, so the timeout below can kill the whole tree. + # `process.kill()` signals the direct child only, and a governed run + # delegates agent phases to Claude Code — so a killed run used to + # leave that grandchild alive, still holding the workspace the next + # call would run in. The CLI's own `deadline_guard` kills its group + # deliberately, but that is the CLI's guard, not this outer one. + start_new_session=True, ) try: out, err = await asyncio.wait_for(process.communicate(), timeout=timeout) except TimeoutError: - process.kill() + _kill_tree(process) await process.wait() raise DriverError( - f"grapharc {argv[0]} did not finish within {timeout}s and was stopped" + f"grapharc {argv[0]} did not finish within {timeout}s and was stopped. " + "Whether the plan ran is recorded in the run directory, not here: " + "read it back with show_graph before reissuing anything." ) from None return process.returncode or 0, out.decode(errors="replace"), err.decode(errors="replace") diff --git a/grapharc/mcp/server.py b/grapharc/mcp/server.py index acb9bab..a6c8d9f 100644 --- a/grapharc/mcp/server.py +++ b/grapharc/mcp/server.py @@ -116,20 +116,31 @@ async def execute( A plan with a mutating kind parks until a human answers `grapharc approve ` out of band; tell the user, do not - answer it yourself. A timeout leaves the plan unexecuted and this - call safe to reissue. + answer it yourself. + + **Do not reissue this call after a timeout without reading the run + back first.** A park that expires unanswered leaves the plan + unexecuted, and that case alone is safe to retry. A timeout during + the *work* is a different thing: the run may have been stopped + partway through, and for a mutating plan that means partway through + changing the tree. `show_graph` on the same `run_dir` says which + happened, and `execute` refuses a plan whose trace shows an + unfinished attempt rather than spending one human approval twice. """ resolved = driver.confine_run_dir(base, run_dir) record = driver.read_plan_record(resolved) mutating = driver.plan_is_mutating(record) + # The park and the work get separate budgets. One budget covering both + # meant a human approving near the end of the park left ~120s for the + # run, and the non-mutating branch was unbounded -- a wedged run held + # the call open forever. See DEFAULT_WORK_TIMEOUT for the reasoning. + work = driver.work_timeout() code, out, err = await driver.run_cli( driver.build_execute_argv( resolved, mutating=mutating, approval_timeout=approval_timeout ), cwd=base, - # The subprocess bounds its own park via --approval-timeout; this - # outer bound only catches a wedged process, generously. - timeout=approval_timeout + 120.0 if mutating else None, + timeout=(approval_timeout + work) if mutating else work, ) document = driver.parse_document(out, command="go") document["mutating"] = mutating diff --git a/grapharc/observe/metrics.py b/grapharc/observe/metrics.py index 8ef3fa2..363b4dd 100644 --- a/grapharc/observe/metrics.py +++ b/grapharc/observe/metrics.py @@ -26,7 +26,7 @@ from grapharc.observe.replay import replay from grapharc.observe.status import node_states -from grapharc.observe.trace import TraceRecorder +from grapharc.observe.trace import LOOP_PHASES, SHAPE_PHASES, TraceRecorder class RunMetrics(BaseModel): @@ -112,13 +112,16 @@ def _label(text: str, limit: int = 120) -> str: #: Phases that describe the run rather than doing work; the executed-path #: fallback must never chain them as if they were steps. -_SHAPE_PHASES = frozenset({"topology", "approval_request", "approval_response"}) +#: Re-exported from `observe.trace`, which owns the vocabulary — `cli.plan` +#: needs the same split and the two must not drift. The local names stay so +#: the drawing logic below reads as it did. +_SHAPE_PHASES = SHAPE_PHASES #: The governed loop's own bookkeeping. A planning round is not a node #: execution — chaining these drew `plan -> admission -> round1 -> plan ...` #: as though the planner's paperwork were the orchestration, which is exactly #: the picture a run whose planning failed used to end on. -_LOOP_PHASES = frozenset({"plan", "admission", "round"}) +_LOOP_PHASES = LOOP_PHASES #: What to draw when a run has no graph to show. Honest about *why* there is #: nothing: a run that never got a graph admitted and built has no topology, diff --git a/grapharc/observe/trace.py b/grapharc/observe/trace.py index fcc9486..ae2a2bc 100644 --- a/grapharc/observe/trace.py +++ b/grapharc/observe/trace.py @@ -82,6 +82,35 @@ class TraceEvent(BaseModel): error: str | None = None +#: Phases that are **not** a node doing work. +#: +#: Three kinds, all bookkeeping: the governed loop's own (`plan`, `admission`, +#: `round`), the shape events a viewer draws a graph from (`topology`, +#: `approval_request`, `approval_response`), and a bare `stop`, which is a +#: driver saying why it finished rather than a path it took. +#: +#: Defined here, beside `TraceEvent`, because two callers need the same answer +#: and had their own copies: `observe.metrics` splits these out to decide what +#: to draw, and `cli.plan` asks whether a run got far enough to have changed +#: the tree. A phase classified one way in one file and the other way in the +#: other is a bug in whichever is wrong, and there is no way to tell which. +#: +#: **A phase that is not listed here reads as a node execution**, which is the +#: safe direction: an unclassified new phase makes `go` refuse a plan it could +#: have run, rather than re-run one it should have refused. Refusing is +#: recoverable with `--again`; re-running a half-finished mutating plan spends +#: a human approval that was given once. +LOOP_PHASES = frozenset({"plan", "admission", "round"}) +SHAPE_PHASES = frozenset({"topology", "approval_request", "approval_response"}) +DRIVER_PHASES = frozenset({"stop"}) +NON_EXECUTION_PHASES = LOOP_PHASES | SHAPE_PHASES | DRIVER_PHASES + + +def began_execution(phase: str) -> bool: + """Whether `phase` is a node doing work, rather than bookkeeping.""" + return phase not in NON_EXECUTION_PHASES + + class TraceRecorder: """Append-only JSONL trace writer with a read-back helper for tests and the CLI.""" diff --git a/grapharc/observe/viewmodel.py b/grapharc/observe/viewmodel.py index bba6aa5..8925744 100644 --- a/grapharc/observe/viewmodel.py +++ b/grapharc/observe/viewmodel.py @@ -37,15 +37,17 @@ from grapharc.observe.metrics import latest_topologies from grapharc.observe.replay import ReplayedRun from grapharc.observe.status import NodeStatus, node_states -from grapharc.observe.trace import TraceEvent +from grapharc.observe.trace import LOOP_PHASES, SHAPE_PHASES, TraceEvent #: Same truncation the Mermaid labels use: flatten whitespace, cap the length. #: An error message is the one free-text string this model carries. _ERROR_LIMIT = 120 -#: Phases that describe the run rather than doing work (mirrors `metrics`). -_SHAPE_PHASES = frozenset({"topology", "approval_request", "approval_response"}) -_LOOP_PHASES = frozenset({"plan", "admission", "round"}) +#: Phases that describe the run rather than doing work. Owned by +#: `observe.trace`; this comment used to say "mirrors `metrics`", which was +#: true and was the problem. +_SHAPE_PHASES = SHAPE_PHASES +_LOOP_PHASES = LOOP_PHASES _NO_GRAPH_NOTE = "no graph ran: no proposal was admitted and built" diff --git a/grapharc/slack/live.py b/grapharc/slack/live.py index 00ad444..f1b98ea 100644 --- a/grapharc/slack/live.py +++ b/grapharc/slack/live.py @@ -29,7 +29,7 @@ from grapharc.observe.metrics import to_mermaid from grapharc.observe.replay import NodeExecution, ReplayedRun, replay from grapharc.observe.status import NodeState, node_states -from grapharc.observe.trace import TailRecorder, TraceEvent +from grapharc.observe.trace import SHAPE_PHASES, TailRecorder, TraceEvent from grapharc.slack.format import fence, mermaid_live_url, truncate #: How many trailing sub-step events the flat feed shows for a run with no @@ -249,7 +249,9 @@ def _sub_event_line(event: TraceEvent) -> str: #: Phases that describe the run rather than doing work; never shown as feed. -_SHAPE_PHASES = frozenset({"topology", "approval_request", "approval_response"}) +#: Owned by `observe.trace` — a phase this file thinks is paperwork while +#: another thinks it is work is a bug in one of them, with nothing to say which. +_SHAPE_PHASES = SHAPE_PHASES def _goal(run: ReplayedRun) -> str | None: diff --git a/tests/test_go_rerun.py b/tests/test_go_rerun.py index 7201464..0bb3329 100644 --- a/tests/test_go_rerun.py +++ b/tests/test_go_rerun.py @@ -18,6 +18,7 @@ from __future__ import annotations import json +import re from pathlib import Path from grapharc.cli.main import main @@ -202,3 +203,255 @@ def test_bare_go_still_skips_an_executed_plan(tmp_path, capsys, monkeypatch): assert main(["go", "--json"]) == 1 payload = _last_document(capsys.readouterr().out) assert "no unexecuted plan" in payload["error"] + + +# -- an attempt that began and never recorded finishing (#113) -------------- +# +# `executed_run_id` is stamped after `loop.run()` returns, so a run killed +# partway through — the MCP `execute` work budget expiring, a SIGKILL, an +# OOM-kill — never reaches the stamp. The record then says the plan was never +# executed while the tree may already have been changed, and the guard above +# waves a second run through on the strength of one human approval. The trace +# is written as the run proceeds, so it is the only place that evidence lives. + + +def _forget_the_stamp(run_dir: Path) -> str: + """A record as a killed run would leave it: trace written, stamp missing. + + Simulated by removing the stamp rather than by killing a real subprocess, + because the two leave the same artefacts and only one of them is a test + that races. + """ + record = _record(run_dir) + ran = record.pop("executed_run_id") + record.pop("executed_run_ids", None) + record.pop("executed_at", None) + (run_dir / "plan.json").write_text(json.dumps(record, indent=2) + "\n", encoding="utf-8") + return ran + + +def test_an_attempt_that_began_and_never_finished_is_refused(tmp_path, capsys): + """The bug: the record says "never executed", so this used to exit 0 and + run the plan a second time over a tree the first run had half-changed.""" + run_dir = _saved_plan(tmp_path, capsys) + assert main(["go", str(run_dir), "--json"]) == 0 + killed = _forget_the_stamp(run_dir) + before = _runs_in_trace(run_dir) + capsys.readouterr() + + code = main(["go", str(run_dir), "--json"]) + + assert code == 2 + payload = _last_document(capsys.readouterr().out) + assert payload["ok"] is False + assert payload["unfinished_run_id"] == killed + assert "never recorded finishing" in payload["error"] + assert "--again" in payload["error"] + # Refused before anything ran. + assert _runs_in_trace(run_dir) == before + + +def test_the_refusal_points_at_the_trace_that_holds_what_it_did(tmp_path, capsys): + """The tree may have been changed. A refusal that does not say where to + look leaves the reader with no way to find out how far it got.""" + run_dir = _saved_plan(tmp_path, capsys) + assert main(["go", str(run_dir), "--json"]) == 0 + _forget_the_stamp(run_dir) + capsys.readouterr() + + assert main(["go", str(run_dir)]) == 2 + + message = capsys.readouterr().err + assert str(run_dir / "trace.jsonl") in message + assert "may have changed the tree" in message + + +def test_again_still_runs_an_unfinished_plan(tmp_path, capsys): + """The escape hatch is the same one the executed-plan guard offers.""" + run_dir = _saved_plan(tmp_path, capsys) + assert main(["go", str(run_dir), "--json"]) == 0 + _forget_the_stamp(run_dir) + capsys.readouterr() + + assert main(["go", str(run_dir), "--again", "--json"]) == 0 + assert _last_document(capsys.readouterr().out)["executed"] is True + + +def test_planning_paperwork_is_not_mistaken_for_a_half_run(tmp_path, capsys): + """The false positive this guard must not have. `plan` writes its own + events into the same trace — `plan`, `admission`, `round`, `topology` — and + a first `go` would be refused forever if those counted as an execution.""" + run_dir = _saved_plan(tmp_path, capsys) + assert (run_dir / "trace.jsonl").is_file() # paperwork is already there + capsys.readouterr() + + assert main(["go", str(run_dir), "--json"]) == 0 + assert _last_document(capsys.readouterr().out)["executed"] is True + + +# -- the phase vocabulary the guard depends on ------------------------------ + + +def _trace_with(tmp_path: Path, phases: list[str], run_id: str = "killed-run") -> Path: + tmp_path.mkdir(parents=True, exist_ok=True) + trace = tmp_path / "trace.jsonl" + trace.write_text( + "".join( + json.dumps( + { + "ts": "2026-01-01T00:00:00+00:00", + "run_id": run_id, + "graph": "g", + "node": "n", + "phase": phase, + "step": 1, + } + ) + + "\n" + for phase in phases + ), + encoding="utf-8", + ) + return trace + + +def test_only_a_node_doing_work_counts_as_having_begun(tmp_path): + """Pinned deliberately. The loop's bookkeeping, the viewer's shape events + and a bare `stop` all share the trace with node events; treating any of + them as an execution would refuse a plan that had never run a node — a + parked plan a human *denied* being the case that matters most.""" + from grapharc.cli.plan import _unfinished_execution + + paperwork = ["plan", "admission", "round", "topology", "stop"] + assert _unfinished_execution(_trace_with(tmp_path / "a", paperwork), {}) is None + + denied = ["plan", "admission", "round", "approval_request", "approval_response", "stop"] + assert _unfinished_execution(_trace_with(tmp_path / "b", denied), {}) is None + + for phase in ("start", "model", "end", "error"): + trace = _trace_with(tmp_path / phase, ["plan", "admission", phase]) + assert _unfinished_execution(trace, {}) == "killed-run", phase + + +def test_a_run_the_record_already_names_is_not_unfinished(tmp_path): + from grapharc.cli.plan import _unfinished_execution + + trace = _trace_with(tmp_path / "known", ["end"], run_id="r1") + assert _unfinished_execution(trace, {"executed_run_id": "r1"}) is None + assert _unfinished_execution(trace, {"executed_run_ids": ["r1"]}) is None + assert _unfinished_execution(trace, {"executed_run_id": "other"}) == "r1" + + +def test_a_damaged_trace_does_not_wedge_every_plan_beside_it(tmp_path): + """A torn line is not evidence of a half-run, and refusing on one would + make an unreadable file block work it says nothing about.""" + from grapharc.cli.plan import _unfinished_execution + + torn = tmp_path / "trace.jsonl" + torn.parent.mkdir(parents=True, exist_ok=True) + torn.write_text('{"run_id": "r1", "phase": "en', encoding="utf-8") + + assert _unfinished_execution(torn, {}) is None + assert _unfinished_execution(tmp_path / "absent.jsonl", {}) is None + + +# -- the same guard on the documented flow ---------------------------------- + + +def test_bare_go_does_not_re_run_an_unfinished_plan(tmp_path, capsys, monkeypatch): + """`grapharc plan … && grapharc go` is the flow the README teaches, and + bare `go` selects its plan by a different route: `find_unexecuted_plan` + passes over any record carrying an `executed_run_id`, and a killed run + never wrote one. So the half-finished plan looks *unexecuted* to the + selector — the newest candidate, chosen first. + + The refusal happens after the selection either way, which is what makes + this safe, and it is exactly the kind of thing a later refactor of the + selector could quietly undo. Pinned here for that reason. + """ + monkeypatch.chdir(tmp_path) + trace = tmp_path / ".grapharc" / "runs" / "r1" / "trace.jsonl" + assert main(["plan", "investigate", "--scripted", "--trace", str(trace), "--json"]) == 0 + capsys.readouterr() + run_dir = trace.parent + + assert main(["go", "--json"]) == 0 + killed = _forget_the_stamp(run_dir) + before = _runs_in_trace(run_dir) + capsys.readouterr() + + code = main(["go", "--json"]) + + assert code == 2 + payload = _last_document(capsys.readouterr().out) + assert payload["unfinished_run_id"] == killed + # The load-bearing assertion: nothing ran a second time. + assert _runs_in_trace(run_dir) == before + + +def test_the_phase_vocabulary_has_exactly_one_owner(): + """Four modules needed to know which phases are bookkeeping, and each had + its own copy — `observe.metrics`, `observe.viewmodel` (whose comment said + "mirrors `metrics`", which was true and was the problem), `slack.live`, and + the one `cli.plan` added. A phase classified one way in one file and the + other way in another is a bug in whichever is wrong, with nothing in the + tree to say which. + + Identity, not equality: four frozensets that happen to agree today is + exactly the state this assertion exists to rule out. + """ + from grapharc.cli import plan as cli_plan + from grapharc.observe import metrics, trace, viewmodel + from grapharc.slack import live + + assert metrics._LOOP_PHASES is trace.LOOP_PHASES + assert metrics._SHAPE_PHASES is trace.SHAPE_PHASES + assert viewmodel._LOOP_PHASES is trace.LOOP_PHASES + assert viewmodel._SHAPE_PHASES is trace.SHAPE_PHASES + assert live._SHAPE_PHASES is trace.SHAPE_PHASES + assert not hasattr(cli_plan, "_NON_EXECUTION_PHASES"), ( + "cli.plan grew its own copy of the phase vocabulary again" + ) + + +def test_no_module_redefines_the_phase_vocabulary(): + """The identity check above only covers names it knows to look at, so it + cannot notice a *fifth* copy appearing under a new name. This reads the + source instead: `observe.trace` defines the sets, and nothing else does. + """ + package = Path(__file__).resolve().parents[1] / "grapharc" + pattern = re.compile(r"^_?(?:LOOP|SHAPE|DRIVER|NON_EXECUTION)_PHASES\s*=\s*frozenset") + # Files, not file:line — a line number would make this fail on any edit to + # trace.py, which is noise rather than a finding. + owners = sorted( + { + str(path.relative_to(package)) + for path in package.rglob("*.py") + if "__pycache__" not in path.parts + for line in path.read_text(encoding="utf-8").splitlines() + if pattern.match(line) + } + ) + + assert owners == ["observe/trace.py"], ( + f"the phase vocabulary is defined outside observe/trace.py: {owners}" + ) + + +def test_an_unclassified_phase_reads_as_an_execution(): + """The direction the predicate errs in, asserted rather than assumed. + + A bookkeeping phase nobody classified makes `go` refuse a plan it could + have run, which `--again` recovers from. The opposite would re-run a + half-finished mutating plan and spend a human approval given once. + """ + from grapharc.observe import trace + + assert trace.began_execution("a-phase-nobody-has-written-yet") is True + for phase in trace.NON_EXECUTION_PHASES: + assert trace.began_execution(phase) is False, phase + + # The three groups partition the bookkeeping set, with nothing dropped. + assert ( + trace.LOOP_PHASES | trace.SHAPE_PHASES | trace.DRIVER_PHASES + ) == trace.NON_EXECUTION_PHASES diff --git a/tests/test_mcp_execute_timeout.py b/tests/test_mcp_execute_timeout.py new file mode 100644 index 0000000..c7da0f7 --- /dev/null +++ b/tests/test_mcp_execute_timeout.py @@ -0,0 +1,262 @@ +"""The MCP `execute` tool's two budgets, and the kill that enforces them. + +One timeout used to cover both halves of a mutating `execute`: the park, where +a human is being asked, and the work, where an approved graph runs. It was +`approval_timeout + 120s`, so a human who answered near the end of the park +left roughly two minutes for the run itself — and a governed run's agent +phases delegate to Claude Code, which reads files, edits them and verifies. +The likely outcome was not a wedged process being cleaned up; it was a +human-approved, tree-mutating run being SIGKILLed partway through mutating the +tree. The non-mutating branch, meanwhile, passed `timeout=None` and so was not +bounded at all. + +The tool then told the calling agent, in its own docstring, that "a timeout +leaves the plan unexecuted and this call safe to reissue" — which is true of a +park that expired unanswered and false of everything else. + +Issue #113. The park and the work have separate budgets now, the kill reaches +the whole process group rather than the direct child, and the docstring no +longer promises something the mutating path cannot honour. +""" + +from __future__ import annotations + +import asyncio +import inspect +import json +import os +import signal +import sys + +import pytest + +from grapharc.mcp import build_server +from grapharc.mcp.driver import ( + DEFAULT_WORK_TIMEOUT, + DriverError, + _kill_tree, + run_cli, + work_timeout, +) + + +def _unwrap(raw) -> dict: + """FastMCP's call_tool result as the tool's own dict, across SDK shapes. + + Duplicated from test_mcp_gate rather than imported: `tests/` is not a + package, so a cross-module import here would depend on sys.path shape. + """ + if isinstance(raw, tuple) and len(raw) == 2 and isinstance(raw[1], dict): + structured = raw[1] + return structured.get("result", structured) + blocks = raw[0] if isinstance(raw, tuple) else raw + text = "".join(getattr(block, "text", "") for block in blocks) + return json.loads(text) + + +def _mark_mutating(root, run_dir: str): + """Flip the record's verdict, resolving the CLI's root-relative run_dir.""" + from pathlib import Path + + resolved = Path(run_dir) if Path(run_dir).is_absolute() else root / run_dir + plan_file = resolved / "plan.json" + record = json.loads(plan_file.read_text()) + record["mutating"] = True + plan_file.write_text(json.dumps(record, indent=2) + "\n") + return resolved + + +def _capture_timeout(monkeypatch) -> dict: + """Stand in for `run_cli` and record the bound it was handed. + + Installed *after* the plan under test exists: the `plan` tool goes through + the same `run_cli`, so patching earlier would stub out the very call that + writes the run directory. + """ + seen: dict = {} + + async def fake_run_cli(argv, *, cwd, timeout=None): + seen["argv"] = argv + seen["timeout"] = timeout + return 0, json.dumps({"ok": True, "executed": True, "run_id": "r"}), "" + + from grapharc.mcp import driver + + monkeypatch.setattr(driver, "run_cli", fake_run_cli) + return seen + + +# -- the two budgets -------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_the_work_budget_is_separate_from_the_park(tmp_path, monkeypatch): + """The bug: this was `approval_timeout + 120`, so approving late left two + minutes to do the work in.""" + server = build_server(tmp_path) + planned = _unwrap(await server.call_tool("plan", {"goal": "fix", "scripted": True})) + run_dir = _mark_mutating(tmp_path, planned["run_dir"]) + seen = _capture_timeout(monkeypatch) + + await server.call_tool("execute", {"run_dir": str(run_dir), "approval_timeout": 240.0}) + + # The park keeps its own budget; the work gets a whole one beside it. + assert seen["timeout"] == 240.0 + DEFAULT_WORK_TIMEOUT + assert seen["timeout"] != 240.0 + 120.0 + + +@pytest.mark.asyncio +async def test_a_non_mutating_execute_is_bounded_at_all(tmp_path, monkeypatch): + """The other half of the bug: this branch passed `timeout=None`, so a + wedged run held the tool call open with nothing to end it.""" + server = build_server(tmp_path) + planned = _unwrap(await server.call_tool("plan", {"goal": "fix", "scripted": True})) + seen = _capture_timeout(monkeypatch) + + await server.call_tool("execute", {"run_dir": planned["run_dir"]}) + + assert seen["timeout"] is not None + # No park to wait through, so the work budget is the whole bound. + assert seen["timeout"] == DEFAULT_WORK_TIMEOUT + + +# -- the operator's ceiling ------------------------------------------------- + + +def test_the_work_budget_defaults_and_honours_the_environment(): + assert work_timeout({}) == DEFAULT_WORK_TIMEOUT + assert work_timeout({"GRAPHARC_MCP_WORK_TIMEOUT": "60"}) == 60.0 + # Unset and empty both mean "operator said nothing", not "zero". + assert work_timeout({"GRAPHARC_MCP_WORK_TIMEOUT": ""}) == DEFAULT_WORK_TIMEOUT + + +@pytest.mark.parametrize("bad", ["soon", "-1", "0"]) +def test_a_nonsense_work_budget_is_refused_rather_than_replaced(bad): + """Substituting the default for an unreadable override is how a run gets + killed at a limit nobody chose.""" + with pytest.raises(DriverError) as caught: + work_timeout({"GRAPHARC_MCP_WORK_TIMEOUT": bad}) + assert "GRAPHARC_MCP_WORK_TIMEOUT" in str(caught.value) + + +@pytest.mark.asyncio +async def test_the_env_override_reaches_the_call(tmp_path, monkeypatch): + server = build_server(tmp_path) + planned = _unwrap(await server.call_tool("plan", {"goal": "fix", "scripted": True})) + seen = _capture_timeout(monkeypatch) + monkeypatch.setenv("GRAPHARC_MCP_WORK_TIMEOUT", "45") + + await server.call_tool("execute", {"run_dir": planned["run_dir"]}) + + assert seen["timeout"] == 45.0 + + +# -- the kill reaches the whole tree ---------------------------------------- + + +@pytest.mark.asyncio +async def test_the_child_is_started_in_its_own_session(tmp_path, monkeypatch): + """`killpg` needs a group to aim at, and the child only has its own if it + was started with one.""" + seen: dict = {} + real = asyncio.create_subprocess_exec + + async def spy(*argv, **kwargs): + seen.update(kwargs) + return await real(*argv, **kwargs) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", spy) + await run_cli(["--version"], cwd=tmp_path, timeout=30) + + assert seen.get("start_new_session") is True + + +@pytest.mark.asyncio +@pytest.mark.timeout(30) +async def test_a_kill_reaches_a_grandchild(tmp_path): + """The defect this closes: `process.kill()` signals the direct child only, + so a delegated Claude Code process outlived the run that spawned it and + kept holding the workspace. + + A real tree, not a mock: the child spawns a grandchild that writes its pid + and sleeps, and the grandchild must be gone once the group is killed. + """ + pid_file = tmp_path / "grandchild.pid" + script = ( + "import os, subprocess, sys, time\n" + f"kid = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(120)'])\n" + f"open({str(pid_file)!r}, 'w').write(str(kid.pid))\n" + "time.sleep(120)\n" + ) + process = await asyncio.create_subprocess_exec( + sys.executable, "-c", script, cwd=tmp_path, start_new_session=True + ) + for _ in range(200): # wait for the grandchild to exist + if pid_file.is_file() and pid_file.read_text().strip(): + break + await asyncio.sleep(0.05) + grandchild = int(pid_file.read_text().strip()) + os.kill(grandchild, 0) # alive, or this raises + + _kill_tree(process) + await process.wait() + + for _ in range(200): + try: + os.kill(grandchild, 0) + except ProcessLookupError: + break + await asyncio.sleep(0.05) + else: # pragma: no cover — the failure this test exists to catch + os.kill(grandchild, signal.SIGKILL) + pytest.fail(f"grandchild {grandchild} survived the kill") + + +def test_killing_an_already_dead_group_is_not_an_error(): + """The success case arriving early must not raise out of the timeout path.""" + + class Gone: + pid = 2**22 # far above any live pid on a normal machine + + def kill(self): + raise ProcessLookupError + + _kill_tree(Gone()) # no exception + + +# -- what the agent is told ------------------------------------------------- + + +@pytest.mark.asyncio +async def test_the_tool_no_longer_promises_a_timeout_is_safe_to_reissue(tmp_path): + """The docstring is the agent's whole briefing. It said a timeout left the + plan unexecuted and the call safe to reissue, which is true of an + unanswered park and false of a run killed partway through its work.""" + server = build_server(tmp_path) + doc = next(t.description for t in await server.list_tools() if t.name == "execute") + + assert "safe to reissue" not in doc + # And it must say what to do instead of reissuing blindly. + assert "show_graph" in doc + + +@pytest.mark.asyncio +async def test_the_timeout_error_sends_the_reader_to_the_run_directory(tmp_path): + """`DriverError` is what the agent sees when the bound is hit, so it is the + other half of the same briefing.""" + with pytest.raises(DriverError) as caught: + await run_cli(["plan", "wait", "--scripted"], cwd=tmp_path, timeout=0.001) + + message = str(caught.value) + assert "show_graph" in message + assert "did not finish" in message + + +def test_the_default_work_budget_matches_the_slack_path(): + """Both exist for the same reason — an approved run needs minutes, not two + of them — so a reader comparing the two surfaces should find one number. + Read off the module rather than a cwd-relative path.""" + from grapharc.slack import config as slack_config + + assert DEFAULT_WORK_TIMEOUT == 1800.0 + assert '"GRAPHARC_SLACK_WORK_TIMEOUT", "1800"' in inspect.getsource(slack_config)