From a56f5abf35b662e88c151116dc21e556560057ee Mon Sep 17 00:00:00 2001
From: Imran Siddique <imran.siddique@opaque.co>
Date: Fri, 18 Sep 2026 10:09:14 -0700
Subject: [PATCH 1/4] feat: supervise confined agents and serialize live
 release restrictions

Signed-off-by: Imran Siddique <imran.siddique@opaque.co>
---
 .github/workflows/confinement.yml          |  3 +-
 docs/confinement.md                        | 68 +++++++++++++++--
 examples/confinement/adapter.py            | 51 ++++++++++++-
 examples/confinement/agent.py              |  7 +-
 examples/confinement/lifecycle.py          | 43 +++++++++++
 examples/confinement/watchdog.py           | 39 ++++++++++
 requirements/confinement.txt               |  6 ++
 src/cmcp_runtime/audit/chain.py            |  3 +-
 src/cmcp_runtime/observability/otel.py     |  3 +-
 tests/confinement/bridge_worker.py         | 57 ++++++++++++++
 tests/confinement/gateway.py               |  4 +-
 tests/confinement/test_adapter.py          | 14 ++++
 tests/confinement/test_exports.py          | 87 ++++++++++++++++++++++
 tests/confinement/test_linux.py            | 22 ++++++
 tests/confinement/test_policy_lifecycle.py | 60 +++++++++++++++
 tests/confinement/test_supervision.py      | 82 ++++++++++++++++++++
 tests/confinement/upstream.py              | 16 ++++
 17 files changed, 551 insertions(+), 14 deletions(-)
 create mode 100644 examples/confinement/lifecycle.py
 create mode 100644 examples/confinement/watchdog.py
 create mode 100644 requirements/confinement.txt
 create mode 100644 tests/confinement/bridge_worker.py
 create mode 100644 tests/confinement/test_exports.py
 create mode 100644 tests/confinement/test_policy_lifecycle.py
 create mode 100644 tests/confinement/test_supervision.py

diff --git a/.github/workflows/confinement.yml b/.github/workflows/confinement.yml
index 5cba9646..af73319d 100644
--- a/.github/workflows/confinement.yml
+++ b/.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
diff --git a/docs/confinement.md b/docs/confinement.md
index 0bb84d57..39bbf0f9 100644
--- a/docs/confinement.md
+++ b/docs/confinement.md
@@ -98,6 +98,63 @@ 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 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. 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 +175,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
diff --git a/examples/confinement/adapter.py b/examples/confinement/adapter.py
index 1214ed0a..be385fd1 100644
--- a/examples/confinement/adapter.py
+++ b/examples/confinement/adapter.py
@@ -26,6 +26,37 @@ 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")
+            self.process.stdin.write(b".")
+            await self.process.stdin.drain()
+            await asyncio.sleep(0.25)
+
+    async def close(self):
+        if self.process is not None:
+            self.process.stdin.close()
+            await asyncio.wait_for(self.process.communicate(), 18)
+
+
 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 +170,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 +202,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 +224,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 +238,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]:
diff --git a/examples/confinement/agent.py b/examples/confinement/agent.py
index b25ed2f6..fd0f5899 100644
--- a/examples/confinement/agent.py
+++ b/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)
diff --git a/examples/confinement/lifecycle.py b/examples/confinement/lifecycle.py
new file mode 100644
index 00000000..1193182a
--- /dev/null
+++ b/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)
diff --git a/examples/confinement/watchdog.py b/examples/confinement/watchdog.py
new file mode 100644
index 00000000..4302a82d
--- /dev/null
+++ b/examples/confinement/watchdog.py
@@ -0,0 +1,39 @@
+"""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(), 4096) == b"":
+            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())
diff --git a/requirements/confinement.txt b/requirements/confinement.txt
new file mode 100644
index 00000000..3c22ad5b
--- /dev/null
+++ b/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
diff --git a/src/cmcp_runtime/audit/chain.py b/src/cmcp_runtime/audit/chain.py
index 02886efc..363b9b7f 100644
--- a/src/cmcp_runtime/audit/chain.py
+++ b/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:
diff --git a/src/cmcp_runtime/observability/otel.py b/src/cmcp_runtime/observability/otel.py
index e5e2bdfb..a8502214 100644
--- a/src/cmcp_runtime/observability/otel.py
+++ b/src/cmcp_runtime/observability/otel.py
@@ -158,7 +158,8 @@ def export(self, entry: AuditEntry) -> None:
             self._export(tracer, entry)
         except Exception:  # pragma: no cover - defensive
             if not self._warned:
-                logger.debug("OTel audit export failed; further failures are silent", exc_info=True)
+                # Exporter failures can embed private input in exception text.
+                logger.debug("OTel audit export failed; further failures are silent")
                 self._warned = True
 
     def _export(self, tracer: Any, entry: AuditEntry) -> None:
diff --git a/tests/confinement/bridge_worker.py b/tests/confinement/bridge_worker.py
new file mode 100644
index 00000000..e8179fe2
--- /dev/null
+++ b/tests/confinement/bridge_worker.py
@@ -0,0 +1,57 @@
+"""Synthetic bridge process killed by an independent lifecycle observer."""
+
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+from examples.confinement import adapter
+
+from tests.confinement.gateway import make_gateway
+
+
+class NoWatchdog:
+    """Test-only causal mutation; never selected by the reference adapter."""
+
+    process = None
+
+    def __init__(self, command):
+        pass
+
+    async def start(self):
+        pass
+
+    async def pulse(self):
+        await asyncio.Future()
+
+    async def close(self):
+        pass
+
+
+async def main():
+    if sys.argv[3] == "unguarded":
+        adapter.LeaseWatchdog = NoWatchdog
+    root = Path(sys.argv[2])
+    sandbox = adapter.DockerSandbox(sys.argv[1])
+    proxy, dispatch = make_gateway(root / "sink.jsonl")
+    try:
+        await sandbox.create()
+        initial = json.loads(sys.stdin.readline())
+
+        async def observed_dispatch(tool, arguments):
+            result = await dispatch(tool, arguments)
+            watchdog = sandbox._watchdog.process
+            (root / "ready.json").write_text(json.dumps({
+                "container": sandbox.name, "watchdog": watchdog.pid if watchdog else None,
+            }))
+            return result
+
+        await sandbox.execute(initial, observed_dispatch,
+                              {"permitted": "permitted.tool", "public": "public.tool"}, timeout=60)
+    finally:
+        await proxy.aclose()
+        await sandbox.remove()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/tests/confinement/gateway.py b/tests/confinement/gateway.py
index b5c17b9b..07f0877f 100644
--- a/tests/confinement/gateway.py
+++ b/tests/confinement/gateway.py
@@ -20,7 +20,7 @@
 from cmcp_runtime.sink_policy import SinkPolicy
 
 
-def make_gateway(sink: Path, *, public_ceiling="public"):
+def make_gateway(sink: Path, *, public_ceiling="public", audit_store=None, audit_sinks=None):
     upstream = Path(__file__).with_name("upstream.py").resolve()
     spawn = StdioSpawn(sys.executable, (str(upstream), str(sink)),
                        measure_executable(str(upstream)), str(upstream))
@@ -33,7 +33,7 @@ def make_gateway(sink: Path, *, public_ceiling="public"):
         "permitted.tool": "confidential", "public.tool": public_ceiling,
     }, "confidential"))
     session = SessionState(uuid4().hex, max_sensitivity="confidential")
-    audit = AuditChain(session.session_id)
+    audit = AuditChain(session.session_id, store=audit_store, sinks=audit_sinks)
     bundle = PolicyBundle(
         PolicyManifest("1.0.0", "2026-09-18T00:00:00Z", "fixture", "fixture"),
         {"allow.cedar": "permit(principal, action, resource);"}, '{"cMCP": {}}',
diff --git a/tests/confinement/test_adapter.py b/tests/confinement/test_adapter.py
index 2c377f3f..c6c13770 100644
--- a/tests/confinement/test_adapter.py
+++ b/tests/confinement/test_adapter.py
@@ -64,6 +64,20 @@ async def test_output_flood_cleanup_drains_real_child_pipes(monkeypatch):
     # the hosted suite separately verifies that the container is stopped.
     monkeypatch.setattr("examples.confinement.adapter.shutil.which", lambda name: sys.executable)
     monkeypatch.setattr("examples.confinement.adapter.host_preflight", lambda: None)
+    class NoWatchdog:
+        def __init__(self, command):
+            pass
+
+        async def start(self):
+            pass
+
+        async def pulse(self):
+            await asyncio.Future()
+
+        async def close(self):
+            pass
+
+    monkeypatch.setattr("examples.confinement.adapter.LeaseWatchdog", NoWatchdog)
     sandbox = DockerSandbox("sha256:" + "a" * 64)
     sandbox.command = [sys.executable, "-c",
                        "import sys,time; sys.stdin.readline(); "
diff --git a/tests/confinement/test_exports.py b/tests/confinement/test_exports.py
new file mode 100644
index 00000000..a373b567
--- /dev/null
+++ b/tests/confinement/test_exports.py
@@ -0,0 +1,87 @@
+"""Observe real audit persistence, emitted spans and failure diagnostics."""
+
+import json
+import logging
+import sqlite3
+from dataclasses import asdict
+from uuid import uuid4
+
+import pytest
+
+from cmcp_runtime.audit.chain import AuditChain
+from cmcp_runtime.audit.store import SqliteAuditStore
+from cmcp_runtime.observability.otel import OtelAuditExporter
+from tests.confinement.gateway import make_gateway
+
+
+def test_failing_audit_observer_does_not_log_private_repr_or_exception(caplog):
+    secret = "synthetic-private-" + uuid4().hex
+
+    class Observer:
+        def __repr__(self):
+            return secret
+
+        def __call__(self, entry):
+            raise RuntimeError(secret)
+
+    with caplog.at_level(logging.DEBUG):
+        chain = AuditChain("safe-session", sinks=[Observer()])
+    assert len(chain.entries) == 1
+    assert "Audit sink" in caplog.text
+    assert secret not in caplog.text
+
+
+def test_otel_failure_diagnostics_do_not_include_exception_payload(caplog):
+    secret = "synthetic-private-" + uuid4().hex
+
+    class BrokenTracer:
+        def start_as_current_span(self, *args, **kwargs):
+            raise RuntimeError(secret)
+
+    exporter = OtelAuditExporter()
+    exporter._tracer = BrokenTracer()
+    with caplog.at_level(logging.DEBUG):
+        chain = AuditChain("safe-session", sinks=[exporter])
+    assert len(chain.entries) == 1
+    assert "OTel audit export failed" in caplog.text
+    assert secret not in caplog.text
+
+
+@pytest.mark.parametrize("mode", ["echo", "error", "malformed", "stderr"])
+async def test_gateway_payload_absent_from_actual_audit_spans_and_logs(tmp_path, caplog, mode):
+    pytest.importorskip("opentelemetry.sdk")
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+    from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+
+    secret = "canary-" + uuid4().hex
+    recording = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(recording))
+    exporter = OtelAuditExporter()
+    exporter._tracer = provider.get_tracer("cmcp.confinement")
+    store = SqliteAuditStore(tmp_path / "audit.db")
+    sink = tmp_path / "received.jsonl"
+    proxy, dispatch = make_gateway(sink, audit_store=store, audit_sinks=[exporter])
+    try:
+        with caplog.at_level(logging.DEBUG):
+            result = await dispatch("permitted.tool", {"value": secret, "mode": mode})
+            denied = await dispatch("public.tool", {"value": secret, "mode": mode})
+            await proxy.aclose()
+        assert result["allowed"] == (mode in {"echo", "stderr"})
+        assert not denied["allowed"]
+        assert secret in sink.read_text()  # positive: the source really reached the tool
+        chain = json.dumps([asdict(entry) for entry in proxy._audit.entries])
+        with sqlite3.connect(tmp_path / "audit.db") as reader:
+            persisted = "\n".join(row[0] for row in reader.execute("SELECT payload FROM audit_entries"))
+        provider.force_flush()
+        spans = recording.get_finished_spans()
+        assert len(spans) == len(proxy._audit.entries) >= 3
+        exported = "\n".join(span.to_json() for span in spans)
+        for observed in (chain, persisted, exported, caplog.text):
+            assert secret not in observed
+        assert any(entry.request_payload_hash for entry in proxy._audit.entries)
+    finally:
+        await proxy.aclose()
+        store.close()
+        provider.shutdown()
diff --git a/tests/confinement/test_linux.py b/tests/confinement/test_linux.py
index 51539d46..086ec9a1 100644
--- a/tests/confinement/test_linux.py
+++ b/tests/confinement/test_linux.py
@@ -9,6 +9,7 @@
 
 import pytest
 from examples.confinement import adapter
+from examples.confinement.lifecycle import PolicyGate
 
 from tests.confinement.gateway import make_gateway
 
@@ -82,6 +83,19 @@ async def run_case(tmp_path, monkeypatch, *, mutation=None, mode="normal", unava
     sink = tmp_path / (uuid4().hex + ".jsonl")
     canary = "canary-" + uuid4().hex
     proxy, dispatch = make_gateway(sink, public_ceiling="confidential" if mutation == "sink" else "public")
+    if mode == "policy":
+        gate = PolicyGate(dispatch, {"permitted.tool", "public.tool"})
+        calls = 0
+
+        async def dispatch(tool, arguments):
+            nonlocal calls
+            result = await gate(tool, arguments)
+            calls += 1
+            if calls == 1:
+                await gate.replace(1, set())
+            elif calls == 2:
+                await gate.replace(2, {"permitted.tool", "public.tool"})
+            return result
     if unavailable:
         async def dispatch(tool, arguments):
             raise ConnectionError("gateway unavailable")
@@ -137,6 +151,14 @@ async def test_confinement_and_fresh_restart(tmp_path, monkeypatch):
         assert stats["stderr_bytes"] > 0
 
 
+async def test_live_operator_policy_restricts_and_restores_without_label_reset(tmp_path, monkeypatch):
+    observation, stats, refused = await run_case(tmp_path, monkeypatch, mode="policy")
+    assert not refused
+    require_confinement(observation)
+    assert observation["permitted"] == 2
+    assert stats["allowed"] == 2 and stats["denied"] == 2
+
+
 @pytest.mark.parametrize("mutation", ["network", "filesystem", "logging", "sink"])
 async def test_removed_restriction_is_detected_at_independent_sink(tmp_path, monkeypatch, mutation):
     observation, _, refused = await run_case(tmp_path, monkeypatch, mutation=mutation)
diff --git a/tests/confinement/test_policy_lifecycle.py b/tests/confinement/test_policy_lifecycle.py
new file mode 100644
index 00000000..5f831e6f
--- /dev/null
+++ b/tests/confinement/test_policy_lifecycle.py
@@ -0,0 +1,60 @@
+import asyncio
+import json
+
+import pytest
+from examples.confinement.adapter import Refused
+from examples.confinement.lifecycle import PolicyGate
+
+from tests.confinement.gateway import make_gateway
+
+
+async def test_live_cutover_waits_for_admitted_call_then_blocks_queued_release(tmp_path):
+    sink = tmp_path / "received.jsonl"
+    proxy, dispatch = make_gateway(sink)
+    entered, finish = asyncio.Event(), asyncio.Event()
+
+    async def held_dispatch(tool, arguments):
+        entered.set()
+        await finish.wait()
+        return await dispatch(tool, arguments)
+
+    gate = PolicyGate(held_dispatch, {"permitted.tool", "public.tool"})
+    try:
+        first = asyncio.create_task(gate("permitted.tool", {"value": "first"}))
+        await entered.wait()
+        update = asyncio.create_task(gate.replace(1, set()))
+        await asyncio.sleep(0)
+        queued = asyncio.create_task(gate("permitted.tool", {"value": "queued"}))
+        assert not update.done()
+        finish.set()
+        assert (await first)["allowed"]
+        assert await update == 1
+        assert not (await queued)["allowed"]
+        assert not (await gate("permitted.tool", {"value": "later"}))["allowed"]
+        assert [json.loads(line)["arguments"]["value"] for line in sink.read_text().splitlines()] == ["first"]
+        await gate.replace(2, {"permitted.tool", "public.tool"})
+        # Restoring an alias never weakens the original cMCP classification.
+        assert not (await gate("public.tool", {"value": "still confidential"}))["allowed"]
+        assert (await gate("permitted.tool", {"value": "restored"}))["allowed"]
+    finally:
+        await proxy.aclose()
+
+
+@pytest.mark.parametrize("revision,tools", [(0, set()), (True, set()), (1, {"unknown"}), (1, ["permitted.tool"])])
+async def test_invalid_policy_update_closes_admission_until_newer_valid_revision(tmp_path, revision, tools):
+    sink = tmp_path / "received.jsonl"
+    proxy, dispatch = make_gateway(sink)
+    gate = PolicyGate(dispatch, {"permitted.tool"})
+    try:
+        with pytest.raises(Refused):
+            await gate.replace(revision, tools)
+        assert not (await gate("permitted.tool", {"value": "must not arrive"}))["allowed"]
+        assert not sink.exists()
+        await gate.replace(2, {"permitted.tool"})
+        assert (await gate("permitted.tool", {"value": "recovered"}))["allowed"]
+        with pytest.raises(Refused):
+            await gate.replace(1, {"permitted.tool"})
+        assert not (await gate("permitted.tool", {"value": "rollback"}))["allowed"]
+        assert len(sink.read_text().splitlines()) == 1
+    finally:
+        await proxy.aclose()
diff --git a/tests/confinement/test_supervision.py b/tests/confinement/test_supervision.py
new file mode 100644
index 00000000..08112757
--- /dev/null
+++ b/tests/confinement/test_supervision.py
@@ -0,0 +1,82 @@
+"""Kill/pause the real bridge after plaintext delivery, observe Docker externally."""
+
+import asyncio
+import json
+import os
+import signal
+import sys
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+from examples.confinement.adapter import DockerSandbox
+
+from tests.confinement.test_linux import observers
+
+pytestmark = pytest.mark.skipif(not os.environ.get("CMCP_CONFINEMENT_IMAGE"),
+                                reason="requires native Linux Docker CI")
+
+
+async def running(observer, container):
+    # Empty successful listing means removed; command failure must propagate.
+    value = await observer.docker("ps", "--filter", "name=^/" + container + "$", "--format", "{{.Names}}")
+    return container.encode() in value.splitlines()
+
+
+@pytest.mark.parametrize("failure", ["kill", "pause", "watchdog", "unguarded"])
+async def test_independent_lease_stops_container_after_bridge_failure(tmp_path, failure):
+    image = os.environ["CMCP_CONFINEMENT_IMAGE"]
+    observer = DockerSandbox(image)
+    canary = "canary-" + uuid4().hex
+    metadata = None
+    async with observers() as (received, addresses):
+        bridge = await asyncio.create_subprocess_exec(
+            sys.executable, "-m", "tests.confinement.bridge_worker", image, str(tmp_path), failure,
+            stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL,
+            stderr=asyncio.subprocess.DEVNULL, start_new_session=True,
+        )
+        try:
+            bridge.stdin.write(json.dumps({"canary": canary, "mode": "linger", **addresses}).encode() + b"\n")
+            await bridge.stdin.drain()
+            bridge.stdin.close()
+            async with asyncio.timeout(15):
+                while metadata is None:
+                    try:
+                        metadata = json.loads((tmp_path / "ready.json").read_text())
+                    except (FileNotFoundError, json.JSONDecodeError):
+                        assert bridge.returncode is None, "bridge exited before admitting data"
+                        await asyncio.sleep(0.05)
+            container = metadata["container"]
+            assert await running(observer, container)
+            # Independent sink confirms this was a plaintext-bearing session.
+            assert canary in (tmp_path / "sink.jsonl").read_text()
+            if failure == "watchdog":
+                os.kill(metadata["watchdog"], signal.SIGKILL)
+            elif failure == "pause":
+                os.kill(bridge.pid, signal.SIGSTOP)
+            else:
+                bridge.kill()
+            if failure == "unguarded":
+                await asyncio.sleep(3)
+                stopped = not await running(observer, container)
+                assert not stopped, "removing watchdog must expose the surviving container"
+                with pytest.raises(AssertionError):
+                    assert stopped
+            else:
+                async with asyncio.timeout(10):
+                    while await running(observer, container):
+                        await asyncio.sleep(0.1)
+                stopped = True
+            assert all(not values for values in received.values())
+            if report := os.environ.get("CMCP_CONFINEMENT_EVIDENCE"):
+                with Path(report).open("a", encoding="utf-8") as artifact:
+                    artifact.write(json.dumps({"lifecycle_failure": failure,
+                                               "stopped": stopped, "network_deliveries": 0}) + "\n")
+        finally:
+            if bridge.returncode is None:
+                bridge.kill()
+            await bridge.wait()
+            if metadata is not None:
+                containers = await observer.docker("ps", "--all", "--format", "{{.Names}}")
+                if metadata["container"].encode() in containers.splitlines():
+                    await observer.docker("rm", "--force", metadata["container"])
diff --git a/tests/confinement/upstream.py b/tests/confinement/upstream.py
index a96e6cb7..4217228b 100644
--- a/tests/confinement/upstream.py
+++ b/tests/confinement/upstream.py
@@ -14,7 +14,23 @@
     elif method == "tools/call":
         with Path(sys.argv[1]).open("a", encoding="utf-8") as sink:
             sink.write(json.dumps(request["params"]) + "\n")
+        args = request["params"]["arguments"]
+        mode = args.get("mode")
+        if mode == "error":
+            sys.stdout.write(json.dumps({"jsonrpc": "2.0", "id": request["id"],
+                                         "error": {"code": -32000, "message": args["value"]}}) + "\n")
+            sys.stdout.flush()
+            continue
+        if mode == "malformed":
+            sys.stdout.write(args["value"] + "\n")
+            sys.stdout.flush()
+            continue
+        if mode == "stderr":
+            sys.stderr.write(args["value"] + "\n")
+            sys.stderr.flush()
         result = {"content": [{"type": "text", "text": "recorded"}]}
+        if mode == "echo":
+            result["content"][0]["text"] = args["value"]
     else:
         result = {}
     if "id" in request:

From cd619862f53864fda1dc58c3f4b69cc4c72356ab Mon Sep 17 00:00:00 2001
From: Imran Siddique <imran.siddique@opaque.co>
Date: Fri, 18 Sep 2026 14:47:21 -0700
Subject: [PATCH 2/4] fix(claim): refuse an unrecognised enforcement mode
 instead of signing advisory

_build_policy mapped any value outside enforcing, advisory and silent to
advisory, so a claim built from one asserted a policy evaluation that never
happened. It now raises, matching _build_runtime's refusal of an unknown
provider (AUDIT-003). The gateway's own path passes a validated
EnforcementMode, so this was reachable only by direct callers.

Closes #654

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Signed-off-by: Imran Siddique <imran.siddique@opaque.co>
---
 CHANGELOG.md                          |  8 ++++++++
 src/cmcp_runtime/audit/trace_claim.py | 17 +++++++++++++++--
 tests/unit/test_trace_claim.py        | 25 +++++++++++++++++++++++++
 3 files changed, 48 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 39b9ca74..b535e3c1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### 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.
+
 - **Nothing in a claim showed whether the kill switch was armed, and a refusal
   left no evidence.** A claim with `kill_switch_triggered: false` read the same
   whether the switch was enabled and did not trip or was never enabled, and a
diff --git a/src/cmcp_runtime/audit/trace_claim.py b/src/cmcp_runtime/audit/trace_claim.py
index df15e1fd..0ce43724 100644
--- a/src/cmcp_runtime/audit/trace_claim.py
+++ b/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,
     )
 
diff --git a/tests/unit/test_trace_claim.py b/tests/unit/test_trace_claim.py
index 48bceb3f..ad3c7d09 100644
--- a/tests/unit/test_trace_claim.py
+++ b/tests/unit/test_trace_claim.py
@@ -7,6 +7,7 @@
 import pathlib
 
 import jsonschema
+import pytest
 
 from cmcp_runtime.audit.chain import AuditChain
 from cmcp_runtime.audit.keys import SigningKey
@@ -456,6 +457,30 @@ def test_build_runtime_all_known_providers_accepted():
         _build_runtime(report)  # must not raise
 
 
+# ── #654: unknown enforcement mode rejected ─────────────────────────────────────
+
+
+@pytest.mark.parametrize("mode", ["enforce", "declared", "audit-only", "ENFORCING", "", "Enforcing"])
+def test_build_policy_unknown_enforcement_mode_raises(mode):
+    """#654: an unrecognised mode was signed as "advisory", asserting an evaluation."""
+    from cmcp_runtime.audit.trace_claim import PolicyBundleInfo, _build_policy
+
+    bundle = PolicyBundleInfo(hash="sha256:" + "a" * 64, enforcement_mode=mode, policy_version="1")
+    with pytest.raises(ValueError, match="not in the allowed set"):
+        _build_policy(bundle)
+
+
+@pytest.mark.parametrize(
+    ("mode", "expected"),
+    [("enforcing", "enforce"), ("advisory", "advisory"), ("silent", "silent")],
+)
+def test_build_policy_known_enforcement_modes_map(mode, expected):
+    from cmcp_runtime.audit.trace_claim import PolicyBundleInfo, _build_policy
+
+    bundle = PolicyBundleInfo(hash="sha256:" + "a" * 64, enforcement_mode=mode, policy_version="1")
+    assert _build_policy(bundle).enforcement_mode == expected
+
+
 # ── tool_transcript entries (#126) ──────────────────────────────────────────────
 
 

