fix(runtime): resolve Wave 3-S audit findings (P2-1, P2-2, P2-3)

- P2-1: reject non-process PID values (None, 0, negative integers) in pid_alive
  without invoking underlying process probe.
- P2-2: truthfully represent production external bridge executions as uncontained,
  external, non-authoritative grants carrying sanitized endpoint metadata.
- P2-3: reject writable_extra overlay bindings over protected system roots and
  their descendants while preserving legitimate scratch destinations.

core/platform_compat.py

@@ -94,7 +94,13 @@ def pid_alive(pid: Optional[int]) -> bool:
     the process it is checking. We instead open the process and read its exit
     code via the Win32 API.
     """
-    if not pid:
+    if pid is None:
+        return False
+    try:
+        pid_int = int(pid)
+    except (TypeError, ValueError):
+        return False
+    if pid_int <= 0:
         return False
     if IS_WINDOWS:
         import ctypes
@@ -104,7 +110,7 @@ def pid_alive(pid: Optional[int]) -> bool:
         STILL_ACTIVE = 259
         kernel32 = ctypes.windll.kernel32
         handle = kernel32.OpenProcess(
-            PROCESS_QUERY_LIMITED_INFORMATION, False, int(pid)
+            PROCESS_QUERY_LIMITED_INFORMATION, False, pid_int
         )
         if not handle:
             return kernel32.GetLastError() != 87  # ERROR_INVALID_PARAMETER: PID absent
@@ -116,7 +122,7 @@ def pid_alive(pid: Optional[int]) -> bool:
         finally:
             kernel32.CloseHandle(handle)
     try:
-        os.kill(pid, 0)
+        os.kill(pid_int, 0)
         return True
     except ProcessLookupError:
         return False

src/agent_tools/subprocess_tools.py

@@ -653,6 +653,16 @@ class HostShellTool:
             requested_timeout = 30
         timeout = max(1, min(requested_timeout, 120))
 
+        from src import containment
+        from src.tool_execution import agent_cwd
+
+        sanitized_endpoint = f"{parsed.scheme}://{parsed.netloc}{parsed.path}" if parsed.scheme and parsed.netloc else "host_shell_bridge"
+        owner = str(ctx.get("session_id") or ctx.get("owner") or "host_shell")
+        spec = _owned_spec(agent_cwd(), ctx.get("subproc_env"), timeout)
+        grant = containment.declare_external_bridge(spec, owner=owner, endpoint=sanitized_endpoint)
+        boundary = grant.to_dict()
+        boundary["executed"] = False
+
         request_body: dict[str, object] = {"timeout": timeout}
         request_id = ""
         if job_id:
@@ -676,6 +686,8 @@ class HostShellTool:
                     return {
                         "error": f"host_shell: bridge returned HTTP {resp.status_code}",
                         "exit_code": 1,
+                        "host_bridge": "tui",
+                        "containment": boundary,
                     }
                 data = resp.json()
 
@@ -702,6 +714,8 @@ class HostShellTool:
                             return {
                                 "error": f"host_shell: bridge returned HTTP {poll.status_code}",
                                 "exit_code": 1,
+                                "host_bridge": "tui",
+                                "containment": boundary,
                             }
                         data = poll.json()
                         if not isinstance(data, dict):
@@ -732,15 +746,16 @@ class HostShellTool:
                 task.add_done_callback(_HOST_SHELL_CANCEL_TASKS.discard)
             raise
         except Exception as e:
-            return {"error": f"host_shell: bridge call failed: {e}", "exit_code": 1}
+            return {"error": f"host_shell: bridge call failed: {e}", "exit_code": 1, "containment": boundary}
 
         if not isinstance(data, dict):
-            return {"error": "host_shell: bridge returned invalid payload", "exit_code": 1}
+            return {"error": "host_shell: bridge returned invalid payload", "exit_code": 1, "containment": boundary}
         if data.get("error"):
             return {
                 "error": _truncate(str(data["error"]), MAX_OUTPUT_CHARS),
                 "exit_code": 1,
                 "host_bridge": "tui",
+                "containment": boundary,
             }
         stdout = str(data.get("stdout") or data.get("output") or "")
         stderr = str(data.get("stderr") or "")
@@ -754,15 +769,18 @@ class HostShellTool:
                 "error": "host_shell: bridge returned an invalid exit_code",
                 "exit_code": 1,
                 "host_bridge": "tui",
+                "containment": boundary,
             }
         exit_code = raw_exit_code
         output = stdout.rstrip()
         if stderr.strip():
             output = (output + "\nSTDERR: " + stderr.strip()).strip() if output else "STDERR: " + stderr.strip()
+        boundary["executed"] = True
         result = {
             "output": _truncate(output, MAX_OUTPUT_CHARS) or "(no output)",
             "exit_code": exit_code,
             "host_bridge": "tui",
+            "containment": boundary,
         }
         for key in ("detached", "job_id", "status", "running", "finished", "cwd"):
             if key in data:

src/containment.py

@@ -121,9 +121,35 @@ _DEATH_POLL_S = 0.05
 # private /tmp or the workspace itself with a host directory would undo the
 # namespace from inside the argv that builds it.
 _RESERVED_BIND_DESTS = frozenset({
-    "/", "/tmp", "/proc", "/dev", "/sys", WORKSPACE_MOUNT,
+    "/", "/tmp", "/home", "/proc", "/dev", "/sys", WORKSPACE_MOUNT,
 })
 
+# System hierarchies where a writable overlay would invalidate the boundary
+# established by bubblewrap. Reject both exact roots and all descendants.
+_PROTECTED_WRITABLE_HIERARCHIES = frozenset({
+    "/etc",
+    "/usr",
+    "/bin",
+    "/sbin",
+    "/lib",
+    "/lib64",
+    "/proc",
+    "/dev",
+    "/sys",
+    "/root",
+    WORKSPACE_MOUNT,
+})
+
+
+def _is_protected_writable_destination(path: str) -> bool:
+    normalized = os.path.abspath(path)
+    if normalized in _RESERVED_BIND_DESTS:
+        return True
+    for root in _PROTECTED_WRITABLE_HIERARCHIES:
+        if normalized == root or normalized.startswith(root.rstrip(os.sep) + os.sep):
+            return True
+    return False
+
 # WORKSPACE_MOUNT is re-exported from src.constants: where the workspace is
 # mounted inside a namespace is a property of the tool contract, not of this
 # module, and two definitions of it would be two contracts.
@@ -217,6 +243,7 @@ class ContainmentGrant:
     pgid: Optional[int] = None
     namespace_pid: Optional[int] = None
     namespace_start_token: Optional[str] = None
+    endpoint: Optional[str] = None
 
     @property
     def contained(self) -> bool:
@@ -229,7 +256,7 @@ class ContainmentGrant:
         Deliberately omits ``env``: it is part of the boundary but it is also
         where credentials live, and a tool result is model-visible.
         """
