feat(runtime): add the containment boundary and its failure contract

Agent-reachable execution has 25 independent spawn sites and no single
place deciding where a process runs or under what limits. All three
consequences are visible on this SHA. When bwrap is absent the workspace
namespace degrades to a regex that rewrites /workspace to the real path,
and nothing in the tool result says which one you got. No spawn site
passes start_new_session, so a wall-clock kill reaches the wrapper shell
and leaves its backgrounded grandchildren running while reporting the
process killed. Teardown stops at SIGTERM without ever checking death.

src/containment.py gives those paths one boundary. acquire() establishes
containment or refuses -- a string rewrite is not a mechanism it can
select -- and the grant states which dimensions actually hold, which were
best-effort and are missing, and which were required and are missing.
run() enforces the wall clock and the output cap. release() signals the
process group, escalates to SIGKILL, and reports dead only for a group it
observed go empty.

Containment never sees the command: acquire() takes a workspace and
limits, and the command text only reaches run(). Nothing in a request can
widen a boundary it is never shown.

CONTAINMENT_MODE chooses between refusing an unestablishable required
dimension and recording it. It ships report-only, so landing this changes
no behaviour on a host without bwrap -- which is every host today.

No call sites move here; they follow on this branch. The configuration
reference is regenerated because the page records src/constants.py line
numbers and the new path constant shifts two of them.

src/constants.py

@@ -75,6 +75,7 @@ APP_KEY_FILE = os.path.join(DATA_DIR, ".app_key")
 EMBEDDING_ENDPOINT_FILE = os.path.join(DATA_DIR, "embedding_endpoint.json")
 COOKBOOK_STATE_FILE = os.path.join(DATA_DIR, "cookbook_state.json")
 BG_JOBS_FILE = os.path.join(DATA_DIR, "bg_jobs.json")
+CONTAINMENT_STATE_FILE = os.path.join(DATA_DIR, "containment_grants.json")
 VAULT_FILE = os.path.join(DATA_DIR, "vault.json")
 TIDY_CALENDAR_STATE_FILE = os.path.join(DATA_DIR, "tidy_calendar_state.json")
 SKILLS_FILE = os.path.join(DATA_DIR, "skills.json")

tests/test_containment_contract.py

