diff --git a/CHANGELOG.md b/CHANGELOG.md
index 13cc7d2..76b5406 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,49 @@ All notable changes to this project are documented here. Format follows
 
 ## [Unreleased]
 
+### Security
+- The Cosmos store writes spans with `create_item`, which fails atomically if
+  the id exists. It read first and then upserted, so two writers of one span
+  id could both see it absent and the second silently replaced the first.
+- The authorization engine denies a rule with no condition instead of failing
+  an `assert` outside its fail-closed path (and the asserts on the policy and
+  gateway paths are explicit checks, which `python -O` does not strip).
+
+### Fixed
+- Documentation: what "durably emitted" means per store; the trust penalties
+  that actually apply; that the evasion page is written against the local
+  shield; which keys `sentinel.yaml` accepts; that the non-bypassability test
+  covers the proxy's own path, not network topology; a positioning claim of
+  "the same class of guarantee" as CaMeL. `docs/design-notes.md` lists the
+  catalogue findings kept as they are, and why.
+- The rule-based attack classifier labels any server's tools by the verbs in
+  their names (`delete_*`, `refundCharge`, `sendSlackMessage`...); it knew only
+  the demo's tools, so an operator's own blocks were mostly "unclassified".
+- The before/after baseline uses the same driver (scripted or live model) and
+  web mode as the protected run; it was always the scripted transcript.
+- The dashboard's kill chain and verdict come from the run's data. They named
+  `web_fetch`, `block-untrusted-origin` and "the exfiltration" whatever the
+  run did. Its trust gauge uses the configured quarantine threshold (now in
+  `/capabilities` and the trust endpoint) instead of a hard-coded 40.
+- The examples no longer say an external `/mcp` run appears in the live view;
+  it is listed in the history tab.
+- Cosmos DB persistence could not start: its async Azure clients need aiohttp,
+  which is not installed, and setting `AZURE_COSMOS_ENDPOINT` failed at startup
+  with a bare import error. Install `sentinel-prox[cosmos]`; without it the
+  error now says so. An unused duplicate Cosmos factory was removed.
+- Restoring the run index after a restart no longer loads every span of every
+  trace to guess a scenario label from redacted input (it never matched).
+- The gateway logs an error if the MCP SDK stops exposing the session index it
+  reads to release ended sessions, instead of silently never releasing them;
+  a test pins that SDK attribute.
+- Replay reports a duplicated span id as a duplicate; it was reported as a
+  cycle, which with one parent per span it cannot be.
+- SQLite stores each span's timestamp in an indexed column (existing databases
+  are migrated in place), so listing traces and retention purges no longer
+  parse every span's JSON while holding the store lock.
+- The test suite's fake Cosmos container refuses queries other than the one it
+  can answer; it ignored the SQL entirely.
+
 ## [0.2.2] — 2026-10-19
 
 **Upgrade from 0.2.1 or earlier.** Read *Changed (breaking)* first: catalogue
diff --git a/README.md b/README.md
index 6ba9af1..fb5ddf7 100644
--- a/README.md
+++ b/README.md
@@ -78,7 +78,7 @@ protects you from.
 | **Quarantine is recorded, not enforced** (default) | The trust score never recovers, so agents making only allowed calls cross the threshold within tens of calls. By default that crossing is logged, not acted on. `SENTINEL_ENFORCE_QUARANTINE=1` enforces it, for every agent that shares a credential. |
 | **Downstream servers: HTTP only** | No stdio servers, no per-server credentials or OAuth, one shared session per server, and no reconnect. Most published MCP servers are stdio. |
 | **State lives in one process** | Trust scores, MCP sessions, the event feed, the run index and policies replaced through `POST /tenants/{t}/policy` are in memory. A restart forgets API policy reloads (put durable policy in `SENTINEL_POLICY_FILE` / `SENTINEL_TENANT_POLICIES`), and two replicas share none of it, which is why the Azure template runs one. |
-| **Cosmos persistence is partial** | With Cosmos DB, spans are stored and replayable, but run ownership, the run index after a restart, and retention purging exist only for the default SQLite store. |
+| **Cosmos persistence is partial** | It needs `pip install 'sentinel-prox[cosmos]'` (for aiohttp). With Cosmos DB, spans are stored and replayable, but run ownership, the run index after a restart, and retention purging exist only for the default SQLite store. |
 | **Some state is visible across tenants** | Event ids on `/events` are one global counter, so a tenant can infer how busy the deployment is. `/downstream` and `/capabilities` describe the whole deployment to any authenticated caller. |
 | **Calls the SDK rejects are not recorded** | Arguments that fail a tool's input schema are refused by the MCP SDK before SENTINEL sees the call, so they leave no forensic record. |
 
@@ -205,8 +205,13 @@ access-controlled forensic store. There is a test asserting the synthetic SSN an
 API key never appear in log output.
 
 Precedence is **CLI flag > environment variable > config file > default**, so a
-container can override a checked-in file. Every key has an env equivalent
-(`SENTINEL_MCP_SERVERS`, `SENTINEL_POLICY_FILE`, …) — see [`.env.example`](./.env.example).
+container can override a checked-in file. The file takes only these keys, and an
+unknown key is an error: `servers`, `policy`, `catalogue_strict`,
+`catalogue_approvals`, `real_web_fetch`, `api_token` and
+`downstream_timeout_seconds` (each also an environment variable:
+`SENTINEL_MCP_SERVERS`, `SENTINEL_POLICY_FILE`, ...), plus `host`, `port`,
+`dashboard`, `log_level` and `log_format` (also `sentinel serve` flags). Every other
+setting is environment-only; see [`.env.example`](./.env.example).
 
 ### Writing policy
 
@@ -358,7 +363,8 @@ to wake). It runs in **anonymous demo mode with authentication disabled**, so tr
 it as a public sandbox, not as an example of a secured deployment: everything in it
 is synthetic and anyone can drive it. A real deployment must set
 `SENTINEL_API_TOKEN`. A poisoned page induces the agent to email a synthetic customer record to an
-attacker; the Layer-1 filter misses the obfuscated variant and authorization blocks it
+attacker; the Layer-1 filter misses the obfuscated variant (written to slip past its
+local patterns, so the miss is staged) and authorization blocks it
 anyway. The taint comes from the page: the bundled policy marks the records service's
 output as trusted (`trusted_output`), so the record lookup alone taints nothing, and
 without the page the agent never attempts the email. The attacker's domain is also
diff --git a/docs/COMPETITIVE_POSITIONING.md b/docs/COMPETITIVE_POSITIONING.md
index 40a2324..6ff5cfb 100644
--- a/docs/COMPETITIVE_POSITIONING.md
+++ b/docs/COMPETITIVE_POSITIONING.md
@@ -64,8 +64,11 @@ solved with provable security**, vs 84% undefended.
 
 **Our defensible edge vs CaMeL: deployability.** CaMeL requires rearchitecting the agent into
 a dual-LLM interpreter pattern — a research architecture, hard to retrofit onto an existing
-product. SENTINEL gets the same class of guarantee by *topology*, with an unmodified agent.
-That is the commercially important difference, and it is real.
+product. SENTINEL gets a coarser guarantee of the same kind by *topology*, with an unmodified
+agent: taint is tracked per MCP session and per tool result, not per value inside
+the agent's program, so it blocks more benign flows than CaMeL would and cannot
+follow data across sessions. Deployability is the commercially important
+difference, and it is real; equivalence of guarantees is not claimed.
 
 **Their edge over us: evidence.** They have a number. We have a demo. See §4.
 