-        return {
+        data = {
             "id": self.id,
             "mechanism": self.mechanism,
             "mode": self.mode,
@@ -242,6 +269,9 @@ class ContainmentGrant:
             "requested": sorted(self.spec.requested),
             "network": self.spec.network,
         }
+        if self.endpoint:
+            data["endpoint"] = self.endpoint
+        return data
 
 
 @dataclass(frozen=True)
@@ -480,6 +510,7 @@ def _write_record(grant: ContainmentGrant) -> None:
         "pgid": grant.pgid,
         "namespace_pid": grant.namespace_pid,
         "namespace_start_token": grant.namespace_start_token,
+        "endpoint": grant.endpoint,
         "acquired_at": time.time(),
         "released_at": None,
         "release": None,
@@ -581,6 +612,9 @@ def _validate_spec(spec: ContainmentSpec) -> ContainmentSpec:
     for path in writable + readonly:
         if path in _RESERVED_BIND_DESTS:
             raise ValueError(f"containment: refusing to bind over reserved path {path}")
+    for path in writable:
+        if _is_protected_writable_destination(path):
+            raise ValueError(f"containment: refusing to bind over reserved path {path}")
     return replace(spec, workspace=workspace, writable_extra=writable, readonly_extra=readonly)
 
 
@@ -770,6 +804,7 @@ def declare_external_bridge(
         mode=CONTAINMENT_MODE,
         spec=spec,
         external=True,
+        endpoint=endpoint,
     )
     logger.info(
         "containment: grant %s is external (%s); nothing local contains it",

src/tool_execution.py

@@ -348,9 +348,24 @@ def _text_write_to_binary_artifact_result(content: str) -> tuple[str, Dict] | No
 
 async def _route_tool_via_bridge(tool: str, content: str, session_id: Optional[str], client_runtime_context: Optional[Dict]):
     import base64
+    from urllib.parse import urlparse
     bridge = _client_bridge(client_runtime_context)
     if bridge is None:
         return tool, {"error": f"{tool}: TUI host bridge is not available", "exit_code": 1}
+
+    url = str(bridge.get("url") or "").strip()
+    parsed = urlparse(url)
+    sanitized_endpoint = f"{parsed.scheme}://{parsed.netloc}{parsed.path}" if parsed.scheme and parsed.netloc else "tui_bridge"
+    from src import containment
+    spec = containment.agent_spec(agent_cwd(), {}, int(_BRIDGE_TOOL_TIMEOUT_S))
+    grant = containment.declare_external_bridge(
+        spec,
+        owner=str(session_id or "tui_bridge"),
+        endpoint=sanitized_endpoint,
+    )
+    boundary = grant.to_dict()
+    boundary["executed"] = False
+
     if tool == "bash":
         from src.agent_tools.subprocess_tools import _host_shell_requires_detach, _host_shell_should_auto_poll
 
@@ -401,9 +416,13 @@ async def _route_tool_via_bridge(tool: str, content: str, session_id: Optional[s
                     "output": "host job still running; poll the returned job_id",
                     "exit_code": 0,
                 }
+        if isinstance(result, dict):
+            b = dict(boundary)
+            b["executed"] = (result.get("exit_code") == 0 or (isinstance(result.get("exit_code"), int) and not result.get("error")))
+            result["containment"] = b
         return desc, result
     if tool == "python":
-        return "python: (client)", await _bridge_post(
+        py_res = await _bridge_post(
             bridge,
             "/run",
             {
@@ -414,6 +433,11 @@ async def _route_tool_via_bridge(tool: str, content: str, session_id: Optional[s
             timeout_s=_BRIDGE_TOOL_TIMEOUT_S,
             err_prefix="python",
         )
+        if isinstance(py_res, dict):
+            b = dict(boundary)
+            b["executed"] = (py_res.get("exit_code") == 0 or (isinstance(py_res.get("exit_code"), int) and not py_res.get("error")))
+            py_res["containment"] = b
+        return "python: (client)", py_res
     if tool == "grep":
         stripped = content.strip()
         try:

tests/test_containment_contract.py

@@ -505,3 +505,40 @@ async def test_an_external_bridge_grant_claims_nothing_and_cannot_be_run_locally
     with pytest.raises(ValueError, match="does not own"):
         await containment.run(grant, "echo hello")
     assert no_spawn == []
+
+
+@pytest.mark.parametrize("protected_dest", [
+    "/etc",
+    "/etc/ssl",
+    "/usr",
+    "/usr/local",
+    "/bin",
+    "/bin/sh",
+    "/sbin",
+    "/lib",
+    "/lib64",
+    "/proc",
+    "/proc/sys",
+    "/dev",
+    "/dev/shm",
+    "/sys",
+    "/root",
+    "/root/.ssh",
+    "/home",
+    "/workspace",
+    "/workspace/sub",
+])
+def test_writable_extra_rejects_protected_system_roots_and_descendants(monkeypatch, workspace, protected_dest):
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    spec = spec_for(workspace, writable_extra=(protected_dest,))
+    with pytest.raises(ValueError, match="reserved path"):
+        containment.acquire(spec, owner="session-1")
+
+
+def test_writable_extra_accepts_legitimate_scratch_destinations(monkeypatch, workspace):
+    install(monkeypatch, mechanism("fake", 10, containment.DIMENSIONS))
+    spec = spec_for(workspace, writable_extra=("/var/scratch", "/tmp/custom_scratch", "/home/testuser/scratch"))
+    grant = containment.acquire(spec, owner="session-1")
+    assert "/var/scratch" in grant.spec.writable_extra
+    assert "/tmp/custom_scratch" in grant.spec.writable_extra
+    assert "/home/testuser/scratch" in grant.spec.writable_extra

tests/test_containment_process_tree.py

@@ -376,3 +376,55 @@ def test_an_unenforceable_required_limit_refuses_instead_of_crashing_the_spawn(w
     with pytest.raises(containment.ContainmentUnavailable) as caught:
         containment.acquire(spec, owner="session-9")
     assert caught.value.missing == frozenset({containment.MEMORY})
+
+
+def test_pid_alive_rejects_non_process_pids(monkeypatch):
+    calls = []
+    real_kill = os.kill
+
+    def fake_kill(pid, sig):
+        calls.append((pid, sig))
+        return real_kill(pid, sig)
+
+    monkeypatch.setattr(os, "kill", fake_kill)
+
+    # Required: None, 0, and any negative integers must return False
+    assert pid_alive(None) is False
+    assert pid_alive(0) is False
+    assert pid_alive(-1) is False
+    assert pid_alive(-42) is False
+    assert pid_alive(-9999) is False
+    assert pid_alive("not_a_pid") is False
+
+    # The underlying process probe (os.kill) must NEVER be invoked for pid <= 0
+    assert calls == []
+
+    # Normal positive-PID behavior remains covered
+    my_pid = os.getpid()
+    assert pid_alive(my_pid) is True
+    assert (my_pid, 0) in calls
+
+
+def test_pid_alive_retains_conservative_liveness_on_eperm(monkeypatch):
+    def fake_kill(pid, sig):
+        raise PermissionError(1, "Operation not permitted")
+
+    monkeypatch.setattr(os, "kill", fake_kill)
+
+    # Positive PID with EPERM is conservatively considered alive
+    assert pid_alive(1) is True
+    assert pid_alive(99999) is True
+
+    # But non-process values still immediately return False without calling probe
+    assert pid_alive(None) is False
+    assert pid_alive(0) is False
+    assert pid_alive(-1) is False
+    assert pid_alive(-100) is False
+
+
+def test_pid_alive_reports_false_on_process_lookup_error(monkeypatch):
+    def fake_kill(pid, sig):
+        raise ProcessLookupError(3, "No such process")
+
+    monkeypatch.setattr(os, "kill", fake_kill)
+    assert pid_alive(99999999) is False

tests/test_production_external_bridge.py

@@ -0,0 +1,197 @@
+"""Tests verifying truthful representation of production external bridge execution.
+
+P2-2 invariant: external bridge execution != local containment.
+"""
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from src import containment
+from src.agent_tools import subprocess_tools
+from src import tool_execution as _te
+
+
+class _FakeResponse:
+    def __init__(self, data, status_code=200):
+        self._data = data
+        self.status_code = status_code
+        self.text = "error detail" if status_code >= 400 else ""
+
+    def json(self):
+        return self._data
+
+
+class _FakeAsyncClient:
+    def __init__(self, *args, **kwargs):
+        pass
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return None
+
+    async def post(self, url, *args, **kwargs):
+        return _FakeResponse({"stdout": "remote stdout", "stderr": "", "exit_code": 0})
+
+
+@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.mark.asyncio
+async def test_host_shell_creates_uncontained_external_record(monkeypatch):
+    """1. Production bridge execution creates an external/uncontained record.
+    2. It cannot be interpreted as contained."""
+    monkeypatch.setattr(subprocess_tools.httpx, "AsyncClient", _FakeAsyncClient)
+
+    tool = subprocess_tools.HostShellTool()
+    ctx = {
+        "client_runtime_context": {
+            "host_shell_bridge": {
+                "url": "http://127.0.0.1:17654/run",
+                "token": "secret-bridge-token",
+            }
+        },
+        "session_id": "test-session-host-shell",
+    }
+    result = await tool.execute('{"command": "echo host"}', ctx)
+
+    assert result["exit_code"] == 0
+    assert result["output"] == "remote stdout"
+    assert "containment" in result
+    c = result["containment"]
+
+    # Invariant: external bridge execution != local containment
+    assert c["external"] is True
+    assert c["contained"] is False
+    assert c["mechanism"] == "external_bridge"
+    assert c["enforced"] == []
+    assert c["executed"] is True
+
+    # Server-owned metadata identifies endpoint without secrets
+    assert c.get("endpoint") == "http://127.0.0.1:17654/run"
+    assert "secret-bridge-token" not in str(c)
+
+
+@pytest.mark.asyncio
+async def test_host_shell_failure_does_not_become_containment_or_effect_evidence(monkeypatch):
+    """5. Bridge failure does not become successful containment/effect evidence."""
+    class _FailingClient(_FakeAsyncClient):
+        async def post(self, url, *args, **kwargs):
+            return _FakeResponse({"error": "bridge exploded"}, status_code=500)
+
+    monkeypatch.setattr(subprocess_tools.httpx, "AsyncClient", _FailingClient)
+
+    tool = subprocess_tools.HostShellTool()
+    ctx = {
+        "client_runtime_context": {
+            "host_shell_bridge": {
+                "url": "http://127.0.0.1:17654/run",
+                "token": "secret-bridge-token",
+            }
+        },
+        "session_id": "test-session-host-shell-fail",
+    }
+    result = await tool.execute('{"command": "echo fail"}', ctx)
+
+    assert result["exit_code"] == 1
+    assert "bridge returned HTTP 500" in result["error"]
+    assert "containment" in result
+    c = result["containment"]
+
+    assert c["external"] is True
+    assert c["contained"] is False
+    assert c["enforced"] == []
+    # Failure means not executed
+    assert c["executed"] is False
+
+
+@pytest.mark.asyncio
+async def test_routed_bash_and_python_via_bridge_creates_uncontained_external_record():
+    """Prove _route_tool_via_bridge generates truthful external records for bash & python."""
+    bridge_ctx = {
+        "surface": "odysseus-tui",
+        "host_shell_bridge": {
+            "url": "http://127.0.0.1:17654/run",
+            "token": "bridge-token",
+        },
+    }
+
+    async def fake_bridge_post(bridge, path, payload, **kwargs):
+        return {"stdout": "bridge out", "stderr": "", "exit_code": 0}
+
+    from tests.runtime_evidence_helpers import server_authorized_executor
+
+    with patch.object(_te, "_bridge_post", fake_bridge_post), \
+         patch.object(_te, "_owner_is_admin", lambda owner: True):
+        # Routed bash
+        desc, result = await server_authorized_executor(_te.execute_tool_block)(
+            SimpleNamespace(tool_type="bash", content="ls -la"),
+            session_id="session-routed-bash",
+            client_runtime_context=bridge_ctx,
+            security_context=_te.NO_TOOL_SECURITY_CONTEXT,
+        )
+        assert result["exit_code"] == 0
+        assert "containment" in result
+        cb = result["containment"]
+        assert cb["external"] is True
+        assert cb["contained"] is False
+        assert cb["mechanism"] == "external_bridge"
+        assert cb["enforced"] == []
+        assert cb["executed"] is True
+        assert cb.get("endpoint") == "http://127.0.0.1:17654/run"
+
+        # Routed python
+        desc_py, result_py = await server_authorized_executor(_te.execute_tool_block)(
+            SimpleNamespace(tool_type="python", content="print('hi')"),
+            session_id="session-routed-py",
+            client_runtime_context=bridge_ctx,
+            security_context=_te.NO_TOOL_SECURITY_CONTEXT,
+        )
+        assert result_py["exit_code"] == 0
+        assert "containment" in result_py
+        cp = result_py["containment"]
+        assert cp["external"] is True
+        assert cp["contained"] is False
+        assert cp["mechanism"] == "external_bridge"
+        assert cp["enforced"] == []
+        assert cp["executed"] is True
+
+
+@pytest.mark.asyncio
+async def test_external_record_does_not_grant_authority(tmp_path):
+    """3. The record does not grant execution authority."""
+    spec = containment.agent_spec(str(tmp_path), {}, 5)
+    grant = containment.declare_external_bridge(
+        spec, owner="auth-test-session", endpoint="http://127.0.0.1:17654/run"
+    )
+
+    assert grant.external is True
+    assert grant.contained is False
+
+    # Attempting to use this grant to run local command must be rejected
+    with pytest.raises(ValueError, match="backend does not own"):
+        await containment.run(grant, "id")
+
+
+@pytest.mark.asyncio
+async def test_native_local_bash_python_behavior_unchanged(tmp_path, monkeypatch):
+    """4. Native local Bash/Python behavior is unchanged."""
+    tool_bash = subprocess_tools.BashTool()
+    ctx = {
+        "session_id": "native-session",
+    }
+    result = await tool_bash.execute("echo 'native run'", ctx)
+    assert result["exit_code"] == 0
+    assert "native run" in result["output"]
+    assert "containment" in result
+    c = result["containment"]
+    assert c["external"] is False
+    assert c["mechanism"] in ("bubblewrap", "process_group")
+    assert c["executed"] is True