From bad751becca0ec5c42d70062e5c9fe5ee1380e85 Mon Sep 17 00:00:00 2001
From: Imran Siddique <imran.siddique@opaque.co>
Date: Sat, 19 Sep 2026 06:28:10 -0700
Subject: [PATCH 3/4] fix(confinement): require bounded watchdog
 acknowledgements

Signed-off-by: Imran Siddique <imran.siddique@opaque.co>
---
 docs/confinement.md                   |  8 ++++++--
 examples/confinement/adapter.py       | 18 +++++++++++++++---
 examples/confinement/watchdog.py      |  7 ++++++-
 tests/confinement/bridge_worker.py    | 14 ++++++++++++++
 tests/confinement/test_adapter.py     | 19 +++++++++++++++++++
 tests/confinement/test_supervision.py | 13 ++++++++++---
 6 files changed, 70 insertions(+), 9 deletions(-)

diff --git a/docs/confinement.md b/docs/confinement.md
index 39bbf0f9..2d11375f 100644
--- a/docs/confinement.md
+++ b/docs/confinement.md
@@ -132,10 +132,14 @@ 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 the watchdog separately, and inspect Docker from
+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. Live policy tests include an in-flight call, a queued call, revocation,
+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.
 
diff --git a/examples/confinement/adapter.py b/examples/confinement/adapter.py
index be385fd1..90688f21 100644
--- a/examples/confinement/adapter.py
+++ b/examples/confinement/adapter.py
@@ -47,14 +47,26 @@ async def pulse(self):
         while True:
             if self.process.returncode is not None:
                 raise Refused("watchdog unavailable")