@@ -0,0 +1,509 @@
+"""The containment API's own invariants.
+
+These pin the contract rather than any one mechanism, so they run identically on
+a host with bubblewrap and one without: every test substitutes
+``containment.MECHANISMS`` with fake mechanisms whose availability and provided
+dimensions are stated in the test. No real sandbox, no real process.
+
+The one thing these tests must prove above all others: a request that cannot be
+contained does not execute. That is asserted by recording every spawn attempt
+and showing the list is empty.
+"""
+
+import asyncio
+
+import pytest
+
+from src import containment
+
+
+@pytest.fixture(autouse=True)
+def _isolated_store(tmp_path, monkeypatch):
+    """Keep grant records out of ./data for every test in this module."""
+    store = tmp_path / "containment_grants.json"
+    monkeypatch.setattr(containment, "_store_path", lambda: store)
+    return store
+
+
+@pytest.fixture
+def workspace(tmp_path):
+    path = tmp_path / "ws"
+    path.mkdir()
+    return str(path)
+
+
+@pytest.fixture
+def no_spawn(monkeypatch):
+    """Record spawn attempts and refuse them, so "did not execute" is provable."""
+    attempts = []
+
+    async def _refuse(*args, **kwargs):
+        attempts.append(args)
+        raise AssertionError("containment spawned a process it should not have")
+
+    monkeypatch.setattr(asyncio, "create_subprocess_exec", _refuse)
+    return attempts
+
+
+def mechanism(name, rank, provides, *, available=True):
+    return containment.Mechanism(
+        name=name,
+        rank=rank,
+        available=lambda: available,
+        provides=lambda spec, _provides=frozenset(provides): _provides,
+    )
+
+
+def install(monkeypatch, *mechanisms):
+    monkeypatch.setattr(containment, "MECHANISMS", tuple(mechanisms))
+
+
+def enforcing(monkeypatch):
+    monkeypatch.setattr(containment, "CONTAINMENT_MODE", containment.MODE_ENFORCING)
+
+
+def report_only(monkeypatch):
+    monkeypatch.setattr(containment, "CONTAINMENT_MODE", containment.MODE_REPORT_ONLY)
+
+
+def spec_for(workspace, **kwargs):
+    kwargs.setdefault("env", {"PATH": "/usr/bin"})
+    kwargs.setdefault("wall_clock_s", 5)
+    return containment.ContainmentSpec(workspace=workspace, **kwargs)
+
+
+ALL = tuple(sorted(containment.DIMENSIONS))
+
+
+# ── The postcondition ───────────────────────────────────────────────────────
+@pytest.mark.parametrize("provided", [
+    frozenset(containment.DEFAULT_REQUIRED),
+    containment.DIMENSIONS,
+])
+def test_required_is_always_a_subset_of_enforced(monkeypatch, workspace, provided):
+    """spec.required <= grant.enforced, for any mechanism that can satisfy it."""
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("fake", 10, provided))
+    grant = containment.acquire(spec_for(workspace), owner="session-1")
+    assert grant.spec.required <= grant.enforced
+    assert grant.contained is True
+    assert grant.unenforced_required == ()
+
+
+def test_enforced_never_exceeds_what_the_spec_requested(monkeypatch, workspace):
+    """A grant is never a superset of its spec: unrequested dimensions are not claimed."""
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("generous", 10, containment.DIMENSIONS))
+    grant = containment.acquire(spec_for(workspace), owner="session-1")
+    # network/memory/process_count were not asked for, so they are not enforced
+    # even though the mechanism offers them.
+    assert grant.enforced == frozenset(containment.DEFAULT_REQUIRED)
+    assert containment.NETWORK not in grant.enforced
+    assert grant.degraded == ()
+
+
+def test_enforced_is_always_within_the_known_dimension_set(monkeypatch, workspace):
+    """A mechanism cannot invent a dimension the API does not define."""
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism(
+        "liar", 10, frozenset(containment.DEFAULT_REQUIRED) | {"telepathy"},
+    ))
+    grant = containment.acquire(spec_for(workspace), owner="session-1")
+    assert grant.enforced <= containment.DIMENSIONS
+
+
+# ── Deterministic failure: the request does not execute ─────────────────────
+async def test_uncontainable_request_does_not_execute(monkeypatch, workspace, no_spawn):
+    """The headline contract: no mechanism for a required dimension → no process.
+
+    Written as a call site would use the API — acquire, then run — so the proof
+    covers the whole path and not just the raising function.
+    """
+    enforcing(monkeypatch)
+    # The only mechanism available cannot do filesystem, which is required.
+    install(monkeypatch, mechanism(
+        "group_only", 10, {containment.PROCESS_TREE, containment.WALL_CLOCK},
+    ))
+
+    spec = containment.agent_spec(workspace, {"PATH": "/usr/bin"}, 5)
+    try:
+        grant = containment.acquire(spec, owner="session-1")
+    except containment.ContainmentUnavailable as exc:
+        result = containment.unavailable_tool_result(exc, tool="bash")
+    else:  # pragma: no cover - the point of the test is that this is unreachable
+        result = await containment.run(grant, "echo hello")
+
+    assert no_spawn == [], "an uncontainable request reached a spawn"
+    assert result["exit_code"] == 1
+    assert "command not executed" in result["error"]
+    assert "filesystem" in result["error"]
+    assert result["containment"]["contained"] is False
+    assert result["containment"]["executed"] is False
+    assert result["containment"]["unenforced_required"] == ["filesystem"]
+    # A run that could not be contained is distinguishable from a contained run
+    # that failed: there is no output key at all, and `executed` is explicit.
+    assert "output" not in result
+
+
+def test_unavailable_names_every_missing_dimension_and_the_mechanism_tried(
+    monkeypatch, workspace,
+):
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("group_only", 10, {containment.WALL_CLOCK}))
+    spec = containment.agent_spec(workspace, {}, 5)
+    with pytest.raises(containment.ContainmentUnavailable) as caught:
+        containment.acquire(spec, owner="session-1")
+    assert caught.value.missing == frozenset({
+        containment.FILESYSTEM, containment.PROCESS_TREE,
+    })
+    assert caught.value.mechanism_tried == "group_only"
+    assert "containment unavailable" in str(caught.value)
+
+
+def test_no_mechanism_at_all_still_refuses_rather_than_running(
+    monkeypatch, workspace, no_spawn,
+):
+    """With nothing available there is no weaker thing to fall back to."""
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("absent", 10, containment.DIMENSIONS, available=False))
+    with pytest.raises(containment.ContainmentUnavailable) as caught:
+        containment.acquire(containment.agent_spec(workspace, {}, 5), owner="s")
+    assert caught.value.mechanism_tried == "none"
+    assert no_spawn == []
+
+
+async def test_a_forged_grant_cannot_buy_a_spawn(monkeypatch, workspace, no_spawn):
+    """run() re-checks the postcondition at the point of effect.
+
+    A grant is a plain record, so a caller could construct one claiming
+    dimensions it does not have. run() refuses it rather than trusting the
+    record it was handed.
+    """
+    enforcing(monkeypatch)
+    spec = containment.agent_spec(workspace, {}, 5)
+    forged = containment.ContainmentGrant(
+        id="forged",
+        mechanism="bubblewrap",
+        workspace=workspace,
+        enforced=frozenset({containment.WALL_CLOCK}),
+        degraded=(),
+        unenforced_required=(),            # the lie: claims nothing is missing
+        owner="session-1",
+        mode=containment.MODE_ENFORCING,
+        spec=spec,
+    )
+    with pytest.raises(containment.ContainmentUnavailable):
+        await containment.run(forged, "echo hello")
+    assert no_spawn == []
+
+
+# ── Best-effort dimensions degrade, they do not refuse ──────────────────────
+def test_unavailable_best_effort_dimension_is_reported_not_refused(monkeypatch, workspace):
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("fake", 10, containment.DEFAULT_REQUIRED))
+    spec = containment.agent_spec(
+        workspace, {}, 5, network=containment.NETWORK_NONE, max_memory_bytes=1 << 30,
+    )
+    grant = containment.acquire(spec, owner="session-1")
+    assert grant.contained is True
+    assert grant.degraded == (containment.MEMORY, containment.NETWORK)
+    # And it is reported as fact in the model-visible block, never as permission.
+    block = grant.to_dict()
+    assert block["degraded"] == ["memory", "network"]
+    assert block["contained"] is True
+    assert "env" not in block
+
+
+def test_a_degraded_dimension_is_never_also_enforced(monkeypatch, workspace):
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("fake", 10, containment.DEFAULT_REQUIRED))
+    spec = containment.agent_spec(workspace, {}, 5, max_processes=16)
+    grant = containment.acquire(spec, owner="session-1")
+    assert set(grant.degraded).isdisjoint(grant.enforced)
+
+
+# ── Mechanism selection: strongest first, command-independent ───────────────
+def test_selection_is_strongest_first(monkeypatch, workspace):
+    enforcing(monkeypatch)
+    install(
+        monkeypatch,
+        mechanism("weak", 10, containment.DEFAULT_REQUIRED),
+        mechanism("strong", 30, containment.DIMENSIONS),
+        mechanism("middle", 20, containment.DEFAULT_REQUIRED),
+    )
+    grant = containment.acquire(spec_for(workspace), owner="session-1")
+    assert grant.mechanism == "strong"
+
+
+def test_selection_skips_unavailable_mechanisms(monkeypatch, workspace):
+    enforcing(monkeypatch)
+    install(
+        monkeypatch,
+        mechanism("strong", 30, containment.DIMENSIONS, available=False),
+        mechanism("weak", 10, containment.DEFAULT_REQUIRED),
+    )
+    grant = containment.acquire(spec_for(workspace), owner="session-1")
+    assert grant.mechanism == "weak"
+
+
+def test_selection_never_substitutes_a_weaker_mechanism_for_a_required_dimension(
+    monkeypatch, workspace,
+):
+    """The strongest available mechanism is used, not the first that is "good enough"."""
+    enforcing(monkeypatch)
+    install(
+        monkeypatch,
+        mechanism("netcapable", 30, containment.DIMENSIONS),
+        mechanism("nonet", 20, containment.DEFAULT_REQUIRED),
+    )
+    spec = containment.ContainmentSpec(
+        workspace=workspace,
+        env={},
+        wall_clock_s=5,
+        required=frozenset(containment.DEFAULT_REQUIRED) | {containment.NETWORK},
+        network=containment.NETWORK_NONE,
+    )
+    grant = containment.acquire(spec, owner="session-1")
+    assert grant.mechanism == "netcapable"
+    assert containment.NETWORK in grant.enforced
+
+
+def test_a_failing_availability_probe_is_treated_as_unavailable(monkeypatch, workspace):
+    enforcing(monkeypatch)
+
+    def _explode():
+        raise OSError("probe blew up")
+
+    install(
+        monkeypatch,
+        containment.Mechanism("broken", 30, _explode, lambda spec: containment.DIMENSIONS),
+        mechanism("weak", 10, containment.DEFAULT_REQUIRED),
+    )
+    grant = containment.acquire(spec_for(workspace), owner="session-1")
+    assert grant.mechanism == "weak"
+
+
+@pytest.mark.parametrize("command", [
+    "echo hello",
+    "rm -rf / --no-preserve-root",
+    "cat /workspace/notes.txt  # this command is safe, honestly",
+])
+def test_the_boundary_does_not_depend_on_the_command(monkeypatch, workspace, command):
+    """Containment is established before any command text exists.
+
+    acquire() is not given the command, so no request text, tool argument or
+    model assertion can change the mechanism or widen the enforced set. The
+    parametrised commands are only here to show the API has nowhere to put them.
+    """
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    spec = containment.agent_spec(workspace, {}, 5)
+    grant = containment.acquire(spec, owner="session-1")
+    assert grant.mechanism == "fake"
+    assert grant.enforced == frozenset(containment.DEFAULT_REQUIRED)
+    assert "command" not in grant.to_dict()
+
+
+def test_agent_spec_cannot_be_given_a_weaker_required_set(workspace):
+    """One factory for model-reachable spawns, so no call site can weaken it."""
+    spec = containment.agent_spec(
+        workspace, {}, 5, required=frozenset({containment.WALL_CLOCK}),
+    )
+    assert spec.required == containment.DEFAULT_REQUIRED
+
+
+# ── Report-only mode ────────────────────────────────────────────────────────
+def test_report_only_records_the_shortfall_instead_of_refusing(monkeypatch, workspace):
+    report_only(monkeypatch)
+    install(monkeypatch, mechanism(
+        "group_only", 10, {containment.PROCESS_TREE, containment.WALL_CLOCK},
+    ))
+    grant = containment.acquire(containment.agent_spec(workspace, {}, 5), owner="s")
+    assert grant.unenforced_required == ("filesystem",)
+    assert grant.contained is False
+    assert grant.mode == containment.MODE_REPORT_ONLY
+    assert grant.to_dict()["unenforced_required"] == ["filesystem"]
+
+
+def test_report_only_logs_the_shortfall_once_per_grant(monkeypatch, workspace, caplog):
+    report_only(monkeypatch)
+    install(monkeypatch, mechanism("group_only", 10, {containment.WALL_CLOCK}))
+    with caplog.at_level("WARNING", logger="src.containment"):
+        grant = containment.acquire(containment.agent_spec(workspace, {}, 5), owner="s")
+    messages = [record.getMessage() for record in caplog.records]
+    assert sum("NOT contained" in message for message in messages) == 1
+    assert grant.id in messages[0]
+
+
+def test_the_two_modes_differ_only_in_whether_the_shortfall_refuses(monkeypatch, workspace):
+    install(monkeypatch, mechanism("group_only", 10, {containment.WALL_CLOCK}))
+    spec = containment.agent_spec(workspace, {}, 5)
+
+    report_only(monkeypatch)
+    reported = containment.acquire(spec, owner="s")
+
+    enforcing(monkeypatch)
+    with pytest.raises(containment.ContainmentUnavailable) as caught:
+        containment.acquire(spec, owner="s")
+
+    assert frozenset(reported.unenforced_required) == caught.value.missing
+
+
+def test_report_only_is_the_shipped_default():
+    """Pinned deliberately: merging this must not change behaviour on a host
+    without bubblewrap. Flipping it is a one-line diff, reviewed as one."""
+    assert containment.CONTAINMENT_MODE == containment.MODE_REPORT_ONLY
+
+
+# ── Spec validation: caller bugs raise in both modes ────────────────────────
+@pytest.mark.parametrize("mode", [containment.MODE_ENFORCING, containment.MODE_REPORT_ONLY])
+@pytest.mark.parametrize("overrides, fragment", [
+    ({"required": frozenset({"telepathy"})}, "unknown required dimension"),
+    ({"required": frozenset({containment.NETWORK})}, "does not request it"),
+    ({"required": frozenset({containment.MEMORY})}, "does not request it"),
+    ({"wall_clock_s": 0}, "must be positive"),
+    ({"wall_clock_s": -1}, "must be positive"),
+    ({"max_output_bytes": 0}, "max_output_bytes must be positive"),
+    ({"max_memory_bytes": 0}, "max_memory_bytes must be a positive int"),
+    ({"max_processes": -4}, "max_processes must be a positive int"),
+    ({"network": "maybe"}, "network must be"),
+    ({"env": {"A": 1}}, "env keys and values must be str"),
+    ({"env": {"A": "x\x00y"}}, "must not contain NUL"),
+    ({"writable_extra": ("relative/path",)}, "must be absolute"),
+    ({"writable_extra": ("/",)}, "reserved path"),
+    ({"readonly_extra": ("/workspace",)}, "reserved path"),
+])
+def test_a_malformed_spec_raises_in_both_modes(
+    monkeypatch, workspace, mode, overrides, fragment,
+):
+    monkeypatch.setattr(containment, "CONTAINMENT_MODE", mode)
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    spec = spec_for(workspace, **overrides)
+    with pytest.raises(ValueError, match=fragment):
+        containment.acquire(spec, owner="session-1")
+
+
+@pytest.mark.parametrize("bad_workspace, fragment", [
+    ("", "non-empty path"),
+    ("relative/ws", "must be absolute"),
+])
+def test_a_malformed_workspace_raises(monkeypatch, bad_workspace, fragment):
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    spec = containment.ContainmentSpec(
+        workspace=bad_workspace, env={}, wall_clock_s=5,
+    )
+    with pytest.raises(ValueError, match=fragment):
+        containment.acquire(spec, owner="session-1")
+
+
+def test_a_workspace_that_is_not_a_directory_raises(monkeypatch, tmp_path):
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    missing = tmp_path / "nope"
+    spec = containment.ContainmentSpec(workspace=str(missing), env={}, wall_clock_s=5)
+    with pytest.raises(ValueError, match="not a directory"):
+        containment.acquire(spec, owner="session-1")
+
+
+def test_a_grant_without_an_owner_raises(monkeypatch, workspace):
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    with pytest.raises(ValueError, match="needs an owner"):
+        containment.acquire(spec_for(workspace), owner="   ")
+
+
+def test_the_child_environment_cannot_be_edited_after_acquire(monkeypatch, workspace):
+    """env is part of the boundary, so the caller's dict is copied and frozen."""
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    caller_env = {"PATH": "/usr/bin"}
+    grant = containment.acquire(
+        spec_for(workspace, env=caller_env), owner="session-1",
+    )
+    caller_env["LD_PRELOAD"] = "/tmp/evil.so"
+    assert dict(grant.spec.env) == {"PATH": "/usr/bin"}
+    with pytest.raises(TypeError):
+        grant.spec.env["LD_PRELOAD"] = "/tmp/evil.so"
+
+
+# ── Durable records: one owner, one record ─────────────────────────────────
+def test_a_grant_is_recorded_with_its_owner_and_declared_limits(
+    monkeypatch, workspace, _isolated_store,
+):
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    spec = containment.agent_spec(workspace, {}, 7, max_processes=8)
+    grant = containment.acquire(spec, owner="session-42")
+
+    active = containment.active_grants()
+    assert [record["id"] for record in active] == [grant.id]
+    record = active[0]
+    assert record["owner"] == "session-42"
+    assert record["wall_clock_s"] == 7
+    assert record["max_processes"] == 8
+    assert record["required"] == sorted(containment.DEFAULT_REQUIRED)
+    assert record["pid"] is None
+
+
+def test_releasing_a_grant_with_no_process_clears_it_from_active(monkeypatch, workspace):
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    grant = containment.acquire(containment.agent_spec(workspace, {}, 5), owner="s")
+    outcome = containment.release(grant)
+    assert outcome.dead is True
+    assert outcome.escalated is False
+    assert outcome.survivors == ()
+    assert containment.active_grants() == []
+
+
+def test_forget_drops_a_record(monkeypatch, workspace):
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    grant = containment.acquire(containment.agent_spec(workspace, {}, 5), owner="s")
+    containment.forget(grant.id)
+    assert containment.active_grants() == []
+
+
+def test_an_unwritable_store_does_not_take_out_execution(monkeypatch, workspace, caplog):
+    """The record is observability, not a containment dimension.
+
+    Refusing an authorized command because a journal file could not be written
+    would be a worse failure than running it, so this degrades loudly and the
+    grant still describes the boundary accurately.
+    """
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+
+    def _explode(*args, **kwargs):
+        raise OSError("read-only filesystem")
+
+    monkeypatch.setattr(containment, "atomic_write_json", _explode)
+    with caplog.at_level("WARNING", logger="src.containment"):
+        grant = containment.acquire(containment.agent_spec(workspace, {}, 5), owner="s")
+    assert grant.contained is True
+    assert any("could not persist" in record.getMessage() for record in caplog.records)
+
+
+def test_a_corrupt_store_does_not_take_out_execution(monkeypatch, workspace, _isolated_store):
+    enforcing(monkeypatch)
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    _isolated_store.write_text("{ this is not json", encoding="utf-8")
+    grant = containment.acquire(containment.agent_spec(workspace, {}, 5), owner="s")
+    assert [record["id"] for record in containment.active_grants()] == [grant.id]
+
+
+# ── Execution that leaves the box is declared, not pretended ───────────────
+async def test_an_external_bridge_grant_claims_nothing_and_cannot_be_run_locally(
+    monkeypatch, workspace, no_spawn,
+):
+    enforcing(monkeypatch)
+    spec = containment.agent_spec(workspace, {}, 5)
+    grant = containment.declare_external_bridge(
+        spec, owner="session-1", endpoint="http://127.0.0.1:8777/exec",
+    )
+    assert grant.mechanism == "external_bridge"
+    assert grant.enforced == frozenset()
+    assert grant.external is True
+    assert grant.contained is False
+    assert grant.to_dict()["external"] is True
+    with pytest.raises(ValueError, match="does not own"):
+        await containment.run(grant, "echo hello")
+    assert no_spawn == []

