fix(proving-ground): a run releases what it started

.software-factory/ratchet.yaml

@@ -12,8 +12,8 @@ rules:
     - packages/agent-proving-ground/agent_proving_ground/assertions/api.py:62
     - packages/agent-proving-ground/agent_proving_ground/assertions/files.py:669
     - packages/agent-proving-ground/agent_proving_ground/drivers/_provider.py:217
-    - packages/agent-proving-ground/agent_proving_ground/runner.py:578
-    - packages/agent-proving-ground/agent_proving_ground/runner.py:594
+    - packages/agent-proving-ground/agent_proving_ground/runner.py:580
+    - packages/agent-proving-ground/agent_proving_ground/runner.py:596
     - packages/agent-proving-ground/tests/unit/test_builtin_scenario_labels.py:10
     - packages/runner/logion_runner/_json.py:51
   L1.COMPLEXITY_CEILING:

packages/agent-proving-ground/agent_proving_ground/runner.py

@@ -383,6 +383,7 @@ def _make_run_id(self) -> str:
 
     async def run(self) -> ScenarioResult:
         result: ScenarioResult | None = None
+        started_world: World | None = None
         phase_results: list[dict] = []
         all_assertion_results: list[AssertionOutcome] = []
         self.timeline.event(
@@ -391,7 +392,7 @@ async def run(self) -> ScenarioResult:
         try:
             await self.api.start()
             self.timeline.event("api.started", api_adapter=self.api.name)
-            world = await self.api.create_world(
+            world = started_world = await self.api.create_world(
                 self.run_id,
                 self.scenario.name,
                 [a.id for a in self.scenario.agents],
@@ -476,6 +477,7 @@ async def run(self) -> ScenarioResult:
                 status=result.status,
             )
             await self._stop_agents()
+            await self._run_teardown_hooks(started_world)
             await self.api.stop()
             await self.artifacts.flush()
             await self.timeline.flush()
@@ -927,6 +929,66 @@ async def _run_assertions(
         )
         return results
 
+    async def _run_teardown_hooks(self, world: World | None) -> None:
+        """Release what the scenario started, whatever the run decided.
+
+        A teardown hook produces no evidence and cannot change a result:
+        a run that passed had already passed, and a run that failed is not
+        rescued by a clean exit. What it exists to stop is the opposite
+        failure — a gate that seals evidence about isolated roles and
+        leaves the roles running on the operator's machine afterwards.
+
+        Failures here are logged and swallowed. A teardown that could fail
+        a run would give a scenario two ways to go red, only one of which
+        is about the product.
+        """
+        import asyncio
+        import subprocess
+
+        if world is None or not self.scenario.teardown_hooks:
+            return
+        bindings = _scenario_bindings(world)
+        env = {
+            **os.environ,
+            **bindings,
+            "LOGION_PUBLIC_REPO_PATH": str(world.root_dir),
+        }
+        env.pop("LOGION_API_KEY", None)
+        env.pop("LOGION_PROVING_GROUND_API_KEY", None)
+        for spec in self.scenario.teardown_hooks:
+            hook = os.path.expandvars(spec.hook)
+            if not hook.startswith("/"):
+                hook = _resolve_hook_path(hook, world.root_dir)
+            args = [
+                os.path.expandvars(_resolve_scenario_value(a, bindings))
+                for a in spec.args
+            ]
+            cmd = [hook, *args]
+            if Path(hook).suffix == ".py":
+                cmd = [sys.executable, hook, *args]
+            self.timeline.event("run.teardown.started", hook=spec.hook)
+            try:
+                proc = await asyncio.to_thread(
+                    subprocess.run,
+                    cmd,
+                    cwd=world.root_dir,
+                    env=env,
+                    capture_output=True,
+                    text=True,
+                    timeout=spec.timeout_seconds,
+                )
+            except (subprocess.TimeoutExpired, OSError) as exc:
+                self.timeline.event(
+                    "run.teardown.failed", hook=spec.hook, error=str(exc)
+                )
+                continue
+            self.timeline.event(
+                "run.teardown.completed",
+                hook=spec.hook,
+                exit_code=proc.returncode,
+                stderr=proc.stderr[:500] if proc.returncode else "",
+            )
+
     async def _run_local_hook(
         self, phase: PhaseSpec, world: World
     ) -> JsonObject:

packages/agent-proving-ground/agent_proving_ground/scenarios/builtin/local_multi_agent_node.yaml

@@ -252,4 +252,14 @@ final_assertions:
       manifest: "${AGENT_CONSUMER_WORKSPACE}/evidence/identity.json"
   - type: sandbox.cross_volume_canary_unreadable
     params:
-      manifest: "${AGENT_AUDITOR_WORKSPACE}/evidence/canaries.json"
+      manifest: "${AGENT_AUDITOR_WORKSPACE}/evidence/canaries.json"
+# The scenario starts the roles and is therefore what stops them. Runs
+# after every phase and after the final assertions, whatever the run
+# decided, and produces no evidence: it releases the containers and the
+# network, and `compose down` without `-v` leaves the named volumes the
+# restart evidence depends on exactly where they are.
+teardown_hooks:
+  - hook: packages/agent-proving-ground/scripts/capture_local_node_evidence.py
+    args:
+      - teardown
+      - "${AGENT_CONSUMER_WORKSPACE}/evidence/teardown.json"

packages/agent-proving-ground/agent_proving_ground/scenarios/schema.py

@@ -181,6 +181,22 @@ class ExecutionRequirements(BaseModel):
     driver_models: dict[str, list[str]] = Field(default_factory=dict)
 
 
+class TeardownHookSpec(BaseModel):
+    """A command the run releases its own resources with.
+
+    Declared by the scenario rather than known to the runner: the runner
+    has no business knowing what a compose project is, and a scenario that
+    starts something outside the run directory is the only thing that knows
+    how to stop it.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    hook: str
+    args: list[str] = Field(default_factory=list)
+    timeout_seconds: int = 300
+
+
 class ScenarioSpec(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -202,6 +218,8 @@ class ScenarioSpec(BaseModel):
     agents: list[AgentSpec]
     phases: list[PhaseSpec]
     final_assertions: list[AssertionSpec] = Field(default_factory=list)
+    #: Run after every phase, whatever the run decided, and never evidence.
+    teardown_hooks: list[TeardownHookSpec] = Field(default_factory=list)
 
     @field_validator("name")
     @classmethod

packages/agent-proving-ground/scripts/capture_local_node_evidence.py

@@ -730,6 +730,42 @@ def capture_harness_use(out: Path) -> None:
     out.write_text(json.dumps({"harness_runs": results}, indent=2) + "\n")
 
 
+def capture_teardown(out: Path) -> None:
+    """Stop the node the run started and record what survived it.
+
+    The scenario brings roles up and, before this existed, the last thing
+    it did to the node was bring it up again — so a sealed gate about
+    isolated roles left the roles on the operator's machine until their
+    wall-clock timeout killed them, and left the dead containers after
+    that. Named volumes are kept: ``node.sh down`` is ``compose down``
+    without ``-v``, so role state survives a teardown exactly as the
+    restart evidence says it does.
+    """
+    repo_root = NODE_DIR.parent.parent
+    down = subprocess.run(
+        ["make", "node-dev-down"],
+        capture_output=True,
+        text=True,
+        cwd=repo_root,
+        check=False,
+    )
+    survivors = {role: _container_id(role) for role in ("consumer", "auditor")}
+    out.write_text(
+        json.dumps(
+            {
+                "teardown": {
+                    "down_exit_code": down.returncode,
+                    "surviving_container_ids": survivors,
+                }
+            },
+            indent=2,
+            sort_keys=True,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+
+
 def main() -> int:
     capture = sys.argv[1]
     out = Path(sys.argv[2]).resolve()
@@ -750,6 +786,8 @@ def main() -> int:
         capture_restart(out)
     elif capture == "harness_use":
         capture_harness_use(out)
+    elif capture == "teardown":
+        capture_teardown(out)
     elif capture == "selective_reset":
         cred = Path(sys.argv[3]) if len(sys.argv) > 3 else None
         capture_selective_reset(out, cred)

packages/agent-proving-ground/tests/unit/test_local_hook_runner.py

@@ -88,3 +88,103 @@ async def test_local_hook_gets_package_pythonpath(tmp_path: Path) -> None:
 
     assert result["status"] == "completed"
     assert world.data["scenario_vars"]["VALUE"] == "ok"
+
+
+def _teardown_scenario(hook: Path, marker: Path) -> ScenarioSpec:
+    return ScenarioSpec.model_validate({
+        "name": "teardown_runs",
+        "description": "test teardown hooks",
+        "kind": "rig",
+        "agents": [{"id": "agent1", "role": "tester"}],
+        "phases": [
+            {"id": "phase1", "actor": "agent1", "goal": ""},
+        ],
+        "teardown_hooks": [
+            {"hook": str(hook), "args": [str(marker)]},
+        ],
+    })
+
+
+def _runner(scenario: ScenarioSpec, tmp_path: Path) -> ScenarioRunner:
+    return ScenarioRunner(
+        scenario=scenario,
+        api=DummyApiAdapter(),
+        driver_factory=AgentDriverFactory({}),
+        artifacts=ArtifactStore(tmp_path / "artifacts"),
+        assertions=AssertionRegistry(),
+        timeline=Timeline(tmp_path / "timeline.jsonl"),
+    )
+
+
+async def test_teardown_hook_releases_what_the_run_started(
+    tmp_path: Path,
+) -> None:
+    """The hook runs after the phases, without being a phase."""
+    marker = tmp_path / "released"
+    hook = tmp_path / "teardown.py"
+    hook.write_text(
+        "#!/usr/bin/env python3\n"
+        "import sys\n"
+        "open(sys.argv[1], 'w').write('down')\n",
+        encoding="utf-8",
+    )
+    hook.chmod(0o755)
+    scenario = _teardown_scenario(hook, marker)
+    runner = _runner(scenario, tmp_path)
+    world = World(
+        run_id="r1",
+        base_url="http://example.test",
+        root_dir=tmp_path,
+        data={},
+    )
+
+    await runner._run_teardown_hooks(world)
+
+    assert marker.read_text(encoding="utf-8") == "down"
+
+
+async def test_a_failing_teardown_hook_does_not_raise(tmp_path: Path) -> None:
+    """A teardown that fails must not give a scenario a second way to go red.
+
+    The run's verdict was decided before this ran. Letting the release path
+    raise would turn an operator's dirty machine into a failed measurement.
+    """
+    hook = tmp_path / "teardown.py"
+    hook.write_text(
+        "#!/usr/bin/env python3\nimport sys\nsys.exit(3)\n",
+        encoding="utf-8",
+    )
+    hook.chmod(0o755)
+    scenario = _teardown_scenario(hook, tmp_path / "unused")
+    runner = _runner(scenario, tmp_path)
+    world = World(
+        run_id="r1",
+        base_url="http://example.test",
+        root_dir=tmp_path,
+        data={},
+    )
+
+    await runner._run_teardown_hooks(world)
+
+    events = (tmp_path / "timeline.jsonl").read_text(encoding="utf-8")
+    assert "run.teardown.completed" in events
+
+
+async def test_no_world_means_nothing_was_started_to_release(
+    tmp_path: Path,
+) -> None:
+    """A run that never built a world started nothing this can release."""
+    marker = tmp_path / "released"
+    hook = tmp_path / "teardown.py"
+    hook.write_text(
+        "#!/usr/bin/env python3\n"
+        "import sys\n"
+        "open(sys.argv[1], 'w').write('down')\n",
+        encoding="utf-8",
+    )
+    hook.chmod(0o755)
+    runner = _runner(_teardown_scenario(hook, marker), tmp_path)
+
+    await runner._run_teardown_hooks(None)
+
+    assert not marker.exists()