Supervise confined agents and validate emitted enforcement claims

.github/workflows/confinement.yml

@@ -20,8 +20,9 @@ jobs:
           python-version: '3.12'
       - name: Install pinned dependencies
         run: |
-          python -m pip install --require-hashes -r requirements/dev.txt
+          python -m pip install --require-hashes -r requirements/confinement.txt
           python -m pip install --no-deps -e .
+          python -c "import opentelemetry.sdk.trace"
       - name: Check reference adapter and fixtures
         run: ruff check examples/confinement tests/confinement
       - name: Build adversarial fixture

CHANGELOG.md

@@ -7,8 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Accept emitted `gateway.call_log_summary` and call-graph `edges_represent`
+  fields in the TRACE claim schema. Both remain optional for older claims;
+  malformed values and undeclared properties remain rejected.
+
 ### Security
 
+- **An unrecognised enforcement mode was signed as `advisory`.** `_build_policy`
+  fell back to `advisory` for any value outside `enforcing`, `advisory` and
+  `silent`, so a claim built from such a value asserted that the policy had been
+  evaluated. The gateway's own path passes a validated `EnforcementMode`, so this
+  was reachable only by callers building claims directly. Claim construction now
+  refuses the value, as `_build_runtime` already refuses an unknown provider.
+  Reported by @saintmalik in #654.
+
 - Add an opt-in exact-output disclosure contract and library adapter. Independently
   scoped owner signatures bind output bytes, source labels, workload, recipient,
   purpose, policy and validity. Durable pre-delivery consumption rejects replay

docs/confinement.md

@@ -98,6 +98,67 @@ work. Restart, malformed stdout, attempted policy replacement, unavailable
 gateway, oversized stdout, stderr flood, crash and timeout cases test bounded
 failure and termination. No real secrets or external recipients are used.
 
+## Bridge failure and live policy restrictions
+
+Before plaintext admission, the running container gets a separate host watchdog.
+The bridge sends a payload-free heartbeat every 250 ms over a private inherited
+pipe. EOF or a two-second missed-heartbeat lease makes the watchdog stop the
+container. The watchdog runs in a separate process session, so a killed or paused
+bridge cannot prevent that cleanup. A watchdog failure also terminates the
+bridge exchange. The bound is the lease plus Docker stop latency; a responsive
+trusted host/daemon is required. The watcher retries a failed stop up to three
+times with a five-second timeout per attempt.
+
+The watcher is armed after attaching to the container and before supplying its
+initial frame. A bridge crash before that point can leave an empty container;
+the adapter does not provide a host-wide orphan collector. Simultaneous loss of
+the bridge and watchdog, a dead Docker daemon or a host crash requires external
+supervision. Container isolation remains in place independently of either
+process. The watcher contains the container identifier, not the agent payload.
+
+`examples/confinement/lifecycle.py` adds a trusted operator `PolicyGate` around
+the cMCP dispatch callback. It can revoke or restore originally approved tools
+without replacing the gateway, resetting sensitivity or expanding its original
+sink ceilings. A monotonically increasing revision and an async lock serialize
+the cutover with actual dispatch. An update is acknowledged only after earlier
+admitted calls finish. Calls queued behind that update see the new restriction.
+An invalid or rolled-back update closes future admission until a valid newer
+revision arrives. This operator API is not exposed to agent frames.
+
+Cutover cannot undo an earlier tool side effect. The reference supports this
+restrictive overlay, not arbitrary hot replacement of catalogs, classifiers,
+Cedar bundles or sink ceilings. A new deployment configuration still requires
+a new session. Revision state is process-local and is not rollback-resistant
+storage.
+
+Hosted lifecycle tests kill and pause a real bridge after an independent tool
+sink observes its canary, kill or pause the watchdog separately, and inspect Docker from
+another process. A synthetic mutation disables the watcher and requires the
+adversarial container to survive bridge death, demonstrating why the watcher is
+needed. Each pulse requires an acknowledgment within one second, including
+the pipe write and drain. A paused watchdog therefore causes the surviving
+bridge to stop the container. A second mutation removes this acknowledgment
+check and requires the container to survive a paused watcher. Simultaneous
+loss of both processes remains outside this guarantee. Live policy tests include an in-flight call, a queued call, revocation,
+restoration and invalid/stale revisions; the Docker fixture also exercises
+revocation/restoration without a session reset.
+
+## Audit and telemetry probes
+
+The confinement job installs the hash-pinned optional SDK from
+`requirements/confinement.txt`. Its tests send synthetic payloads through the
+real gateway and stdio upstream, then inspect persisted SQLite audit entries,
+in-memory chain entries, real SDK-exported spans and Python logs. Success,
+payload-bearing upstream errors, malformed stdout and stderr echoing all have
+positive tool-delivery controls. Payload hashes and fixed decision metadata
+remain present; literal canary payloads must be absent from these observed sinks.
+
+Two regression tests cover audit-observer and OTel export exceptions containing
+private data. Failure diagnostics now omit observer representations and
+tracebacks. The tests reproduced both leaks before that change. An arbitrary
+plugin can still write its own logs, and metadata/hashes may themselves be
+sensitive; these probes do not certify all collectors, exporters or encodings.
+
 ## Limits and remaining issue work
 
 The trusted boundary includes the host kernel, Docker daemon and local socket,
