chore: bump version to 1.6.2

121 warnings -> 0. Two upstream deprecations get narrowly-scoped
filterwarnings entries (the mcp streamablehttp_client rename — adoption
deliberately rides the v2 migration since the new entry point's call
shape changes again there; the starlette httpx TestClient notice). The
one real RuntimeWarning is fixed at the source: tests that mock
asyncio.run_coroutine_threadsafe handed real coroutines to a stub that
never awaited them, GC-firing 'coroutine was never awaited' inside
whatever unrelated test ran later (the same cross-test bleed mechanism
as the CI closed-stream spew — per-test filterwarnings markers cannot
catch it, which is why two such markers existed and still leaked). A
shared _dispatch_stub now closes real coroutines before returning the
canned future; the obsolete markers are removed.

Dockerfile

@@ -8,7 +8,7 @@ FROM python:3.14-slim
 LABEL org.opencontainers.image.title="turnstone" \
       org.opencontainers.image.description="Multi-node AI orchestration platform"
 
-COPY --from=ghcr.io/astral-sh/uv:0.11.19 /uv /usr/local/bin/uv
+COPY --from=ghcr.io/astral-sh/uv:0.11.21 /uv /usr/local/bin/uv
 
 # Remove the slim image's man page exclusion so man-db has actual content
 RUN rm -f /etc/dpkg/dpkg.cfg.d/docker

docs/architecture.md

@@ -702,7 +702,7 @@ agent_model = "claude"
 
 Each `[models.*]` entry produces a `ModelConfig` with a `provider` field
 (default: `"openai"`). Supported values: `"openai"`, `"anthropic"`, `"google"`,
-and `"openai-compatible"`.
+`"openai-compatible"`, and `"anthropic-compatible"`.
 
 **Per-model sampling overrides:** Each model can specify `temperature`,
 `max_tokens`, and `reasoning_effort` to override the global defaults from