diff --git a/docs/design-notes.md b/docs/design-notes.md
new file mode 100644
index 0000000..8d8db12
--- /dev/null
+++ b/docs/design-notes.md
@@ -0,0 +1,35 @@
+# Design notes: findings kept as they are
+
+The 2026-10-17 findings catalogue listed about 180 items. Most were fixed; see the
+CHANGELOG. These were examined and deliberately left as they are. Each says why,
+so a later review can disagree with the reasoning rather than rediscover the item.
+
+| ID | What the catalogue said | Why it stays |
+|---|---|---|
+| PP-04, GW-9 | Taint lives in one MCP session; a new session starts clean. | The proxy has no identity for "the same task" across sessions. Carrying taint per credential would deny every guarded action for an agent forever after one untrusted read. Documented in the README's *Known limitations*. |
+| PP-06 | Declassification never clears taint already in the session. | Declassification vouches for one value, not for what the agent did with earlier untrusted content. Clearing the session on it would let any declassifiable tool launder the lineage. |
+| PP-11, DEMO-24 | The demo baseline and the Phase 1 skeleton reach tools without `_intercept`. | Both exist to show what happens without SENTINEL. The baseline is reachable only through the demo endpoints, which answer 404 unless `SENTINEL_DASHBOARD=1`; neither is reachable from `/mcp`. |
+| PP-14 | The per-session lock is held across the downstream call. | It is what keeps a session's provenance updates ordered. The call is now bounded by `SENTINEL_DOWNSTREAM_TIMEOUT_SECONDS`, so the wait is bounded too. |
+| PP-15 | The union-vs-walk test uses the union's own derivation rule. | True, and now stated in the test. It verifies the incremental computation; the rule (each call derives from everything observed before it) is the design. |
+| TRUST-2 | The trust score never recovers. | Why quarantine is recorded, not enforced, by default. Documented in *Known limitations*. |
+| TRUST-3 | Quarantine flags are set before the event is written. | Fail-closed: if the write fails the agent is still quarantined, and the log line is written first. Reversing it would let an agent through when recording fails. |
+| TRUST-5 | Only some events cost trust. | Catalogue refusals are the downstream's change, the user's own prompt is not agent behaviour, and the local shield is a heuristic. Now stated in `trust/config.yaml`. |
+| TRUST-6, CM-7, GW-2 | Trust and run state are per process. | Shared state would need a store SENTINEL does not have. The Azure template runs one replica; documented. |
+| AUTHZ-14 | "Deny-overrides" is vacuous with deny-only rules. | The term still describes the behaviour correctly (any deny wins); it is simply trivially satisfied. |
+| AUTHZ-17 | A module named `ast` shadows the standard library's name. | It is only ever imported as `sentinel.authorization.ast`. Renaming it would break imports for anyone using the package API, for no behaviour change. |
+| DCC-18 (part) | `sentinel check` fails on flag-only findings that `serve` accepts. | A readiness check should not pass findings an operator still has to triage. |
+| DCC-22 (part) | No way to authenticate to downstream servers. | A missing feature, not a defect; listed in *Known limitations*. |
+| CM-13, GW-19 | Global event ids reveal activity volume; `/downstream` and `/capabilities` are deployment-wide. | Per-tenant counters would change the event-cursor protocol. Listed in *Known limitations*. |
+| PP-16 | Calls the MCP SDK rejects for schema reasons are not recorded. | They never reach SENTINEL. Recording them would mean turning off the SDK's validation and doing it inside the proxy; listed in *Known limitations*. |
+| COSMOS-1 | The Cosmos path has no ownership, restore or retention. | A larger piece of work on a paid path; listed in *Known limitations*. Cosmos can now at least start (it needs the `cosmos` extra). |
+| TCD-04, OWN-6 | Real-HTTP tests are order-dependent and can hang. | The open teardown defect in `docs/mcp-streamable-http-teardown.md`, tracked by a strict xfail and the quarantined CI job. |
+| RED-1 (rest) | Redaction keeps tool names and deny reasons in the clear. | Argument names and values are redacted. Tool names and reasons are what an investigator reads first, and they come from the catalogue and the policy, not from the caller's data. A caller-invented tool name is still logged with control characters escaped. |
+| AUTHZ-15 (rest) | No static check of predicate variables or config types. | Loading now warns about comparands the config does not declare and config keys that shadow call fields, refuses duplicate rule ids, and a scalar where a collection is expected denies at evaluation. A full type checker for config values is not worth its complexity here. |
+| DCC-23 (rest) | `verify_catalogue` and `DownstreamTopology.server_of`/`tool_names` are unused in production. | Small, tested, public helpers for anyone building on the package; the truly unused `build_web_server` was removed. |
+| TCD-26 | The codebase's history starts with a large root commit. | A fact about the repository's history; nothing to change in the code. |
+
+## Not done
+
+| ID | What | Status |
+|---|---|---|
+| GW-12 | The browser session cookie is the raw bearer token; there is no server-side session, expiry or revocation. | Open. The cookie is `HttpOnly` and `SameSite=Strict`, and the dashboard is off unless `SENTINEL_DASHBOARD=1`, but a server-side session store remains to be built. |
diff --git a/examples/README.md b/examples/README.md
index 1198606..83fae5a 100644
--- a/examples/README.md
+++ b/examples/README.md
@@ -70,8 +70,9 @@ python examples/connect_over_http.py    # a real mcp client drives the hero atta
 SENTINEL-specific code** — to connect to `/mcp`, list tools, and run the hero
 kill chain. The poisoned-lineage `send_email` comes back `isError=True` with
 `SENTINEL blocked …`. Swap in your own client the same way; to your agent,
-SENTINEL is just an MCP tool server. Each connection is its own `live-mcp` trace
-on the dashboard.
+SENTINEL is just an MCP tool server. Each connection is its own `live-mcp` run,
+listed in the dashboard's history tab (the live view follows dashboard-launched
+runs only).
 
 For a **real LLM agent loop** (the model decides; SENTINEL is the backstop):
 
diff --git a/examples/connect_over_http.py b/examples/connect_over_http.py
index c37e050..756bfce 100644
--- a/examples/connect_over_http.py
+++ b/examples/connect_over_http.py
@@ -22,8 +22,9 @@
 What you'll see: tool discovery and the legit read calls succeed, and the
 poisoned-lineage ``send_email`` to the attacker is BLOCKED at the MCP boundary
 with a clean error — the same result as the in-process demo, now over real HTTP.