@@ -118,12 +179,11 @@ Canary observations cover these probes, not all encodings, kernel interfaces or
 covert channels. Timing, traffic shape, resource contention, shared hardware and
 other side channels remain untested. Crash tests verify termination and configured
 hard limits; they do not scan host crash services or prove memory erasure.
-Abrupt loss of the host bridge is not a full supervisor/recovery protocol: the
-container retains its isolation and limits, but needs external reaping if the
-bridge cannot execute cleanup. Live operator-policy replacement is unsupported;
-stop the old session and construct a new one. Broader audit/export adversarial
-coverage, host lifecycle supervision and deployment-specific custody evidence
-remain tracked in #659 rather than implied by a passing reference run.
+The watchdog covers loss of the bridge while the watcher and daemon survive;
+it is not a host-wide supervisor/recovery protocol. Live restrictions preserve
+the existing gateway state, but arbitrary policy replacement remains unsupported.
+Broader exporter/plugin adversarial coverage, host-wide recovery and
+deployment-specific custody evidence remain outside this reference claim.
 
 Docker behavior references:
 [container execution](https://docs.docker.com/engine/containers/run/) and

examples/confinement/adapter.py

@@ -26,6 +26,49 @@ class Refused(RuntimeError):
     """Fixed messages only: untrusted bytes must not reach host diagnostics."""
 
 
+class LeaseWatchdog:
+    """Separate process; contains only a container identifier, never plaintext."""
+
+    def __init__(self, command):
+        self.command = command
+        self.process = None
+
+    async def start(self):
+        self.process = await asyncio.create_subprocess_exec(
+            sys.executable, "-I", str(Path(__file__).with_name("watchdog.py")),
+            *self.command,
+            stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.DEVNULL, start_new_session=True,
+        )
+        if await asyncio.wait_for(self.process.stdout.readline(), 5) != b"ready\n":
+            raise Refused("watchdog admission failed")
+
+    async def pulse(self):
+        while True:
+            if self.process.returncode is not None:
+                raise Refused("watchdog unavailable")
+            try:
+                # Drain alone only proves pipe capacity, not supervisor liveness.
+                async with asyncio.timeout(1):
+                    self.process.stdin.write(b".")
+                    await self.process.stdin.drain()
+                    if await self.process.stdout.readexactly(1) != b".":
+                        raise Refused("watchdog invalid acknowledgement")
+            except (TimeoutError, OSError, asyncio.IncompleteReadError) as exc:
+                raise Refused("watchdog acknowledgement unavailable") from exc
+            await asyncio.sleep(0.25)
+
+    async def close(self):
+        if self.process is not None:
+            self.process.stdin.close()
+            try:
+                await asyncio.wait_for(self.process.communicate(), 18)
+            except TimeoutError:
+                self.process.kill()
+                await self.process.communicate()
+                raise Refused("watchdog cleanup timed out") from None
+
+
 def check_core_pattern(pattern: str) -> None:
     # RLIMIT_CORE is ignored for piped kernel core handlers (core(5)).
     if not pattern.strip() or pattern.lstrip().startswith("|"):
@@ -139,6 +182,8 @@ async def execute(
             stderr=asyncio.subprocess.PIPE, limit=MAX_FRAME,
         )
         stats = {"allowed": 0, "denied": 0, "stderr_bytes": 0}
+        watchdog = LeaseWatchdog([*self.command, "stop", "--time", "0", self.name])
+        self._watchdog = watchdog
 
         async def drain() -> None:
             while chunk := await process.stderr.read(4096):
@@ -169,9 +214,20 @@ async def exchange() -> None:
             if process.returncode:
                 raise Refused("agent failed")
 
-        tasks = [asyncio.create_task(exchange()), asyncio.create_task(drain())]
+        tasks = []
+        work = None
         try:
-            await asyncio.wait_for(asyncio.gather(*tasks), timeout)
+            # Attach is running but no plaintext has been supplied. Arm the
+            # independent watcher before admitting the initial frame.
+            await watchdog.start()
+            tasks = [asyncio.create_task(exchange()), asyncio.create_task(drain()),
+                     asyncio.create_task(watchdog.pulse())]
+            work = asyncio.gather(*tasks[:2])
+            done, _ = await asyncio.wait([work, tasks[2]], timeout=timeout,
+                                         return_when=asyncio.FIRST_COMPLETED)
+            if work not in done:
+                raise Refused("watchdog failed or session expired")
+            await work
             return stats
         except Exception:
             # Do not expose parser values, exception messages, or agent output.
@@ -180,6 +236,8 @@ async def exchange() -> None:
             for task in tasks:
                 task.cancel()
             await asyncio.gather(*tasks, return_exceptions=True)
+            if work is not None:
+                await asyncio.gather(work, return_exceptions=True)
             # Kill the container, not merely its attached CLI process.
             try:
                 # Also stop if the attach client failed while the container
@@ -192,6 +250,7 @@ async def exchange() -> None:
                 # remaining pipe buffers: wait() alone can deadlock on a full
                 # asyncio pipe after an output-flood rejection.
                 await process.communicate()
+                await watchdog.close()
 
 
 def decode_request(line: bytes, operations: Mapping[str, str]) -> tuple[str, dict]:

examples/confinement/agent.py

@@ -69,10 +69,15 @@
 
 sys.stderr.write(canary + "\n")
 sys.stderr.flush()
-for operation in ("permitted", "public"):
+operations = ("permitted", "public") if mode != "policy" else ("permitted",) * 3 + ("public",)
+for operation in operations:
     sys.stdout.write(json.dumps({"operation": operation, "arguments": {
         "value": canary, "core_limit": resource.getrlimit(resource.RLIMIT_CORE),
     }}) + "\n")
     sys.stdout.flush()
     if not sys.stdin.readline():
+        if mode == "linger":
+            time.sleep(60)  # adversarial child ignores loss of its bridge
         sys.exit(2)
+if mode == "linger":
+    time.sleep(60)

examples/confinement/lifecycle.py

@@ -0,0 +1,43 @@
+"""Reference operator gate for live restriction of an existing cMCP session."""
+
+import asyncio
+
+from examples.confinement.adapter import Refused
+
+
+class PolicyGate:
+    """Serialize cutover with dispatch; never replace the underlying cMCP state.
+
+    This overlay can restrict or re-enable originally configured tools. It
+    cannot expand catalog/sink ceilings or lower the session classification.
+    The operator method is not exposed on the agent's stdio protocol.
+    """
+
+    def __init__(self, dispatch, approved_tools):
+        self._dispatch = dispatch
+        self._approved = frozenset(approved_tools)
+        self._allowed = self._approved
+        self._revision = 0
+        self._closed = False
+        self._lock = asyncio.Lock()
+
+    async def replace(self, revision, allowed_tools):
+        async with self._lock:
+            # A failed update leaves future calls denied until a valid newer
+            # update arrives. Never silently keep an unexpectedly broad policy.
+            self._closed = True
+            if (type(revision) is not int or revision <= self._revision
+                    or not isinstance(allowed_tools, (set, frozenset))
+                    or not all(isinstance(t, str) for t in allowed_tools)
+                    or not allowed_tools <= self._approved):
+                raise Refused("invalid operator policy revision")
+            self._allowed = frozenset(allowed_tools)
+            self._revision = revision
+            self._closed = False
+            return self._revision
+
+    async def __call__(self, tool, arguments):
+        async with self._lock:
+            if self._closed or tool not in self._allowed:
+                return {"allowed": False, "response": None}
+            return await self._dispatch(tool, arguments)

examples/confinement/watchdog.py

@@ -0,0 +1,44 @@
+"""Independent, plaintext-free lease watcher for a running agent container.
+
+The bridge holds the only write end of stdin. EOF or a missed heartbeat stops
+the container even when the bridge cannot execute its finally block.
+"""
+
+import os
+import select
+import subprocess
+import sys
+
+LEASE_SECONDS = 2.0
+
+
+def main():
+    command = sys.argv[1:]
+    # Arguments are supplied only by the trusted adapter; never by the agent.
+    sys.stdout.buffer.write(b"ready\n")
+    sys.stdout.buffer.flush()
+    while True:
+        ready, _, _ = select.select([sys.stdin.fileno()], [], [], LEASE_SECONDS)
+        if not ready or os.read(sys.stdin.fileno(), 1) != b".":
+            break
+        try:
+            sys.stdout.buffer.write(b".")
+            sys.stdout.buffer.flush()
+        except OSError:
+            break
+    # A surviving, responsive host/daemon is required. Retry transient failure;
+    # never log Docker output (nor accept payloads on this channel).
+    for _ in range(3):
+        try:
+            result = subprocess.run(command, stdin=subprocess.DEVNULL,
+                                    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                                    timeout=5, check=False)
+            if result.returncode == 0:
+                return 0
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

requirements/confinement.txt

@@ -0,0 +1,6 @@
+# Real optional OTel export in the confinement job; base dependencies stay pinned.
+-r dev.txt
+opentelemetry-sdk==1.44.0 \
+    --hash=sha256:df081c4c6bcfdb1211e3e86140376792643128a25f8d72d1d27675936e7e96ad
+opentelemetry-semantic-conventions==0.65b0 \
+    --hash=sha256:1cacde7b0ad306f84c5ef08c3dbe1bbaf20165bba6f8bff43b670e555a086bcb

schemas/trace-claim.schema.json

@@ -172,6 +172,10 @@
               "additionalProperties": false,
               "required": ["compliance_domains_touched", "cross_boundary_events"],
               "properties": {
+                "edges_represent": {
+                  "type": "string",
+                  "description": "Meaning of recorded edges; temporal adjacency does not establish causality."
+                },
                 "compliance_domains_touched": {
                   "type": "array", "items": { "type": "string" }
                 },
@@ -182,6 +186,16 @@
             }
           }
         },