tests/test_containment_process_tree.py

@@ -0,0 +1,378 @@
+"""Teardown through the containment boundary, against real processes.
+
+The headline case is the one that fails on an unmodified baseline: a command
+that backgrounds a grandchild and then times out leaves the grandchild running,
+while the tool result claims "process killed". These tests pin that the boundary
+signals the whole process group and verifies death before reporting it.
+
+POSIX only — the Windows path walks the tree with ``taskkill /T /F`` and has no
+host here to run on, which is stated in the PR rather than skipped silently.
+"""
+
+import os
+import signal
+import subprocess
+import sys
+import time
+from dataclasses import replace
+
+import pytest
+
+from core.platform_compat import pid_alive
+from src import containment
+
+pytestmark = pytest.mark.skipif(
+    sys.platform.startswith("win"), reason="POSIX process groups; Windows path untested here"
+)
+
+
+@pytest.fixture(autouse=True)
+def _isolated_store(tmp_path, monkeypatch):
+    store = tmp_path / "containment_grants.json"
+    monkeypatch.setattr(containment, "_store_path", lambda: store)
+    return store
+
+
+@pytest.fixture(autouse=True)
+def _real_process_group_mechanism(monkeypatch):
+    """Pin the mechanism to the real POSIX process group.
+
+    Not a fake: this is the mechanism shipped in ``MECHANISMS``, selected by
+    name so the test behaves the same on a host that happens to have bubblewrap
+    installed. Filesystem containment is bubblewrap's job and is not what these
+    tests are about.
+    """
+    selected = [item for item in containment.MECHANISMS if item.name == "process_group"]
+    assert selected, "process_group mechanism disappeared from MECHANISMS"
+    monkeypatch.setattr(containment, "MECHANISMS", tuple(selected))
+    monkeypatch.setattr(containment, "CONTAINMENT_MODE", containment.MODE_ENFORCING)
+
+
+@pytest.fixture
+def workspace(tmp_path):
+    path = tmp_path / "ws"
+    path.mkdir()
+    return str(path)
+
+
+def tree_spec(workspace, **kwargs):
+    """A spec requiring exactly what a process group can give."""
+    kwargs.setdefault("env", {"PATH": "/usr/bin:/bin:/usr/sbin:/sbin"})
+    kwargs.setdefault("wall_clock_s", 1)
+    return containment.ContainmentSpec(
+        workspace=workspace,
+        required=frozenset({containment.PROCESS_TREE, containment.WALL_CLOCK}),
+        **kwargs,
+    )
+
+
+def read_pid(path, *, timeout=5.0):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            text = path.read_text(encoding="utf-8").strip()
+        except OSError:
+            text = ""
+        if text.isdigit():
+            return int(text)
+        time.sleep(0.02)
+    raise AssertionError(f"{path} never received a pid")
+
+
+def gone(pid, *, timeout=5.0):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if not pid_alive(pid):
+            return True
+        time.sleep(0.02)
+    return not pid_alive(pid)
+
+
+# ── The regression this lane exists to close ────────────────────────────────
+async def test_a_timeout_leaves_no_surviving_grandchild(tmp_path, workspace):
+    """A backgrounded grandchild does not survive the wall-clock kill.
+
+    On a baseline spawn site the wrapper shell is killed with ``proc.kill()``
+    and the grandchild keeps running, unowned and unreaped, while the tool
+    result says the process was killed.
+    """
+    pidfile = tmp_path / "grandchild.pid"
+    command = f"bash -c 'sleep 60 & echo $! > {pidfile}'; sleep 60"
+
+    grant = containment.acquire(tree_spec(workspace), owner="session-1")
+    result = await containment.run(grant, command)
+
+    grandchild = read_pid(pidfile)
+    assert result.timed_out is True
+    assert gone(grandchild), f"grandchild {grandchild} survived the timeout kill"
+    assert result.release is not None
+    assert result.release.dead is True
+    assert result.release.survivors == ()
+
+
+async def test_the_timeout_outcome_is_observed_not_asserted(tmp_path, workspace):
+    """``dead`` reflects a verified empty process group, not a signal that was sent."""
+    pidfile = tmp_path / "child.pid"
+    command = f"echo $$ > {pidfile}; sleep 60"
+    grant = containment.acquire(tree_spec(workspace), owner="session-1")
+    result = await containment.run(grant, command)
+    leader = read_pid(pidfile)
+    assert result.timed_out is True
+    assert result.release.dead is True
+    assert gone(leader)
+    assert containment._group_present(result.grant.pgid) is False
+
+
+async def test_a_clean_exit_tears_down_anything_left_behind(tmp_path, workspace):
+    """A command that returns while leaving a background process does not leak it.
+
+    The leftover closes its inherited pipes (``>/dev/null 2>&1``) so the command
+    really does complete: a background process still holding the output pipes
+    keeps the grant open until the wall clock, which is the previous test's case
+    rather than this one's.
+    """
+    pidfile = tmp_path / "leftover.pid"
+    command = f"sleep 60 >/dev/null 2>&1 & echo $! > {pidfile}; exit 0"
+    grant = containment.acquire(tree_spec(workspace, wall_clock_s=10), owner="session-1")
+    result = await containment.run(grant, command)
+    leftover = read_pid(pidfile)
+    assert result.timed_out is False
+    assert result.exit_code == 0
+    assert gone(leftover), f"background process {leftover} outlived its grant"
+    assert result.release.dead is True
+
+
+# ── Escalation ──────────────────────────────────────────────────────────────
+def test_release_escalates_to_sigkill_and_reports_only_verified_death(tmp_path, workspace):
+    """A SIGTERM-ignoring tree is escalated, and ``dead`` is set only once gone."""
+    # The child announces itself only after installing the handler. Without that
+    # the test races process startup and sometimes measures a child that was
+    # still using the default SIGTERM disposition.
+    ready = tmp_path / "ignoring-sigterm"
+    code = (
+        "import signal, time\n"
+        "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n"
+        f"open({str(ready)!r}, 'w').write('x')\n"
+        "time.sleep(60)\n"
+    )
+    proc = subprocess.Popen(
+        [sys.executable, "-c", code],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+        start_new_session=True,
+    )
+    try:
+        deadline = time.monotonic() + 10
+        while time.monotonic() < deadline and not ready.exists():
+            time.sleep(0.02)
+        assert ready.exists(), "child never installed its SIGTERM handler"
+        grant = containment.acquire(tree_spec(workspace), owner="session-1")
+        grant = replace(grant, pid=proc.pid, pgid=os.getpgid(proc.pid))
+        outcome = containment.release(grant, grace_s=0.3)
+        proc.wait(timeout=5)
+        assert outcome.escalated is True
+        assert outcome.dead is True
+        assert outcome.survivors == ()
+    finally:
+        if proc.poll() is None:  # pragma: no cover - only on an unexpected failure
+            proc.kill()
+            proc.wait(timeout=5)
+
+
+def test_release_does_not_escalate_a_cooperative_tree(workspace):
+    proc = subprocess.Popen(
+        [sys.executable, "-c", "import time; time.sleep(60)"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+        start_new_session=True,
+    )
+    try:
+        grant = containment.acquire(tree_spec(workspace), owner="session-1")
+        grant = replace(grant, pid=proc.pid, pgid=os.getpgid(proc.pid))
+        outcome = containment.release(grant, grace_s=2.0)
+        proc.wait(timeout=5)
+        assert outcome.dead is True
+        assert outcome.escalated is False
+    finally:
+        if proc.poll() is None:  # pragma: no cover
+            proc.kill()
+            proc.wait(timeout=5)
+
+
+def test_a_surviving_tree_keeps_its_record_active(workspace, monkeypatch):
+    """A record moves to released only on verified death.
+
+    With teardown unable to signal anything, ``release`` must report
+    ``dead=False`` with the survivors named, and must not mark the grant
+    released — the inverse of marking a job killed without checking.
+    """
+    proc = subprocess.Popen(
+        [sys.executable, "-c", "import time; time.sleep(60)"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+        start_new_session=True,
+    )
+    try:
+        grant = containment.acquire(tree_spec(workspace), owner="session-1")
+        grant = replace(grant, pid=proc.pid, pgid=os.getpgid(proc.pid))
+        monkeypatch.setattr(containment, "_signal_tree", lambda *args, **kwargs: None)
+        outcome = containment.release(grant, grace_s=0.2)
+        assert outcome.dead is False
+        assert outcome.escalated is True
+        assert proc.pid in outcome.survivors
+        active = {record["id"] for record in containment.active_grants()}
+        assert grant.id in active
+        assert pid_alive(proc.pid) is True
+    finally:
+        proc.kill()
+        proc.wait(timeout=5)
+
+
+def test_release_never_signals_the_servers_own_process_group(workspace, monkeypatch):
+    """If setsid had not applied, killpg would take the server down with the child.
+
+    Driven by handing teardown our own group id, which is exactly the state a
+    failed setsid would leave behind.
+    """
+    sent = []
+    monkeypatch.setattr(os, "killpg", lambda pgid, sig: sent.append((pgid, sig)))
+    monkeypatch.setattr(os, "kill", lambda pid, sig: sent.append(("pid", pid, sig)))
+    containment._signal_tree(os.getpid(), os.getpgid(0), signal.SIGTERM)
+    assert all(entry[0] != os.getpgid(0) for entry in sent), sent
+    assert sent == [("pid", os.getpid(), signal.SIGTERM)]
+
+
+def test_our_own_group_is_never_reported_as_a_childs_group():
+    assert containment._group_present(os.getpgid(0)) is False
+
+
+# ── run(): the contained happy path ─────────────────────────────────────────
+async def test_run_returns_output_exit_code_and_a_released_record(workspace):
+    grant = containment.acquire(tree_spec(workspace, wall_clock_s=10), owner="session-9")
+    result = await containment.run(grant, "echo contained; exit 3")
+    assert result.stdout == "contained\n"
+    assert result.exit_code == 3
+    assert result.timed_out is False
+    assert result.output_truncated is False
+    assert result.grant.pid is not None
+    assert containment.active_grants() == []
+
+
+async def test_run_executes_in_the_workspace_and_with_the_declared_env_only(workspace):
+    grant = containment.acquire(
+        tree_spec(workspace, wall_clock_s=10, env={"PATH": "/usr/bin:/bin", "MARK": "yes"}),
+        owner="session-9",
+    )
+    result = await containment.run(grant, 'pwd; echo "MARK=$MARK"; echo "HOME=${HOME:-unset}"')
+    assert result.exit_code == 0
+    assert os.path.realpath(workspace) == os.path.realpath(result.stdout.splitlines()[0])
+    assert "MARK=yes" in result.stdout
+    # The child env is exactly what the spec declared, never an implicit
+    # inherit, so the server's own environment does not leak into it.
+    assert "HOME=unset" in result.stdout
+
+
+async def test_run_caps_output_and_says_so(workspace):
+    grant = containment.acquire(
+        tree_spec(workspace, wall_clock_s=10, max_output_bytes=16), owner="session-9",
+    )
+    result = await containment.run(grant, "printf 'x%.0s' $(seq 1 500); echo")
+    assert result.output_truncated is True
+    assert len(result.stdout.encode("utf-8")) <= 16
+
+
+async def test_run_accepts_an_argv_command_without_a_shell(workspace):
+    grant = containment.acquire(tree_spec(workspace, wall_clock_s=10), owner="session-9")
+    result = await containment.run(
+        grant, [sys.executable, "-c", "print('argv path')"], argv=True,
+    )
+    assert result.exit_code == 0
+    assert result.stdout.strip() == "argv path"
+
+
+async def test_run_feeds_stdin_when_given(workspace):
+    grant = containment.acquire(tree_spec(workspace, wall_clock_s=10), owner="session-9")
+    result = await containment.run(grant, "cat", stdin=b"piped\n")
+    assert result.exit_code == 0
+    assert result.stdout == "piped\n"
+
+
+@pytest.mark.parametrize("command, argv", [("   ", False), ([], True)])
+async def test_run_rejects_an_empty_command(workspace, command, argv):
+    grant = containment.acquire(tree_spec(workspace, wall_clock_s=10), owner="session-9")
+    with pytest.raises(ValueError, match="empty"):
+        await containment.run(grant, command, argv=argv)
+
+
+# ── Resource limits, where the platform provides them ───────────────────────
+async def test_a_process_count_limit_is_applied_to_the_child(workspace):
+    """RLIMIT_NPROC is set in the child, so the ceiling is real where it is claimed."""
+    spec = containment.ContainmentSpec(
+        workspace=workspace,
+        env={"PATH": "/usr/bin:/bin"},
+        wall_clock_s=10,
+        required=frozenset({
+            containment.PROCESS_TREE, containment.WALL_CLOCK, containment.PROCESS_COUNT,
+        }),
+        max_processes=64,
+    )
+    grant = containment.acquire(spec, owner="session-9")
+    assert containment.PROCESS_COUNT in grant.enforced
+    result = await containment.run(
+        grant,
+        [sys.executable, "-c",
+         "import resource; print(resource.getrlimit(resource.RLIMIT_NPROC))"],
+        argv=True,
+    )
+    assert result.exit_code == 0
+    assert result.stdout.strip() == "(64, 64)"
+
+
+async def test_a_memory_limit_is_claimed_only_where_it_can_be_applied(workspace):
+    """The claim and the reality agree, on whichever platform this runs.
+
+    macOS reports an infinite RLIMIT_AS hard limit and then refuses to lower it,
+    so `memory` must come back unenforced there rather than enforced-and-crashing.
+    Written to assert the consistency rather than the platform, so it is a real
+    test on Linux and a real test here.
+    """
+    limit = 2 * 1024 * 1024 * 1024
+    spec = containment.ContainmentSpec(
+        workspace=workspace,
+        env={"PATH": "/usr/bin:/bin"},
+        wall_clock_s=10,
+        required=frozenset({containment.PROCESS_TREE, containment.WALL_CLOCK}),
+        max_memory_bytes=limit,
+    )
+    grant = containment.acquire(spec, owner="session-9")
+    probe = [sys.executable, "-c",
+             "import resource; print(resource.getrlimit(resource.RLIMIT_AS)[0])"]
+    result = await containment.run(grant, probe, argv=True)
+    assert result.exit_code == 0, result.stderr
+    if containment.MEMORY in grant.enforced:
+        assert result.stdout.strip() == str(limit)
+    else:
+        assert containment.MEMORY in grant.degraded
+        assert result.stdout.strip() != str(limit)
+
+
+def test_an_unenforceable_required_limit_refuses_instead_of_crashing_the_spawn(workspace):
+    """A limit this platform cannot apply is refused at acquire, not in preexec_fn.
+
+    Skipped where the platform *can* apply it, since then there is nothing to
+    refuse.
+    """
+    if containment._ADDRESS_SPACE_LIMIT_SUPPORTED:
+        pytest.skip("this platform can lower RLIMIT_AS, so there is no shortfall")
+    spec = containment.ContainmentSpec(
+        workspace=workspace,
+        env={"PATH": "/usr/bin:/bin"},
+        wall_clock_s=10,
+        required=frozenset({
+            containment.PROCESS_TREE, containment.WALL_CLOCK, containment.MEMORY,
+        }),
+        max_memory_bytes=2 * 1024 * 1024 * 1024,
+    )
+    with pytest.raises(containment.ContainmentUnavailable) as caught:
+        containment.acquire(spec, owner="session-9")
+    assert caught.value.missing == frozenset({containment.MEMORY})

website/configuration-reference.md

@@ -52,7 +52,7 @@ The source tree reads **108** `ODYSSEUS_*` variables: 78 an operator may want to
 | Variable | Default | Read in | What it does |
 |---|---|---|---|
 | `ODYSSEUS_DATA_DIR` | `get_default_data_dir()` | `src/constants.py:56` (+1 more) | Root directory for every persisted file. Prefer this over the per-path overrides; the rest of `src/constants.py` derives from it. |
-| `ODYSSEUS_MAIL_ATTACHMENTS_DIR` | `os.path.join(DATA_DIR, 'mail-attachments')` | `src/constants.py:102` | Dedicated override for the mail attachment store, which otherwise lives under the data directory. |
+| `ODYSSEUS_MAIL_ATTACHMENTS_DIR` | `os.path.join(DATA_DIR, 'mail-attachments')` | `src/constants.py:103` | Dedicated override for the mail attachment store, which otherwise lives under the data directory. |
 
 ### Model routing and providers
 
@@ -167,7 +167,7 @@ The source tree reads **108** `ODYSSEUS_*` variables: 78 an operator may want to
 
 | Variable | Default | Read in | What it does |
 |---|---|---|---|
-| `ODYSSEUS_INTERNAL_BASE` | *unset* | `src/constants.py:178` | Base URL the in-app tool layer uses for loopback HTTP calls. Set it when the app is not reachable at the port it thinks it is bound to. |
+| `ODYSSEUS_INTERNAL_BASE` | *unset* | `src/constants.py:179` | Base URL the in-app tool layer uses for loopback HTTP calls. Set it when the app is not reachable at the port it thinks it is bound to. |
 | `ODYSSEUS_INTERNAL_TOKEN` | *unset* | `core/middleware.py:20` | Security-relevant. Token that lets the in-app tool layer reach admin-gated routes over loopback. Unset generates a fresh per-process token, which is what you want unless something outside the process needs the same value. |
 
 ### Integrations (Claude, Codex)