-            self.process.stdin.write(b".")
-            await self.process.stdin.drain()
+            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()
-            await asyncio.wait_for(self.process.communicate(), 18)
+            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:
diff --git a/examples/confinement/watchdog.py b/examples/confinement/watchdog.py
index 4302a82d..7279f685 100644
--- a/examples/confinement/watchdog.py
+++ b/examples/confinement/watchdog.py
@@ -19,7 +19,12 @@ def main():
     sys.stdout.buffer.flush()
     while True:
         ready, _, _ = select.select([sys.stdin.fileno()], [], [], LEASE_SECONDS)
-        if not ready or os.read(sys.stdin.fileno(), 4096) == b"":
+        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).
diff --git a/tests/confinement/bridge_worker.py b/tests/confinement/bridge_worker.py
index e8179fe2..6fe73f06 100644
--- a/tests/confinement/bridge_worker.py
+++ b/tests/confinement/bridge_worker.py
@@ -28,9 +28,23 @@ async def close(self):
         pass
 
 
+class UnacknowledgedWatchdog(adapter.LeaseWatchdog):
+    """Test-only mutation that mistakes successful pipe writes for liveness."""
+
+    async def pulse(self):
+        while True:
+            if self.process.returncode is not None:
+                raise adapter.Refused("watchdog unavailable")
+            self.process.stdin.write(b".")
+            await self.process.stdin.drain()
+            await asyncio.sleep(0.25)
+
+
 async def main():
     if sys.argv[3] == "unguarded":
         adapter.LeaseWatchdog = NoWatchdog