@@ -765,6 +765,63 @@ model = "qwen-3.5-vl"
 supports_vision = true
 ```
 
+**Anthropic-compatible local servers (vLLM `/v1/messages`):** the
+`"anthropic-compatible"` provider drives local servers that expose
+Anthropic's Messages API for arbitrary checkpoints — vLLM's
+`/v1/messages` endpoint, which requires a release with thinking-block
+support in the Anthropic endpoint (post-2026-02-28; verified against
+v0.22.1rc1). The lane reuses `AnthropicProvider` in compat mode: same
+wire translation as the real Anthropic lane, but every model resolves to
+the `_ANTHROPIC_COMPAT_DEFAULT` capabilities (200K context, 64K output,
+`token_param=max_tokens`, `thinking_mode=none`, no native
+web_search/tool_search, no vision) — the static Claude table never
+applies to local checkpoints. `base_url` is required — the server root
+WITHOUT `/v1` (the Anthropic SDK appends `/v1/messages`); a trailing
+`/v1` pasted out of openai-compatible habit is stripped automatically,
+and an empty value fails at client construction rather than falling
+back to the commercial endpoint. Set a
+placeholder `api_key` (e.g. `"dummy"`) for unauthenticated servers. Tool calling
+needs the server started with `--enable-auto-tool-choice
+--tool-call-parser <family>` plus the matching reasoning parser.
+Per-model capability overrides opt in to what the checkpoint actually
+supports:
+
+```toml
+[models.vllm-claude]
+provider = "anthropic-compatible"
+base_url = "http://localhost:8000"   # no /v1 — the SDK appends /v1/messages
+api_key = "dummy"
+model = "deepseek-ai/DeepSeek-V4-Flash"
+
+[models.vllm-claude.capabilities]
+supports_vision = true                    # multimodal checkpoints only
+supports_mid_conversation_system = true   # template-dependent
+context_window = 131072
+```
+
+The reasoning toggle does NOT use Anthropic's `thinking` request param.
+Toggle it through the chat template instead: set `{"chat_template_kwargs":
+{"thinking": false}}` as extra body params in the admin Models
+server-compat section (for this provider the section shows only the
+extra-body field — server type, API surface, and thinking mode are
+openai-compatible-only knobs); the provider forwards it via the SDK's
+`extra_body`.
+
+Verified quirks of vLLM's Anthropic endpoint:
+
+* The `thinking` request param is silently dropped — use
+  `chat_template_kwargs` (above) to control reasoning.
+* `stop_sequences` cut the raw stream wherever the text appears —
+  including inside thinking — and report `end_turn` with
+  `stop_sequence=None`. Turnstone does not send stop sequences from
+  this provider.
+* No cache telemetry: `usage` carries input/output token counts only
+  (no `cache_creation_input_tokens` / `cache_read_input_tokens`).
+* Images require a multimodal checkpoint — text-only models return a
+  500 on image blocks, so `supports_vision` stays opt-in per model.
+* Mid-conversation `role: "system"` turns are template-dependent —
+  opt in per model via `supports_mid_conversation_system`.
+
 **Database model definitions:** On server entry points, models can also be
 defined in the `model_definitions` table (admin Models tab). DB models support
 the same per-model sampling overrides. Config.toml models override DB models

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "turnstone"
-version = "1.6.0"
+version = "1.6.2"
 description = "Multi-node AI orchestration platform with tool use, agent routing, and cluster simulation."
 readme = "README.md"
 license = "Apache-2.0"
@@ -26,7 +26,7 @@ dependencies = [
     "openai>=2.37",
     "anthropic>=0.108",  # claude-fable-5 support; hard runtime floor is 0.105 (mid-conversation system blocks)
     "httpx>=0.28",
-    "mcp>=1.27",
+    "mcp>=1.27,<2",  # v2 is a breaking rewrite (2.0.0a1 live 2026-06-11; stable ~2026-07-27) — streamablehttp_client removed, 2-tuple transport, snake_case types; migrate deliberately
     "starlette>=1.0.1",  # PYSEC-2026-161: host-header path-injection in URL reconstruction (auth-bypass on apps comparing reconstructed URL paths)
     "uvicorn>=0.34",
     "sse-starlette>=2.0",
@@ -94,6 +94,15 @@ include = [
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 markers = ["live: requires a running LLM backend"]
+filterwarnings = [
+    # mcp v1 deprecates streamablehttp_client for an entry point whose call
+    # shape only settles in v2 — adoption rides the deliberate v2 migration
+    # (pin capped <2); silence exactly this message until then.
+    "ignore:Use `streamable_http_client` instead",
+    # starlette deprecates the httpx-backed TestClient; revisit at the next
+    # starlette floor bump.
+    "ignore:Using `httpx` with `starlette.testclient` is deprecated",
+]
 
 [tool.ruff]
 target-version = "py311"

tests/test_app_js.py

@@ -285,6 +285,38 @@ def test_system_turn_dedups_against_history_by_event_id() -> None:
         "replayHistory (and the live handler) must record system-turn ids for the dedup set."
     )
 
+    # Pin the wiring on BOTH read paths, scoped to its method — a refactor that
+    # keeps the Set but drops the live-handler consultation (or the
+    # replayHistory-side record) silently re-opens the double-render while the
+    # file-global checks above still pass.
+    live_start = body.index('case "system_turn":')
+    # End at the NEXT switch case, not the first ``break;`` — the dedup-skip
+    # path breaks before the ``.add(``, so a ``break;``-bounded slice would
+    # drop the record half and false-fail the ``.add(`` assertion below.
+    # Whitespace-tolerant so a reformat can't silently break the bound.
+    next_case = re.search(r'\n\s*case "', body[live_start + 1 :])
+    assert next_case, (
+        "no switch case found after system_turn to bound the pin slice — if "
+        "system_turn became the last case, re-anchor this pin's end marker."
+    )
+    live_block = body[live_start : live_start + 1 + next_case.start()]
+    assert re.search(r"_renderedSystemEventIds[\s\S]*?\.\s*has\(", live_block), (
+        "the live system_turn handler must CONSULT the dedup set (skip an id "
+        "already painted from /history), not merely reference the Set elsewhere."
+    )
+    assert re.search(r"_renderedSystemEventIds[\s\S]*?\.\s*add\(", live_block), (
+        "the live system_turn handler must RECORD the id it renders so a later "
+        "/history re-render (clear_ui) doesn't repaint it."
+    )
+
+    replay_start = _pane_method_offset(body, "replayHistory")
+    replay_end = _pane_method_offset(body, "_attachRetryToLastAssistant")
+    replay_block = body[replay_start:replay_end]
+    assert re.search(r"_renderedSystemEventIds[\s\S]*?\.\s*add\(", replay_block), (
+        "replayHistory must record each replayed system row's event_id so the "
+        "live system_turn handler can dedup against it."
+    )
+
 
 def test_retry_walk_skips_operator_context_cards() -> None:
     """Interactive twin of the coord retry-skip guard.
@@ -396,9 +428,11 @@ def test_phase8_mcp_error_helpers_defined() -> None:
     (interactive consent / forbidden / operator card) moved into the shared
     interactive module with the Pane.  The consent-badge state
     (``_pendingConsentServers`` / ``_onConsentDetected``) stays in the
-    standalone shell — it drives the settings-gear badge — and the pane reaches
-    it through the ``host.onConsentDetected`` seam (a no-op in the console,
-    which has no gear badge).  Pin both halves and the seam."""
+    standalone shell — it drives the rail's Manage-row badge — and the pane
+    reaches it through the ``host.onConsentDetected`` seam.  The shared host
+    bridges that seam to the standalone via ``window.TS_APP.onConsentDetected``
+    (undefined on the console, so it stays a no-op there).  Pin both halves and
+    the bridge."""
     inter = _INTERACTIVE_JS.read_text(encoding="utf-8")
     assert "function tryParseMcpError" in inter
     assert "function buildMcpErrorEmbed" in inter
@@ -408,14 +442,62 @@ def test_phase8_mcp_error_helpers_defined() -> None:
     assert "onConsentDetected(s)" in inter, (
         "the pane must notify consent through host.onConsentDetected"
     )
+    # The shared host bridges the seam to the standalone subsystem (feature-
+    # detected, so the console — which never defines the hook — no-ops).
+    assert "window.TS_APP.onConsentDetected(server)" in inter, (
+        "the shared interactive host must bridge onConsentDetected to the TS_APP seam"
+    )
     app = _APP_JS.read_text(encoding="utf-8")
     assert "_pendingConsentServers" in app
     assert "function _onConsentDetected" in app
-    assert "onConsentDetected(server)" in app, (
-        "STANDALONE_HOST must wire host.onConsentDetected -> _onConsentDetected"
+    assert "window.TS_APP.onConsentDetected = _onConsentDetected" in app, (
+        "the standalone must expose _onConsentDetected on the TS_APP seam for the pane bridge"
     )
 
 
+def test_consent_badge_drives_rail_manage_row() -> None:
+    """The pending-consent badge was re-homed off the retired settings gear
+    (``#settings-btn``, deleted in the L-shell renovation, which silently made
+    the badge invisible) onto the rail's Manage > Connections row.  Classic
+    app.js can't import the ESM rail module, so it drives the rail's generic
+    ``setRowBadge`` hook through the ``window.TS_SHELL`` bridge — keyed on the
+    standalone's Connections tab.  Pin the new lane and the absence of the dead
+    gear lookup."""
+    app = _APP_JS.read_text(encoding="utf-8")
+    # The badge refresh must drive the rail bridge, not the deleted gear.
+    assert 'getElementById("settings-btn")' not in app, (
+        "the consent badge must no longer target the retired #settings-btn gear"
+    )
+    assert "shell.setRowBadge(_CONSENT_BADGE_TAB" in app, (
+        "_refreshConsentBadge must drive the rail Manage-row badge via the TS_SHELL bridge"
+    )
+    assert 'const _CONSENT_BADGE_TAB = "connections"' in app, (
+        "the standalone badge rides the Connections Manage tab (its MCP surface)"
+    )
+    # The hydrate + clear paths must still funnel through the single refresh.
+    assert "function loadPendingConsents" in app and "_refreshConsentBadge()" in app
+
+
+def test_media_player_activation_not_duplicated_in_standalone() -> None:
+    """The media-player activation (``_loadHls`` / ``_activatePlayer`` + the
+    click/keydown delegate) moved into the shared interactive pane so BOTH the
+    standalone server and the console activate the Play button.  The standalone
+    app.js must NOT keep its own copy — a duplicate document-level listener
+    would double-fire on the standalone (two players swapped in) while the lift
+    is what fixed the console (where app.js was never the host).  Pin the
+    standalone clean so the stale copy can't drift back in."""
+    app = _APP_JS.read_text(encoding="utf-8")
+    for name in ("_loadHls", "_activatePlayer", "_isHlsUrl", "media-play-btn"):
+        assert name not in app, (
+            f"standalone app.js must not re-declare the lifted media player "
+            f"({name!r}) — it lives in shared_static/interactive.js now"
+        )
+    # The lift target carries the real implementation (the click delegate too).
+    inter = _INTERACTIVE_JS.read_text(encoding="utf-8")
+    assert "function _activatePlayer(" in inter
+    assert "activateMediaPlayButton(btn)" in inter
+
+
 def test_phase8_settings_panel_handlers_defined() -> None:
     """The settings modal exposes four entry points that the inline
     ``onclick`` attributes in index.html depend on. Renaming or
@@ -676,7 +758,9 @@ def test_phase8_css_classes_present_in_stylesheet() -> None:
     connections render in the Admin pane's Connections panel (#view-admin), not a
     floating dialog — so #settings-overlay / #settings-box are no longer pinned.
     The revoke confirm's chrome moved to /shared/hatch.css with the dialog-tier
-    conversion, so no #revoke-mcp-* rule is pinned here either."""
+    conversion, so no #revoke-mcp-* rule is pinned here either.  The pending-
+    consent badge moved off the retired settings gear onto the rail's Manage row
+    (shell.css `.rail-badge`), so `.settings-consent-badge` is gone from here."""
     css = _STYLE_CSS.read_text(encoding="utf-8")
     for selector in [
         ".mcp-error-card",
@@ -684,9 +768,13 @@ def test_phase8_css_classes_present_in_stylesheet() -> None:
         ".mcp-error-action-btn",
         ".mcp-scope-pill",
         ".settings-revoke-btn",
-        ".settings-consent-badge",
     ]:
         assert selector in css, f"Missing CSS rule for {selector}"
+    # The dead gear-badge rule must be GONE (its host #settings-btn was retired).
+    assert ".settings-consent-badge" not in css, (
+        "the retired settings-gear consent badge CSS must be removed "
+        "(the badge now lives on the rail Manage row — shell.css .rail-badge)"
+    )
 
 
 def test_phase8_consent_url_prefix_check_in_click_handler() -> None:

tests/test_coordinator_page.py

@@ -8,6 +8,8 @@ visitor lands on the page but all API calls fail).
 
 from __future__ import annotations
 
+import re
+
 import pytest
 from starlette.applications import Starlette
 from starlette.routing import Route
@@ -401,6 +403,49 @@ def test_coord_dedups_system_turn_against_history_by_event_id():
         "false-skip after clear_ui / replay_truncated."
     )
 
+    # The seam must be wired on BOTH read paths, not merely present somewhere
+    # in the file — a refactor that keeps the Set but drops the live-handler
+    # consultation (or the history-side record) silently re-opens the
+    # double-render. Scope each assertion to its block so the wiring, not the
+    # bare symbol, is pinned. (A dedupe-neutered factory — guard short-circuited
+    # to ``false`` — still contains ``renderedSystemEventIds.has(`` and so
+    # passes the file-global checks above; these slice checks catch it.)
+    sys_case = body.index('case "system_turn":')
+    # End at the NEXT switch case, not the first ``break;`` — the dedup-skip
+    # ``...has(sysEid)) break;`` is itself a break that precedes the ``.add(``,
+    # so a ``break;``-bounded slice would drop the record half.
+    # Whitespace-tolerant so a reformat can't silently break the bound.
+    next_case = re.search(r'\n\s*case "', body[sys_case + 1 :])
+    assert next_case, (
+        "no switch case found after system_turn to bound the pin slice — if "
+        "system_turn became the last case, re-anchor this pin's end marker."
+    )
+    live_block = body[sys_case : sys_case + 1 + next_case.start()]
+    assert "renderedSystemEventIds.has(" in live_block, (
+        "the live system_turn handler must CONSULT the dedup set (skip an id "
+        "already painted from /history) — not just reference the Set elsewhere."
+    )
+    assert "renderedSystemEventIds.add(" in live_block, (
+        "the live system_turn handler must RECORD the id it renders so a later "
+        "/history re-render (clear_ui) doesn't repaint it."
+    )
+
+    # The history render path must seed the set from each replayed system row's
+    # event_id, so a subsequent live replay of the same id is skipped.  Bound
+    # the slice structurally — from the system-role branch to the next role
+    # branch in the same chain (falling back to a generous window when it's
+    # the last branch) — so adding comments/fields inside the branch can't
+    # false-fail a pin that only cares about the wiring.
+    assert 'role === "system"' in body
+    sys_replay = body.index('role === "system"', body.index("refetchHistory"))
+    next_role = re.search(r"role\s*===", body[sys_replay + 1 :])
+    replay_end = sys_replay + 1 + next_role.start() if next_role else sys_replay + 1500
+    replay_window = body[sys_replay:replay_end]
+    assert "renderedSystemEventIds.add(" in replay_window, (
+        "the history render's system-role branch must record each replayed "
+        "turn's event_id so the live system_turn handler can dedup against it."
+    )
+
 
 def test_coord_retry_walk_skips_operator_context_cards():
     """Retry must NOT regenerate a stale assistant turn when the last DOM row is

tests/test_interactive_pane_js.py

@@ -184,6 +184,42 @@ def test_approval_keyboard_shortcuts_wired() -> None:
     )
 
 
+def test_media_playback_lifted_and_pane_owned() -> None:
+    """The media Play affordance is rendered by the pane (buildPlayButton /
+    buildMediaEmbed), so its activation must live in the pane too — the old
+    standalone wired a DOCUMENT-level click/keydown listener in app.js, which
+    the console host never loaded (so the button was dead in console-hosted
+    panes).  The fix mirrors the approval-keydown pattern: a pane-owned listener
+    on this.el, root-scoped via closest(".media-play-btn").  Pin both the
+    lifted helpers and the pane wiring so the document-level regression can't
+    silently come back."""
+    body = _INTERACTIVE.read_text(encoding="utf-8")
+    # The lifted activation machinery now lives in the shared module.
+    for fn in (
+        "function _loadHls(",
+        "function _isHlsUrl(",
+        "function _activatePlayer(",
+        "function activateMediaPlayButton(",
+    ):
+        assert fn in body, f"media player helper must be lifted into the pane: {fn}"
+    # The HLS vendor is fetched by absolute /shared/ URL (resolves in BOTH the
+    # standalone server and the console, where /shared is mounted at the root).
+    assert 'script.src = "/shared/hls-1.6.16/hls.min.js";' in body
+    # Pane-owned + root-scoped — NOT a document-level delegated listener.
+    assert 'this.el.addEventListener("click"' in body, (
+        "media play must be wired on this.el (pane-owned), not document"
+    )
+    assert 'e.target.closest(".media-play-btn")' in body, (
+        "the play handler must be root-scoped via closest, not a document-wide id"
+    )
+    assert "activateMediaPlayButton(btn)" in body
+    collapsed = _strip_comments(body)
+    assert 'document.addEventListener("click"' not in collapsed, (
+        "the pane must not register a document-level click delegate — that is "
+        "the standalone regression that left console panes dead"
+    )
+
+
 def test_controller_terminal_dead_state() -> None:
     """Lifecycle round 2: the console controller must STOP reconnect-polling a
     session that is gone (closed / evicted / node restarted) — three consecutive

tests/test_mcp_client.py

@@ -4,9 +4,10 @@ from __future__ import annotations
 
 import asyncio
 import concurrent.futures
+import inspect
 import json
 import time
-from contextlib import AsyncExitStack
+from contextlib import AsyncExitStack, suppress
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -26,6 +27,26 @@ from turnstone.core.tools import INTERACTIVE_TOOLS, TOOLS, merge_mcp_tools
 # ---------------------------------------------------------------------------
 
 
+def _dispatch_stub(mock_future: MagicMock) -> Any:
+    """Stand-in for ``asyncio.run_coroutine_threadsafe`` in sync-bridge tests.
+
+    Closes the never-scheduled coroutine before handing back the canned
+    future — a mocked dispatch never awaits it, and an unawaited coroutine
+    GC-fires "coroutine ... was never awaited" inside whatever unrelated
+    test happens to be running when collection finally occurs (cross-test
+    bleed that per-test filterwarnings markers cannot catch).
+    """
+
+    def _rct(coro: Any, _loop: Any) -> MagicMock:
+        # Only real coroutines need (or survive) closing — several tests
+        # dispatch a plain MagicMock return value through this seam.
+        if inspect.iscoroutine(coro):
+            coro.close()
+        return mock_future
+
+    return _rct
+
+
 def _fake_mcp_tool(name: str = "search", description: str = "Search stuff") -> MagicMock:
     """Create a mock MCP tool object matching the SDK's Tool type."""
     tool = MagicMock()
@@ -153,8 +174,24 @@ def running_loop_mgr():
     try:
         yield mgr, loop, thread
     finally:
+        # Drain BEFORE stopping: a task left pending (or finished-but-
+        # unretrieved) on a stopped loop becomes cross-test global state —
+        # asyncio reports it at GC time, mid-suite, onto whatever stream
+        # pytest has attached THEN (the "I/O operation on closed file"
+        # spew), and a silently-abandoned loop thread keeps running
+        # manager code against torn-down mocks.
+        async def _cancel_pending() -> None:
+            tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
+            for t in tasks:
+                t.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+        with suppress(Exception):
+            asyncio.run_coroutine_threadsafe(_cancel_pending(), loop).result(timeout=5)
         loop.call_soon_threadsafe(loop.stop)
-        thread.join(timeout=2)
+        thread.join(timeout=5)
+        assert not thread.is_alive(), "mcp test loop thread failed to stop within 5s"
+        loop.close()
 
 
 # ---------------------------------------------------------------------------
@@ -2024,6 +2061,37 @@ class TestShutdownCleanup:
         assert mgr._resource_map == {}
         assert mgr._prompt_map == {}
 
+    def test_shutdown_closes_owned_loop_and_clears_refs(self):
+        """When the manager owns the loop thread, shutdown must close the loop
+        (selector resources leak otherwise) and drop both refs; a second
+        shutdown is then a clean no-op."""
+        import threading as _threading
+
+        mgr = MCPClientManager({})
+        loop = asyncio.new_event_loop()
+        thread = _threading.Thread(target=loop.run_forever, daemon=True)
+        thread.start()
+        mgr._loop = loop
+        mgr._thread = thread
+
+        mgr.shutdown()
+        assert loop.is_closed()
+        assert mgr._loop is None
+        assert mgr._thread is None
+        mgr.shutdown()  # idempotent
+
+    def test_shutdown_leaves_unowned_loop_open(self):
+        """Tests (and any embedder) that wire ``_loop`` directly without a
+        thread own the loop's lifecycle — shutdown must not close it."""
+        mgr = MCPClientManager({})
+        loop = asyncio.new_event_loop()
+        mgr._loop = loop
+        try:
+            mgr.shutdown()
+            assert not loop.is_closed()
+        finally:
+            loop.close()
+
 
 # ---------------------------------------------------------------------------
 # TCP probe and unreachable server handling
@@ -2181,7 +2249,7 @@ class TestFutureCancellation:
         mock_future = MagicMock()
         mock_future.result.side_effect = concurrent.futures.TimeoutError()
         with (
-            patch("asyncio.run_coroutine_threadsafe", return_value=mock_future),
+            patch("asyncio.run_coroutine_threadsafe", new=_dispatch_stub(mock_future)),
             pytest.raises(TimeoutError, match="timed out"),
         ):
             mgr.call_tool_sync("mcp__test__search", {"query": "x"}, timeout=1)
@@ -2192,7 +2260,7 @@ class TestFutureCancellation:
         mock_future = MagicMock()
         mock_future.result.side_effect = concurrent.futures.TimeoutError()
         with (
-            patch("asyncio.run_coroutine_threadsafe", return_value=mock_future),
+            patch("asyncio.run_coroutine_threadsafe", new=_dispatch_stub(mock_future)),
             pytest.raises(TimeoutError, match="timed out"),
         ):
             mgr.read_resource_sync("file:///a.txt", timeout=1)
@@ -2203,7 +2271,7 @@ class TestFutureCancellation:
         mock_future = MagicMock()
         mock_future.result.side_effect = concurrent.futures.TimeoutError()
         with (
-            patch("asyncio.run_coroutine_threadsafe", return_value=mock_future),
+            patch("asyncio.run_coroutine_threadsafe", new=_dispatch_stub(mock_future)),
             pytest.raises(TimeoutError, match="timed out"),
         ):
             mgr.get_prompt_sync("mcp__test__review", timeout=1)
@@ -2216,7 +2284,7 @@ class TestFutureCancellation:
         mock_future.result.side_effect = concurrent.futures.TimeoutError()
         with (
             patch.object(mgr, "_refresh_all", return_value=MagicMock()),
-            patch("asyncio.run_coroutine_threadsafe", return_value=mock_future),
+            patch("asyncio.run_coroutine_threadsafe", new=_dispatch_stub(mock_future)),
             pytest.raises(TimeoutError, match="timed out"),
         ):
             mgr.refresh_sync(timeout=1)
@@ -2331,8 +2399,6 @@ class TestCircuitBreaker:
         assert "srv" not in mgr._circuit_open_until
         assert "srv" not in mgr._circuit_trip_count
 
-    @pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning")
-    @pytest.mark.filterwarnings("ignore:coroutine.*was never awaited:RuntimeWarning")
     def test_call_tool_sync_records_failure_on_timeout(self):
         mgr = MCPClientManager({"test": {"type": "stdio", "command": "echo"}})
         mock_session = MagicMock()
@@ -2343,7 +2409,7 @@ class TestCircuitBreaker:
         mock_future = MagicMock()
         mock_future.result.side_effect = concurrent.futures.TimeoutError()
         with (
-            patch("asyncio.run_coroutine_threadsafe", return_value=mock_future),
+            patch("asyncio.run_coroutine_threadsafe", new=_dispatch_stub(mock_future)),
             pytest.raises(TimeoutError),
         ):
             mgr.call_tool_sync("mcp__test__ping", {}, timeout=1)
@@ -2363,7 +2429,7 @@ class TestCircuitBreaker:
         mock_result.isError = False
         mock_future = MagicMock()
         mock_future.result.return_value = mock_result
-        with patch("asyncio.run_coroutine_threadsafe", return_value=mock_future):
+        with patch("asyncio.run_coroutine_threadsafe", new=_dispatch_stub(mock_future)):
             mgr.call_tool_sync("mcp__test__ping", {}, timeout=5)
         assert mgr._consecutive_failures.get("test") is None
 
@@ -2380,7 +2446,7 @@ class TestCircuitBreaker:
         mock_future = MagicMock()
         mock_future.result.side_effect = BrokenPipeError("dead")
         with (
-            patch("asyncio.run_coroutine_threadsafe", return_value=mock_future),
+            patch("asyncio.run_coroutine_threadsafe", new=_dispatch_stub(mock_future)),
             pytest.raises(BrokenPipeError),
         ):
             mgr.call_tool_sync("mcp__test__ping", {}, timeout=5)
@@ -2414,7 +2480,7 @@ class TestCircuitBreaker:
         mock_future = MagicMock()
         mock_future.result.side_effect = McpError(ErrorData(code=-32601, message="tool not found"))
         with (
-            patch("asyncio.run_coroutine_threadsafe", return_value=mock_future),
+            patch("asyncio.run_coroutine_threadsafe", new=_dispatch_stub(mock_future)),
             pytest.raises(McpError),
         ):
             mgr.call_tool_sync("mcp__test__ping", {}, timeout=5)
@@ -2706,11 +2772,22 @@ class TestCBAutoReconnectRefresh:
             patch.object(mgr, "_refresh_server", side_effect=_refresh),
         ):
             session = mgr._cb_auto_reconnect("srv")
-            # Wait for the scheduled refresh task to actually run on the loop.
+            # Wait for the scheduled refresh task to actually run on the loop,
+            # then for the tracked task to DRAIN — exiting the patch context
+            # while the task is still in flight would hand the un-patched
+            # method to its tail.
             assert refresh_event.wait(timeout=5), "refresh task was not scheduled"
+            deadline = time.time() + 5
+            while mgr._background_tasks and time.time() < deadline:
+                time.sleep(0.02)
+            assert not mgr._background_tasks, "background refresh task never drained"
         assert session is new_session
 
-    def test_auto_reconnect_swallows_refresh_failure(self, running_loop_mgr):
+    def test_auto_reconnect_retrieves_and_logs_refresh_failure(self, running_loop_mgr):
+        """A refresh failure must be RETRIEVED and logged by the task's
+        done-callback — not abandoned for asyncio to report as "Task exception
+        was never retrieved" at GC time (which lands on whatever stream pytest
+        has attached by then: the closed-file CI spew)."""
         import threading as _threading
 
         mgr, _loop, _thread = running_loop_mgr
@@ -2728,10 +2805,32 @@ class TestCBAutoReconnectRefresh:
         with (
             patch.object(mgr, "_connect_one", side_effect=_connect_one),
             patch.object(mgr, "_refresh_server", side_effect=_refresh_failing),
+            patch("turnstone.core.mcp_client.log") as mock_log,
         ):
-            # Must not raise — refresh failures are non-fatal.
+            # Must not raise — refresh failures are non-fatal to the caller.
             session = mgr._cb_auto_reconnect("srv")
-            # Background refresh actually started and exception was swallowed
-            # by the task without affecting the synchronous caller.
             assert refresh_started.wait(timeout=5), "refresh task was not scheduled"
+            # Poll for the WARNING while the patch is still active — gating on
+            # set-emptiness alone would race the un-patch (review-caught: the
+            # warning could land on the restored real logger).
+            deadline = time.time() + 5
+            warn_calls = []
+            while not warn_calls and time.time() < deadline:
+                warn_calls = [
+                    c for c in mock_log.warning.call_args_list if "MCP background" in str(c.args[0])
+                ]
+                time.sleep(0.02)
+            # The tracked task must also fully drain (emptiness now implies
+            # "done AND reported" — discard is the callback's LAST step).
+            deadline = time.time() + 5
+            while mgr._background_tasks and time.time() < deadline:
+                time.sleep(0.02)
+            assert not mgr._background_tasks, "background refresh task never drained"
         assert session is new_session
+        assert warn_calls, (
+            "the refresh failure must be logged by the done-callback, not left "
+            "for GC-time reporting"
+        )
+        exc = warn_calls[0].kwargs.get("exc_info")
+        assert isinstance(exc, RuntimeError)
+        assert "catalog fetch broke" in str(exc)

tests/test_model_probe.py

@@ -164,6 +164,25 @@ class TestProbeModelEndpoint:
         assert result["server_type"] == "anthropic"
         assert result["context_window"] == 1000000
 
+    @patch("turnstone.core.providers.create_client")
+    def test_anthropic_compatible_server_type(self, mock_cc: MagicMock) -> None:
+        m = _mock_model("deepseek-ai/DeepSeek-V4-Flash")
+        mock_cc.return_value = _mock_client(m)
+
+        result = probe_model_endpoint("anthropic-compatible", "http://localhost:8000", "dummy")
+        assert result["reachable"] is True
+        assert result["server_type"] == "anthropic-compatible"
+        assert result["context_window"] is None
+
+    @patch("turnstone.core.providers.create_client")
+    def test_anthropic_compatible_max_model_len(self, mock_cc: MagicMock) -> None:
+        m = _mock_model("deepseek-ai/DeepSeek-V4-Flash", max_model_len=131072)
+        mock_cc.return_value = _mock_client(m)
+
+        result = probe_model_endpoint("anthropic-compatible", "http://localhost:8000", "dummy")
+        assert result["context_window"] == 131072
+        assert result["server_type"] == "anthropic-compatible"
+
     @patch("turnstone.core.providers.create_client")
     def test_connection_failure(self, mock_cc: MagicMock) -> None:
         mock_cc.side_effect = OSError("Connection refused")

tests/test_models_changed_event.py

@@ -142,6 +142,24 @@ def test_create_emits_models_changed(storage: SQLiteBackend) -> None:
     assert collector.emit_models_changed.call_count == 1
 
 
+def test_create_accepts_anthropic_compatible_provider(storage: SQLiteBackend) -> None:
+    """anthropic-compatible passes the _MODEL_PROVIDERS enum check."""
+    client, collector = _make_client(storage)
+    resp = client.post(
+        "/v1/api/admin/model-definitions",
+        json={
+            "alias": "vllm-messages",
+            "model": "deepseek-ai/DeepSeek-V4-Flash",
+            "provider": "anthropic-compatible",
+            "base_url": "http://localhost:8000",
+            "api_key": "dummy",
+            "context_window": 131072,
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert collector.emit_models_changed.call_count == 1
+
+
 def test_update_emits_models_changed(storage: SQLiteBackend) -> None:
     _seed(storage, definition_id="m1", alias="local")
     client, collector = _make_client(storage)

tests/test_orphan_cleanup.py

@@ -0,0 +1,232 @@
+"""Orphan-conversation scan + purge (the ``turnstone-admin orphan-conversations`` verb).
+
+Orphans are conversation rows whose ``workstreams`` row is gone — written by
+historical unregistered paths or by the delete-during-inflight race (a late
+tool-result save re-creating rows after ``delete_workstream``).
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+
+import pytest
+
+from turnstone.admin import _cmd_orphan_conversations
+
+
+def _orphan(backend, ws_id: str, n: int = 2) -> None:
+    """Persist *n* conversation rows for *ws_id* WITHOUT registering it."""
+    for i in range(n):
+        backend.save_message(ws_id, "user" if i % 2 == 0 else "assistant", f"m{i}")
+
+
+def _blob(backend, payload: bytes, origin: str = "upload") -> str:
+    """Save a content-addressed attachment; each save bumps the refcount."""
+    aid = hashlib.sha256(payload).hexdigest()
+    backend.save_attachment(aid, "f.txt", "text/plain", len(payload), "text", payload, origin)
+    return aid
+
+
+class TestOrphanScan:
+    def test_clean_db_has_no_orphans(self, backend):
+        backend.register_workstream("live1")
+        backend.save_message("live1", "user", "hello")
+        assert backend.list_orphan_conversations() == []
+
+    def test_orphan_reported_with_stats(self, backend):
+        _orphan(backend, "ghost1", n=3)
+        scan = backend.list_orphan_conversations()
+        assert len(scan) == 1
+        entry = scan[0]
+        assert entry["ws_id"] == "ghost1"
+        assert entry["rows"] == 3
+        assert entry["first"] <= entry["last"]
+        assert entry["attachment_refs"] == 0
+
+    def test_scan_counts_attachment_refs(self, backend):
+        _orphan(backend, "ghost2", n=1)
+        msg_id = backend.save_message("ghost2", "user", "with attachment")
+        aid = _blob(backend, b"orphan-bytes")
+        backend.set_message_attachments("ghost2", msg_id, [aid])
+        scan = backend.list_orphan_conversations()
+        assert scan[0]["attachment_refs"] == 1
+
+    def test_scan_is_oldest_first(self, backend):
+        _orphan(backend, "newer")
+        _orphan(backend, "older")
+        # Timestamps are insertion-ordered ISO text; rewrite to force ordering.
+        import sqlalchemy as sa
+
+        from turnstone.core.storage._schema import conversations
+
+        with backend._conn() as conn:
+            conn.execute(
+                sa.update(conversations)
+                .where(conversations.c.ws_id == "older")
+                .values(timestamp="2020-01-01T00:00:00")
+            )
+            conn.commit()
+        scan = backend.list_orphan_conversations()
+        assert [o["ws_id"] for o in scan] == ["older", "newer"]
+
+
+class TestOrphanPurge:
+    def test_purge_deletes_only_orphans(self, backend):
+        backend.register_workstream("live1")
+        backend.save_message("live1", "user", "keep me")
+        _orphan(backend, "ghost1", n=4)
+        result = backend.delete_orphan_conversations(["ghost1"])
+        assert result == {"workstreams": 1, "rows": 4, "released_refs": 0, "skipped": 0}
+        assert backend.list_orphan_conversations() == []
+        assert len(backend.load_messages("live1")) == 1
+
+    def test_purge_skips_reregistered_ws(self, backend):
+        """A ws_id that gained a workstreams row between scan and purge survives."""
+        _orphan(backend, "ghost1", n=2)
+        scan = [o["ws_id"] for o in backend.list_orphan_conversations()]
+        backend.register_workstream("ghost1")
+        result = backend.delete_orphan_conversations(scan)
+        assert result["skipped"] == 1
+        assert result["workstreams"] == 0
+        assert result["rows"] == 0
+        assert len(backend.load_messages("ghost1")) == 2
+
+    def test_purge_releases_refcounts_and_prunes_at_zero(self, backend):
+        _orphan(backend, "ghost1", n=1)
+        msg_id = backend.save_message("ghost1", "user", "img")
+        aid = _blob(backend, b"only-orphan-referenced")
+        backend.set_message_attachments("ghost1", msg_id, [aid])
+        result = backend.delete_orphan_conversations(["ghost1"])
+        assert result["released_refs"] == 1
+        assert backend.get_attachment(aid) is None
+
+    def test_purge_keeps_blob_shared_with_live_ws(self, backend):
+        payload = b"shared-bytes"
+        backend.register_workstream("live1")
+        live_msg = backend.save_message("live1", "user", "live ref")
+        aid_live = _blob(backend, payload)
+        backend.set_message_attachments("live1", live_msg, [aid_live])
+
+        _orphan(backend, "ghost1", n=1)
+        ghost_msg = backend.save_message("ghost1", "user", "ghost ref")
+        aid_ghost = _blob(backend, payload)  # same content hash; refcount -> 2
+        backend.set_message_attachments("ghost1", ghost_msg, [aid_ghost])
+        assert aid_live == aid_ghost
+
+        result = backend.delete_orphan_conversations(["ghost1"])
+        assert result["released_refs"] == 1
+        row = backend.get_attachment(aid_live)
+        assert row is not None
+        assert row["refcount"] == 1
+
+    def test_purge_sweeps_config_rows(self, backend):
+        _orphan(backend, "ghost1", n=1)
+        backend.save_workstream_config("ghost1", {"model": "x"})
+        backend.delete_orphan_conversations(["ghost1"])
+        import sqlalchemy as sa
+
+        from turnstone.core.storage._schema import workstream_config
+
+        with backend._conn() as conn:
+            left = conn.execute(
+                sa.select(sa.func.count()).where(workstream_config.c.ws_id == "ghost1")
+            ).scalar()
+        assert left == 0
+
+    def test_purge_dedupes_input(self, backend):
+        """Duplicate ws_ids must not inflate counts or bind params."""
+        _orphan(backend, "ghost1", n=2)
+        result = backend.delete_orphan_conversations(["ghost1", "ghost1", "ghost1"])
+        assert result["workstreams"] == 1
+        assert result["rows"] == 2
+        assert result["skipped"] == 0
+
+    def test_purge_unknown_ws_counts_skipped(self, backend):
+        """An input with no rows and no workstream is reported, not purged."""
+        result = backend.delete_orphan_conversations(["nope-never-existed"])
+        assert result == {"workstreams": 0, "rows": 0, "released_refs": 0, "skipped": 1}
+
+    def test_purge_chunks_large_input(self, backend, monkeypatch):
+        """IN-lists are chunked (SQLite bind-parameter limits) without losing rows."""
+        import turnstone.core.storage._utils as storage_utils
+
+        monkeypatch.setattr(storage_utils, "_PURGE_CHUNK", 2)
+        for i in range(5):
+            _orphan(backend, f"ghost{i}", n=1)
+        result = backend.delete_orphan_conversations([f"ghost{i}" for i in range(5)])
+        assert result["workstreams"] == 5
+        assert result["rows"] == 5
+        assert backend.list_orphan_conversations() == []
+
+    def test_purge_empty_list_is_noop(self, backend):
+        assert backend.delete_orphan_conversations([]) == {
+            "workstreams": 0,
+            "rows": 0,
+            "released_refs": 0,
+            "skipped": 0,
+        }
+
+
+class TestAdminVerb:
+    """The CLI handler over a real (ephemeral) backend."""
+
+    def _args(self, **kw) -> argparse.Namespace:
+        return argparse.Namespace(delete=False, yes=False, **kw)
+
+    def test_scan_reports_and_does_not_delete(self, backend, monkeypatch, capsys):
+        _orphan(backend, "ghost1", n=2)
+        monkeypatch.setattr("turnstone.admin._get_storage", lambda args: backend)
+        _cmd_orphan_conversations(self._args())
+        out = capsys.readouterr().out
+        assert "ghost1" in out
+        assert "--delete" in out
+        assert len(backend.load_messages("ghost1")) == 2
+
+    def test_delete_yes_purges(self, backend, monkeypatch, capsys):
+        _orphan(backend, "ghost1", n=2)
+        monkeypatch.setattr("turnstone.admin._get_storage", lambda args: backend)
+        ns = self._args()
+        ns.delete = True
+        ns.yes = True
+        _cmd_orphan_conversations(ns)
+        out = capsys.readouterr().out
+        assert "Purged 2" in out
+        assert backend.list_orphan_conversations() == []
+
+    def test_delete_confirmation_abort(self, backend, monkeypatch, capsys):
+        _orphan(backend, "ghost1", n=1)
+        monkeypatch.setattr("turnstone.admin._get_storage", lambda args: backend)
+        monkeypatch.setattr("builtins.input", lambda prompt: "n")
+        ns = self._args()
+        ns.delete = True
+        with pytest.raises(SystemExit):
+            _cmd_orphan_conversations(ns)
+        assert len(backend.load_messages("ghost1")) == 1
+
+    def test_delete_summary_reports_partial_skip(self, backend, monkeypatch, capsys):
+        """Mixed batch: summary shows ACTUAL purge counts plus the skipped clause."""
+        _orphan(backend, "ghost1", n=2)
+        _orphan(backend, "ghost2", n=3)
+        real_list = backend.list_orphan_conversations
+
+        def list_then_register():
+            scan = real_list()
+            backend.register_workstream("ghost2")  # wins the scan-to-purge race
+            return scan
+
+        monkeypatch.setattr(backend, "list_orphan_conversations", list_then_register)
+        monkeypatch.setattr("turnstone.admin._get_storage", lambda args: backend)
+        ns = self._args()
+        ns.delete = True
+        ns.yes = True
+        _cmd_orphan_conversations(ns)
+        out = capsys.readouterr().out
+        assert "Purged 2 row(s) across 1 workstream(s)" in out
+        assert "skipped 1 re-registered" in out
+        assert len(backend.load_messages("ghost2")) == 3
+
+    def test_clean_db_message(self, backend, monkeypatch, capsys):
+        monkeypatch.setattr("turnstone.admin._get_storage", lambda args: backend)
+        _cmd_orphan_conversations(self._args())
+        assert "No orphan conversation rows." in capsys.readouterr().out

tests/test_provider_anthropic_compat.py

@@ -0,0 +1,357 @@
+"""Tests for the ``anthropic-compatible`` provider lane.
+
+Local servers (vLLM) expose Anthropic's ``/v1/messages`` wire surface for
+arbitrary checkpoints.  The lane reuses ``AnthropicProvider`` with
+``compat=True``: identical message translation, but capabilities come from
+``_ANTHROPIC_COMPAT_DEFAULT`` for every model (the static Claude table
+never applies), native server-side tools are not injected, and operator
+``server_compat["extra_body"]`` overrides ride the Anthropic SDK's
+``extra_body`` — the channel for vLLM's ``chat_template_kwargs`` reasoning
+toggle.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from tests._session_helpers import make_session as _make_session
+from turnstone.core.providers._anthropic import AnthropicProvider
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _capture_client() -> MagicMock:
+    """Build a fake Anthropic client whose ``messages.stream`` records kwargs."""
+    stream_ctx = MagicMock()
+    stream_ctx.__enter__ = MagicMock(return_value=iter([]))
+    stream_ctx.__exit__ = MagicMock(return_value=False)
+    client = MagicMock()
+    client.messages.stream.return_value = stream_ctx
+    return client
+
+
+_WEB_SEARCH_FUNCTION_TOOL = {
+    "type": "function",
+    "function": {
+        "name": "web_search",
+        "description": "Search the web",
+        "parameters": {
+            "type": "object",
+            "properties": {"query": {"type": "string"}},
+            "required": ["query"],
+        },
+    },
+}
+
+
+# ===========================================================================
+# TestCompatCapabilities
+# ===========================================================================
+
+
+class TestCompatCapabilities:
+    """Capability resolution on the compat lane."""
+
+    def test_compat_capability_defaults(self) -> None:
+        provider = AnthropicProvider(compat=True)
+        caps = provider.get_capabilities("deepseek-ai/DeepSeek-V4-Flash")
+        assert caps.token_param == "max_tokens"
+        assert caps.thinking_mode == "none"
+        assert caps.supports_web_search is False
+        assert caps.supports_tool_search is False
+        assert caps.supports_vision is False
+        assert caps.supports_reasoning_replay is True
+        assert caps.supports_temperature is True
+
+    def test_claude_id_does_not_pick_up_static_table(self) -> None:
+        """A Claude-named local checkpoint must not inherit Claude API caps."""
+        provider = AnthropicProvider(compat=True)
+        caps = provider.get_capabilities("claude-opus-4-6")
+        assert caps.context_window == 200000
+        assert caps.thinking_mode == "none"
+        assert caps.supports_web_search is False
+        # The real lane still resolves the static entry.
+        real_caps = AnthropicProvider().get_capabilities("claude-opus-4-6")
+        assert real_caps.context_window == 1000000
+        assert real_caps.thinking_mode == "adaptive"
+
+
+# ===========================================================================
+# TestCompatWireShape
+# ===========================================================================
+
+
+class TestCompatWireShape:
+    """Body-inspecting tests on the kwargs handed to ``messages.stream``."""
+
+    def setup_method(self) -> None:
+        self.provider = AnthropicProvider(compat=True)
+
+    @patch("turnstone.core.providers._anthropic._ensure_anthropic")
+    def test_compat_no_web_search_swap_no_temp_force(self, mock_ensure: MagicMock) -> None:
+        """No native web_search swap, no temperature=1 forcing, max_tokens param."""
+        client = _capture_client()
+        list(
+            self.provider.create_streaming(
+                client=client,
+                model="deepseek-ai/DeepSeek-V4-Flash",
+                messages=[{"role": "user", "content": "hi"}],
+                tools=[_WEB_SEARCH_FUNCTION_TOOL],
+                temperature=0.6,
+            )
+        )
+        kwargs = client.messages.stream.call_args[1]
+        sent_tools = kwargs["tools"]
+        assert sent_tools == [
+            {
+                "name": "web_search",
+                "description": "Search the web",
+                "input_schema": {
+                    "type": "object",
+                    "properties": {"query": {"type": "string"}},
+                    "required": ["query"],
+                },
+            }
+        ]
+        assert all(t.get("type") != "web_search_20250305" for t in sent_tools)
+        assert kwargs["temperature"] == 0.6
+        assert "thinking" not in kwargs
+        assert "max_tokens" in kwargs
+        assert "max_completion_tokens" not in kwargs
+
+    @patch("turnstone.core.providers._anthropic._ensure_anthropic")
+    def test_extra_params_passthrough_to_extra_body(self, mock_ensure: MagicMock) -> None:
+        """server_compat extra_body (chat_template_kwargs) reaches the SDK."""
+        client = _capture_client()
+        list(
+            self.provider.create_streaming(
+                client=client,
+                model="deepseek-ai/DeepSeek-V4-Flash",
+                messages=[{"role": "user", "content": "hi"}],
+                extra_params={"chat_template_kwargs": {"thinking": False}},
+            )
+        )
+        kwargs = client.messages.stream.call_args[1]
+        assert kwargs["extra_body"] == {"chat_template_kwargs": {"thinking": False}}
+
+    @patch("turnstone.core.providers._anthropic._ensure_anthropic")
+    def test_internal_keys_not_leaked(self, mock_ensure: MagicMock) -> None:
+        """Real-lane request bodies stay byte-identical with thinking overrides.
+
+        ``thinking_budget_tokens`` is consumed by ``_reasoning_params`` and
+        must never surface as wire ``extra_body`` — a leaked key would change
+        every real-Anthropic request that threads a thinking override.
+        Negative-tested: fails when the ``_INTERNAL_EXTRA_PARAMS`` exclusion
+        is removed from ``_build_thinking_and_kwargs``.
+        """
+        provider = AnthropicProvider()
+        client = _capture_client()
+        list(
+            provider.create_streaming(
+                client=client,
+                model="claude-sonnet-4-5",
+                messages=[{"role": "user", "content": "hi"}],
+                extra_params={"thinking_budget_tokens": 2048},
+            )
+        )
+        kwargs = client.messages.stream.call_args[1]
+        assert "extra_body" not in kwargs
+        assert kwargs["thinking"] == {"type": "enabled", "budget_tokens": 2048}
+
+
+# ===========================================================================
+# TestCompatFactory
+# ===========================================================================
+
+
+class TestCompatFactory:
+    """create_provider / create_client routing for the compat lane."""
+
+    def test_create_provider_anthropic_compatible(self) -> None:
+        from turnstone.core.providers import create_provider
+
+        provider = create_provider("anthropic-compatible")
+        assert provider.provider_name == "anthropic-compatible"
+        assert provider is not create_provider("anthropic")
+        assert create_provider("anthropic-compatible") is provider
+
+    @patch("turnstone.core.providers._anthropic._ensure_anthropic")
+    def test_create_client_anthropic_compatible(self, mock_ensure: MagicMock) -> None:
+        """base_url forwards verbatim; empty api_key is omitted entirely."""
+        from turnstone.core.providers import create_client
+
+        mock_anthropic_cls = MagicMock()
+        mock_mod = MagicMock()
+        mock_mod.Anthropic = mock_anthropic_cls
+        mock_ensure.return_value = mock_mod
+
+        create_client("anthropic-compatible", base_url="http://vllm-host:8000", api_key="")
+        mock_anthropic_cls.assert_called_once_with(base_url="http://vllm-host:8000")
+
+    @patch("turnstone.core.providers._anthropic._ensure_anthropic")
+    def test_create_client_strips_v1_suffix(self, mock_ensure: MagicMock) -> None:
+        """A /v1-suffixed base_url (openai-compatible muscle memory) is
+        normalized for the compat lane — the SDK appends /v1/... itself,
+        so the verbatim URL would request /v1/v1/messages and 404."""
+        from turnstone.core.providers import create_client
+
+        mock_anthropic_cls = MagicMock()
+        mock_mod = MagicMock()
+        mock_mod.Anthropic = mock_anthropic_cls
+        mock_ensure.return_value = mock_mod
+
+        for suffixed in ("http://vllm-host:8000/v1", "http://vllm-host:8000/v1/"):
+            mock_anthropic_cls.reset_mock()
+            create_client("anthropic-compatible", base_url=suffixed, api_key="")
+            mock_anthropic_cls.assert_called_once_with(base_url="http://vllm-host:8000")
+
+        # A base_url that strips to nothing stays verbatim so the typo
+        # fails loudly in httpx instead of silently targeting the SDK's
+        # prod default.
+        mock_anthropic_cls.reset_mock()
+        create_client("anthropic-compatible", base_url="/v1", api_key="")
+        mock_anthropic_cls.assert_called_once_with(base_url="/v1")
+
+    @patch("turnstone.core.providers._anthropic._ensure_anthropic")
+    def test_create_client_requires_base_url(self, mock_ensure: MagicMock) -> None:
+        """Empty base_url fails at construction — the local-only lane must
+        never fall back to the SDK's https://api.anthropic.com default."""
+        from turnstone.core.providers import create_client
+
+        with pytest.raises(ValueError, match="anthropic-compatible requires base_url"):
+            create_client("anthropic-compatible", base_url="", api_key="dummy")
+        mock_ensure.return_value.Anthropic.assert_not_called()
+
+    @patch("turnstone.core.providers._anthropic._ensure_anthropic")
+    def test_create_client_real_lane_base_url_untouched(self, mock_ensure: MagicMock) -> None:
+        """The real anthropic lane forwards base_url verbatim — the /v1
+        normalization is compat-lane-only."""
+        from turnstone.core.providers import create_client
+
+        mock_anthropic_cls = MagicMock()
+        mock_mod = MagicMock()
+        mock_mod.Anthropic = mock_anthropic_cls
+        mock_ensure.return_value = mock_mod
+
+        create_client("anthropic", base_url="http://proxy:9000/v1", api_key="k")
+        mock_anthropic_cls.assert_called_once_with(api_key="k", base_url="http://proxy:9000/v1")
+
+
+# ===========================================================================
+# TestCliScope
+# ===========================================================================
+
+
+class TestCliScope:
+    def test_cli_rejects_compat_provider_id(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """The lane is registry-only — the CLI --provider flag does not grow."""
+        from turnstone import cli
+
+        monkeypatch.setattr(sys, "argv", ["turnstone", "--provider", "anthropic-compatible"])
+        with pytest.raises(SystemExit) as excinfo:
+            cli.main()
+        assert excinfo.value.code == 2
+
+
+# ===========================================================================
+# TestCompatSessionPlumbing
+# ===========================================================================
+
+
+class TestCompatSessionPlumbing:
+    """ChatSession capability merge + extra_params gate for the lane."""
+
+    def test_per_model_capability_override_merge(self, tmp_db: Any) -> None:
+        """Per-model capabilities win over _ANTHROPIC_COMPAT_DEFAULT fields."""
+        from turnstone.core.model_registry import ModelConfig, ModelRegistry
+        from turnstone.core.providers import create_provider
+
+        cfg = ModelConfig(
+            alias="vllm-messages",
+            base_url="http://localhost:8000",
+            api_key="dummy",
+            model="deepseek-ai/DeepSeek-V4-Flash",
+            provider="anthropic-compatible",
+            capabilities={"supports_mid_conversation_system": True, "context_window": 131072},
+        )
+        registry = ModelRegistry(models={"vllm-messages": cfg}, default="vllm-messages")
+        session = _make_session(registry=registry, model_alias="vllm-messages")
+        provider = create_provider("anthropic-compatible")
+        caps = session._resolve_capabilities(
+            provider, "deepseek-ai/DeepSeek-V4-Flash", "vllm-messages"
+        )
+        assert caps.supports_mid_conversation_system is True
+        assert caps.context_window == 131072
+        # Untouched fields keep the compat-lane defaults.
+        assert caps.token_param == "max_tokens"
+        assert caps.thinking_mode == "none"
+        assert caps.supports_web_search is False
+        assert caps.supports_vision is False
+
+    def test_session_extra_params_gate(self, tmp_db: Any) -> None:
+        """server_compat extra_body forwards for the compat lane, not real Anthropic."""
+        from turnstone.core.model_registry import ModelConfig, ModelRegistry
+        from turnstone.core.providers import create_provider
+
+        session = _make_session(reasoning_effort="medium")
+        cfg = ModelConfig(
+            alias="vllm-messages",
+            base_url="http://localhost:8000",
+            api_key="dummy",
+            model="deepseek-ai/DeepSeek-V4-Flash",
+            provider="anthropic-compatible",
+            server_compat={"extra_body": {"chat_template_kwargs": {"thinking": False}}},
+        )
+        session._registry = ModelRegistry(models={"vllm-messages": cfg}, default="vllm-messages")
+        session._model_alias = "vllm-messages"
+
+        session._provider = create_provider("anthropic-compatible")
+        assert session._provider_extra_params() == {"chat_template_kwargs": {"thinking": False}}
+
+        session._provider = create_provider("anthropic")
+        assert session._provider_extra_params() is None
+
+
+# ===========================================================================
+# TestLiveCompatStream
+# ===========================================================================
+
+
+@pytest.mark.live
+@pytest.mark.skipif(
+    not os.environ.get("TURNSTONE_LIVE_ANTHROPIC_COMPAT_URL"),
+    reason="TURNSTONE_LIVE_ANTHROPIC_COMPAT_URL not set",
+)
+class TestLiveCompatStream:
+    """One real streamed turn against a vLLM /v1/messages endpoint."""
+
+    def test_live_compat_streamed_turn(self) -> None:
+        from turnstone.core.providers import create_client, create_provider
+
+        base_url = os.environ["TURNSTONE_LIVE_ANTHROPIC_COMPAT_URL"]
+        model = os.environ.get(
+            "TURNSTONE_LIVE_ANTHROPIC_COMPAT_MODEL", "deepseek-ai/DeepSeek-V4-Flash"
+        )
+        client = create_client("anthropic-compatible", base_url=base_url, api_key="dummy")
+        provider = create_provider("anthropic-compatible")
+        chunks = list(
+            provider.create_streaming(
+                client=client,
+                model=model,
+                messages=[{"role": "user", "content": "Reply with the single word: pong"}],
+                max_tokens=64,
+                extra_params={"chat_template_kwargs": {"thinking": False}},
+            )
+        )
+        content = "".join(c.content_delta or "" for c in chunks)
+        assert content.strip()
+        assert any(c.finish_reason for c in chunks)
+        assert any(c.usage is not None for c in chunks)
+        assert not any(c.reasoning_delta for c in chunks)

tests/test_session.py

@@ -2913,6 +2913,181 @@ class TestMemoryCompositionDeferral:
         assert session._system_composed_with_context is False
 
 
+class TestMemoryAccessTouch:
+    """Access metadata (``access_count`` / ``last_accessed``) moves only when
+    the model actually sees a memory: the injected top-k during composition,
+    and explicit search/get reads via the memory tool.  Save/list and the
+    wider candidate pool must NOT bump the counter.
+    """
+
+    @staticmethod
+    def _access_count(name: str, scope: str = "global", scope_id: str = "") -> int:
+        from turnstone.core.storage import get_storage
+
+        mem = get_storage().get_structured_memory_by_name(name, scope, scope_id)
+        assert mem is not None, f"memory {name!r} not found"
+        return int(mem["access_count"])
+
+    @staticmethod
+    def _save(name: str, content: str) -> None:
+        from turnstone.core.memory import save_structured_memory
+
+        save_structured_memory(name, content, scope="global")
+
+    @staticmethod
+    def _empty_session() -> ChatSession:
+        """A session whose __init__ composed before any memory existed.
+
+        The constructor composes the system prefix once; building it before
+        the memories are saved keeps that first (empty) compose from touching
+        rows, so the tests observe only the turn-driven recompose below.
+        """
+        return _make_session(ws_id="ws-1", user_id="user-1")
+
+    @staticmethod
+    def _compose_turn(session: ChatSession, query: str) -> None:
+        """Drive one user turn's worth of composition.
+
+        Mirrors ``send``: a fresh user turn invalidates the per-turn memory
+        caches, then the prefix recomposes against the new query.
+        """
+        session._invalidate_memory_cache()
+        session.messages.append(turn_from_dict({"role": "user", "content": query}))
+        session._init_system_messages()
+
+    def test_composition_touches_injected_memories(self, tmp_db):
+        session = self._empty_session()
+        self._save("kafka_runbook", "restart the kafka broker pods")
+        self._save("kafka_alerts", "kafka consumer lag alert thresholds")
+        self._compose_turn(session, "how do I restart kafka")
+        # Both query-matching memories were injected, so both got touched once.
+        assert self._access_count("kafka_runbook") == 1
+        assert self._access_count("kafka_alerts") == 1
+
+    def test_composition_skips_unmatched_candidates(self, tmp_db):
+        """The candidate pool is a superset of the injected set — a memory
+        that loses BM25 ranking (no query overlap) must NOT be touched."""
+        session = self._empty_session()
+        self._save("kafka_runbook", "restart the kafka broker pods")
+        self._save("garden_notes", "tomato watering schedule midsummer")
+        self._compose_turn(session, "restart kafka broker pods status")
+        # The matching memory was injected and touched.
+        assert self._access_count("kafka_runbook") == 1
+        # The non-matching one was a candidate but never injected.
+        assert self._access_count("garden_notes") == 0
+        # Sanity: it really was in the visible candidate pool.
+        visible = {m["name"] for m in session._list_visible_memories()}
+        assert "garden_notes" in visible
+
+    def test_composition_touches_each_memory_once_per_turn(self, tmp_db):
+        """``_init_system_messages`` runs many times within a turn (tool
+        results, MCP refresh); the injected set must be touched at most once
+        per memory between user turns, not once per recompose."""
+        session = self._empty_session()
+        self._save("kafka_runbook", "restart the kafka broker pods")
+        self._compose_turn(session, "how do I restart kafka")
+        # Several mid-turn recomposes (no new user turn between them).
+        session._init_system_messages()
+        session._init_system_messages()
+        assert self._access_count("kafka_runbook") == 1
+        # A genuinely new turn lets the same memory be counted again.
+        self._compose_turn(session, "kafka again please")
+        assert self._access_count("kafka_runbook") == 2
+
+    def test_composition_touches_exactly_the_injected_keys(self, tmp_db):
+        """Spy the touch boundary and assert the keys match the names the
+        composer rendered into the ``<memories>`` block — exactly, not the
+        candidate pool."""
+        session = self._empty_session()
+        self._save("kafka_runbook", "restart the kafka broker pods")
+        self._save("garden_notes", "tomato watering schedule midsummer")
+        session._invalidate_memory_cache()
+        session.messages.append(
+            turn_from_dict({"role": "user", "content": "restart kafka broker pods status"})
+        )
+        touched: list[tuple[str, str, str]] = []
+        with patch(
+            "turnstone.core.session.touch_structured_memories",
+            side_effect=lambda keys: touched.extend(keys),
+        ):
+            session._init_system_messages()
+        joined = "\n".join(m["content"] for m in session.system_messages if m["role"] == "system")
+        touched_names = {name for name, _, _ in touched}
+        assert touched_names == {"kafka_runbook"}
+        assert '<memory name="kafka_runbook"' in joined
+        assert '<memory name="garden_notes"' not in joined
+
+    def test_composition_survives_touch_storage_error(self, tmp_db):
+        """A storage blow-up inside the touch must not break composition —
+        the facade swallows it and the memory block still lands."""
+        from turnstone.core.storage import get_storage
+
+        session = self._empty_session()
+        self._save("kafka_runbook", "restart the kafka broker pods")
+        session._invalidate_memory_cache()
+        session.messages.append(
+            turn_from_dict({"role": "user", "content": "how do I restart kafka"})
+        )
+        with patch.object(
+            get_storage(),
+            "touch_structured_memories",
+            side_effect=RuntimeError("storage exploded"),
+        ):
+            session._init_system_messages()
+        joined = "\n".join(m["content"] for m in session.system_messages if m["role"] == "system")
+        assert '<memory name="kafka_runbook"' in joined
+
+    def test_search_action_touches_returned_hits(self, tmp_db):
+        session = self._empty_session()
+        self._save("kafka_runbook", "restart the kafka broker pods")
+        item = session._prepare_memory("call_1", {"action": "search", "query": "kafka"})
+        assert "error" not in item
+        session._exec_memory(item)
+        assert self._access_count("kafka_runbook") == 1
+
+    def test_get_action_touches_fetched_memory(self, tmp_db):
+        session = self._empty_session()
+        self._save("kafka_runbook", "restart the kafka broker pods")
+        item = session._prepare_memory(
+            "call_1", {"action": "get", "name": "kafka_runbook", "scope": "global"}
+        )
+        assert "error" not in item
+        session._exec_memory(item)
+        assert self._access_count("kafka_runbook") == 1
+
+    def test_get_miss_touches_nothing(self, tmp_db):
+        session = self._empty_session()
+        self._save("kafka_runbook", "restart the kafka broker pods")
+        item = session._prepare_memory(
+            "call_1", {"action": "get", "name": "no_such_mem", "scope": "global"}
+        )
+        _, msg = session._exec_memory(item)
+        assert "not found" in msg
+        # The existing row must not be collaterally touched by a miss.
+        assert self._access_count("kafka_runbook") == 0
+
+    def test_list_action_does_not_touch(self, tmp_db):
+        session = self._empty_session()
+        self._save("kafka_runbook", "restart the kafka broker pods")
+        item = session._prepare_memory("call_1", {"action": "list"})
+        session._exec_memory(item)
+        assert self._access_count("kafka_runbook") == 0
+
+    def test_save_action_does_not_touch_access_count(self, tmp_db):
+        """The save action handler itself must not bump ``access_count`` —
+        that counter is read traffic only.  (The recompose a save triggers
+        may surface the row via the composition path; that is exercised by
+        the composition tests.  Suppressed here to isolate the handler.)"""
+        session = self._empty_session()
+        item = session._prepare_memory(
+            "call_1",
+            {"action": "save", "name": "kafka_runbook", "content": "x", "scope": "global"},
+        )
+        with patch.object(session, "_init_system_messages"):
+            session._exec_memory(item)
+        assert self._access_count("kafka_runbook") == 0
+
+
 class TestMetacognitiveBuffers:
     """Nudges drain through advisory channels, not the system message."""
 

tests/test_shell_js.py

@@ -330,6 +330,63 @@ def test_step3_rail_manage_builds_from_admin_seam() -> None:
     assert "aria-expanded" in body, "collapsible group heads must expose aria-expanded"
 
 
+def test_rail_manage_row_badge_hook() -> None:
+    """rail.js owns a GENERIC Manage-row count badge — `setRowBadge(tabKey, count,
+    label?)` stamps a glyph+count chip on a tab row (DS warn `.rail-badge`), and so
+    a COLLAPSED group never hides the signal, mirrors the group's running total onto
+    its head.  rail.js stays agnostic about what the count means (no consent
+    specifics here); a subsystem drives it.  `mountManage` registers the row/head
+    refs and re-applies live counts across a (re)mount.  This is the re-homing
+    target for the MCP consent badge after its settings-gear host was deleted."""
+    body = _RAIL_JS.read_text(encoding="utf-8")
+    assert "export function setRowBadge(tabKey, count, label)" in body, (
+        "rail must export the generic setRowBadge hook (mechanism, not meaning)"
+    )
+    # The chip pairs colour with a glyph (never colour alone — chip-contrast rule).
+    assert "rail-badge" in body and '"⚠"' in body, (
+        "the badge must carry a ⚠ glyph alongside the count (not colour alone)"
+    )
+    # The collapsed-group head must carry the group total so a hidden row's signal
+    # still surfaces — pin the head propagation + the per-group sum.
+    assert "function _groupCount(" in body, "the head badge must sum the group's tab counts"
+    assert "_groupEls" in body and "_rowEls" in body, (
+        "mountManage must register row + owning-group-head refs for the badge hook"
+    )
+    assert "_reapplyBadges()" in body, (
+        "a (re)mount must re-apply any live badge state (the refs are rebuilt)"
+    )
+    # rail.js stays agnostic — the hook takes a generic tabKey/count, with no
+    # consent-specific endpoint, fetch, or branch (an explanatory comment naming a
+    # sample caller is fine; logic is not).  `setRowBadge` itself never fetches.
+    badge_fn = body[body.index("export function setRowBadge") :]
+    badge_fn = badge_fn[: badge_fn.index("\n}\n") + 3]
+    assert "fetch" not in badge_fn and "/v1/" not in badge_fn, (
+        "the generic badge hook must not reach into a subsystem (no fetch / endpoint)"
+    )
+    # No colour-only treatment: the chip uses DS warn tokens AND a glyph.
+    css = _SHELL_CSS.read_text(encoding="utf-8")
+    assert ".rail-badge" in css, "shell.css must carry the .rail-badge chip rule"
+    badge_block = css[css.index(".rail-badge") :]
+    badge_block = badge_block[: badge_block.index("\n.grp") if "\n.grp" in badge_block else 800]
+    assert "var(--warn" in badge_block, (
+        "the badge chip must use the DS --warn family (theme-flips by construction)"
+    )
+    assert "#" not in badge_block, "the badge chip must be token-only (no hex) so themes flip"
+
+
+def test_shell_bridges_setrowbadge_for_classic_subsystems() -> None:
+    """shell.js is the ESM module bridge: a classic-script subsystem (the
+    standalone consent badge in ui/static/app.js) can't import rail.js, so the
+    shell re-exports `setRowBadge` on the `window.TS_SHELL` seam.  Pin the import
+    and the seam so the bridge can't be silently dropped (which would re-break the
+    badge the same way the gear deletion did)."""
+    body = _SHELL_JS.read_text(encoding="utf-8")
+    assert 'setRowBadge } from "./rail.js"' in body, "shell must import setRowBadge from rail.js"
+    assert "notifySessionClosed, setRowBadge }" in body, (
+        "TS_SHELL must expose setRowBadge for classic subsystems (the consent-badge bridge)"
+    )
+
+
 def test_step3_admin_seam_and_thin_show_admin() -> None:
     """admin.js exposes the TS_ADMIN seam (IA + shared perm gate + active-tab +
     openTab) and showAdmin is now a thin delegator that opens the singleton
@@ -615,7 +672,7 @@ def test_step7_live_tab_state_glyphs() -> None:
     )
     assert "this.stateful" in pane, "ShellPane must carry the stateful flag"
     shell = _SHELL_JS.read_text(encoding="utf-8")
-    assert 'import { mountRail, mountManage, glyph } from "./rail.js"' in shell, (
+    assert 'import { mountRail, mountManage, glyph, setRowBadge } from "./rail.js"' in shell, (
         "the shell must import the rail's glyph builder (one source for tab + rail)"
     )
     assert "function stateForWs(" in shell and "function paintConvTabs(" in shell
@@ -749,7 +806,7 @@ def test_shell_marks_pane_dead_on_ws_closed() -> None:
     assert "const notifySessionClosed = (wsId)" in shell
     assert 'pm.getPane("interactive", wsId)' in shell
     assert "p._ctl.markDead()" in shell
-    assert "window.TS_SHELL = { panes: pm, caps, notifySessionClosed }" in shell, (
+    assert "window.TS_SHELL = { panes: pm, caps, notifySessionClosed, setRowBadge }" in shell, (
         "the seam must be exported on TS_SHELL for the console's Tier-1 handler"
     )
     app = _CONSOLE_APP.read_text(encoding="utf-8")

turnstone/__init__.py

@@ -1,3 +1,3 @@
 """turnstone - Multi-node AI orchestration platform with tool use, agent routing, and cluster simulation."""
 
-__version__ = "1.6.0"
+__version__ = "1.6.2"

turnstone/admin.py

@@ -439,6 +439,44 @@ def _discover_console_url() -> str:
 # ---------------------------------------------------------------------------
 
 
+def _cmd_orphan_conversations(args: argparse.Namespace) -> None:
+    """Scan for (and with --delete purge) conversation rows whose workstream is gone."""
+    storage = _get_storage(args)
+    orphans = storage.list_orphan_conversations()
+    if not orphans:
+        print("No orphan conversation rows.")
+        return
+    width = max(len(o["ws_id"]) for o in orphans)
+    print(f"{'ws_id':<{width}}  {'rows':>5}  {'refs':>4}  first       last")
+    for o in orphans:
+        print(
+            f"{o['ws_id']:<{width}}  {o['rows']:>5}  {o['attachment_refs']:>4}  "
+            f"{(o['first'] or '')[:10]}  {(o['last'] or '')[:10]}"
+        )
+    total_rows = sum(o["rows"] for o in orphans)
+    total_refs = sum(o["attachment_refs"] for o in orphans)
+    print(
+        f"\n{len(orphans)} orphan workstream(s), {total_rows} conversation row(s), "
+        f"{total_refs} attachment ref(s)."
+    )
+    if not args.delete:
+        print("Re-run with --delete to purge them.")
+        return
+    if not args.yes:
+        reply = input(f"Delete {total_rows} row(s) across {len(orphans)} workstream(s)? [y/N] ")
+        if reply.strip().lower() not in ("y", "yes"):
+            print("Aborted.", file=sys.stderr)
+            sys.exit(1)
+    result = storage.delete_orphan_conversations([o["ws_id"] for o in orphans])
+    summary = (
+        f"Purged {result['rows']} row(s) across {result['workstreams']} workstream(s); "
+        f"released {result['released_refs']} attachment ref(s)"
+    )
+    if result["skipped"]:
+        summary += f"; skipped {result['skipped']} re-registered workstream(s)"
+    print(summary + ".")
+
+
 def _cmd_rerank_calibrate(args: argparse.Namespace) -> None:
     from turnstone.core.config import get_rerank_instruction
     from turnstone.core.config_store import ConfigStore
@@ -619,6 +657,15 @@ def main() -> None:
         help="Write the calibration onto the model's capabilities",
     )
 
+    p_orph = sub.add_parser(
+        "orphan-conversations",
+        help="Scan (and with --delete purge) conversation rows whose workstream row is gone",
+    )
+    p_orph.add_argument(
+        "--delete", action="store_true", help="Purge the orphans after the scan report"
+    )
+    p_orph.add_argument("--yes", action="store_true", help="Skip the interactive confirmation")
+
     args = parser.parse_args()
     if not args.command:
         parser.print_help()
@@ -638,6 +685,7 @@ def main() -> None:
         "set-node-metadata": _cmd_set_node_metadata,
         "delete-node-metadata": _cmd_delete_node_metadata,
         "export": _cmd_export,
+        "orphan-conversations": _cmd_orphan_conversations,
         "rerank-calibrate": _cmd_rerank_calibrate,
     }
     dispatch[args.command](args)

turnstone/console/server.py

@@ -10210,7 +10210,9 @@ async def admin_import_mcp_config(request: Request) -> JSONResponse:
 # ---------------------------------------------------------------------------
 
 _MODEL_ALIAS_RE = re.compile(r"^[a-zA-Z0-9._-]+$")
-_MODEL_PROVIDERS = frozenset({"openai", "anthropic", "openai-compatible", "google", "xai"})
+_MODEL_PROVIDERS = frozenset(
+    {"openai", "anthropic", "openai-compatible", "anthropic-compatible", "google", "xai"}
+)
 _REASONING_EFFORT_CHOICES = frozenset(
     {"", "none", "minimal", "low", "medium", "high", "xhigh", "max"}
 )

turnstone/console/static/admin.js

@@ -5596,7 +5596,8 @@ function _renderModels(items) {
         ? "model-provider-anthropic"
         : m.provider === "google"
           ? "model-provider-google"
-          : m.provider === "openai-compatible"
+          : m.provider === "openai-compatible" ||
+              m.provider === "anthropic-compatible"
             ? "model-provider-compat"
             : "model-provider-openai";
 
@@ -5896,12 +5897,16 @@ function showEditModelModal(definitionId) {
         ? capsObj.server_compat
         : {};
       // Only extract thinking_mode into the dropdown when the UI can
-      // represent it ("manual" or "").  Values like "adaptive" (Anthropic-
-      // only) stay in the raw capabilities JSON so they aren't silently
-      // lost on save.
+      // represent it ("manual" or "") AND the provider round-trips the
+      // dropdown on save (every provider except anthropic-compatible —
+      // see submitCreateModel).  Unrepresentable values like "adaptive"
+      // and anthropic-compatible rows keep thinking_mode in the raw
+      // capabilities JSON so it isn't silently lost on save.
       const tmVal = capsObj.thinking_mode || "";
       const tmRepresentable = tmVal === "" || tmVal === "manual";
-      if (tmRepresentable) {
+      const tmCaptured =
+        tmRepresentable && (m.provider || "openai") !== "anthropic-compatible";
+      if (tmCaptured) {
         document.getElementById("model-thinking-mode").value = tmVal;
         document.getElementById("model-thinking-param").value =
           capsObj.thinking_param || "";
@@ -5945,7 +5950,7 @@ function showEditModelModal(definitionId) {
       // Remove structured fields from capabilities display — only delete
       // thinking_mode/thinking_param when the UI successfully captured them.
       delete capsObj.server_compat;
-      if (tmRepresentable) {
+      if (tmCaptured) {
         delete capsObj.thinking_mode;
         delete capsObj.thinking_param;
       }
@@ -6017,10 +6022,16 @@ function submitCreateModel() {
     }
   }
 
-  // Thinking mode → capabilities (provider uses this to inject
-  // the correct chat_template_kwargs param automatically).
+  const providerVal = document.getElementById("model-provider").value;
+
+  // Thinking mode → capabilities.  thinking_mode round-trips through the
+  // dropdown for every provider EXCEPT anthropic-compatible, where it
+  // stays in the raw capabilities JSON (mirroring the edit-load lift):
+  // that lane hides the dropdown row and drives reasoning via extra-body
+  // chat_template_kwargs, so a lingering dropdown value must never be
+  // persisted.
   const thinkingMode = document.getElementById("model-thinking-mode").value;
-  if (thinkingMode) {
+  if (providerVal !== "anthropic-compatible" && thinkingMode) {
     caps.thinking_mode = thinkingMode;
     // Preserve thinking_param so Granite/DeepSeek "thinking" key
     // isn't silently reverted to the default "enable_thinking".
@@ -6028,20 +6039,25 @@ function submitCreateModel() {
     if (savedParam) caps.thinking_param = savedParam;
   }
 
-  // Build server_compat from structured fields.  Only meaningful for
-  // openai-compatible aliases — for other providers the section is hidden
+  // Build server_compat from structured fields.  Only meaningful for the
+  // compat lanes (openai-compatible: all fields; anthropic-compatible: the
+  // extra-body JSON only) — for other providers the section is hidden
   // but the form values can linger after a provider switch, so gate the
   // whole block on the active provider to keep persisted state honest.
   const serverCompat = {};
-  const providerVal = document.getElementById("model-provider").value;
   const ebEl = document.getElementById("model-extra-body");
   ebEl.removeAttribute("aria-invalid");
   ebEl.style.borderColor = "";
-  if (providerVal === "openai-compatible") {
-    const serverType = document.getElementById("model-server-type").value;
-    if (serverType) serverCompat.server_type = serverType;
-    const apiSurface = document.getElementById("model-api-surface").value;
-    if (apiSurface) serverCompat.api_surface = apiSurface;
+  if (
+    providerVal === "openai-compatible" ||
+    providerVal === "anthropic-compatible"
+  ) {
+    if (providerVal === "openai-compatible") {
+      const serverType = document.getElementById("model-server-type").value;
+      if (serverType) serverCompat.server_type = serverType;
+      const apiSurface = document.getElementById("model-api-surface").value;
+      if (apiSurface) serverCompat.api_surface = apiSurface;
+    }
     const ebText = ebEl.value.trim();
     if (ebText) {
       try {
@@ -6516,7 +6532,11 @@ function _modelCapsRefreshBaseline() {
   const provider = document.getElementById("model-provider").value;
   const modelName = document.getElementById("model-name").value.trim();
   const banner = document.getElementById("model-autofill");
-  if (!modelName || provider === "openai-compatible") {
+  if (
+    !modelName ||
+    provider === "openai-compatible" ||
+    provider === "anthropic-compatible"
+  ) {
     _modelCapsBaseline = {};
     banner.hidden = true;
     _modelRenderTiles();
@@ -6578,6 +6598,10 @@ const _providerDefaults = {
     urlPlaceholder: "e.g. https://your-provider.com/v1",
     modelPlaceholder: "GLM5",
   },
+  "anthropic-compatible": {
+    urlPlaceholder: "e.g. http://your-vllm-host:8000",
+    modelPlaceholder: "deepseek-ai/DeepSeek-V4-Flash",
+  },
 };
 
 /* Update placeholders when provider changes. */
@@ -6592,8 +6616,18 @@ function _applyProviderDefaults() {
   if (scSection) {
     // hidden attr, not style.display — `.hatch [hidden]` is !important and
     // an inline display can never un-hide it.
-    scSection.hidden = provider !== "openai-compatible";
+    scSection.hidden =
+      provider !== "openai-compatible" && provider !== "anthropic-compatible";
   }
+  // Within the section, server type / API surface / thinking mode are
+  // openai-compatible knobs — the anthropic-compatible lane is configured
+  // through the extra-body JSON alone, so collapse the section to just
+  // that field.
+  const hideOpenaiOnlyRows = provider === "anthropic-compatible";
+  ["model-server-fields-row", "model-thinking-mode-row"].forEach(function (id) {
+    const row = document.getElementById(id);
+    if (row) row.hidden = hideOpenaiOnlyRows;
+  });
 }
 
 /* Populate the model name datalist with known model prefixes for the

turnstone/console/static/index.html

@@ -1577,6 +1577,7 @@
                     <option value="anthropic">anthropic</option>
                     <option value="google">google</option>
                     <option value="openai-compatible">openai-compatible</option>
+                    <option value="anthropic-compatible">anthropic-compatible</option>
                   </select>
                 </div>
                 <div>
@@ -1673,7 +1674,7 @@
 
               <div id="model-server-compat-section" hidden>
                 <div class="sh-section">Server compatibility</div>
-                <div class="field-pair">
+                <div class="field-pair" id="model-server-fields-row">
                   <div>
                     <label for="model-server-type"
                       >Server type
@@ -1702,28 +1703,32 @@
                     </select>
                   </div>
                 </div>
-                <label for="model-thinking-mode"
-                  >Thinking mode
-                  <span class="label-hint">chat-template reasoning</span></label
-                >
-                <select id="model-thinking-mode">
-                  <option value="">None</option>
-                  <option value="manual">Enabled</option>
-                </select>
-                <div id="model-thinking-param-row" hidden>
-                  <label for="model-thinking-param"
-                    >Template param name
+                <div id="model-thinking-mode-row">
+                  <label for="model-thinking-mode"
+                    >Thinking mode
                     <span class="label-hint"
-                      >Granite/DeepSeek use "thinking"</span
+                      >chat-template reasoning</span
                     ></label
                   >
-                  <input
-                    type="text"
-                    id="model-thinking-param"
-                    class="sh-mono"
-                    value="enable_thinking"
-                    placeholder="enable_thinking"
-                  />
+                  <select id="model-thinking-mode">
+                    <option value="">None</option>
+                    <option value="manual">Enabled</option>
+                  </select>
+                  <div id="model-thinking-param-row" hidden>
+                    <label for="model-thinking-param"
+                      >Template param name
+                      <span class="label-hint"
+                        >Granite/DeepSeek use "thinking"</span
+                      ></label
+                    >
+                    <input
+                      type="text"
+                      id="model-thinking-param"
+                      class="sh-mono"
+                      value="enable_thinking"
+                      placeholder="enable_thinking"
+                    />
+                  </div>
                 </div>
                 <label for="model-extra-body"
                   >Extra body params

turnstone/core/mcp_client.py

@@ -57,7 +57,7 @@ from turnstone.core.mcp_oauth import (
 )
 
 if TYPE_CHECKING:
-    from collections.abc import Awaitable, Callable
+    from collections.abc import Awaitable, Callable, Coroutine
 
 log = get_logger("turnstone.mcp")
 
@@ -607,6 +607,13 @@ class MCPClientManager:
         # static-only deployments).
         self._user_pool_eviction_task: asyncio.Task[None] | None = None
 
+        # Strong references to fire-and-forget background tasks (catalog
+        # refreshes etc.).  ``create_task`` alone keeps only a weak ref — an
+        # untracked task can be GC'd mid-flight, and its exception surfaces
+        # as "Task exception was never retrieved" at GC time instead of
+        # being logged where it happened.  See ``_spawn_background``.
+        self._background_tasks: set[asyncio.Task[Any]] = set()
+
     def _ensure_static_state(self, name: str) -> StaticServerState:
         """Get or create the StaticServerState for ``name``.
 
@@ -2712,6 +2719,30 @@ class MCPClientManager:
 
     def shutdown(self) -> None:
         """Close all MCP sessions and stop the background loop."""
+        # Cancel tracked background tasks (catalog refreshes etc.) FIRST —
+        # they are pure auxiliaries, and draining them up front means the
+        # stack teardown below can't race an in-flight refresh.  Submitted
+        # whenever a loop exists — NOT gated on a main-thread truthiness
+        # check of ``_background_tasks``: a spawn queued via
+        # call_soon_threadsafe may not have reached the set yet, but ready
+        # callbacks run in FIFO order, so by the time the drain coroutine
+        # snapshots the set ON the loop, every earlier-queued spawn has
+        # landed.  ``is_running()`` guard: on a stopped loop nothing can
+        # execute the drain — submitting would just stall on the future.
+        if self._loop is not None and self._loop.is_running():
+
+            async def _cancel_background() -> None:
+                tasks = list(self._background_tasks)
+                for task in tasks:
+                    task.cancel()
+                await asyncio.gather(*tasks, return_exceptions=True)
+
+            future = asyncio.run_coroutine_threadsafe(_cancel_background(), self._loop)
+            try:
+                future.result(timeout=10)
+            except Exception:
+                log.debug("Error cancelling MCP background tasks", exc_info=True)
+
         # Cancel the pool eviction task, then close all pool entries
         # before tearing down static-path state. Both run on the
         # mcp-loop so dispatcher coroutines can't race them.
@@ -2767,8 +2798,21 @@ class MCPClientManager:
             self._loop.call_soon_threadsafe(self._loop.stop)
         if self._thread:
             self._thread.join(timeout=5)
+            if self._thread.is_alive():
+                # Closing a still-running loop raises; the daemon thread dies
+                # with the process, so leaving the loop open is the lesser
+                # evil.  Loud because a stuck loop thread is itself a bug.
+                log.warning("MCP loop thread did not stop within 5s; loop left open")
+            else:
+                if self._loop is not None:
+                    self._loop.close()
+                self._loop = None
+                self._thread = None
+        # When no thread was started (tests wire ``_loop`` directly) the loop
+        # is not ours to close — the stop above is all the owner needs.
 
         # Clear all state
+        self._background_tasks.clear()
         self._static_servers.clear()
         self._db_managed.clear()
         self._tools = []
@@ -3327,6 +3371,33 @@ class MCPClientManager:
             # broken the first failure re-trips the circuit immediately.
             self._circuit_open_until.pop(server_name, None)
 
+    def _spawn_background(self, coro: Coroutine[Any, Any, Any], label: str) -> asyncio.Task[Any]:
+        """Schedule *coro* as a tracked background task (loop thread only).
+
+        Holds a strong reference until completion and retrieves the task's
+        outcome in a done-callback: failures are logged once, here, at
+        warning — never deferred to garbage collection, where they surface
+        as "Task exception was never retrieved" on whatever stream happens
+        to be attached at the time.  ``shutdown()`` cancels anything still
+        tracked before stopping the loop.
+        """
+        task = asyncio.create_task(coro)
+        self._background_tasks.add(task)
+
+        def _done(t: asyncio.Task[Any]) -> None:
+            try:
+                if not t.cancelled():
+                    exc = t.exception()
+                    if exc is not None:
+                        log.warning("MCP background %s failed", label, exc_info=exc)
+            finally:
+                # Discard LAST so set-emptiness means "done AND reported" —
+                # a watcher keying on emptiness must never race the warning.
+                self._background_tasks.discard(t)
+
+        task.add_done_callback(_done)
+        return task
+
     def _cb_auto_reconnect(self, server_name: str) -> Any:
         """Attempt reconnection for a disconnected server during half-open probe.
 
@@ -3355,13 +3426,18 @@ class MCPClientManager:
 
         # Schedule catalog refresh on the loop without blocking the caller.
         # The reconnected session is valid for the imminent dispatch; catalog
-        # drift will be reconciled on the loop in the background.
+        # drift will be reconciled on the loop in the background.  The task is
+        # tracked: a refresh FAILURE is logged by the done-callback — this
+        # except only covers the scheduling itself.
         def _schedule_refresh() -> None:
             try:
-                asyncio.create_task(self._refresh_server(server_name))
+                self._spawn_background(
+                    self._refresh_server(server_name),
+                    f"catalog refresh after reconnect for '{server_name}'",
+                )
             except Exception:
                 log.warning(
-                    "Catalog refresh after reconnect failed for '%s'",
+                    "Scheduling catalog refresh after reconnect failed for '%s'",
                     server_name,
                     exc_info=True,
                 )

turnstone/core/model_registry.py

@@ -817,6 +817,14 @@ def probe_model_endpoint(
             if known is not None:
                 result["context_window"] = known["context_window"]
             result["server_type"] = "xai"
+        elif provider == "anthropic-compatible":
+            # No static table for local models; vLLM exposes max_model_len
+            # as an extra field the Anthropic SDK preserves (extra="allow").
+            if inspect_obj is not None:
+                max_len = inspect_obj.model_dump().get("max_model_len")
+                if isinstance(max_len, int) and max_len > 0:
+                    result["context_window"] = max_len
+            result["server_type"] = "anthropic-compatible"
         else:
             # OpenAI-compatible path
             _detect_openai_compat(result, inspect_obj, inspect_id, base_url)

turnstone/core/providers/__init__.py

@@ -46,6 +46,7 @@ _openai_provider = OpenAIResponsesProvider()
 _openai_compat_provider = OpenAIChatCompletionsProvider()
 _xai_provider = XAIProvider()
 _anthropic_provider: LLMProvider | None = None
+_anthropic_compat_provider: LLMProvider | None = None
 _google_provider: LLMProvider | None = None
 
 
@@ -67,8 +68,13 @@ def create_provider(
       endpoints like Mistral cloud, or local servers that expose the
       Responses surface).
 
-    Ignored for non-OpenAI providers.  ``provider_name="openai"`` always
-    uses the Responses API regardless of *api_surface*.
+    Ignored for non-OpenAI providers — both Anthropic lanes
+    (``"anthropic"`` and ``"anthropic-compatible"``) talk to the
+    Messages API regardless of *api_surface*.
+    ``provider_name="anthropic-compatible"`` returns the Anthropic
+    adapter in compat mode (local servers exposing ``/v1/messages``,
+    e.g. vLLM).  ``provider_name="openai"`` always uses the Responses
+    API regardless of *api_surface*.
 
     Note: the ``OpenAIResponsesProvider`` singleton is reused for both
     cloud OpenAI and ``openai-compatible`` + responses, so its
@@ -77,7 +83,7 @@ def create_provider(
     must read ``ModelConfig.provider`` and ``server_compat["api_surface"]``
     rather than ``provider.provider_name``.
     """
-    global _anthropic_provider, _google_provider  # noqa: PLW0603
+    global _anthropic_provider, _anthropic_compat_provider, _google_provider  # noqa: PLW0603
     if provider_name == "openai":
         return _openai_provider
     if provider_name == "openai-compatible":
@@ -98,6 +104,13 @@ def create_provider(
 
                 _anthropic_provider = AnthropicProvider()
             return _anthropic_provider
+    if provider_name == "anthropic-compatible":
+        with _provider_lock:
+            if _anthropic_compat_provider is None:
+                from turnstone.core.providers._anthropic import AnthropicProvider
+
+                _anthropic_compat_provider = AnthropicProvider(compat=True)
+            return _anthropic_compat_provider
     if provider_name == "google":
         with _provider_lock:
             if _google_provider is None:
@@ -107,7 +120,7 @@ def create_provider(
             return _google_provider
     raise ValueError(
         f"Unknown provider: {provider_name!r}. "
-        "Supported: openai, anthropic, google, openai-compatible, xai"
+        "Supported: openai, anthropic, google, openai-compatible, anthropic-compatible, xai"
     )
 
 
@@ -133,19 +146,36 @@ def create_client(provider_name: str, *, base_url: str, api_key: str) -> Any:
         if base_url:
             return OpenAI(base_url=base_url, api_key=resolved_key)
         return OpenAI(api_key=resolved_key)
-    if provider_name == "anthropic":
+    if provider_name in ("anthropic", "anthropic-compatible"):
         from turnstone.core.providers._anthropic import _ensure_anthropic
 
         anthropic = _ensure_anthropic()
         kwargs: dict[str, str] = {}
         if resolved_key is not None:
             kwargs["api_key"] = resolved_key
+        if provider_name == "anthropic-compatible":
+            # The lane targets local /v1/messages servers; without a
+            # base_url the SDK would default to https://api.anthropic.com
+            # and send compat-shaped requests to the commercial API.
+            if not base_url:
+                raise ValueError(
+                    "anthropic-compatible requires base_url (the server root, "
+                    "e.g. http://your-vllm-host:8000)"
+                )
+            # The Anthropic SDK appends /v1/... to base_url, so a
+            # /v1-suffixed URL (the openai-compatible convention) would
+            # request /v1/v1/messages and 404.  Tolerate the suffix.
+            # Keep the verbatim value when stripping would empty it
+            # (base_url of exactly "/v1") so the typo still fails loudly
+            # instead of silently retargeting the SDK's prod default.
+            stripped = base_url.rstrip("/").removesuffix("/v1")
+            base_url = stripped or base_url
         if base_url and base_url != "https://api.anthropic.com":
             kwargs["base_url"] = base_url
         return anthropic.Anthropic(**kwargs)
     raise ValueError(
         f"Unknown provider: {provider_name!r}. "
-        "Supported: openai, anthropic, google, openai-compatible, xai"
+        "Supported: openai, anthropic, google, openai-compatible, anthropic-compatible, xai"
     )
 
 
@@ -153,11 +183,12 @@ def lookup_model_capabilities(provider: str, model: str) -> dict[str, Any] | Non
     """Return static capabilities for a known model, or ``None`` if unknown.
 
     The returned dict has JSON-friendly values (tuples converted to lists).
-    Returns ``None`` for ``openai-compatible`` (no static table for local models).
+    Returns ``None`` for ``openai-compatible`` and ``anthropic-compatible``
+    (no static table for local models).
     """
     import dataclasses
 
-    if provider == "openai-compatible":
+    if provider in ("openai-compatible", "anthropic-compatible"):
         return None
     prov = create_provider(provider)
     caps = prov.get_capabilities(model)

turnstone/core/providers/_anthropic.py

@@ -71,6 +71,11 @@ _WEB_SEARCH_TOOL_TYPE = "web_search_20250305"
 # Tool search: server-side BM25 tool discovery for deferred tools
 _TOOL_SEARCH_TOOL_TYPE = "tool_search_tool_bm25"
 
+# extra_params keys consumed internally (``_reasoning_params``) — never
+# forwarded to the wire ``extra_body``.  Keeps real-Anthropic request
+# bodies byte-identical when a caller threads thinking overrides.
+_INTERNAL_EXTRA_PARAMS = frozenset({"thinking_budget_tokens"})
+
 # -- model capabilities -------------------------------------------------------
 
 _ANTHROPIC_DEFAULT = ModelCapabilities(
@@ -83,6 +88,27 @@ _ANTHROPIC_DEFAULT = ModelCapabilities(
     supports_reasoning_replay=True,
 )
 
+# Anthropic-compatible local servers (vLLM's /v1/messages endpoint):
+# token_param must be "max_tokens" (the only token param the endpoint
+# accepts); the "thinking" request param is not consumed by vLLM — the
+# reasoning toggle rides chat_template_kwargs via extra_body, so
+# thinking_mode stays "none"; supports_reasoning_replay stays True even
+# so, because the endpoint emits and round-trips thinking blocks whenever
+# the chat template enables reasoning (the request param is simply not
+# the switch); native web_search / tool_search server-tool types 400 on
+# vLLM (tools require input_schema); vision is opt-in per model.
+# supports_temperature stays True via the dataclass default.
+_ANTHROPIC_COMPAT_DEFAULT = ModelCapabilities(
+    context_window=200000,
+    max_output_tokens=64000,
+    token_param="max_tokens",
+    thinking_mode="none",
+    supports_web_search=False,
+    supports_tool_search=False,
+    supports_vision=False,
+    supports_reasoning_replay=True,
+)
+
 _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
     # Fable 5: same wire surface as opus-4-8 (adaptive-only thinking, no
     # sampling params, prefill rejected) with one extra constraint — an
@@ -239,13 +265,24 @@ ANTHROPIC_REASONING_BLOCK_TYPES = frozenset({"thinking", "redacted_thinking"})
 
 
 class AnthropicProvider:
-    """Provider for Anthropic's Messages API with native streaming."""
+    """Provider for Anthropic's Messages API with native streaming.
+
+    ``compat=True`` serves Anthropic-compatible local servers (vLLM's
+    ``/v1/messages``): same wire translation, but capabilities come from
+    ``_ANTHROPIC_COMPAT_DEFAULT`` for every model — the static Claude
+    table never applies to local checkpoints.
+    """
+
+    def __init__(self, *, compat: bool = False) -> None:
+        self._compat = compat
 
     @property
     def provider_name(self) -> str:
-        return "anthropic"
+        return "anthropic-compatible" if self._compat else "anthropic"
 
     def get_capabilities(self, model: str) -> ModelCapabilities:
+        if self._compat:
+            return _ANTHROPIC_COMPAT_DEFAULT
         return _lookup_capabilities(model, _ANTHROPIC_CAPABILITIES, _ANTHROPIC_DEFAULT)
 
     # -- web search tool injection -------------------------------------------
@@ -352,6 +389,13 @@ class AnthropicProvider:
             if effort:
                 kwargs["output_config"] = {"effort": effort}
 
+        # Operator server_compat extra_body overrides (e.g. chat_template_kwargs
+        # for anthropic-compatible local servers) ride the SDK's extra_body.
+        if extra_params:
+            wire_extra = {k: v for k, v in extra_params.items() if k not in _INTERNAL_EXTRA_PARAMS}
+            if wire_extra:
+                kwargs["extra_body"] = wire_extra
+
         return kwargs
 
     # -- message conversion --------------------------------------------------

turnstone/core/session.py

@@ -86,6 +86,7 @@ from turnstone.core.memory import (
     search_visible_structured_memories,
     set_message_attachments,
     set_workstream_alias,
+    touch_structured_memories,
     update_workstream_title,
 )
 from turnstone.core.memory_relevance import (
@@ -1031,6 +1032,10 @@ class ChatSession:
         # tool results) and the recent-context string is identical across
         # them.  Invalidated on user-turn append and on memory write/delete.
         self._mem_search_cache: dict[tuple[str, str, int], list[dict[str, str]]] = {}
+        # Per-turn dedup for composition touches: ``_init_system_messages`` runs
+        # many times within a turn, so the injected set is touched at most once
+        # per memory per turn.  Cleared alongside the search cache.
+        self._touched_memory_keys: set[tuple[str, str, str]] = set()
         self._ws_id = ws_id or uuid.uuid4().hex
         self._title_generated = False
         self._read_files: set[str] = set()
@@ -2873,6 +2878,9 @@ class ChatSession:
                 candidates=len(visible_mems),
                 injected=len(relevant),
             )
+            # Access metadata tracks what the model actually saw — touch the
+            # injected top-k, not the candidate pool.
+            self._touch_injected_memories(relevant)
             if relevant:
                 dev_parts.append("")
                 dev_parts.append(build_memory_context(relevant))
@@ -3127,7 +3135,10 @@ class ChatSession:
 
         Forwards operator-supplied ``server_compat["extra_body"]`` overrides
         (``skip_special_tokens``, ``reasoning_format``, or explicit
-        ``chat_template_kwargs``) to the OpenAI SDK ``extra_body``.  Operators
+        ``chat_template_kwargs``) to the OpenAI SDK ``extra_body`` on the
+        OpenAI-shaped lanes, and to the Anthropic SDK ``extra_body`` on the
+        anthropic-compatible lane (the channel for vLLM's
+        ``chat_template_kwargs`` reasoning toggle).  Operators
         running gpt-oss-style local templates that consume ``reasoning_effort``
         from ``chat_template_kwargs`` should set it explicitly under
         ``server_compat["extra_body"]["chat_template_kwargs"]``.
@@ -3143,9 +3154,11 @@ class ChatSession:
         from turnstone.core.server_compat import merge_server_compat
 
         prov = provider or self._provider
-        # Only OpenAI-shaped providers consume extra_body.  Anthropic/Google
-        # have their own param paths handled inside their providers.
-        if prov.provider_name not in ("openai", "openai-compatible"):
+        # extra_body consumers: the OpenAI-shaped providers, plus the
+        # anthropic-compatible lane (server_compat extra_body rides the
+        # Anthropic SDK's extra_body).  Real Anthropic and Google keep
+        # their own param paths handled inside their providers.
+        if prov.provider_name not in ("openai", "openai-compatible", "anthropic-compatible"):
             return None
         extra = merge_server_compat(None, self._get_server_compat(model_alias))
         return extra or None
@@ -7439,6 +7452,7 @@ class ChatSession:
     def _invalidate_memory_cache(self) -> None:
         """Drop the per-turn search cache; call on user-turn append + memory writes."""
         self._mem_search_cache.clear()
+        self._touched_memory_keys.clear()
 
     def _select_memory_candidates(self, context: str) -> tuple[list[dict[str, str]], str]:
         """Pick the candidate set fed into BM25 ranking.
@@ -7476,6 +7490,38 @@ class ChatSession:
             return extra, "recency"
         return search_hits + extra, ("union" if extra else "search")
 
+    @staticmethod
+    def _memory_keys(rows: list[dict[str, str]]) -> list[tuple[str, str, str]]:
+        """Build ``(name, scope, scope_id)`` touch keys from memory rows.
+
+        The storage read helpers return ``SELECT *`` rows, so all three
+        columns are present.
+        """
+        return [(r.get("name", ""), r.get("scope", ""), r.get("scope_id", "")) for r in rows]
+
+    def _touch_injected_memories(self, rows: list[dict[str, str]]) -> None:
+        """Touch the memories injected into the system prefix this turn.
+
+        ``_init_system_messages`` recomposes many times per turn; gate on the
+        per-turn touched-key set so each surfaced memory is counted at most
+        once between user turns.  Best-effort: the facade swallows storage
+        errors, so a failed touch never breaks composition.
+        """
+        fresh = [k for k in self._memory_keys(rows) if k not in self._touched_memory_keys]
+        if not fresh:
+            return
+        self._touched_memory_keys.update(fresh)
+        touch_structured_memories(fresh)
+
+    def _touch_read_memories(self, rows: list[dict[str, str]]) -> None:
+        """Touch memories returned by an explicit memory-tool read.
+
+        A search/get is a distinct user-driven access each time it runs, so
+        these are counted unconditionally (not subject to the composition
+        per-turn dedup).  Best-effort via the facade.
+        """
+        touch_structured_memories(self._memory_keys(rows))
+
     def _check_metacognitive_nudge(self, user_message: str) -> tuple[str, str] | None:
         """Check if a metacognitive nudge should fire for *user_message*.
 
@@ -11489,6 +11535,7 @@ class ChatSession:
                         found_scope = scope
                         break
                 if mem:
+                    self._touch_read_memories([mem])
                     content = mem.get("content", "")
                     desc = mem.get("description", "")
                     mem_type = mem.get("type", "")
@@ -11566,6 +11613,7 @@ class ChatSession:
                     result_count=len(rows),
                     query=item["query"][:120],
                 )
+                self._touch_read_memories(rows)
                 if rows:
                     lines = []
                     for m in rows:

turnstone/core/storage/_postgresql.py

@@ -114,17 +114,19 @@ from turnstone.core.storage._utils import (
 from turnstone.core.storage._utils import (
     escape_like as _escape_like,
 )
+from turnstone.core.storage._utils import (
+    find_orphan_conversations,
+    prepare_provider_data_for_save,
+    purge_orphan_conversations,
+    release_attachment_refs,
+    sanitize_text,
+)
 from turnstone.core.storage._utils import (
     normalize_search_terms as _normalize_search_terms,
 )
 from turnstone.core.storage._utils import (
     parse_attachment_refs as _parse_attachment_refs,
 )
-from turnstone.core.storage._utils import (
-    prepare_provider_data_for_save,
-    release_attachment_refs,
-    sanitize_text,
-)
 from turnstone.core.storage._utils import (
     reconstruct_messages as _reconstruct_messages,
 )
@@ -909,6 +911,16 @@ class PostgreSQLBackend:
             conn.commit()
             return result.rowcount > 0
 
+    def list_orphan_conversations(self) -> list[dict[str, Any]]:
+        with self._conn() as conn:
+            return find_orphan_conversations(conn)
+
+    def delete_orphan_conversations(self, ws_ids: list[str]) -> dict[str, int]:
+        with self._conn() as conn:
+            result = purge_orphan_conversations(conn, ws_ids)
+            conn.commit()
+            return result
+
     # -- Workstream attachments (content-addressed, refcounted) ----------------
 
     def save_attachment(

turnstone/core/storage/_protocol.py

@@ -633,6 +633,29 @@ class StorageBackend(Protocol):
         """Delete a workstream and all its conversations + config."""
         ...
 
+    def list_orphan_conversations(self) -> list[dict[str, Any]]:
+        """Conversation ws_ids with no ``workstreams`` row.
+
+        One dict per orphan workstream — keys ``ws_id``, ``rows``, ``first``,
+        ``last`` (ISO text timestamps), ``attachment_refs`` — ordered
+        oldest-first.  Read-only; feeds the ``turnstone-admin
+        orphan-conversations`` maintenance verb.
+        """
+        ...
+
+    def delete_orphan_conversations(self, ws_ids: list[str]) -> dict[str, int]:
+        """Purge conversation rows for the *ws_ids* that are STILL orphaned.
+
+        Orphan-ness is enforced inside the DELETE itself (correlated
+        ``NOT EXISTS`` against ``workstreams``) and refcounts are released
+        from its ``RETURNING`` — a ws_id registered before or during the
+        purge keeps both its rows and its refcounts.  Sweeps the purged
+        ws_ids' ``workstream_config`` / ``workstream_overrides`` rows.
+        Returns counts keyed ``workstreams``, ``rows``, ``released_refs``,
+        ``skipped`` (distinct inputs not purged).
+        """
+        ...
+
     def list_workstreams(
         self,
         node_id: str | None = None,

turnstone/core/storage/_sqlite.py

@@ -114,17 +114,19 @@ from turnstone.core.storage._utils import (
 from turnstone.core.storage._utils import (
     escape_like as _escape_like,
 )
+from turnstone.core.storage._utils import (
+    find_orphan_conversations,
+    prepare_provider_data_for_save,
+    purge_orphan_conversations,
+    release_attachment_refs,
+    sanitize_text,
+)
 from turnstone.core.storage._utils import (
     normalize_search_terms as _normalize_search_terms,
 )
 from turnstone.core.storage._utils import (
     parse_attachment_refs as _parse_attachment_refs,
 )
-from turnstone.core.storage._utils import (
-    prepare_provider_data_for_save,
-    release_attachment_refs,
-    sanitize_text,
-)
 from turnstone.core.storage._utils import (
     reconstruct_messages as _reconstruct_messages,
 )
@@ -1055,6 +1057,16 @@ class SQLiteBackend:
             conn.commit()
             return result.rowcount > 0
 
+    def list_orphan_conversations(self) -> list[dict[str, Any]]:
+        with self._conn() as conn:
+            return find_orphan_conversations(conn)
+
+    def delete_orphan_conversations(self, ws_ids: list[str]) -> dict[str, int]:
+        with self._conn() as conn:
+            result = purge_orphan_conversations(conn, ws_ids)
+            conn.commit()
+            return result
+
     # -- Workstream attachments (content-addressed, refcounted) ----------------
 
     def save_attachment(

turnstone/core/storage/_utils.py

@@ -12,7 +12,13 @@ import sqlalchemy as sa
 
 from turnstone.core.attachments import unreadable_placeholder
 from turnstone.core.log import get_logger
-from turnstone.core.storage._schema import workstream_attachments
+from turnstone.core.storage._schema import (
+    conversations,
+    workstream_attachments,
+    workstream_config,
+    workstream_overrides,
+    workstreams,
+)
 from turnstone.core.trajectory import (
     AttachmentRef,
     ContentBlock,
@@ -173,6 +179,119 @@ def release_attachment_refs(conn: Any, attachment_ids: list[str]) -> None:
     )
 
 
+def find_orphan_conversations(conn: Any) -> list[dict[str, Any]]:
+    """Conversation ws_ids that have no ``workstreams`` row, with row stats.
+
+    Orphans come from writers that persisted without a registered workstream:
+    historically the pre-unification CLI/server paths, and the
+    delete-during-inflight race (a late tool-result save re-creating rows
+    after ``delete_workstream``).  Read-only; ordered oldest-first.  Each
+    entry carries the attachment-ref count so a purge's refcount release is
+    visible before it happens.
+    """
+    anti_join = conversations.outerjoin(workstreams, conversations.c.ws_id == workstreams.c.ws_id)
+    rows = conn.execute(
+        sa.select(
+            conversations.c.ws_id,
+            sa.func.count().label("row_count"),
+            sa.func.min(conversations.c.timestamp).label("first"),
+            sa.func.max(conversations.c.timestamp).label("last"),
+        )
+        .select_from(anti_join)
+        .where(workstreams.c.ws_id.is_(None))
+        .group_by(conversations.c.ws_id)
+        .order_by(sa.func.min(conversations.c.timestamp))
+    ).fetchall()
+    # Ref counts in ONE pass over the orphan rows that carry attachments —
+    # not a query per orphan workstream, so the scan stays proportional to
+    # orphan ROW count.
+    ref_counts: dict[str, int] = {}
+    ref_rows = conn.execute(
+        sa.select(conversations.c.ws_id, conversations.c.attachments)
+        .select_from(anti_join)
+        .where(
+            sa.and_(
+                workstreams.c.ws_id.is_(None),
+                conversations.c.attachments.is_not(None),
+            )
+        )
+    ).fetchall()
+    for ws_id, refs in ref_rows:
+        ref_counts[ws_id] = ref_counts.get(ws_id, 0) + len(parse_attachment_refs(refs))
+    return [
+        {
+            "ws_id": ws_id,
+            "rows": int(row_count),
+            "first": first,
+            "last": last,
+            "attachment_refs": ref_counts.get(ws_id, 0),
+        }
+        for ws_id, row_count, first, last in rows
+    ]
+
+
+# IN-list chunk size for the purge statements — mirrors the storage layer's
+# existing bulk chunking (SQLite bind-parameter limits).
+_PURGE_CHUNK = 500
+
+
+def purge_orphan_conversations(conn: Any, ws_ids: list[str]) -> dict[str, int]:
+    """Delete conversation rows for the *ws_ids* that are STILL orphans.
+
+    Orphan-ness is enforced INSIDE the DELETE itself (a correlated
+    ``NOT EXISTS`` against ``workstreams``), and the refcounts to release
+    come from the DELETE's ``RETURNING`` — so refs are released for exactly
+    the rows that were deleted.  A ws_id registered at any point before the
+    DELETE statement keeps both its rows AND its refcounts; there is no
+    pre-count/delete window to underflow.  (Needs ``DELETE .. RETURNING``:
+    PostgreSQL, or SQLite ≥ 3.35.)
+
+    Input is de-duplicated and all IN-lists are chunked.  ``skipped`` =
+    distinct input ws_ids not purged (registered before/during the purge, or
+    no rows).  The purged ws_ids' ``workstream_config`` /
+    ``workstream_overrides`` rows are swept.  Caller owns commit.
+    """
+    distinct = list(dict.fromkeys(ws_ids))
+    if not distinct:
+        return {"workstreams": 0, "rows": 0, "released_refs": 0, "skipped": 0}
+    ref_ids: list[str] = []
+    purged_ws: set[str] = set()
+    rows_deleted = 0
+    for i in range(0, len(distinct), _PURGE_CHUNK):
+        chunk = distinct[i : i + _PURGE_CHUNK]
+        returned = conn.execute(
+            sa.delete(conversations)
+            .where(
+                sa.and_(
+                    conversations.c.ws_id.in_(chunk),
+                    ~sa.exists(
+                        sa.select(workstreams.c.ws_id).where(
+                            workstreams.c.ws_id == conversations.c.ws_id
+                        )
+                    ),
+                )
+            )
+            .returning(conversations.c.ws_id, conversations.c.attachments)
+        ).fetchall()
+        for ws_id, refs in returned:
+            purged_ws.add(ws_id)
+            rows_deleted += 1
+            if refs:
+                ref_ids.extend(parse_attachment_refs(refs))
+    release_attachment_refs(conn, ref_ids)
+    swept = sorted(purged_ws)
+    for i in range(0, len(swept), _PURGE_CHUNK):
+        chunk = swept[i : i + _PURGE_CHUNK]
+        conn.execute(sa.delete(workstream_config).where(workstream_config.c.ws_id.in_(chunk)))
+        conn.execute(sa.delete(workstream_overrides).where(workstream_overrides.c.ws_id.in_(chunk)))
+    return {
+        "workstreams": len(purged_ws),
+        "rows": rows_deleted,
+        "released_refs": len(ref_ids),
+        "skipped": len(distinct) - len(purged_ws),
+    }
+
+
 def attachment_to_content_part(att: dict[str, Any]) -> dict[str, Any] | None:
     """Convert a stored attachment row into an OpenAI-style content part.
 

turnstone/shared_static/interactive.js

@@ -630,6 +630,28 @@ class Pane {
       }
     });
 
+    // Click-to-play for media embeds.  Pane-owned (on this.el) and root-scoped
+    // via closest(".media-play-btn") so every embedded L-shell pane activates
+    // its own players — the old standalone wired this via a document-level
+    // delegated listener in app.js, which the console host never loaded (so the
+    // Play button was dead in console-hosted panes).  Enter on a focused button
+    // routes through the same path, mirroring the approval keydown above.
+    this.el.addEventListener("click", (e) => {
+      const btn = e.target.closest(".media-play-btn");
+      if (!btn) return;
+      e.preventDefault();
+      activateMediaPlayButton(btn);
+    });
+    this.el.addEventListener("keydown", (e) => {
+      if (e.key !== "Enter") return;
+      const btn = e.target.closest(".media-play-btn");
+      if (!btn || btn.disabled) return;
+      // Single-path activation: preventDefault stops the browser's native
+      // Enter-to-click from dispatching a second activation behind ours.
+      e.preventDefault();
+      activateMediaPlayButton(btn);
+    });
+
     // No pane header: the workstream name, persona, and state are shown by the
     // tab and the rail (Workspaces); the --skip-permissions banner lands in
     // messagesEl (see the host warningTarget).  The standalone split-pane
@@ -2693,6 +2715,143 @@ function _tryPrettyJson(text) {
   return _redactApiKeys(JSON.stringify(obj, null, 2));
 }
 
+// ---------------------------------------------------------------------------
+//  HLS lazy-loader + click-to-play (lifted from the standalone app.js so
+//  console-hosted panes activate media too).  Follows the mermaid.js
+//  lazy-load pattern in /shared/renderer.js: the vendor is fetched by absolute
+//  /shared/ URL on first use, so it resolves in BOTH the standalone server and
+//  the console (where /shared is mounted at the root and node-proxied panes
+//  also reach it via /node/{id}/shared/).
+// ---------------------------------------------------------------------------
+let _hlsState = "idle";
+let _hlsQueue = [];
+
+function _loadHls(callback) {
+  if (_hlsState === "ready") {
+    callback();
+    return;
+  }
+  _hlsQueue.push(callback);
+  if (_hlsState === "loading") return;
+  _hlsState = "loading";
+  const script = document.createElement("script");
+  script.src = "/shared/hls-1.6.16/hls.min.js";
+  script.onload = function () {
+    _hlsState = "ready";
+    const q = _hlsQueue;
+    _hlsQueue = [];
+    for (let i = 0; i < q.length; i++) q[i]();
+  };
+  script.onerror = function () {
+    _hlsState = "idle";
+    const q = _hlsQueue;
+    _hlsQueue = [];
+    // Fall through — _activatePlayer will use stream_url since Hls is undefined
+    for (let i = 0; i < q.length; i++) q[i]();
+  };
+  document.head.appendChild(script);
+}
+
+function _isHlsUrl(url) {
+  return typeof url === "string" && /\.m3u8(\?|$)/i.test(url);
+}
+
+function _activatePlayer(btn) {
+  const url = btn.dataset.streamUrl;
+  const hlsUrl = btn.dataset.hlsUrl;
+  const isAudio = btn.dataset.audioOnly === "true";
+  const directStream = btn.dataset.directStream === "true";
+
+  const player = document.createElement(isAudio ? "audio" : "video");
+  player.controls = true;
+  player.autoplay = true;
+  player.className = "media-player";
+
+  // Held so the error handler can tear the instance down before the player
+  // node is replaced — otherwise its listeners/loader timers run detached.
+  let hls = null;
+
+  // Prefer direct stream when the source supports it; fall back to HLS
+  // only when transcoding is needed.
+  if (directStream && url) {
+    player.src = url;
+  } else if (
+    hlsUrl &&
+    !isAudio &&
+    typeof Hls !== "undefined" &&
+    Hls.isSupported()
+  ) {
+    hls = new Hls();
+    hls.loadSource(hlsUrl);
+    hls.attachMedia(player);
+  } else if (
+    hlsUrl &&
+    !isAudio &&
+    player.canPlayType("application/vnd.apple.mpegurl")
+  ) {
+    player.src = hlsUrl;
+  } else {
+    player.src = url;
+  }
+
+  player.addEventListener("error", function () {
+    if (hls) {
+      hls.destroy();
+      hls = null; // media error events can repeat — never double-destroy
+    }
+    const card = player.closest(".media-embed");
+    const titleEl = card ? card.querySelector(".media-card-title") : null;
+    const label = titleEl ? ": " + titleEl.textContent : "";
+
+    const err = document.createElement("div");
+    err.className = "media-player-error";
+    err.setAttribute("role", "alert");
+    err.textContent = "Failed to load stream" + label;
+
+    const retry = document.createElement("button");
+    retry.className = "media-play-btn";
+    retry.type = "button";
+    retry.dataset.streamUrl = url;
+    retry.dataset.hlsUrl = hlsUrl || "";
+    retry.dataset.audioOnly = String(isAudio);
+    retry.dataset.directStream = String(directStream);
+    retry.setAttribute("aria-label", "Retry" + label);
+    retry.appendChild(document.createTextNode("▶ Retry"));
+
+    const container = document.createElement("div");
+    container.appendChild(err);
+    container.appendChild(retry);
+    player.replaceWith(container);
+  });
+
+  btn.replaceWith(player);
+}
+
+// Activate a clicked/Enter-pressed play button: show the loading affordance,
+// then ensure hls.js is loaded before swapping in the player when the source
+// needs it.  The pane wires this from a root-scoped this.el listener.
+function activateMediaPlayButton(btn) {
+  btn.disabled = true;
+  const labelEl = btn.querySelector("span:last-child");
+  if (labelEl) {
+    labelEl.textContent = "Loading…";
+  } else {
+    btn.textContent = "▶ Loading…";
+  }
+
+  const hlsUrl = btn.dataset.hlsUrl;
+  const isAudio = btn.dataset.audioOnly === "true";
+
+  // If HLS URL present and not audio, ensure hls.js is loaded first
+  if (hlsUrl && !isAudio && _isHlsUrl(hlsUrl)) {
+    _loadHls(function () {
+      _activatePlayer(btn);
+    });
+  } else {
+    _activatePlayer(btn);
+  }
+}
+
 function buildMediaCard(item) {
   const card = document.createElement("div");
   card.className = "media-card";
@@ -3244,10 +3403,18 @@ function createInteractivePane(root, wsId, opts) {
     warningTarget(pane) {
       return pane.messagesEl;
     },
-    // MCP re-consent surfaces inline in the pane card; the console has no
-    // settings-gear badge to drive (a future console consent surface can hook
-    // here).
-    onConsentDetected() {},
+    // MCP re-consent surfaces inline in the pane card; the STANDALONE additionally
+    // drives a Manage-row attention badge — bridged through the TS_APP seam so the
+    // shared factory stays deployment-agnostic (the console doesn't define the
+    // hook, so this stays a no-op there).
+    onConsentDetected(server) {
+      if (
+        window.TS_APP &&
+        typeof window.TS_APP.onConsentDetected === "function"
+      ) {
+        window.TS_APP.onConsentDetected(server);
+      }
+    },
   };
 
   const pane = new Pane(wsId, {

turnstone/shared_static/rail.js

@@ -39,6 +39,24 @@ export function glyph(state) {
   return el;
 }
 
+/** A small count chip for a Manage row/group head — glyph + count (never colour
+ *  alone), DS warn vocabulary (shell.css `.rail-badge`).  The count rides the
+ *  text; the ⚠ glyph is aria-hidden because the supplied `label` already names
+ *  the condition for assistive tech.  Returns a detached span the caller mounts. */
+function badge(count, label) {
+  const el = document.createElement("span");
+  el.className = "rail-badge";
+  const g = document.createElement("span");
+  g.className = "rail-badge-glyph";
+  g.setAttribute("aria-hidden", "true");
+  g.textContent = "⚠"; // ⚠ — pairs the colour with a glyph (chip-contrast rule)
+  const n = document.createElement("b");
+  n.textContent = String(count);
+  el.append(g, n);
+  if (label) el.setAttribute("aria-label", label);
+  return el;
+}
+
 /** Derive a node's overall state glyph from its workstream mix + health. */
 function nodeState(info) {
   if (!info.reachable) return "error";
@@ -325,6 +343,80 @@ export function mountRail(sections, caps) {
 
 // ---- Manage section (admin IA → collapsible discovery groups) --------------
 
+// Generic Manage-row badge state.  A subsystem (e.g. the standalone consent
+// badge) drives a count onto a tab row by KEY via `setRowBadge`; rail.js stays
+// agnostic about what the count means.  Kept module-level so a `mountManage`
+// rebuild (the IA is static, but the section re-mounts on shell init) re-applies
+// the live counts rather than dropping them.  `null` label = clear.
+const _rowBadges = new Map(); // tabKey -> { count, label }
+// Rebuilt each mount: the row <button> + owning group key for every tab.  The
+// group's head <button> (where a collapsed group surfaces its summed badge —
+// a collapsed group hides its rows, so the head must carry the signal) is
+// looked up via _groupEls, not duplicated here.
+let _rowEls = new Map(); // tabKey -> { row, group }
+let _groupEls = new Map(); // group key -> { head, tabKeys: [] }
+
+/** Paint (or clear) the badge slot inside a host button, keyed by a stable
+ *  `.rail-badge` child so repeated calls replace rather than stack. */
+function _applyBadge(host, count, label) {
+  if (!host) return;
+  const existing = host.querySelector(":scope > .rail-badge");
+  if (!count) {
+    if (existing) existing.remove();
+    return;
+  }
+  const fresh = badge(count, label);
+  if (existing) host.replaceChild(fresh, existing);
+  else host.append(fresh);
+}
+
+/** Sum the live badge counts for a group's tabs — drives the head badge so a
+ *  COLLAPSED group still shows that something inside it needs attention. */
+function _groupCount(groupKey) {
+  const grp = _groupEls.get(groupKey);
+  if (!grp) return 0;
+  let total = 0;
+  for (const tabKey of grp.tabKeys) {
+    const b = _rowBadges.get(tabKey);
+    if (b) total += b.count;
+  }
+  return total;
+}
+
+/**
+ * Generic Manage-row badge hook.  `setRowBadge(tabKey, count, label?)` stamps a
+ * count chip on that tab's row and, so the signal survives a collapsed group,
+ * mirrors the group's total onto its head.  `count` 0 (or falsy) clears the row.
+ * Drives off the refs `mountManage` registered; a no-op before the first mount
+ * (the consent subsystem re-drives it after `_refreshConsentBadge`).
+ *
+ * rail.js owns the MECHANISM only — callers own the meaning (no consent specifics
+ * here), matching the rail's seam-driven posture.
+ */
+export function setRowBadge(tabKey, count, label) {
+  const n = Number(count) || 0;
+  if (n > 0) _rowBadges.set(tabKey, { count: n, label: label || "" });
+  else _rowBadges.delete(tabKey);
+  const ref = _rowEls.get(tabKey);
+  if (!ref) return; // not mounted yet (or gated away) — state is kept for remount
+  _applyBadge(ref.row, n, label);
+  // The collapsed-group head mirrors the group's running total + its own label.
+  const total = _groupCount(ref.group);
+  const grp = _groupEls.get(ref.group);
+  _applyBadge(
+    grp && grp.head,
+    total,
+    total ? total + " in " + ref.group + " awaiting attention" : "",
+  );
+}
+
+/** Re-stamp every stored badge after a (re)mount rebuilt the row/head refs.
+ *  `setRowBadge` self-guards a missing ref and recomputes each head total, so
+ *  replaying the stored rows lands both rows and heads at their correct sums. */
+function _reapplyBadges() {
+  for (const [tabKey, b] of _rowBadges) setRowBadge(tabKey, b.count, b.label);
+}
+
 /**
  * Build the rail's Manage groups from the admin IA seam (admin.js exposes
  * `window.TS_ADMIN`).  Each group is a collapsible `.grp` whose head toggles
@@ -374,6 +466,9 @@ export function mountManage(root, paneManager) {
   const activeTab = adminOpen && TS.getActiveTab ? TS.getActiveTab() : null;
 
   const rowByTab = new Map(); // tab -> its row <button>, for active-state sync
+  // Rebuild the badge ref maps for this mount (the previous DOM is gone).
+  _rowEls = new Map();
+  _groupEls = new Map();
 
   ia.forEach((group) => {
     const tabs = group.tabs.filter((t) => allowed(t.tab));
@@ -404,6 +499,8 @@ export function mountManage(root, paneManager) {
     count.className = "gcount";
     count.textContent = String(tabs.length);
     head.append(chev, name, count);
+    // Register the head so a collapsed group can carry its tabs' badge total.
+    _groupEls.set(group.group, { head, tabKeys: tabs.map((t) => t.tab) });
 
     const items = document.createElement("div");
     items.className = "grp-items";
@@ -421,6 +518,9 @@ export function mountManage(root, paneManager) {
         if (TS.openTab) TS.openTab(t.tab);
       });
       rowByTab.set(t.tab, row);
+      // Register the row + its owning group for the badge hook (the group's
+      // head element is resolved through _groupEls when needed).
+      _rowEls.set(t.tab, { row, group: group.group });
       items.append(row);
     }
 
@@ -434,6 +534,9 @@ export function mountManage(root, paneManager) {
     root.append(grp);
   });
 
+  // Re-apply any live row badges a subsystem set before/across this (re)mount.
+  _reapplyBadges();
+
   // Single writer for the Manage active-row: the row for the current admin tab
   // carries `.active`.  admin.js notifies on every switchAdminTab; seed it here
   // for an already-open (restored) Admin pane.

turnstone/shared_static/shell.css

@@ -997,6 +997,38 @@
   box-shadow: inset 2px 0 0 var(--ink-4);
 }
 
+/* Manage-row attention badge (rail.js `setRowBadge` / `badge`) — a small count
+   chip pinned to the row tail (and to a COLLAPSED group's head, so a hidden row
+   never hides the signal).  DS warn vocabulary: the ⚠ glyph PLUS the tinted
+   chip carry the meaning (never colour alone), so it reads at chip size and
+   flips themes by construction (the --warn family is defined per theme). */
+.rail-badge {
+  display: inline-flex;
+  align-items: center;
+  gap: 3px;
+  margin-left: auto; /* push to the row/head tail past the flex-1 label */
+  padding: 0 5px;
+  border-radius: var(--r-sm);
+  background: var(--warn-tint);
+  border: 1px solid var(--warn-tint-border);
+  color: var(--warn);
+  font-size: 10px;
+  line-height: 1.5;
+  font-variant-numeric: tabular-nums;
+  flex: none;
+}
+/* The group head already auto-spaces with its own gap, and its --warn count
+   should not fight the dim --ink-4 tab count beside it. */
+.grp-head .rail-badge {
+  margin-left: 6px;
+}
+.rail-badge .rail-badge-glyph {
+  font-weight: 700;
+}
+.rail-badge b {
+  font-weight: 600;
+}
+
 /* ===== Admin pane — adopts #view-admin (the 18 tabpanels).  The in-pane
    sidebar is retired (the rail's Manage groups navigate), so #view-admin fills
    the pane body and the content host renders full-width. ===== */

turnstone/shared_static/shell.js

@@ -24,7 +24,7 @@
    ========================================================================== */
 
 import { PaneManager, ShellPane, openPopupMenu } from "./pane.js";
-import { mountRail, mountManage, glyph } from "./rail.js";
+import { mountRail, mountManage, glyph, setRowBadge } from "./rail.js";
 import { authFetch } from "./auth.js";
 // The interactive pane is a real ES module beside us in /shared (step 5a) — the
 // shell imports it directly, and it exists in every deployment.  The coordinator
@@ -880,7 +880,11 @@ async function mountShell() {
     const p = pm.getPane("interactive", wsId);
     if (p && p._ctl && p._ctl.markDead) p._ctl.markDead();
   };
-  window.TS_SHELL = { panes: pm, caps, notifySessionClosed };
+  // `setRowBadge` lets a classic-script subsystem (the standalone consent badge
+  // in ui/static/app.js) stamp a count chip on a Manage row without importing the
+  // ESM rail module — the shell is its module bridge.  Generic: the rail owns the
+  // chip mechanism, the caller owns what the count means.
+  window.TS_SHELL = { panes: pm, caps, notifySessionClosed, setRowBadge };
 
   // Login fan-out: app.js owns the single window.onLoginSuccess (the Tier-1
   // reconnect, set at load).  Wrap it in a tiny registry so EVERY conversational

turnstone/ui/static/app.js

@@ -1473,17 +1473,19 @@ function connectGlobalSSE() {
 }
 
 // ===========================================================================
-//  12. MCP consent badge (standalone settings-gear pending-consent indicator)
+//  12. MCP consent badge (standalone pending-consent indicator)
 //
 //  The tool-output / media / MCP-error / verdict renderers that used to live in
 //  this section moved to shared_static/interactive.js with the Pane.  What
-//  stays here is the standalone consent-badge subsystem: the gear badge lives
-//  in this shell's header, so the pane only notifies it (host.onConsentDetected
-//  -> _onConsentDetected) and the dashboard hydrates it via loadPendingConsents.
+//  stays here is the standalone consent-badge subsystem: it owns the pending set
+//  and drives the rail's Manage > Connections row badge (via the TS_SHELL bridge
+//  — `setRowBadge`).  An interactive pane only NOTIFIES it (the shared host
+//  bridges `onConsentDetected` to the TS_APP seam below); `loadPendingConsents`
+//  hydrates it on boot.  The settings-gear it used to hang on is retired.
 // ===========================================================================
 
 // Module-level set of servers with an unresolved consent prompt; drives the
-// gear-icon badge so the user has a stable signal that re-consent is pending
+// Manage-row badge so the user has a stable signal that re-consent is pending
 // after the inline card scrolls out of view.
 const _pendingConsentServers = new Set();
 
@@ -1528,32 +1530,26 @@ function loadPendingConsents() {
     });
 }
 
+// The Manage tab the pending-consent badge rides on.  The standalone's Manage IA
+// (TS_ADMIN.ia, below) is a single Extensions > Connections tab where MCP server
+// connections live; the badge surfaces there (and, when that group is collapsed,
+// on its head — the rail handles that).  The retired settings-gear it used to
+// hang on is gone with the L-shell renovation.
+const _CONSENT_BADGE_TAB = "connections";
+
 function _refreshConsentBadge() {
-  const btn = document.getElementById("settings-btn");
-  if (!btn) return;
-  let existing = btn.querySelector(".settings-consent-badge");
   const n = _pendingConsentServers.size;
-  // Keep the visible badge and the accessible name in lockstep so screen-
-  // reader users get the same pending-consent signal that sighted users
-  // get from the red dot. The badge itself stays aria-hidden because the
-  // count is already reflected in the button's aria-label/title.
-  if (n === 0) {
-    if (existing) existing.remove();
-    btn.setAttribute("aria-label", "Settings");
-    btn.setAttribute("title", "Settings");
-    return;
-  }
-  if (!existing) {
-    existing = document.createElement("span");
-    existing.className = "settings-consent-badge";
-    existing.setAttribute("aria-hidden", "true");
-    btn.appendChild(existing);
-  }
-  existing.textContent = String(n);
+  // Drive the rail's generic Manage-row badge through the shell bridge (classic
+  // app.js can't import the ESM rail module).  The chip's own ⚠ glyph + count
+  // carry the signal; `label` keeps the accessible name in lockstep.  A no-op
+  // before the rail mounts — `loadPendingConsents` re-drives it after boot.
+  const shell = window.TS_SHELL;
+  if (!shell || typeof shell.setRowBadge !== "function") return;
   const label =
-    "Settings (" + n + " MCP consent" + (n === 1 ? "" : "s") + " pending)";
-  btn.setAttribute("aria-label", label);
-  btn.setAttribute("title", label);
+    n === 0
+      ? ""
+      : n + " MCP server" + (n === 1 ? "" : "s") + " awaiting consent";
+  shell.setRowBadge(_CONSENT_BADGE_TAB, n, label);
 }
 
 /**
@@ -1569,141 +1565,6 @@ function _refreshConsentBadge() {
  * Render the action card for an MCP error envelope.  Mirrors the
  * media-embed pattern: visible card on top, collapsible raw JSON below.
  */
-// ---------------------------------------------------------------------------
-//  HLS lazy-loader (follows the mermaid.js lazy-load pattern in
-//  /shared/renderer.js)
-// ---------------------------------------------------------------------------
-let _hlsState = "idle";
-let _hlsQueue = [];
-
-function _loadHls(callback) {
-  if (_hlsState === "ready") {
-    callback();
-    return;
-  }
-  _hlsQueue.push(callback);
-  if (_hlsState === "loading") return;
-  _hlsState = "loading";
-  const script = document.createElement("script");
-  script.src = "/shared/hls-1.6.16/hls.min.js";
-  script.onload = function () {
-    _hlsState = "ready";
-    const q = _hlsQueue;
-    _hlsQueue = [];
-    for (let i = 0; i < q.length; i++) q[i]();
-  };
-  script.onerror = function () {
-    _hlsState = "idle";
-    const q = _hlsQueue;
-    _hlsQueue = [];
-    // Fall through — _activatePlayer will use stream_url since Hls is undefined
-    for (let i = 0; i < q.length; i++) q[i]();
-  };
-  document.head.appendChild(script);
-}
-
-function _isHlsUrl(url) {
-  return typeof url === "string" && /\.m3u8(\?|$)/i.test(url);
-}
-
-// ---------------------------------------------------------------------------
-//  Click-to-play delegated handler (follows img-placeholder pattern)
-// ---------------------------------------------------------------------------
-function _activatePlayer(btn) {
-  const url = btn.dataset.streamUrl;
-  const hlsUrl = btn.dataset.hlsUrl;
-  const isAudio = btn.dataset.audioOnly === "true";
-  const directStream = btn.dataset.directStream === "true";
-
-  const player = document.createElement(isAudio ? "audio" : "video");
-  player.controls = true;
-  player.autoplay = true;
-  player.className = "media-player";
-
-  // Prefer direct stream when the source supports it; fall back to HLS
-  // only when transcoding is needed.
-  if (directStream && url) {
-    player.src = url;
-  } else if (
-    hlsUrl &&
-    !isAudio &&
-    typeof Hls !== "undefined" &&
-    Hls.isSupported()
-  ) {
-    const hls = new Hls();
-    hls.loadSource(hlsUrl);
-    hls.attachMedia(player);
-  } else if (
-    hlsUrl &&
-    !isAudio &&
-    player.canPlayType("application/vnd.apple.mpegurl")
-  ) {
-    player.src = hlsUrl;
-  } else {
-    player.src = url;
-  }
-
-  player.addEventListener("error", function () {
-    const card = player.closest(".media-embed");
-    const titleEl = card ? card.querySelector(".media-card-title") : null;
-    const label = titleEl ? ": " + titleEl.textContent : "";
-
-    const err = document.createElement("div");
-    err.className = "media-player-error";
-    err.setAttribute("role", "alert");
-    err.textContent = "Failed to load stream" + label;
-
-    const retry = document.createElement("button");
-    retry.className = "media-play-btn";
-    retry.type = "button";
-    retry.dataset.streamUrl = url;
-    retry.dataset.hlsUrl = hlsUrl || "";
-    retry.dataset.audioOnly = String(isAudio);
-    retry.dataset.directStream = String(directStream);
-    retry.setAttribute("aria-label", "Retry" + label);
-    retry.appendChild(document.createTextNode("\u25b6 Retry"));
-
-    const container = document.createElement("div");
-    container.appendChild(err);
-    container.appendChild(retry);
-    player.replaceWith(container);
-  });
-
-  btn.replaceWith(player);
-}
-
-document.addEventListener("click", function (e) {
-  const btn = e.target.closest(".media-play-btn");
-  if (!btn) return;
-  e.preventDefault();
-  btn.disabled = true;
-  const labelEl = btn.querySelector("span:last-child");
-  if (labelEl) {
-    labelEl.textContent = "Loading\u2026";
-  } else {
-    btn.textContent = "\u25b6 Loading\u2026";
-  }
-
-  const hlsUrl = btn.dataset.hlsUrl;
-  const isAudio = btn.dataset.audioOnly === "true";
-
-  // If HLS URL present and not audio, ensure hls.js is loaded first
-  if (hlsUrl && !isAudio && _isHlsUrl(hlsUrl)) {
-    _loadHls(function () {
-      _activatePlayer(btn);
-    });
-  } else {
-    _activatePlayer(btn);
-  }
-});
-
-document.addEventListener("keydown", function (e) {
-  if (e.key !== "Enter") return;
-  const btn = e.target.closest(".media-play-btn");
-  if (!btn) return;
-  btn.click();
-});
-
 function _announce(text) {
   const el = document.getElementById("toast");
   if (!el) return;
@@ -2365,6 +2226,11 @@ window.TS_APP.onRender = function (cb) {
 };
 window.TS_APP.bucketByParent = bucketByParent;
 window.TS_APP.boot = boot;
+// Live MCP-consent notifications from an interactive pane (the shared pane host
+// bridges its `onConsentDetected` here when this seam exists; the console leaves
+// it undefined, so the pane no-ops there).  Adds the server to the pending set
+// and re-paints the Manage-row badge.
+window.TS_APP.onConsentDetected = _onConsentDetected;
 
 // --- Manage seam: one Connections tab (MCP server connections) -------------
 const _CONN_IA = [

turnstone/ui/static/style.css

@@ -1822,16 +1822,3 @@ audio.media-player {
 .settings-revoke-btn:hover {
   background: rgba(255, 0, 0, 0.05);
 }
-.settings-consent-badge {
-  display: inline-block;
-  margin-left: 4px;
-  background: var(--red);
-  color: var(--bg);
-  border-radius: 8px;
-  font-size: 10px;
-  padding: 1px 5px;
-  vertical-align: top;
-  font-weight: 600;
-  line-height: 1.4;
-  font-family: var(--font-ui);
-}

uv.lock

@@ -2324,7 +2324,7 @@ wheels = [
 
 [[package]]
 name = "turnstone"
-version = "1.6.0"
+version = "1.6.2"
 source = { editable = "." }
 dependencies = [
     { name = "alembic" },
@@ -2385,7 +2385,7 @@ requires-dist = [
     { name = "httpx", specifier = ">=0.28" },
     { name = "httpx-sse", specifier = ">=0.4" },
     { name = "lacme", specifier = ">=1.0.5" },
-    { name = "mcp", specifier = ">=1.27" },
+    { name = "mcp", specifier = ">=1.27,<2" },
     { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.14" },
     { name = "openai", specifier = ">=2.37" },
     { name = "psycopg", extras = ["binary"], specifier = ">=3.2" },