-Open the dashboard (http://127.0.0.1:8765) to watch the live run appear as a
-``live-mcp`` trace.
+The session is recorded as a ``live-mcp`` run: open the dashboard
+(http://127.0.0.1:8765), go to the history tab and select it to replay its trace.
+The live view follows runs launched from the dashboard itself.
 """
 from __future__ import annotations
 
diff --git a/examples/run_live_agent.py b/examples/run_live_agent.py
index 368c3ce..ff784c1 100644
--- a/examples/run_live_agent.py
+++ b/examples/run_live_agent.py
@@ -21,7 +21,8 @@
     # ...or with NO credentials, using the offline susceptible stand-in model:
     python examples/run_live_agent.py --offline
 
-Watch http://localhost:8765 — the run appears live as a `live-mcp` trace.
+The session is recorded as a `live-mcp` run; find it in the dashboard's history
+tab (http://localhost:8765). The live view follows dashboard-launched runs only.
 
 Exit status: 0 when SENTINEL blocked at least one action, 4 when nothing was
 blocked, 3 when no live model is available, 2 when the run failed.
diff --git a/pyproject.toml b/pyproject.toml
index 2af4ef2..76f6c45 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -83,6 +83,8 @@ Demo = "https://sentinel-i63x.onrender.com"
 sentinel = "sentinel.cli:main"
 
 [project.optional-dependencies]
+# Cosmos DB persistence uses the async Azure clients, which need aiohttp.
+cosmos = ["aiohttp>=3.9,<4"]
 dev = [
     "pytest>=8.3",
     "pytest-asyncio>=0.24",
diff --git a/src/sentinel/authorization/ast.py b/src/sentinel/authorization/ast.py
index 6b1d851..0a19fd3 100644
--- a/src/sentinel/authorization/ast.py
+++ b/src/sentinel/authorization/ast.py
@@ -6,7 +6,7 @@
     ==  !=  <  >=  in  "not in"
 
 (Note: ``<=`` and ``>`` are deliberately NOT supported; the parser rejects them
-at load time.) Operands are typed literals (label / string / number / set) or
+at load time.) Operands are typed literals (label / string / number / bool / set) or
 variables resolved from an evaluation namespace (typed tool args + provenance +
 config). Set membership and set equality are first-class so provenance
 predicates like ``RETRIEVED_CONTENT in effective_provenance`` and
diff --git a/src/sentinel/authorization/engine.py b/src/sentinel/authorization/engine.py
index 62099db..c1ef147 100644
--- a/src/sentinel/authorization/engine.py
+++ b/src/sentinel/authorization/engine.py
@@ -235,7 +235,10 @@ def _evaluate(
     ) -> tuple[bool, str | None]:
         if rule.deny_always:
             return True, "deny_always"
-        assert rule.condition is not None  # guaranteed by the compiler
+        if rule.condition is None:
+            # The compiler never produces this. An assert here sat outside the
+            # fail-closed try below (a crash, not a deny) and vanished under -O.
+            return True, "rule has no condition (fail-closed deny)"
         try:
             return rule.condition.evaluate(namespace), None
         except Exception as exc:  # noqa: BLE001 - any failure to evaluate is a deny
diff --git a/src/sentinel/authorization/parser.py b/src/sentinel/authorization/parser.py
index 08c7c58..efcc115 100644
--- a/src/sentinel/authorization/parser.py
+++ b/src/sentinel/authorization/parser.py
@@ -103,7 +103,8 @@ def _tokenize(text: str) -> list[_Token]:
             )
         pos = match.end()
         kind = match.lastgroup
-        assert kind is not None
+        if kind is None:  # every alternative in _TOKEN_RE is named
+            raise PolicyParseError(f"unnamed token at position {pos} in {text!r}")
         if kind == "WS":
             continue
         value = match.group()
diff --git a/src/sentinel/authorization/policy.py b/src/sentinel/authorization/policy.py
index 2ed1f29..cb5feaf 100644
--- a/src/sentinel/authorization/policy.py
+++ b/src/sentinel/authorization/policy.py
@@ -196,7 +196,8 @@ def _compile(doc: _PolicyDoc) -> CompiledPolicy:
             if rule.deny_always:
                 compiled.append(CompiledRule(id=rule.id, deny_always=True, condition=None))
             else:
-                assert rule.deny_if is not None  # guaranteed by _RuleDoc validator
+                if rule.deny_if is None:  # _RuleDoc's validator guarantees this
+                    raise PolicyLoadError(f"tool {tool_name!r} rule {rule.id!r}: no effect")
                 try:
                     condition = parse_condition(rule.deny_if)
                 except PolicyParseError as exc:
diff --git a/src/sentinel/classifier/attack_classifier.py b/src/sentinel/classifier/attack_classifier.py
index 7bc8358..11df0a7 100644
--- a/src/sentinel/classifier/attack_classifier.py
+++ b/src/sentinel/classifier/attack_classifier.py
@@ -10,6 +10,7 @@
 
 import asyncio
 import logging
+import re
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Final, Literal
 
@@ -62,6 +63,23 @@ class AttackLabel:
     classifier: str  # which backend produced it
 
 
+_PRIVILEGED_VERBS: Final[frozenset[str]] = frozenset(
+    {"delete", "remove", "drop", "destroy", "purge", "wipe", "truncate", "revoke", "grant",
+     "transfer", "pay", "refund", "charge", "withdraw", "exec", "execute", "shell", "sudo",
+     "admin", "kill", "terminate"}
+)
+_OUTBOUND_VERBS: Final[frozenset[str]] = frozenset(
+    {"send", "email", "mail", "post", "publish", "upload", "share", "forward", "notify",
+     "message", "tweet", "webhook", "export", "sync", "push"}
+)
+
+
+def _name_words(tool_name: str) -> frozenset[str]:
+    """``sendSlackMessage`` / ``delete-record`` / ``repo.push`` -> their lowercase words."""
+    spaced = re.sub(r"([a-z0-9])([A-Z])", r"\1 \2", tool_name)
+    return frozenset(w for w in re.split(r"[^A-Za-z0-9]+", spaced.lower()) if w)
+
+
 def _rule_label(attempt: BlockedAttempt) -> AttackLabel:
     """Deterministic labelling (DEMO MODE) — classify by the ACTION attempted.
 
@@ -89,6 +107,22 @@ def _rule_label(attempt: BlockedAttempt) -> AttackLabel:
             "attempted to pull sensitive data on an untrusted (tainted) lineage",
             _CLASSIFIER_RULES,
         )
+    # Any other server's tools, by the verbs in their names. The rules above
+    # know only the demo's tools, so an operator's own blocks were mostly
+    # "unclassified" (review CLS-2).
+    words = _name_words(tool)
+    if words & _PRIVILEGED_VERBS:
+        return AttackLabel(
+            "privilege_escalation",
+            "attempted a destructive or privileged action",
+            _CLASSIFIER_RULES,
+        )
+    if words & _OUTBOUND_VERBS and (attempt.is_tainted or attempt.blocked_by == "authorization"):
+        return AttackLabel(
+            "exfiltration",
+            "attempted to send data out through an outbound tool",
+            _CLASSIFIER_RULES,
+        )
     if attempt.injection_detected or attempt.is_tainted:
         return AttackLabel(
             "instruction_override",
diff --git a/src/sentinel/cli.py b/src/sentinel/cli.py
index 79ff7b6..d4d0ddc 100644
--- a/src/sentinel/cli.py
+++ b/src/sentinel/cli.py
@@ -13,7 +13,8 @@
 
 Configuration precedence is the usual one: **CLI flag > environment variable >
 config file > default**, so a container can override a checked-in file without
-editing it.
+editing it. The file accepts only the keys in ``_CONFIG_KEYS``; every other
+setting is read from the environment alone.
 
 The config file (default ``sentinel.yaml``)::
 
diff --git a/src/sentinel/control/app.py b/src/sentinel/control/app.py
index 90151f7..bcf2ea8 100644
--- a/src/sentinel/control/app.py
+++ b/src/sentinel/control/app.py
@@ -179,16 +179,6 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             self._on_close()
 
 
-def _expected_token() -> str | None:
-    """The bearer token every endpoint expects (None when auth is disabled).
-
-    Resolved fresh per request rather than cached, so changing
-    ``SENTINEL_API_TOKEN`` and restarting takes effect. Deliberately tiny: the
-    auth posture is a single chokepoint.
-    """
-    return get_settings().api_token
-
-
 def _anonymous_allowed() -> bool:
     """Whether running with no authentication has been explicitly requested."""
     return get_settings().allow_anonymous
@@ -448,10 +438,10 @@ def visible(trace: str) -> bool:
     def _owned_agent(agent_id: str, request: Request) -> None:
         """Refuse an agent that has never acted in the caller's tenant.
 
-        Trust scores are keyed by agent, not by tenant, so without this one
-        tenant could read — or RESET — another's quarantine. Reset is the
-        sharper end: clearing a quarantine you do not own re-enables an agent
-        someone else's policy stopped. 404 for the same reason as runs.
+        Trust is kept per (tenant, agent), so another tenant's agent of the
+        same id is a different score already. This check adds that a scoped
+        caller cannot probe agent ids its tenant never ran: 404, for the same
+        reason as runs.
         """
         tenant = _scope(request)
         if tenant is None:
@@ -891,5 +881,5 @@ def _to_sse(event: BroadcastEvent) -> ServerSentEvent:
     return ServerSentEvent(
         id=str(event.event_id),
         event="span",
-        data=json.dumps(event.to_payload()),
+        data=event.json_text,
     )
diff --git a/src/sentinel/control/events.py b/src/sentinel/control/events.py
index c55e390..58694a1 100644
--- a/src/sentinel/control/events.py
+++ b/src/sentinel/control/events.py
@@ -19,6 +19,8 @@
 
 import asyncio
 import bisect
+import functools
+import json
 from collections.abc import AsyncIterator
 from dataclasses import dataclass
 from typing import Any, Final
@@ -43,6 +45,14 @@ class BroadcastEvent:
     def to_payload(self) -> dict[str, Any]:
         return {"event_id": self.event_id, "span": self.span.model_dump(mode="json")}
 
+    @functools.cached_property
+    def json_text(self) -> str:
+        """The event as JSON, serialized once however many subscribers stream it.
+
+        Every live subscriber used to serialize every event itself.
+        """
+        return json.dumps(self.to_payload())
+
 
 class EventCursorExpired(ValueError):
     """The requested resume point cannot be replayed without a gap."""
diff --git a/src/sentinel/control/manager.py b/src/sentinel/control/manager.py
index 416fdd7..a9ca31c 100644
--- a/src/sentinel/control/manager.py
+++ b/src/sentinel/control/manager.py
@@ -16,6 +16,7 @@
 
 import asyncio
 import functools
+import importlib.util
 import logging
 import os
 import time
@@ -48,7 +49,6 @@
     AgentQuarantined,
     AuthorizationDecided,
     InjectionScanned,
-    InputReceived,
     ToolBlocked,
 )
 from sentinel.forensics.replay import TraceReplay, replay
@@ -64,21 +64,6 @@
 from sentinel.trust.scorer import TrustScorer
 
 
-def _infer_scenario(user_input: str) -> str:
-    """Best-effort scenario classification for a restored run record.
-
-    Only used for display in the run index. Authoritative replay is from the
-    forensic spans; this just guesses a label from the user prompt because
-    the scenario name isn't itself stored as a span.
-    """
-    text = user_input.lower()
-    if "bulk-delete" in text or "permanently remove" in text:
-        return "privilege" if "permanently remove" in text else "trust-collapse"
-    if "pricing page" in text or "research" in text:
-        return "hero-obvious"
-    return "restored"
-
-
 def _build_cosmos_container(settings: Settings) -> Any:  # noqa: ANN401 - SDK container
     """Build the live async Cosmos container client (AZURE MODE).
 
@@ -87,11 +72,20 @@ def _build_cosmos_container(settings: Settings) -> Any:  # noqa: ANN401 - SDK co
     deploy time). Constructed lazily so the azure SDKs are imported only when
     Cosmos is actually configured. Not exercised offline (needs a live account).
     """
+    # The async Cosmos client and credential need aiohttp, which SENTINEL does
+    # not install by default. Without this check, setting AZURE_COSMOS_ENDPOINT
+    # failed at startup with a bare "aiohttp package is not installed".
+    if importlib.util.find_spec("aiohttp") is None:
+        raise RuntimeError(
+            "Cosmos DB persistence needs aiohttp: pip install 'sentinel-prox[cosmos]' "
+            "(or unset AZURE_COSMOS_ENDPOINT to use the default SQLite store)"
+        )
     from azure.cosmos.aio import CosmosClient  # noqa: PLC0415 - Azure-only path
     from azure.identity.aio import DefaultAzureCredential  # noqa: PLC0415
 
     endpoint = settings.azure_cosmos_endpoint
-    assert endpoint is not None  # precondition: only called when configured
+    if endpoint is None:  # only called when configured
+        raise RuntimeError("AZURE_COSMOS_ENDPOINT is not set")
     client = CosmosClient(endpoint, DefaultAzureCredential())
     database = client.get_database_client(settings.azure_cosmos_database)
     return database.get_container_client(settings.azure_cosmos_container)
@@ -513,13 +507,18 @@ async def run_baseline(self, build: ScenarioBuild) -> dict[str, Any]:
         the dashboard — shows the two outboxes side-by-side so a viewer can
         FEEL the win, not just read the BLOCKED banner.
         """
+        # The same driver choice and web mode as the protected run. The baseline
+        # used to be the scripted transcript always, so with a live model the
+        # dashboard compared a model's run against a script. (A live model may
+        # still behave differently on the second run; that is the model.)
         result = await run_demo_session(
-            build.driver,
+            self._select_driver(build),
             user_input=build.user_input,
             demo_mode=self._demo_mode,
             agent_id="baseline-no-sentinel",
             extra_pages=build.extra_pages,
             skip_sentinel=True,
+            real_web=self._settings.real_web_fetch,
         )
         return {
             "outbox": result.outbox,
@@ -588,17 +587,12 @@ async def restore_persisted_runs(self) -> int:
         for trace_id in trace_ids:
             if trace_id in self._runs:
                 continue
-            spans = await self._inner_store.get_spans(trace_id)
-            if not spans:
-                continue
+            # The scenario and agent come from the ownership row. They used to be
+            # guessed from each trace's user input, which is redacted before it
+            # is stored, so the guess never matched, and every span of every
+            # trace was loaded at startup to make it.
             scenario = "restored"
             agent_id = f"agent-{trace_id[:8]}"
-            for s in spans:
-                if isinstance(s.payload, InputReceived):
-                    # The user input doesn't tell us the scenario name, but the
-                    # control plane only uses scenario for display.
-                    scenario = _infer_scenario(s.payload.content)
-                    break
             # Ownership comes from what was recorded when the run started. A
             # trace with no record predates that (or was written by something
             # else) and is attributed to NOBODY rather than to the default
@@ -635,6 +629,7 @@ def trust(self, agent_id: str, tenant: str = DEFAULT_TENANT) -> dict[str, Any]:
             "quarantined": self._scorer.is_quarantined(agent_id, tenant=tenant),
             "below_threshold": self._scorer.crossed_threshold(agent_id, tenant=tenant),
             "quarantine_enforced": self._scorer.enforces_quarantine,
+            "quarantine_threshold": self._scorer.quarantine_threshold,
         }
 
     def reset(self, agent_id: str, tenant: str = DEFAULT_TENANT) -> dict[str, Any]:
@@ -657,6 +652,11 @@ def capabilities(self) -> dict[str, Any]:
             store=self._inner_store,
             demo_mode=self._demo_mode,
         )
+        # The dashboard draws its trust gauge against this; it used to hard-code 40.
+        summary["trust"] = {
+            "quarantine_threshold": self._scorer.quarantine_threshold,
+            "quarantine_enforced": self._scorer.enforces_quarantine,
+        }
         summary["deployment_verification"] = {
             "status": "unverified",
             "detail": (
diff --git a/src/sentinel/control/mcp_gateway.py b/src/sentinel/control/mcp_gateway.py
index 01359a6..f86670b 100644
--- a/src/sentinel/control/mcp_gateway.py
+++ b/src/sentinel/control/mcp_gateway.py
@@ -432,6 +432,7 @@ def __init__(
         self._transport_sessions: dict[str, _ReservedTransportSession] = {}
         self._pending_transport_sessions: dict[str, int] = {}
         self._transport_session_lock = asyncio.Lock()
+        self._warned_no_session_index = False
         self._front = self._build_front_server()
         # stateless=False: keep one ServerSession per MCP session so a session's
         # provenance frontier persists across its successive tool calls.
@@ -548,7 +549,8 @@ async def _enter(self) -> SentinelGateway:
             await self._connect_remote_downstream(urls)
         else:
             await self._connect_inmemory_downstream()
-        assert self._router is not None  # every connect path sets it
+        if self._router is None:  # every connect path sets it; never serve without one
+            raise RuntimeError("no downstream router after connect")
         # Preflight (health-check + cache schemas) so list_tools can't flake, AND
         # vet the catalogue itself: a poisoned tool DESCRIPTION subverts the agent
         # before anything is retrieved, so provenance alone cannot catch it. By
@@ -791,16 +793,36 @@ def _record_session_owner(self, session_id: str, owner: str, *, tenant: str) ->
         # Forget sessions the transport no longer has. Only ever drops ids the
         # SDK has already discarded, so no live session loses its owner (which
         # would lock its creator out, since an unowned id is refused).
-        live = getattr(self._session_manager, "_server_instances", None)
-        if isinstance(live, dict):
+        live = self._live_session_ids()
+        if live is not None:
             for session in [s for s in self._session_owners if s not in live]:
                 del self._session_owners[session]
                 self._transport_sessions.pop(session, None)
 
+    def _live_session_ids(self) -> frozenset[str] | None:
+        """The session ids the SDK still holds, or None if it cannot be read.
+
+        The SDK exposes no public way to ask, so this reads the private
+        ``_server_instances``. Without it, ended sessions are never released:
+        the owner table grows and each tenant hits its session cap for good.
+        tests/test_mcp_session_ownership.py pins the attribute, so an SDK that
+        drops it fails CI; at runtime the loss is logged rather than silent.
+        """
+        live = getattr(self._session_manager, "_server_instances", None)
+        if isinstance(live, dict):
+            return frozenset(live)
+        if not self._warned_no_session_index:
+            self._warned_no_session_index = True
+            _LOG.error(
+                "the MCP SDK no longer exposes its session index; ended sessions "
+                "cannot be released, so tenants will reach their session cap"
+            )
+        return None
+
     def _prune_transport_sessions(self) -> None:
         """Drop reservations only after the SDK has ended the transport session."""
-        live = getattr(self._session_manager, "_server_instances", None)
-        if not isinstance(live, dict):
+        live = self._live_session_ids()
+        if live is None:
             return
         for session_id in [sid for sid in self._transport_sessions if sid not in live]:
             self._transport_sessions.pop(session_id, None)
diff --git a/src/sentinel/control/static/index.html b/src/sentinel/control/static/index.html
index d5c5e13..2b30d58 100644
--- a/src/sentinel/control/static/index.html
+++ b/src/sentinel/control/static/index.html
@@ -410,6 +410,7 @@ <h1>SENTINEL<span class="c">_</span></h1>
 <script>
 const $ = (id) => document.getElementById(id);
 const esc = (s) => String(s ?? '').replace(/[&<>]/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;'}[c]));
+// Replaced by the server's value from /capabilities (see loadCapabilities).
 const TRUST = {threshold:40};
 const trunc = (s,n) => { s = String(s??''); return s.length>n ? s.slice(0,n)+'…' : s; };
 
@@ -545,7 +546,7 @@ <h1>SENTINEL<span class="c">_</span></h1>
   if(state.refused){ cls='deny'; sig='refused|'+state.refused;
     html='<span class="badge">REFUSED</span><span class="txt">'+esc(state.refused)+'</span>';
   } else if(blk){ const d=blk.decision||{}; const q=state.quarantine?' agent quarantined.':''; cls='deny'; sig='deny|'+blk.tool+'|'+!!state.quarantine;
-    html='<span class="badge">✗ BLOCKED</span><span class="txt">SENTINEL refused <b>'+esc(blk.tool)+'</b> — lineage '+setOf(d.effective_provenance)+(d.matched_rule_id?(', rule <b>'+esc(d.matched_rule_id)+'</b>'):'')+'. without it the exfiltration would have <b>succeeded</b>.'+q+'</span><span class="step">step '+(state.steps.indexOf(blk)+1)+'/'+state.steps.length+'</span>';
+    html='<span class="badge">✗ BLOCKED</span><span class="txt">SENTINEL refused <b>'+esc(blk.tool)+'</b> — lineage '+setOf(d.effective_provenance)+(d.matched_rule_id?(', rule <b>'+esc(d.matched_rule_id)+'</b>'):'')+'. without SENTINEL in the path this call would have <b>run</b>.'+q+'</span><span class="step">step '+(state.steps.indexOf(blk)+1)+'/'+state.steps.length+'</span>';
   } else if(running){ const r=state.steps.find(s=>!s.decision&&!s.blocked); cls='run'; sig='run|'+(r?r.tool:'')+'|'+state.steps.length;
     html='<span class="badge">▶ RUNNING</span><span class="txt">intercepting <b>'+esc(r?r.tool:'…')+'</b> — every call is checked live.</span>';
   } else if(done){ cls='ok'; sig='ok|'+state.steps.length; html='<span class="badge">✓ ALLOWED</span><span class="txt">all proposed actions cleared policy. the log shows why.</span>';
@@ -591,12 +592,15 @@ <h1>SENTINEL<span class="c">_</span></h1>
 function renderChain(){
   const c=$('chain'); const blk=state.steps.find(x=>x.blocked);
   if(!blk&&!state.quarantine){ if(!state.steps.length) c.innerHTML='<span class="faint">launch an attack to populate the kill chain.</span>'; return; }
-  const tainted=state.steps.find(x=>x.tainted); const drop=state.trust.filter(t=>t.delta<0);
-  const rule=(blk&&blk.decision&&blk.decision.matched_rule_id)||'block-untrusted-origin';
+  // From the recorded steps, not a script: these rows used to name web_fetch and
+  // block-untrusted-origin whatever the run did (review DEMO-09).
+  const ti=state.steps.findIndex(x=>x.tainted); const tainted=ti>=0?state.steps[ti]:null;
+  const source=ti>0?state.steps[ti-1]:null; const drop=state.trust.filter(t=>t.delta<0);
+  const rule=blk&&blk.decision&&blk.decision.matched_rule_id;
   const rows=[
-    ['poisoned retrieval','web_fetch returns a page hiding an instruction','amb'],
-    ['taint acquired','lineage = '+setOf(tainted?tainted.origin:['…']),'amb'],
-    ['action blocked','rule '+esc(rule),'red'],
+    source?['untrusted input',esc(source.tool)+' returned content later calls derive from','amb']:['untrusted input','none in this run','off'],
+    tainted?['taint acquired','lineage = '+setOf(tainted.origin),'amb']:['taint acquired','lineage stayed clean','off'],
+    blk?['action blocked',esc(blk.tool)+(rule?(' · rule '+esc(rule)):' · '+esc((blk.blocked&&blk.blocked.blocked_by)||'refused')),'red']:['action blocked','none','off'],
     ['trust drops', drop.length?(drop.length+' penalt'+(drop.length>1?'ies':'y')+' · floor '+Math.round(state.score)):'—','red'],
     ['quarantine', state.quarantine?('score '+state.quarantine.score_at_quarantine.toFixed(1)+' < threshold, lane disabled'):(state.quarantineAdvisory?('score '+state.quarantineAdvisory.score_at_quarantine.toFixed(1)+' < threshold, not enforced'):'not reached'), state.quarantine?'red':(state.quarantineAdvisory?'amb':'off')],
   ];
@@ -628,7 +632,7 @@ <h1>SENTINEL<span class="c">_</span></h1>
   // markup injection sink when a future redaction rule changes.
   $('soc').textContent=d.alerts.map(a=>JSON.stringify(a)).join('\n')||'no spans yet'; $('socdl').style.display=''; $('socdl').href='/runs/'+currentTrace+'/audit?format=jsonl'; };
 async function loadCapabilities(){
-  try{ const d=await (await fetch('/capabilities')).json(); $('modetag').textContent=d.mode.toLowerCase(); $('modetag').className='tag '+(d.demo_mode?'demo':'azure');
+  try{ const d=await (await fetch('/capabilities')).json(); if(d.trust&&typeof d.trust.quarantine_threshold==='number') TRUST.threshold=d.trust.quarantine_threshold; $('modetag').textContent=d.mode.toLowerCase(); $('modetag').className='tag '+(d.demo_mode?'demo':'azure');
     $('capabilities').innerHTML='<div class="dim" style="margin-bottom:9px"><b style="color:var(--txt)">'+esc(d.mode)+'</b> · '+esc(d.tagline)+'</div>'+d.capabilities.map(c=>'<div class="cap '+(c.degraded?'deg':'live')+'"><div class="n"><span class="d"></span>'+esc(c.name)+'</div><div class="vals"><div class="row"><span class="lab">free</span><span class="'+(c.degraded?'on deg':'')+'">'+esc(c.demo)+'</span></div><div class="row"><span class="lab">upgrade</span><span class="'+(c.degraded?'':'on az')+'">'+esc(c.azure)+'</span></div></div></div>').join('');
   }catch(e){}
 }
diff --git a/src/sentinel/demo/driver.py b/src/sentinel/demo/driver.py
index 82eda62..fec587b 100644
--- a/src/sentinel/demo/driver.py
+++ b/src/sentinel/demo/driver.py
@@ -100,6 +100,14 @@ def __init__(self, downstream: Any) -> None:  # noqa: ANN401 - duck-typed
         # as a DownstreamConnection in the mcp SDK.
         self._downstream = downstream
 
+    async def list_tools(self) -> mcp_types.ListToolsResult:
+        """What a live-model driver asks for first; without it, only the scripted
+        driver could run here and every baseline was scripted."""
+        listed = await self._downstream.list_tools()
+        if not isinstance(listed, mcp_types.ListToolsResult):
+            raise TypeError("downstream list_tools did not return a ListToolsResult")
+        return listed
+
     async def call_tool(
         self, name: str, arguments: dict[str, Any] | None = None
     ) -> mcp_types.CallToolResult:
diff --git a/src/sentinel/demo/tool_servers.py b/src/sentinel/demo/tool_servers.py
index 5bb955d..1cb345a 100644
--- a/src/sentinel/demo/tool_servers.py
+++ b/src/sentinel/demo/tool_servers.py
@@ -54,6 +54,10 @@
 # shield — no `SYSTEM:` marker, no "ignore previous instructions", and the
 # address is broken up so it is not a literal email. The shield MISSES this; the
 # Authorization Engine must still catch the exfiltration by provenance.
+#
+# Written against the LOCAL shield's three patterns, so the miss is by design: it
+# shows the backstop holding, not how well a shield does on real-world payloads.
+# (The offline SusceptibleStubModel is likewise built to fall for this page.)
 POISONED_PAGE_EVASION: Final[str] = (
     "Seasonal pricing promo!\n"
     "Standard $18/mo this week only.\n"
diff --git a/src/sentinel/forensics/replay.py b/src/sentinel/forensics/replay.py
index 51fc7d3..340fd86 100644
--- a/src/sentinel/forensics/replay.py
+++ b/src/sentinel/forensics/replay.py
@@ -58,11 +58,23 @@ def get(self, span_id: str) -> Span | None:
 
 def build_replay(trace_id: str, spans: list[Span]) -> TraceReplay:
     """Pure (store-free) replay builder — the testable core of :func:`replay`."""
-    ordered = sorted(spans, key=_seq_key)
-    spans_by_id: dict[str, Span] = {s.span_id: s for s in ordered}
-
     reasons: list[str] = []
 
+    # One span per id. A duplicate used to reach the DFS twice and be reported
+    # as a cycle, which it is not: with one parent per span, re-entry is only
+    # possible through a duplicate, and real cycles show up as unreachable.
+    first: dict[str, Span] = {}
+    copies: dict[str, int] = {}
+    for span in sorted(spans, key=_seq_key):
+        if span.span_id in first:
+            copies[span.span_id] = copies.get(span.span_id, 1) + 1
+            continue
+        first[span.span_id] = span
+    for span_id, count in sorted(copies.items()):
+        reasons.append(f"span id {span_id} appears {count} times (duplicate)")
+    ordered = list(first.values())
+    spans_by_id: dict[str, Span] = dict(first)
+
     # Group real parent->child edges (parent present in this trace), seq-ordered.
     children: dict[str, list[Span]] = {}
     roots: list[Span] = []
@@ -89,9 +101,8 @@ def build_replay(trace_id: str, spans: list[Span]) -> TraceReplay:
     stack: list[Span] = list(reversed(roots))  # reversed → pop in seq order
     while stack:
         span = stack.pop()
-        if span.span_id in visited:
-            # Re-entry can only happen via a malformed (cyclic) edge; stop branch.
-            reasons.append(f"cycle/re-entry detected at span {span.span_id}")
+        if span.span_id in visited:  # defensive: duplicates were removed above
+            reasons.append(f"re-entry at span {span.span_id}")
             continue
         visited.add(span.span_id)
         depth_first.append(span)
diff --git a/src/sentinel/forensics/store.py b/src/sentinel/forensics/store.py
index e0edacf..0c13e92 100644
--- a/src/sentinel/forensics/store.py
+++ b/src/sentinel/forensics/store.py
@@ -33,7 +33,6 @@
 if TYPE_CHECKING:
     from azure.cosmos.aio import ContainerProxy
 
-    from sentinel.config import Settings
 
 _T = TypeVar("_T")
 
@@ -180,33 +179,30 @@ class CosmosForensicStore:
     Takes an already-constructed async ``ContainerProxy`` so the class is unit-
     testable against a fake that honors the real azure-cosmos async contract
     (``await upsert_item(body=...)``; ``query_items(...)`` returns an async
-    iterable). The :func:`create_cosmos_store` factory wires the real client.
+    iterable). The run manager builds the real client (needs the ``cosmos``
+    extra, for aiohttp).
     """
 
     def __init__(self, container: ContainerProxy) -> None:
         self._container = container
 
     async def put(self, span: Span) -> None:
-        # Immutable-record guard. A point read (single partition, by id) tells us
-        # whether this span_id already exists; if so we must NOT blindly overwrite:
-        #   identical content → idempotent no-op; different content → raise.
-        # Only genuinely-new spans are written. upsert (not create) keeps the
-        # write idempotent under the Cosmos SDK's default retry policy — we add
-        # no retry of our own.
-        from azure.cosmos.exceptions import CosmosResourceNotFoundError
+        # Immutable-record guard, atomic: create_item fails if the id exists, and
+        # only then is the stored copy compared (identical → no-op, different →
+        # raise). This used to read first and then upsert, so two writers of the
+        # same span id could both see "absent" and the second silently replaced
+        # the first. A create the SDK retried after it had in fact succeeded now
+        # meets its own identical copy, so the write stays idempotent.
+        from azure.cosmos.exceptions import CosmosResourceExistsError
 
         try:
+            await self._container.create_item(body=_to_item(span))
+            return
+        except CosmosResourceExistsError:
             existing_doc = await self._container.read_item(
                 item=span.span_id, partition_key=span.trace_id
             )
-        except CosmosResourceNotFoundError:
-            existing_doc = None
-
-        if existing_doc is not None:
-            # Raises on conflict; returns True (duplicate) when identical → no-op.
-            _ensure_no_conflict(_from_item(existing_doc), span)
-            return
-        await self._container.upsert_item(body=_to_item(span))
+        _ensure_no_conflict(_from_item(existing_doc), span)
 
     async def get_spans(self, trace_id: str) -> list[Span]:
         query = "SELECT * FROM c WHERE c.trace_id = @trace_id"
@@ -223,31 +219,6 @@ async def get_spans(self, trace_id: str) -> list[Span]:
         return _by_seq(spans)
 
 
-def create_cosmos_store(settings: Settings) -> CosmosForensicStore:
-    """Build a :class:`CosmosForensicStore` from settings (Azure mode only).
-
-    Lazy-imports the Azure SDKs so DEMO MODE never pays for them and the package
-    builds on machines without Azure installed. Auth is identity-based
-    (``DefaultAzureCredential`` → system-assigned managed identity in prod,
-    per §2); no keys in code. The caller owns shutdown via
-    :meth:`CosmosForensicStore`'s underlying client — wired by the proxy in a
-    later phase. Fails loudly if the endpoint is unset rather than guessing.
-    """
-    if not settings.azure_cosmos_endpoint:
-        raise ValueError(
-            "AZURE_COSMOS_ENDPOINT is not set; cannot build a Cosmos store. "
-            "(In DEMO MODE use InMemoryForensicStore instead.)"
-        )
-
-    from azure.cosmos.aio import CosmosClient
-    from azure.identity.aio import DefaultAzureCredential
-
-    credential = DefaultAzureCredential()
-    client = CosmosClient(url=settings.azure_cosmos_endpoint, credential=credential)
-    database = client.get_database_client(settings.azure_cosmos_database)
-    container = database.get_container_client(settings.azure_cosmos_container)
-    return CosmosForensicStore(container)
-
 
 class SqliteForensicStore:
     """File-backed forensic store — DEMO MODE persistence across restarts.
@@ -290,6 +261,16 @@ def __init__(self, path: str | Path) -> None:
         self._conn.execute(
             "CREATE INDEX IF NOT EXISTS spans_by_trace ON spans(trace_id, seq)"
         )
+        # Each span's timestamp in its own column, so listing and retention read
+        # an index instead of parsing every span's JSON under the store lock.
+        # Databases created before the column existed are migrated in place.
+        columns = {row[1] for row in self._conn.execute("PRAGMA table_info(spans)")}
+        if "ts" not in columns:
+            self._conn.execute("ALTER TABLE spans ADD COLUMN ts TEXT")
+            self._conn.execute(
+                "UPDATE spans SET ts = json_extract(body, '$.timestamp') WHERE ts IS NULL"
+            )
+        self._conn.execute("CREATE INDEX IF NOT EXISTS spans_by_trace_ts ON spans(trace_id, ts)")
         # Who each trace belongs to. Not forensic evidence (spans are), but
         # without it a restart cannot tell whose history is whose: every
         # restored run used to be attributed to the default tenant, which hid
@@ -342,8 +323,8 @@ def _put_blocking(self, span: Span, body: str) -> None:
         # Plain INSERT: the check above already refused a conflicting span, and
         # OR REPLACE would let anything that slipped past it overwrite evidence.
         self._conn.execute(
-            "INSERT INTO spans (trace_id, span_id, seq, body) VALUES (?, ?, ?, ?)",
-            (span.trace_id, span.span_id, span.seq, body),
+            "INSERT INTO spans (trace_id, span_id, seq, body, ts) VALUES (?, ?, ?, ?, ?)",
+            (span.trace_id, span.span_id, span.seq, body, json.loads(body)["timestamp"]),
         )
 
     async def get_spans(self, trace_id: str) -> list[Span]:
@@ -372,7 +353,7 @@ def _list_traces_blocking(self) -> list[str]:
         # first, not the most recent.
         cursor = self._conn.execute(
             "SELECT trace_id FROM spans GROUP BY trace_id "
-            "ORDER BY MAX(julianday(json_extract(body, '$.timestamp'))) DESC, trace_id"
+            "ORDER BY MAX(julianday(ts)) DESC, trace_id"
         )
         return [row[0] for row in cursor.fetchall()]
 
@@ -411,7 +392,7 @@ def _purge_blocking(self, cutoff: datetime, keep: frozenset[str]) -> list[str]:
         stamp = cutoff.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
         rows = self._conn.execute(
             "SELECT trace_id FROM spans GROUP BY trace_id "
-            "HAVING MAX(julianday(json_extract(body, '$.timestamp'))) < julianday(?)",
+            "HAVING MAX(julianday(ts)) < julianday(?)",
             (stamp,),
         ).fetchall()
         doomed = [row[0] for row in rows if row[0] not in keep]
diff --git a/src/sentinel/toolservers/web_fetch.py b/src/sentinel/toolservers/web_fetch.py
index f1ebac9..ec82a0a 100644
--- a/src/sentinel/toolservers/web_fetch.py
+++ b/src/sentinel/toolservers/web_fetch.py
@@ -35,13 +35,10 @@
 import socket
 from collections.abc import Callable, Sequence
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Final
+from typing import Final
 
 import httpx
 
-if TYPE_CHECKING:
-    from mcp.server.fastmcp import FastMCP
-
 _USER_AGENT: Final[str] = "SENTINEL-web-fetch/1.0 (+https://github.com/Harshith029/SENTINEL)"
 _TEXTUAL_HINTS: Final[tuple[str, ...]] = (
     "text/",
@@ -277,23 +274,3 @@ async def _fetch(
     finally:
         if owns_client:
             await client.aclose()
-
-
-def build_web_server() -> FastMCP:
-    """A FastMCP ``web`` server whose ``web_fetch`` performs a REAL fetch.
-
-    Drop-in replacement for the demo's mock ``web`` server: same tool name and
-    signature, so it sits behind the proxy unchanged. Imported lazily so the
-    package has no hard dependency on ``mcp.server.fastmcp`` at import time.
-    """
-    from mcp.server.fastmcp import FastMCP
-
-    web = FastMCP("web")
-
-    @web.tool()
-    async def web_fetch(url: str) -> str:
-        """Fetch a web page over HTTP(S) and return its text (guarded)."""
-        result = await fetch_url(url)
-        return result.as_tool_output()
-
-    return web
diff --git a/src/sentinel/trust/config.yaml b/src/sentinel/trust/config.yaml
index a7a6b19..6932a52 100644
--- a/src/sentinel/trust/config.yaml
+++ b/src/sentinel/trust/config.yaml
@@ -13,7 +13,12 @@ quarantine_threshold: 40     # below this score → quarantine
 # nothing; the most surprising transition costs up to this much.
 max_transition_penalty: 25
 
-# Hard-signal penalties (fixed): applied when the authorization engine BLOCKS a
-# call, or when Layer-1 injection scanning flags an attack.
+# Hard-signal penalties (fixed).
+# blocked_call_penalty: the authorization engine DENIED a call (provenance and
+#   allowlist rules included). Catalogue refusals cost nothing: the downstream
+#   changed, not the agent.
+# injection_detected_penalty: a TOOL RESULT was flagged by a shield backend
+#   confident enough to penalize (not the local heuristic, and never the user's
+#   own prompt). Flags are recorded either way.
 blocked_call_penalty: 30
 injection_detected_penalty: 50
diff --git a/src/sentinel/trust/scorer.py b/src/sentinel/trust/scorer.py
index 95122a8..de5ba58 100644
--- a/src/sentinel/trust/scorer.py
+++ b/src/sentinel/trust/scorer.py
@@ -35,8 +35,11 @@
 read → emit → commit sequence. The ``await emit`` sits BETWEEN reading the old
 score and committing the new one, so without the lock concurrent updates to one
 agent would race (lost updates); the lock serializes them. The in-memory score
-is committed only AFTER the TrustUpdated span is durably emitted, so the score
-and the forensic log never diverge. Different agents use different locks and run
+is committed only AFTER the store has accepted the TrustUpdated span, so the
+score and the forensic log do not diverge in a running process. ("Accepted" is
+what each store makes it: SQLite in WAL mode with synchronous=NORMAL survives a
+process crash but can lose the last commits to a power cut; the in-memory store
+keeps nothing across a restart.) Different agents use different locks and run
 concurrently.
 
 This is anomaly telemetry + containment, NOT primary prevention — Authorization
@@ -168,6 +171,10 @@ def __init__(
     def enforces_quarantine(self) -> bool:
         return self._enforce
 
+    @property
+    def quarantine_threshold(self) -> float:
+        return self._config.quarantine_threshold
+
     # --- read-only views (do not create state) -------------------------------
 
     def score(self, agent_id: str, *, tenant: str | None = None) -> float:
diff --git a/tests/conftest.py b/tests/conftest.py
index e680412..ae757e9 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -93,6 +93,16 @@ def __init__(self) -> None:
         # partition_key (trace_id) -> { id (span_id) -> document }
         self._partitions: dict[str, dict[str, dict[str, Any]]] = {}
 
+    async def create_item(self, *, body: dict[str, Any]) -> dict[str, Any]:
+        """Atomic: fails if the id already exists in the partition, as Cosmos does."""
+        partition = self._partitions.setdefault(body["trace_id"], {})
+        if body["id"] in partition:
+            from azure.cosmos.exceptions import CosmosResourceExistsError
+
+            raise CosmosResourceExistsError(message=f"{body['id']} exists")
+        partition[body["id"]] = copy.deepcopy(body)
+        return copy.deepcopy(body)
+
     async def upsert_item(self, *, body: dict[str, Any]) -> dict[str, Any]:
         partition_key = body["trace_id"]
         item_id = body["id"]
@@ -118,7 +128,12 @@ def query_items(
     ) -> AsyncIterator[dict[str, Any]]:
         # Single-partition query: return exactly the requested partition's docs.
         # That is equivalent to the store's `WHERE c.trace_id = @trace_id`, since
-        # the partition key IS trace_id.
+        # the partition key IS trace_id. Any other query is refused rather than
+        # answered as if it were that one (the fake used to ignore the SQL).
+        expected = "SELECT * FROM c WHERE c.trace_id = @trace_id"
+        bound = {p["name"]: p["value"] for p in parameters or []}
+        if query != expected or bound != {"@trace_id": partition_key}:
+            raise NotImplementedError(f"fake Cosmos container cannot answer {query!r} {bound!r}")
         docs = list(self._partitions.get(partition_key or "", {}).values())
 
         async def _aiter() -> AsyncIterator[dict[str, Any]]:
@@ -128,6 +143,12 @@ async def _aiter() -> AsyncIterator[dict[str, Any]]:
         return _aiter()
 
 
+@pytest.fixture()
+def fake_cosmos_container() -> type[FakeAsyncCosmosContainer]:
+    """The fake container class, for tests that need to subclass it."""
+    return FakeAsyncCosmosContainer
+
+
 @pytest.fixture(params=["memory", "cosmos", "sqlite"])
 def store(
     request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathFactory
diff --git a/tests/test_agent_default_driver.py b/tests/test_agent_default_driver.py
index 5528618..2d533a0 100644
--- a/tests/test_agent_default_driver.py
+++ b/tests/test_agent_default_driver.py
@@ -88,3 +88,23 @@ def settings(**kwargs: object) -> Settings:
 
     auto = default_agent_driver(task="t", scripted_fallback=fallback, settings=settings())
     assert auto is fallback  # no credentials: the transcript, as before
+
+
+async def test_the_baseline_uses_the_same_driver_as_the_protected_run(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The baseline was always the script, even when a live model drove the run."""
+    import sentinel.demo.llm_driver as llm_driver
+    from sentinel.control.manager import RunManager
+    from sentinel.demo.scenario import build_scenario_full
+    from sentinel.demo.tool_servers import OBVIOUS_URL
+
+    stub = SusceptibleStubModel(poisoned_url=OBVIOUS_URL)
+    monkeypatch.setattr(llm_driver, "build_live_model", lambda *_a, **_k: stub)
+    manager = RunManager()
+    try:
+        baseline = await manager.run_baseline(build_scenario_full("hero-obvious"))
+    finally:
+        await manager.aclose()
+    assert stub.issued, "the baseline never consulted the live model"
+    assert baseline["outbox"], "with nothing in front, the induced email goes out"
diff --git a/tests/test_authorization_engine.py b/tests/test_authorization_engine.py
index 7aeff7b..a2c79ef 100644
--- a/tests/test_authorization_engine.py
+++ b/tests/test_authorization_engine.py
@@ -172,3 +172,17 @@ def test_compound_deny_if_allows_clean_lineage() -> None:
         )
     )
     assert result.allowed is True