+        "call_log_summary": {
+          "type": "object",
+          "additionalProperties": false,
+          "required": ["total_calls", "tools_called", "suspicious_sequences_detected"],
+          "properties": {
+            "total_calls": { "type": "integer", "minimum": 0 },
+            "tools_called": { "type": "array", "items": { "type": "string" } },
+            "suspicious_sequences_detected": { "type": "integer", "minimum": 0 }
+          }
+        },
         "catalog": {
           "type": "object",
           "additionalProperties": false,

src/cmcp_runtime/audit/chain.py

@@ -294,7 +294,8 @@ def _notify_sinks(self, entry: AuditEntry) -> None:
             try:
                 sink(entry)
             except Exception:
-                logger.debug("Audit sink %r failed", sink, exc_info=True)
+                # Sink repr and exception text may contain private payloads.
+                logger.debug("Audit sink failed; entry remains durable")
 
     @property
     def chain_root(self) -> str:

src/cmcp_runtime/audit/trace_claim.py

@@ -425,11 +425,24 @@ def _build_runtime(report: AttestationReportInfo) -> RuntimeInfo:
     return RuntimeInfo(platform=platform, measurement=measurement, nonce=nonce)  # type: ignore[arg-type]
 
 
+_ENFORCEMENT_MODE_MAP = {"enforcing": "enforce", "advisory": "advisory", "silent": "silent"}
+
+
 def _build_policy(bundle: PolicyBundleInfo) -> PolicyInfo:
-    mode_map = {"enforcing": "enforce", "advisory": "advisory", "silent": "silent"}
+    # Every mode the claim can carry asserts that something evaluated the policy.
+    # An unrecognised value is refused rather than signed as one of them, the
+    # same way _build_runtime refuses an unknown provider (AUDIT-003).
+    mode = _ENFORCEMENT_MODE_MAP.get(bundle.enforcement_mode)
+    if mode is None:
+        raise ValueError(
+            f"Enforcement mode {bundle.enforcement_mode!r} is not in the allowed set "
+            f"{sorted(_ENFORCEMENT_MODE_MAP)}. "
+            "Rejecting claim construction rather than signing a policy posture "
+            "that was never evaluated."
+        )
     return PolicyInfo(
         bundle_hash=bundle.hash,
-        enforcement_mode=mode_map.get(bundle.enforcement_mode, "advisory"),  # type: ignore[arg-type]
+        enforcement_mode=mode,  # type: ignore[arg-type]
         version=bundle.policy_version,
     )