+    elif sys.argv[3] == "watchdog-pause-unguarded":
+        adapter.LeaseWatchdog = UnacknowledgedWatchdog
     root = Path(sys.argv[2])
     sandbox = adapter.DockerSandbox(sys.argv[1])
     proxy, dispatch = make_gateway(root / "sink.jsonl")
diff --git a/tests/confinement/test_adapter.py b/tests/confinement/test_adapter.py
index c6c13770..b2657c00 100644
--- a/tests/confinement/test_adapter.py
+++ b/tests/confinement/test_adapter.py
@@ -123,3 +123,22 @@ def test_inspection_refuses_weakened_profile(section, key, value):
     mutant[section][key] = value
     with pytest.raises(Refused):
         verify_container(mutant)
+
+
+@pytest.mark.parametrize("reply", [None, b"bad\n"])
+async def test_watchdog_requires_bounded_ack_even_when_pipe_drains(reply):
+    from types import SimpleNamespace
+    from unittest.mock import AsyncMock, Mock
+
+    from examples.confinement.adapter import LeaseWatchdog
+
+    reader = asyncio.StreamReader()
+    if reply is not None:
+        reader.feed_data(reply)
+    writer = SimpleNamespace(write=Mock(), drain=AsyncMock())
+    watcher = LeaseWatchdog([])
+    watcher.process = SimpleNamespace(returncode=None, stdin=writer, stdout=reader)
+    with pytest.raises(Refused, match="watchdog"):
+        await asyncio.wait_for(watcher.pulse(), 2)
+    writer.write.assert_called_once_with(b".")
+    writer.drain.assert_awaited_once()
diff --git a/tests/confinement/test_supervision.py b/tests/confinement/test_supervision.py
index 08112757..fac00410 100644
--- a/tests/confinement/test_supervision.py
+++ b/tests/confinement/test_supervision.py
@@ -5,6 +5,7 @@
 import os
 import signal
 import sys
