R4 (partial): isolated-engine-only forwarder guard (gate item 5)

Land only the containment-refusal half of R4: the guard every SOR forwarder
must pass before it does anything — assert engine != local, or refuse. It moves
no traffic, opens no socket, spawns no engine, and builds no SSH chain.

- forwarder.py: assert_isolated (raises on local and on any engine off the
  manifest allow-list), isolation_prefix (docker/multipass exec argv, no host
  path), ForwarderPlan (construction IS the gate — no plan for local/unknown/
  no-container). Allow-list imported from provenance so forwarder and manifest
  cannot disagree about what counts as isolated.
- There is no code path in this module that returns an exec prefix for the host;
  containment is structural, not a runtime flag.

The traffic-moving half of R4 (gate item 1: 3-hop self-traffic e2e delivery +
per-hop pcaps across the grid) is deliberately NOT implemented — HELD for a live
isolated engine + grid. This guard is fully verifiable offline.

Acceptance (gate item 5) green: local/unknown/no-container refused; isolated
engine accepted with the correct isolation prefix. Python forwarder suite 12
passed; full SOR Python suite 49 passed.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

cmd_chat/sor/forwarder.py

@@ -0,0 +1,100 @@
+"""R4 (partial) — containment guard for the SOR nested-SSH forwarder.
+
+This module lands **only instrument-validation gate item 5**: forwarders are
+*isolated-engine-only*. It is the guard that every forwarder MUST pass before it
+does anything — ``assert engine != local`` or refuse. It moves no traffic, opens
+no socket, spawns no engine, and builds no SSH chain.
+
+The traffic-moving half of R4 (gate item 1: a 3-hop self-traffic circuit across
+the grid delivering a known payload e2e, with per-hop pcaps) is intentionally
+**NOT** implemented here — it is HELD for a live isolated engine + grid (see
+OVERSEER-STATUS.md). Committing an unrunnable forwarder would violate the build
+discipline; this guard, by contrast, is fully verifiable offline: constructing a
+forwarder plan for a ``local`` (or unknown) engine refuses, for an isolated
+engine it yields the isolation exec prefix.
+
+Containment law (CLAUDE.md §Containment #1) is encoded structurally: there is no
+code path in this module that returns an exec prefix for the host. ``local`` —
+the explicit warned exception the *chat* sandbox allows (``bridge.py:540``) — is
+never valid for a SOR run. The allow-list is imported from ``provenance`` so the
+forwarder and the run manifest can never disagree about what counts as isolated.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import List
+
+from cmd_chat.sor.provenance import ISOLATED_ENGINES
+
+# The host. Never a valid engine for a SOR forwarder — its whole purpose is to
+# keep every forwarder/circuit process off the host (blast-radius containment).
+LOCAL_ENGINE = "local"
+
+
+class ContainmentError(RuntimeError):
+    """A forwarder was asked to run on a non-isolated engine (or with no
+    container/VM to exec into). Raised instead of running — the code refuses."""
+
+
+def assert_isolated(engine: str) -> None:
+    """Raise :class:`ContainmentError` unless ``engine`` is an isolated engine.
+
+    This is the ``assert engine != local`` half of the R4 acceptance check
+    (instrument-validation gate item 5), hardened to also reject any engine not
+    on the manifest allow-list (unknown engines are refused, not assumed safe)."""
+    if engine == LOCAL_ENGINE:
+        raise ContainmentError(
+            "containment: SOR forwarders never run on the host "
+            f"({LOCAL_ENGINE!r}); isolated engine required (one of {ISOLATED_ENGINES})"
+        )
+    if engine not in ISOLATED_ENGINES:
+        raise ContainmentError(
+            f"containment: engine {engine!r} is not an isolated engine "
+            f"(allowed: {ISOLATED_ENGINES})"
+        )
+
+
+def isolation_prefix(engine: str, container: str) -> List[str]:
+    """The argv prefix that runs a command *inside* the named isolated engine —
+    the same shape the chat sandbox uses (``bridge.py:_exec_prefix``), minus the
+    ``local`` host exception, which is refused here. The caller appends the actual
+    program + args as separate argv items (never a shell-interpolated string), so
+    circuit descriptors can't inject shell metacharacters.
+
+    Refuses (raises) on a non-isolated engine or a missing container name."""
+    assert_isolated(engine)
+    if not container:
+        raise ContainmentError(
+            f"containment: no container/VM name to exec into for engine {engine!r}"
+        )
+    if engine == "docker":
+        return ["docker", "exec", "-i", container]
+    if engine == "multipass":
+        return ["multipass", "exec", container, "--"]
+    # Unreachable: assert_isolated already constrained `engine` to ISOLATED_ENGINES
+    # and both members are handled above. Refuse rather than fall through silently.
+    raise ContainmentError(f"containment: no isolation prefix for engine {engine!r}")
+
+
+@dataclass(frozen=True)
+class ForwarderPlan:
+    """A validated, inert description of *where* a hop forwarder would run. Its
+    construction is the containment gate: a plan for a ``local``/unknown engine,
+    or with no container, cannot be built. It carries no credentials, opens no
+    connection, and moves no bytes — the traffic-moving forwarder that would
+    consume it (gate item 1) is HELD for the human + a live grid."""
+
+    engine: str
+    container: str
+
+    def __post_init__(self) -> None:
+        assert_isolated(self.engine)
+        if not self.container:
+            raise ContainmentError(
+                f"containment: no container/VM name for engine {self.engine!r}"
+            )
+
+    def exec_prefix(self) -> List[str]:
+        """The isolated exec argv prefix for this plan (validated at build time)."""
+        return isolation_prefix(self.engine, self.container)

tests/test_sor_forwarder.py

@@ -0,0 +1,76 @@
+"""R4 (partial) — forwarder containment-guard acceptance check.
+
+Acceptance predicate (instrument-validation gate item 5, the offline-verifiable
+half of the roadmap R4 check): a SOR forwarder is isolated-engine-only —
+``assert engine != local`` or it refuses to run. This suite pins exactly that:
+a plan/prefix for ``local`` (or any non-isolated / unknown engine, or a missing
+container) is refused; an isolated engine kind is accepted and yields the
+correct isolation exec prefix.
+
+The traffic-moving half of R4 (gate item 1: 3-hop e2e delivery + per-hop pcaps
+across the grid) is deliberately NOT covered here — it is HELD for a live
+isolated engine + grid. This guard moves no traffic and spawns no engine, so it
+is fully validatable offline.
+"""
+
+import pytest
+
+from cmd_chat.sor.forwarder import (
+    ISOLATED_ENGINES,
+    LOCAL_ENGINE,
+    ContainmentError,
+    ForwarderPlan,
+    assert_isolated,
+    isolation_prefix,
+)
+
+
+def test_local_engine_is_refused():
+    # The core containment invariant: never the host.
+    with pytest.raises(ContainmentError):
+        assert_isolated(LOCAL_ENGINE)
+    with pytest.raises(ContainmentError):
+        ForwarderPlan(engine="local", container="c0")
+    with pytest.raises(ContainmentError):
+        isolation_prefix("local", "c0")
+
+
+@pytest.mark.parametrize("engine", ["", "host", "podman", "vmware", "LOCAL", "docker "])
+def test_unknown_or_non_isolated_engines_refused(engine):
+    # Anything not on the manifest allow-list is refused, not assumed safe.
+    with pytest.raises(ContainmentError):
+        assert_isolated(engine)
+    with pytest.raises(ContainmentError):
+        ForwarderPlan(engine=engine, container="c0")
+
+
+@pytest.mark.parametrize("engine", list(ISOLATED_ENGINES))
+def test_isolated_engines_are_accepted(engine):
+    # An isolated engine passes the gate and builds a plan.
+    assert_isolated(engine)  # does not raise
+    plan = ForwarderPlan(engine=engine, container="node-1")
+    prefix = plan.exec_prefix()
+    assert isinstance(prefix, list) and prefix
+    # The container name is present as its own argv item (no shell interpolation).
+    assert "node-1" in prefix
+
+
+def test_exec_prefix_shapes_match_the_isolation_convention():
+    assert isolation_prefix("docker", "n") == ["docker", "exec", "-i", "n"]
+    assert isolation_prefix("multipass", "n") == ["multipass", "exec", "n", "--"]
+
+
+def test_missing_container_is_refused():
+    with pytest.raises(ContainmentError):
+        ForwarderPlan(engine="docker", container="")
+    with pytest.raises(ContainmentError):
+        isolation_prefix("docker", "")
+
+
+def test_no_prefix_path_ever_targets_the_host():
+    # Structural check: for every isolated engine, the exec prefix's leading
+    # argv is the engine tool itself (docker/multipass), never a bare host shell.
+    for engine in ISOLATED_ENGINES:
+        prefix = isolation_prefix(engine, "c")
+        assert prefix[0] == engine
+        assert prefix != []  # a host-shell prefix (empty) is never produced