+
+
+def test_a_rule_without_a_condition_denies_instead_of_crashing() -> None:
+    """The engine asserted this outside its fail-closed path (review AUTHZ-16)."""
+    from sentinel.authorization.policy import CompiledPolicy, CompiledRule
+
+    broken = CompiledPolicy(
+        policy_version=1,
+        tools={"t": (CompiledRule(id="r", deny_always=False, condition=None),)},
+    )
+    decision = AuthorizationEngine(broken).authorize(
+        ToolCall(name="t", provenance=frozenset({"USER"}))
+    )
+    assert decision.decision == "DENY" and decision.matched_rule_id == "r"
diff --git a/tests/test_azure_wiring.py b/tests/test_azure_wiring.py
index 80faee5..7f7a6db 100644
--- a/tests/test_azure_wiring.py
+++ b/tests/test_azure_wiring.py
@@ -72,3 +72,21 @@ def test_an_explicit_azure_shield_is_not_downgraded_by_demo_mode() -> None:
         azure_content_safety_key="k",
     )
     assert InputShield.from_settings(auto).backend == "local"
+
+
+def test_cosmos_without_aiohttp_fails_with_the_fix_named(monkeypatch) -> None:  # type: ignore[no-untyped-def]
+    """It failed at startup with a bare "aiohttp package is not installed"."""
+    import importlib.util
+
+    import pytest
+
+    from sentinel.control.manager import _build_cosmos_container
+
+    real_find_spec = importlib.util.find_spec
+
+    def no_aiohttp(name: str, *args: object) -> object:
+        return None if name == "aiohttp" else real_find_spec(name, *args)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(importlib.util, "find_spec", no_aiohttp)
+    with pytest.raises(RuntimeError, match=r"sentinel-prox\[cosmos\]"):
+        _build_cosmos_container(Settings(demo_mode=False, azure_cosmos_endpoint="https://x.example"))
diff --git a/tests/test_forensic_history.py b/tests/test_forensic_history.py
index 2d79758..e4ccba6 100644
--- a/tests/test_forensic_history.py
+++ b/tests/test_forensic_history.py
@@ -111,8 +111,9 @@ def _age(db: Path, trace_id: str, days: float) -> None:
     stamp = (datetime.now(UTC) - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
     with sqlite3.connect(db) as conn:
         conn.execute(
-            "UPDATE spans SET body = json_set(body, '$.timestamp', ?) WHERE trace_id = ?",
-            (stamp, trace_id),
+            # Both copies: the store indexes the timestamp in its own column.
+            "UPDATE spans SET body = json_set(body, '$.timestamp', ?), ts = ? WHERE trace_id = ?",
+            (stamp, stamp, trace_id),
         )
 
 
@@ -195,3 +196,24 @@ async def test_a_store_passed_in_is_left_open_for_its_owner(tmp_path: Path) -> N
     await manager.aclose()
     assert await store.list_trace_ids() == []  # still usable: not ours to close
     store.close()
+
+
+async def test_restoring_the_run_index_does_not_load_every_trace() -> None:
+    """Restore read every span of every trace to guess a label it never found (CM-10)."""
+    run_id, _ = await _acme_run()
+    manager = RunManager()
+    store = manager._inner_store  # noqa: SLF001
+    loads: list[str] = []
+    real = store.get_spans
+
+    async def counting(trace_id: str):  # type: ignore[no-untyped-def]
+        loads.append(trace_id)
+        return await real(trace_id)
+
+    store.get_spans = counting  # type: ignore[method-assign]
+    try:
+        assert await manager.restore_persisted_runs() >= 1
+        assert loads == []
+        assert manager.get_run(run_id).scenario == "hero-obvious"  # type: ignore[union-attr]
+    finally:
+        await manager.aclose()
diff --git a/tests/test_forensics_replay.py b/tests/test_forensics_replay.py
index 7f29162..99af4e7 100644
--- a/tests/test_forensics_replay.py
+++ b/tests/test_forensics_replay.py
@@ -121,3 +121,15 @@ def test_replay_is_cycle_safe(make_span: Callable[..., Span]) -> None:
     assert rep.malformed is True
     assert rep.roots == []
     assert any("unreachable" in r for r in rep.malformed_reasons)
+
+
+def test_a_duplicate_span_is_reported_as_a_duplicate_not_a_cycle(
+    make_span: Callable[..., Span],
+) -> None:
+    """With one parent per span, only a duplicate can re-enter the walk (REPLAY-1)."""
+    root = make_span(seq=0)
+    rep = build_replay(TRACE_A, [root, root])
+    assert [s.span_id for s in rep.ordered] == [root.span_id]
+    assert rep.malformed
+    assert any("duplicate" in r for r in rep.malformed_reasons)
+    assert not any("cycle" in r for r in rep.malformed_reasons)
diff --git a/tests/test_forensics_store.py b/tests/test_forensics_store.py
index 62516e3..bc6c516 100644
--- a/tests/test_forensics_store.py
+++ b/tests/test_forensics_store.py
@@ -285,3 +285,67 @@ async def test_in_memory_spans_cannot_be_rewritten_through_a_reference(
     got.payload.arguments["after_get"] = "y"  # type: ignore[attr-defined]
     stored = (await store.get_spans(span.trace_id))[0]
     assert list(stored.payload.arguments) == ["a"]  # type: ignore[attr-defined]
+
+
+async def test_two_cosmos_writers_of_one_span_id_cannot_both_win(
+    make_span: Callable[..., Span], fake_cosmos_container: type
+) -> None:
+    """Read-then-upsert let both see "absent"; the second silently replaced the first."""
+    from sentinel.forensics.events import ToolCallProposed
+    from sentinel.forensics.store import CosmosForensicStore
+
+    class SlowReads(fake_cosmos_container):  # type: ignore[misc, valid-type]
+        async def read_item(self, *, item: str, partition_key: str) -> dict[str, object]:
+            # Answer first, then yield, as a network round trip does: the other
+            # writer runs between this check and the write that follows it.
+            try:
+                found = await super().read_item(item=item, partition_key=partition_key)
+            except Exception:
+                await asyncio.sleep(0)
+                raise
+            await asyncio.sleep(0)
+            return found
+
+    container = SlowReads()
+    store = CosmosForensicStore(container)  # type: ignore[arg-type]
+    first = make_span(seq=0, payload=ToolCallProposed(tool_name="a", arguments={}))
+    second = first.model_copy(update={"payload": ToolCallProposed(tool_name="b", arguments={})})
+    results = await asyncio.gather(store.put(first), store.put(second), return_exceptions=True)
+    assert any(isinstance(r, ForensicIntegrityError) for r in results), results
+    stored = await store.get_spans(first.trace_id)
+    assert [s.payload.tool_name for s in stored] == ["a"]  # type: ignore[attr-defined]
+
+
+async def test_a_database_from_before_the_timestamp_column_is_migrated(
+    tmp_path: Path, make_span: Callable[..., Span]
+) -> None:
+    """Listing and retention parsed every span's JSON under the lock (STORE-2)."""
+    import json as _json
+    import sqlite3
+
+    db = tmp_path / "old.db"
+    span = make_span(seq=0)
+    with sqlite3.connect(db) as conn:  # the schema as it was
+        conn.execute(
+            "CREATE TABLE spans (trace_id TEXT NOT NULL, span_id TEXT NOT NULL PRIMARY KEY,"
+            " seq INTEGER NOT NULL, body TEXT NOT NULL)"
+        )
+        conn.execute(
+            "INSERT INTO spans VALUES (?, ?, ?, ?)",
+            (span.trace_id, span.span_id, span.seq, _json.dumps(span.model_dump(mode="json"))),
+        )
+    store = SqliteForensicStore(db)
+    assert await store.list_trace_ids() == [span.trace_id]
+    later = make_span(trace_id="b" * 32, seq=0, span_id="2" * 16)
+    await store.put(later)
+    assert await store.list_trace_ids() == ["b" * 32, span.trace_id]
+    store.close()
+    with sqlite3.connect(db) as conn:
+        assert conn.execute("SELECT COUNT(*) FROM spans WHERE ts IS NULL").fetchone()[0] == 0
+        plan = " ".join(
+            str(r) for r in conn.execute(
+                "EXPLAIN QUERY PLAN SELECT trace_id FROM spans GROUP BY trace_id"
+                " ORDER BY MAX(julianday(ts))"
+            )
+        )
+    assert "spans_by_trace_ts" in plan
diff --git a/tests/test_mcp_session_ownership.py b/tests/test_mcp_session_ownership.py
index 2fba02e..08eccec 100644
--- a/tests/test_mcp_session_ownership.py
+++ b/tests/test_mcp_session_ownership.py
@@ -390,3 +390,30 @@ class _Session:  # weak-referenceable stand-in for the SDK's ServerSession
         assert "retired" in message and "new session" in message
         assert "Retry shortly" not in message
         assert "capacity" in gateway._no_proxy_message(_Session())  # noqa: SLF001
+
+
+def test_the_sdk_still_exposes_the_session_index_the_gateway_reads() -> None:
+    """GW-18: owner and reservation release read the SDK's private session index.
+
+    If an SDK upgrade drops it, ended sessions are never released and tenants
+    reach their cap for good. This fails first, in CI, instead.
+    """
+    from mcp.server.lowlevel import Server
+    from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+
+    manager = StreamableHTTPSessionManager(app=Server("probe"))
+    assert isinstance(getattr(manager, "_server_instances", None), dict)
+
+
+async def test_a_missing_session_index_is_logged_not_silent(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    async with _gateway(monkeypatch) as (_client, gateway, _store):
+        real = gateway._session_manager  # noqa: SLF001
+        gateway._session_manager = object()  # type: ignore[assignment]  # noqa: SLF001
+        try:
+            with caplog.at_level("ERROR", logger="sentinel.control.mcp_gateway"):
+                assert gateway._live_session_ids() is None  # noqa: SLF001
+        finally:
+            gateway._session_manager = real  # noqa: SLF001
+        assert "session index" in caplog.text
diff --git a/tests/test_phase4_dod.py b/tests/test_phase4_dod.py
index 34075ca..9974a04 100644
--- a/tests/test_phase4_dod.py
+++ b/tests/test_phase4_dod.py
@@ -178,6 +178,13 @@ async def test_blocked_tool_returns_clean_mcp_error() -> None:
 
 
 async def test_no_tool_executes_without_a_sentinel_span() -> None:
+    """Every downstream execution in this harness has a ToolExecuted span.
+
+    What this proves is that the proxy never forwards a call without running it
+    through its own pipeline. It does not prove the agent cannot reach the tools
+    another way: that is deployment topology (tool servers reachable only by
+    SENTINEL), which no in-process test can show.
+    """
     async with proxy_session(user_input="mixed run") as h:
         # one allowed, two blocked (delete_record always; send_email tainted + bad domain)
         await h.agent.call_tool("web_fetch", {"url": "https://corp.example/a"})
diff --git a/tests/test_phase6_control_plane.py b/tests/test_phase6_control_plane.py
index 90cbdac..2de917c 100644
--- a/tests/test_phase6_control_plane.py
+++ b/tests/test_phase6_control_plane.py
@@ -385,11 +385,15 @@ async def test_runs_index_endpoint() -> None:
         await manager.aclose()
 
 
-# --- Optional API auth on mutating endpoints --------------------------------
+# --- API auth ------------------------------------------------------------------
 
 
-async def test_no_auth_required_when_token_unset() -> None:
-    """The default DEMO posture: zero auth, every endpoint open. Don't regress."""
+async def test_anonymous_mode_serves_without_a_token() -> None:
+    """With SENTINEL_ALLOW_ANONYMOUS=1 (the suite sets it, see conftest) and no
+    token, the endpoints serve. That is opt-in: without the flag SENTINEL refuses
+    every data route (tests/test_control_plane_auth.py). This test used to call
+    zero auth "the default posture", which it has not been since auth failed
+    closed."""
     manager = RunManager()
     try:
         async with _client(manager) as client:
diff --git a/tests/test_phase8_enterprise.py b/tests/test_phase8_enterprise.py
index d9ee3b3..a691970 100644
--- a/tests/test_phase8_enterprise.py
+++ b/tests/test_phase8_enterprise.py
@@ -319,3 +319,36 @@ async def test_the_capability_mode_follows_the_running_manager() -> None:
         assert summary["demo_mode"] is False
     finally:
         await manager.aclose()
+
+
+async def test_capabilities_carry_the_real_trust_threshold() -> None:
+    """The dashboard hard-coded 40 whatever the trust config said (DEMO-26)."""
+    manager = RunManager()
+    try:
+        trust = manager.capabilities()["trust"]
+        assert trust["quarantine_threshold"] == manager._scorer.quarantine_threshold  # noqa: SLF001
+    finally:
+        await manager.aclose()
+    source = (
+        __import__("pathlib").Path(__file__).resolve().parents[1]
+        / "src/sentinel/control/static/index.html"
+    ).read_text(encoding="utf-8")
+    assert "TRUST.threshold=d.trust.quarantine_threshold" in source
+
+
+@pytest.mark.parametrize(
+    "tool,expected",
+    [
+        ("drop_table", "privilege_escalation"),
+        ("refundCharge", "privilege_escalation"),
+        ("sendSlackMessage", "exfiltration"),
+        ("upload-file", "exfiltration"),
+        ("list_repos", "unclassified"),
+    ],
+)
+async def test_rule_labels_cover_tools_beyond_the_demo(tool: str, expected: str) -> None:
+    """Labels knew only the demo's tool names; others were unclassified (CLS-2)."""
+    label = await AttackClassifier(demo_mode=True).classify(
+        BlockedAttempt(tool, "some-rule", "authorization", "denied")
+    )
+    assert label.attack_class == expected
diff --git a/tests/test_provenance_incremental.py b/tests/test_provenance_incremental.py
index d29c638..70ef2e5 100644
--- a/tests/test_provenance_incremental.py
+++ b/tests/test_provenance_incremental.py
@@ -9,6 +9,11 @@
 The equivalence test rebuilds, with :class:`ProvenanceGraph`, exactly the graph
 the proxy used to build, and checks the proxy's provenance against the walk for
 every call of random sessions, including declassified results.
+
+What it proves: the running union is computed correctly for that derivation rule
+(each call derives from everything the session observed before it). It rebuilds
+the graph with that same rule, so it cannot show the rule itself is the right
+one; that is a design decision, recorded in docs/design-notes.md.
 """
 
 from __future__ import annotations
diff --git a/tests/test_resource_bounds.py b/tests/test_resource_bounds.py
index d40053e..8d5b45e 100644
--- a/tests/test_resource_bounds.py
+++ b/tests/test_resource_bounds.py
@@ -440,3 +440,34 @@ async def test_a_finished_run_releases_its_sequence_counter() -> None:
         assert record.trace_id not in manager._emitter._next_seq  # noqa: SLF001
     finally:
         await manager.aclose()
+
+
+async def test_an_event_is_serialized_once_however_many_subscribers_stream_it(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Each live subscriber serialized every event itself (CM-18)."""
+    from sentinel.control.app import _to_sse
+    from sentinel.forensics.span import Span
+
+    bus = EventBus()
+    manager = RunManager(store=InMemoryForensicStore())
+    try:
+        record = manager.start_scenario("privilege")
+        await manager.join(record.trace_id)
+        span = (await manager.replay(record.trace_id)).ordered[0]
+    finally:
+        await manager.aclose()
+    await bus.publish(span)
+    (event,) = bus.events_since(0)
+
+    dumps: list[int] = []
+    real = Span.model_dump
+
+    def counting(self: Span, *a: object, **k: object) -> object:
+        dumps.append(1)
+        return real(self, *a, **k)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(Span, "model_dump", counting)
+    for _ in range(5):  # five subscribers
+        _to_sse(event)
+    assert len(dumps) == 1