+from contextlib import suppress
 from pathlib import Path
 from uuid import uuid4
 
@@ -23,7 +24,8 @@ async def running(observer, container):
     return container.encode() in value.splitlines()
 
 
-@pytest.mark.parametrize("failure", ["kill", "pause", "watchdog", "unguarded"])
+@pytest.mark.parametrize("failure", ["kill", "pause", "watchdog", "unguarded",
+                                     "watchdog-pause", "watchdog-pause-unguarded"])
 async def test_independent_lease_stops_container_after_bridge_failure(tmp_path, failure):
     image = os.environ["CMCP_CONFINEMENT_IMAGE"]
     observer = DockerSandbox(image)
@@ -52,14 +54,16 @@ async def test_independent_lease_stops_container_after_bridge_failure(tmp_path,
             assert canary in (tmp_path / "sink.jsonl").read_text()
             if failure == "watchdog":
                 os.kill(metadata["watchdog"], signal.SIGKILL)
+            elif failure.startswith("watchdog-pause"):
+                os.kill(metadata["watchdog"], signal.SIGSTOP)
             elif failure == "pause":
                 os.kill(bridge.pid, signal.SIGSTOP)
             else:
                 bridge.kill()
-            if failure == "unguarded":
+            if failure in {"unguarded", "watchdog-pause-unguarded"}:
                 await asyncio.sleep(3)
                 stopped = not await running(observer, container)
-                assert not stopped, "removing watchdog must expose the surviving container"
+                assert not stopped, "removing supervision must expose the surviving container"
                 with pytest.raises(AssertionError):
                     assert stopped
             else:
@@ -77,6 +81,9 @@ async def test_independent_lease_stops_container_after_bridge_failure(tmp_path,
                 bridge.kill()
             await bridge.wait()
             if metadata is not None:
+                if failure.startswith("watchdog-pause"):
+                    with suppress(ProcessLookupError):
+                        os.kill(metadata["watchdog"], signal.SIGKILL)
                 containers = await observer.docker("ps", "--all", "--format", "{{.Names}}")
                 if metadata["container"].encode() in containers.splitlines():
                     await observer.docker("rm", "--force", metadata["container"])

From 40e3b2d719dc411361e77191310ce61fc0f37d9a Mon Sep 17 00:00:00 2001
From: Imran Siddique <imran.siddique@opaque.co>
Date: Sun, 20 Sep 2026 16:36:52 -0700
Subject: [PATCH 4/4] fix(schema): validate complete emitted audit summaries

Signed-off-by: Imran Siddique <imran.siddique@opaque.co>
---
 CHANGELOG.md                            |  6 ++
 schemas/trace-claim.schema.json         | 14 +++++
 tests/unit/test_emitted_claim_schema.py | 84 +++++++++++++++++++++++++
 3 files changed, 104 insertions(+)
 create mode 100644 tests/unit/test_emitted_claim_schema.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9180b630..db70dc0d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,12 @@ 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`
diff --git a/schemas/trace-claim.schema.json b/schemas/trace-claim.schema.json
index 09b543f4..88ed8e7a 100644
--- a/schemas/trace-claim.schema.json
+++ b/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,
diff --git a/tests/unit/test_emitted_claim_schema.py b/tests/unit/test_emitted_claim_schema.py
new file mode 100644
index 00000000..3a9500f3
--- /dev/null
+++ b/tests/unit/test_emitted_claim_schema.py
@@ -0,0 +1,84 @@
+"""Validate complete HTTP-emitted claims, including optional audit summaries."""
+
+import copy
+import json
+from pathlib import Path
+
+import jsonschema
+import pytest
+
+from cmcp_runtime.cli import build_server
+from cmcp_runtime.kill_switch import KillSwitchBlockStore
+from tests.unit.test_kill_switch_durable import _bearer, _client, _ctx, _operator
+
+SCHEMA = json.loads(
+    (Path(__file__).parents[2] / "schemas/trace-claim.schema.json").read_text(encoding="utf-8")
+)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("trip", [False, True], ids=["session-close", "operator-trip"])
+async def test_complete_emitted_claim_validates(tmp_path, trip):
+    # Software-only context: this checks the emitted wire contract, not attestation.
+    server = build_server(_ctx(KillSwitchBlockStore(tmp_path / "audit.db")))
+    async with _client(server) as client:
+        if trip:
+            response = await client.post(
+                "/kill-switch/trip",
+                json={"reason": "schema regression", "authorized_by": "oncall"},
+                headers=_operator(),
+            )
+            assert response.status_code == 200, response.text
+            claim = response.json()["claim"]
+        else:
+            response = await client.post(
+                f"/sessions/{server._session.session_id}/close", headers=_bearer()
+            )
+            assert response.status_code == 200, response.text
+            claim = response.json()
+    gateway = claim["gateway"]
+    assert "call_log_summary" in gateway
+    assert "edges_represent" in gateway["call_summary"]["call_graph_summary"]
+    jsonschema.validate(claim, SCHEMA)
+
+    # Each formerly missing declaration is necessary; unknown fields stay refused.
+    for owner, field in [
+        (SCHEMA["properties"]["gateway"], "call_log_summary"),
+        (
+            SCHEMA["properties"]["gateway"]["properties"]["call_summary"]["properties"][
+                "call_graph_summary"
+            ],
+            "edges_represent",
+        ),
+    ]:
+        narrowed = copy.deepcopy(owner)
+        del narrowed["properties"][field]
+        instance = (
+            gateway
+            if field == "call_log_summary"
+            else gateway["call_summary"]["call_graph_summary"]
+        )
+        with pytest.raises(jsonschema.ValidationError):
+            jsonschema.validate(instance, narrowed)
+
+    for field, invalid in [
+        ("total_calls", -1),
+        ("total_calls", "0"),
+        ("tools_called", [1]),
+        ("suspicious_sequences_detected", -1),
+        ("unknown", True),
+    ]:
+        malformed = copy.deepcopy(claim)
+        malformed["gateway"]["call_log_summary"][field] = invalid
+        with pytest.raises(jsonschema.ValidationError):
+            jsonschema.validate(malformed, SCHEMA)
+    for value in [1, None]:
+        malformed = copy.deepcopy(claim)
+        malformed["gateway"]["call_summary"]["call_graph_summary"]["edges_represent"] = value
+        with pytest.raises(jsonschema.ValidationError):
+            jsonschema.validate(malformed, SCHEMA)
+
+    historical = copy.deepcopy(claim)
+    del historical["gateway"]["call_log_summary"]
+    del historical["gateway"]["call_summary"]["call_graph_summary"]["edges_represent"]
+    jsonschema.validate(historical, SCHEMA)