@@ -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,151 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,171 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.
 
 
@@ -233,6 +233,58 @@ def _executed_run_ids(record: dict[str, Any]) -> list[str]:
     return [str(scalar)] if scalar else []
 
 
+
+#: Phases that are *not* a node doing work. The loop's own bookkeeping
+#: (`plan`, `admission`, `round`), the shape events a viewer draws from
+#: (`topology`, `approval_request`, `approval_response`), and a bare `stop`,
+#: which is a driver saying why it finished rather than a path it took --
+#: the same three-way split `observe.metrics` makes to decide what counts as
+#: an execution. Anything else (`start`, `model`, `end`, `error`) is a node
+#: that ran, which is the only thing this file needs to know.
+_NON_EXECUTION_PHASES = frozenset(
+    {
+        "plan",
+        "admission",
+        "round",
+        "topology",
+        "approval_request",
+        "approval_response",
+        "stop",
+    }
+)
+
+
+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
+
+    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 event.phase not in _NON_EXECUTION_PHASES:
+            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 +447,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"))
 
@@ -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")
 
 
@@ -116,20 +116,31 @@ async def execute(
 
         A plan with a mutating kind parks until a human answers
         `grapharc approve <run_dir>` 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