Bump version to 0.5.1

* feat: add ClusterSnapshot for instant console UI state rebuild

The console web UI was SSE-driven with no initial state — reloads and
navigation caused blank/loading gaps while waiting for API re-fetches.

Server-side: GET /v1/api/cluster/snapshot returns the full cluster state
(all nodes with workstreams + overview aggregates) built under a single
lock. The SSE stream now emits this snapshot as the first event on
connect (snapshot taken before listener registration to avoid race).

Frontend: local clusterState object mirrors the snapshot, patched
incrementally by SSE events. View navigation renders from local state
with no API round-trips. Fixes popstate/pushState history corruption
on Back/Forward navigation (pre-existing bug). Stable node sorting
with node_id tie-breaker on both server and client.

SDK: snapshot() method on Python (sync + async) and TypeScript console
clients. ClusterSnapshotEvent in event registries.

* fix: address review feedback and SSE proxy reconnect bug

Copilot review fixes:
- Atomic snapshot+register: new get_snapshot_and_register() acquires
  both state and listener locks, eliminating the event gap between
  snapshot read and listener registration.
- Debounce patch renders: patchClusterState uses requestAnimationFrame
  to batch rapid SSE events into a single recompute+render cycle.
- Fix health type: dict[str, str] → dict[str, Any] on all three
  console schema models (ClusterNodeInfo, NodeDetailResponse,
  ClusterSnapshotNode) since /health payloads contain nested objects.
- TypeScript ClusterSnapshotEvent: use concrete ClusterSnapshotNode[]
  and ClusterOverviewResponse types instead of Record<string, unknown>.

SSE proxy reconnect fix:
- _proxy_sse raw_stream now emits `: proxy-ping` comments every 3s
  when no upstream data arrives, preventing the browser EventSource
  from dropping idle connections. The raw byte passthrough refactor
  (4d11078) removed the proxy's independent keepalive — this restores
  it without reverting to EventSourceResponse.

README.md

@@ -156,7 +156,7 @@ Bridges BLPOP from their per-node queue (priority) then the shared queue. Direct
 | Tool | Description | Auto-approved |
 |------|-------------|:---:|
 | `bash` | Execute shell commands | |
-| `read_file` | Read file contents | yes |
+| `read_file` | Read file contents (text or images with vision models) | yes |
 | `write_file` | Write/create files | |
 | `edit_file` | Fuzzy-match file editing | |
 | `search` | Search files by name/content | yes |

compose.yaml

@@ -207,6 +207,7 @@ services:
       dockerfile: Dockerfile
     profiles:
       - production
+      - cluster
     command:
       - sh
       - -c

docs/architecture.md

@@ -560,21 +560,23 @@ LLMProvider (protocol)
 |------|--------|
 | `StreamChunk` | `content_delta`, `reasoning_delta`, `tool_call_deltas`, `info_delta`, `usage`, `finish_reason` |
 | `CompletionResult` | `content`, `tool_calls`, `finish_reason`, `usage` |
-| `ModelCapabilities` | `context_window`, `max_output_tokens`, `supports_temperature`, `token_param`, `thinking_mode`, `supports_effort`, `supports_web_search` |
+| `ModelCapabilities` | `context_window`, `max_output_tokens`, `supports_temperature`, `token_param`, `thinking_mode`, `supports_effort`, `supports_web_search`, `supports_tool_search`, `supports_vision` |
 | `UsageInfo` | `prompt_tokens`, `completion_tokens`, `total_tokens` |
 
 **OpenAIProvider** (`_openai.py`): passes messages through unchanged (they are
-already in OpenAI format). Model capability lookup table covers
-GPT-5/5.1/5.2, O-series, and search models (`gpt-5-search-api`).
+already in OpenAI format), including multi-part content blocks (text + images)
+in tool results. Model capability lookup table covers GPT-5/5.1/5.2/5.3/5.4,
+O-series, and search models (`gpt-5-search-api`) — all with `supports_vision`.
 For search models, injects `web_search_options` and removes the `web_search`
 function tool (the model always searches). Citations from `url_citation`
 annotations are formatted as footnotes. Unknown models (local servers) get
-permissive defaults and use Tavily for web search.
+permissive defaults with `supports_vision=False` and use Tavily for web search.
 
 **AnthropicProvider** (`_anthropic.py`): converts OpenAI-format messages to
 Anthropic content blocks, maps `system`/`developer` roles to the `system`
 parameter, groups consecutive `tool` result messages into user-role content
-blocks, and translates tool schemas from OpenAI function-calling format to
+blocks (converting `image_url` parts to Anthropic's `image` source format),
+and translates tool schemas from OpenAI function-calling format to
 Anthropic's `input_schema` format. Supports both manual and adaptive thinking
 modes, with effort parameter support for models like Claude Opus 4.6 and
 Sonnet 4.6. Replaces the `web_search` function tool with Anthropic's native
@@ -620,6 +622,18 @@ agent_model = "claude"
 
 Each `[models.*]` entry produces a `ModelConfig` with a `provider` field
 (default: `"openai"`). Supported values: `"openai"` and `"anthropic"`.
+An optional `[models.*.capabilities]` sub-table overrides per-model
+`ModelCapabilities` flags (useful for local models whose capabilities
+cannot be detected programmatically):
+
+```toml
+[models.qwen-vl]
+base_url = "http://localhost:8000/v1"
+model = "qwen-3.5-vl"
+
+[models.qwen-vl.capabilities]
+supports_vision = true
+```
 
 **Lifecycle:**
 1. `load_model_registry()` reads `[models.*]` sections from config.toml and

docs/console.md

@@ -61,6 +61,8 @@ The collector (`turnstone/console/collector.py`) maintains an in-memory snapshot
 
 3. **Poll loop** — fetches `GET /v1/api/dashboard` and `GET /health` from each known node every 10 seconds. Uses `ThreadPoolExecutor(max_workers=50)` for parallelism. Each poll replaces the node's workstream list with the authoritative server data.
 
+A `get_snapshot()` method builds the full cluster state under a single lock acquisition — overview aggregates and per-node workstream lists in one atomic read. This is served both as a REST endpoint and as the initial SSE event on client connect.
+
 ### Thread Safety
 
 All reads and writes to the node/workstream map are protected by a single `threading.Lock`. Query methods acquire the lock, copy data, and release before returning.
@@ -146,6 +148,38 @@ Single node detail with all its workstreams.
 }
 ```
 
+### `GET /v1/api/cluster/snapshot`
+
+Full cluster state in a single response — all nodes with their workstreams plus overview aggregates. Built under a single lock for internal consistency. Used by the browser on initial load and SSE reconnect.
+
+```json
+{
+  "nodes": [
+    {
+      "node_id": "db-west-04",
+      "server_url": "http://10.0.3.4:8080",
+      "max_ws": 10,
+      "reachable": true,
+      "version": "0.3.0",
+      "health": {"status": "ok", "version": "0.3.0"},
+      "aggregate": {"total_tokens": 48200, "total_tool_calls": 156},
+      "workstreams": [
+        {"id": "a1b2c3d4", "name": "perf-db-west", "state": "running", ...}
+      ]
+    }
+  ],
+  "overview": {
+    "nodes": 847,
+    "workstreams": 4219,
+    "states": {"running": 1847, "thinking": 312, "attention": 89, "idle": 1940, "error": 31},
+    "aggregate": {"total_tokens": 12400000, "total_tool_calls": 34200},
+    "version_drift": false,
+    "versions": ["0.3.0"]
+  },
+  "timestamp": 1709294400.0
+}
+```
+
 ### `POST /v1/api/cluster/workstreams/new`
 
 Create a new workstream on a target node. Dispatches a `CreateWorkstreamMessage` through the Redis MQ pipeline — the bridge on the target node picks it up and creates the workstream on the server. Requires `write` scope.
@@ -182,7 +216,7 @@ Creation is asynchronous — the response confirms the MQ message was dispatched
 
 ### `GET /v1/api/cluster/events`
 
-Server-Sent Events stream for real-time cluster updates.
+Server-Sent Events stream for real-time cluster updates. The first event is always a `snapshot` containing the full cluster state (same shape as `GET /v1/api/cluster/snapshot` with an added `type: "snapshot"` field), followed by incremental events:
 
 ```
 data: {"type":"cluster_state","ws_id":"a1b2","node_id":"db-west-04","state":"running"}
@@ -326,7 +360,7 @@ The server UI uses root-relative URLs (`/v1/api/send`, `/static/app.js`, `/share
 
 ### SSE Proxy
 
-SSE streams (`/v1/api/events`, `/v1/api/events/global`) are proxied by creating a per-connection `httpx.AsyncClient(timeout=None)`, streaming the upstream response via `aiter_text()`, parsing SSE framing (`\n\n` delimiters), and re-emitting events through `EventSourceResponse`. Each proxied SSE stream requires its own httpx client since the shared client's 30-second timeout would kill long-lived connections.
+SSE streams (`/v1/api/events`, `/v1/api/events/global`) are proxied as raw byte passthrough — the console opens an `httpx.AsyncClient.stream()` to the upstream server (with `read=None` and `pool=None` timeouts since SSE connections are long-lived) and relays every byte via `StreamingResponse`. This preserves server-side ping comments, event framing, and keepalives verbatim without parsing or re-encoding.
 
 ### Authentication
 
@@ -368,6 +402,8 @@ On submit, `POST /v1/api/cluster/workstreams/new` dispatches the creation reques
 
 All five views receive live updates via SSE — state cards update counts, node rows update metrics, workstream rows update state indicators.
 
+The browser maintains a local `clusterState` object that mirrors the cluster snapshot. It is initialized from the SSE `snapshot` event on connect (or via `GET /v1/api/cluster/snapshot` on initial page load) and updated incrementally by SSE events. View navigation reads from local state — no API round-trips needed after the initial snapshot.
+
 ### 5. Admin Panel
 
 Accessed via the "admin" button in the header (visible when authenticated

docs/diagrams/03-core-engine-classes.puml

@@ -109,6 +109,7 @@ class "ModelCapabilities" as ModelCaps <<frozen>> {
     + supports_effort: bool
     + supports_web_search: bool
     + supports_tool_search: bool
+    + supports_vision: bool
 }
 
 ' ChatSession

docs/diagrams/04-conversation-turn.puml

@@ -112,7 +112,7 @@ group loop [while tool_calls present]
         note right of TP
           Parallel execution:
           bash → Popen + line-by-line streaming
-          read_file → open().read()
+          read_file → open().read() or base64 image
           search → grep subprocess
           edit_file → string replace
           task/plan → _run_agent() sub-loop

docs/diagrams/05-tool-pipeline.puml

@@ -100,7 +100,7 @@ partition "Phase 3: Execute" #E3F2FD {
         if item.denied → return denial message
         else → item["execute"](item)
           ├─ _exec_bash: subprocess.run(["bash", script.sh])
-          ├─ _exec_read_file: open().readlines()
+          ├─ _exec_read_file: open().readlines() or _exec_read_image (base64)
           ├─ _exec_write_file: makedirs + write
           ├─ _exec_edit_file: find_occurrences + replace
           ├─ _exec_search: grep subprocess

docs/diagrams/11-console-data-flow.puml

@@ -89,10 +89,15 @@ deactivate CC
 Browser -> Server : GET /v1/api/cluster/events
 activate Server
 
+Server -> CC : get_snapshot()
+CC --> Server : ClusterSnapshot\n(full current state)
+
 Server -> CC : register_listener(queue)
 note right : Per-client queue.Queue(maxsize=500)\nSSE via EventSourceResponse + run_in_executor()
 
-loop continuous
+Server -> Browser : data: {"type":"snapshot",...}\n(full state as first SSE event)
+
+loop continuous (incremental updates)
     CC -> Server : event via listener queue\n(from any of the 3 threads)
     Server -> Browser : data: {"type":"cluster_state",...}\n\n
 end
@@ -105,6 +110,13 @@ Browser -> Server : connection closed
 Server -> CC : unregister_listener(queue)
 deactivate Server
 
+== Browser REST: Snapshot ==
+
+Browser -> Server : GET /v1/api/cluster/snapshot
+Server -> CC : get_snapshot()
+CC --> Server : ClusterSnapshot\n(full current state)
+Server --> Browser : JSON response
+
 == Browser REST Requests ==
 
 Browser -> Server : GET /v1/api/cluster/overview

docs/diagrams/13-sdk-architecture.puml

@@ -45,6 +45,7 @@ package "turnstone/sdk/ (Python)" {
         + nodes()
         + workstreams()
         + node_detail()
+        + snapshot()
         + create_workstream()
         + stream_cluster_events()
         + login() / logout()
@@ -129,6 +130,7 @@ package "sdk/typescript/ (TypeScript)" {
     class "TurnstoneConsole" as TSConsole <<ts>> {
         + overview()
         + nodes()
+        + snapshot()
         + clusterEvents()
         ...
     }

docs/diagrams/png/03-core-engine-classes.png

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:4512be48a51f7cd1136ea8e3344489a8d225c611cb1b961643b869351de76812
-size 549668
+oid sha256:c53ddce800c59f9432d7a016c7d66282a449555d452b7fe9dd393f4282f08c46
+size 554721

docs/diagrams/png/04-conversation-turn.png

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:24bdc6a83259e4db6aaa24f581bed59d351c7a83e1c282aac123c52b32d9f80d
-size 288250
+oid sha256:e3044c738d6d6853aab5c4990e6c67bab0165eba991a4f5bebdfc4d4a0b305ee
+size 289165

docs/diagrams/png/05-tool-pipeline.png

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:027ad99469d69f1d6b2e73ee802b50a75617cd286392375aa69133c13d3683dc
-size 256347
+oid sha256:282820fe416961e735d050f86ecdc079e29824d2b3c4d5c8c174d0533d41f211
+size 258045

docs/sdk.md

@@ -95,6 +95,7 @@ Both `TurnstoneConsole` (sync) and `AsyncTurnstoneConsole` (async) expose:
 | | `nodes(*, sort, limit, offset)` | `ClusterNodesResponse` |
 | | `workstreams(*, state, node, search, sort, page, per_page)` | `ClusterWorkstreamsResponse` |
 | | `node_detail(node_id)` | `NodeDetailResponse` |
+| | `snapshot()` | `ClusterSnapshotResponse` |
 | | `create_workstream(*, node_id, name, model, initial_message)` | `ConsoleCreateWsResponse` |
 | **Schedules** | `list_schedules()` | `ListSchedulesResponse` |
 | | `create_schedule(*, name, schedule_type, initial_message, ...)` | `ScheduleInfo` |
@@ -146,6 +147,9 @@ SSE events are deserialized into typed dataclasses. Use `event.type` to discrimi
 | `node_lost` | `NodeLostEvent` | `node_id` |
 | `cluster_state` | `ClusterStateEvent` | `ws_id`, `node_id`, `state`, `tokens` |
 | `ws_created` | `ClusterWsCreatedEvent` | `ws_id`, `node_id`, `name` |
+| `ws_closed` | `ClusterWsClosedEvent` | `ws_id` |
+| `ws_rename` | `ClusterWsRenameEvent` | `ws_id`, `name` |
+| `snapshot` | `ClusterSnapshotEvent` | `nodes`, `overview`, `timestamp` |
 
 ### TurnResult
 

docs/tools.md

@@ -189,15 +189,17 @@ Execute a bash command and return stdout + stderr.
 
 ### read_file
 
-Read the contents of a file, returning numbered lines.
+Read the contents of a file, returning numbered lines for text files or
+base64-encoded image data for supported image formats.
 
 | Parameter | Type    | Required | Description |
 |-----------|---------|----------|-------------|
 | `path`    | string  | yes      | Absolute or relative file path. |
-| `offset`  | integer | no       | Line number to start from (1-based, default: 1). |
-| `limit`   | integer | no       | Maximum number of lines to read. Omit for full file. |
+| `offset`  | integer | no       | Line number to start from (1-based, default: 1). Text files only. |
+| `limit`   | integer | no       | Maximum number of lines to read. Omit for full file. Text files only. |
 
-- **What it does**: Reads the file and returns content with line numbers. Must be called before `edit_file` on the same path (the session tracks which files have been read).
+- **What it does**: For text files, reads and returns content with line numbers. For image files (PNG, JPEG, GIF, WebP, BMP, TIFF, ICO), returns image data as multi-part content when the model supports vision, or a text description when it does not. SVG files are read as text. Images larger than 4 MB are rejected. Must be called before `edit_file` on the same path (the session tracks which files have been read).
+- **Vision support**: Controlled by `ModelCapabilities.supports_vision`. All commercial OpenAI and Anthropic models have vision enabled. Local models (vLLM, llama.cpp, NIM) default to off — enable via `[models.*.capabilities] supports_vision = true` in config.toml.
 - **Auto-approve**: Yes.
 - **Agent availability**: `agent` and `task_agent`.
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "turnstone"
-version = "0.4.6"
+version = "0.5.1"
 description = "Multi-node AI orchestration platform with tool use, agent routing, and cluster simulation."
 readme = "README.md"
 license = "BUSL-1.1"

sdk/typescript/src/console.ts

@@ -6,6 +6,7 @@ import type {
   AuthStatusResponse,
   ClusterNodesResponse,
   ClusterOverviewResponse,
+  ClusterSnapshotResponse,
   ClusterWorkstreamsResponse,
   ConsoleCreateWsRequest,
   ConsoleCreateWsResponse,
@@ -33,6 +34,10 @@ export class TurnstoneConsole extends BaseClient {
     return this.request("GET", "/v1/api/cluster/overview");
   }
 
+  async snapshot(): Promise<ClusterSnapshotResponse> {
+    return this.request("GET", "/v1/api/cluster/snapshot");
+  }
+
   async nodes(opts?: NodesOptions): Promise<ClusterNodesResponse> {
     return this.request("GET", "/v1/api/cluster/nodes", {
       params: {

sdk/typescript/src/events.ts

@@ -1,3 +1,5 @@
+import type { ClusterOverviewResponse, ClusterSnapshotNode } from "./types.js";
+
 // ---------------------------------------------------------------------------
 // Server SSE events
 // ---------------------------------------------------------------------------
@@ -191,6 +193,13 @@ export interface ClusterWsRenameEvent {
   name: string;
 }
 
+export interface ClusterSnapshotEvent {
+  type: "snapshot";
+  nodes: ClusterSnapshotNode[];
+  overview: ClusterOverviewResponse;
+  timestamp: number;
+}
+
 /** Discriminated union of all console cluster SSE event types. */
 export type ClusterEvent =
   | NodeJoinedEvent
@@ -198,7 +207,8 @@ export type ClusterEvent =
   | ClusterStateEvent
   | ClusterWsCreatedEvent
   | ClusterWsClosedEvent
-  | ClusterWsRenameEvent;
+  | ClusterWsRenameEvent
+  | ClusterSnapshotEvent;
 
 // ---------------------------------------------------------------------------
 // Type guards

sdk/typescript/src/index.ts

@@ -55,6 +55,7 @@ export type {
   ClusterWsCreatedEvent,
   ClusterWsClosedEvent,
   ClusterWsRenameEvent,
+  ClusterSnapshotEvent,
 } from "./events.js";
 
 export {
@@ -97,6 +98,8 @@ export type {
   ClusterOverviewResponse,
   ClusterNodeInfo,
   ClusterNodesResponse,
+  ClusterSnapshotNode,
+  ClusterSnapshotResponse,
   ClusterWorkstreamInfo,
   ClusterWorkstreamsResponse,
   NodeDetailResponse,

sdk/typescript/src/types.ts

@@ -244,6 +244,23 @@ export interface NodeDetailResponse {
   aggregate: ClusterAggregate;
 }
 
+export interface ClusterSnapshotNode {
+  node_id: string;
+  server_url: string;
+  max_ws: number;
+  reachable: boolean;
+  version: string;
+  health: Record<string, string>;
+  aggregate: Record<string, number>;
+  workstreams: ClusterWorkstreamInfo[];
+}
+
+export interface ClusterSnapshotResponse {
+  nodes: ClusterSnapshotNode[];
+  overview: ClusterOverviewResponse;
+  timestamp: number;
+}
+
 export interface ConsoleCreateWsRequest {
   node_id?: string;
   name?: string;

tests/test_console.py

@@ -1,5 +1,6 @@
 """Tests for turnstone.console — collector and HTTP server."""
 
+import asyncio
 import json
 import queue
 from unittest.mock import MagicMock
@@ -445,6 +446,44 @@ class TestCollectorQueries:
     def test_get_node_detail_not_found(self, populated_collector):
         assert populated_collector.get_node_detail("nonexistent") is None
 
+    def test_get_snapshot_empty(self):
+        c = _make_collector()
+        snap = c.get_snapshot()
+        assert snap["nodes"] == []
+        assert snap["overview"]["nodes"] == 0
+        assert snap["overview"]["workstreams"] == 0
+        assert snap["overview"]["states"]["running"] == 0
+        assert "timestamp" in snap
+
+    def test_get_snapshot_with_nodes(self, populated_collector):
+        snap = populated_collector.get_snapshot()
+        assert len(snap["nodes"]) == 2
+        assert snap["overview"]["nodes"] == 2
+        assert snap["overview"]["workstreams"] == 3
+        assert snap["overview"]["states"]["running"] == 1
+        assert snap["overview"]["states"]["attention"] == 1
+        assert snap["overview"]["states"]["idle"] == 1
+        assert snap["overview"]["aggregate"]["total_tokens"] == 17000
+        assert snap["timestamp"] > 0
+        # Each node should embed its workstreams
+        node_ids = {n["node_id"] for n in snap["nodes"]}
+        assert node_ids == {"node-a", "node-b"}
+        for n in snap["nodes"]:
+            if n["node_id"] == "node-a":
+                assert len(n["workstreams"]) == 2
+            elif n["node_id"] == "node-b":
+                assert len(n["workstreams"]) == 1
+
+    def test_get_snapshot_consistency(self, populated_collector):
+        """Snapshot overview should match get_overview()."""
+        snap = populated_collector.get_snapshot()
+        overview = populated_collector.get_overview()
+        assert snap["overview"]["nodes"] == overview["nodes"]
+        assert snap["overview"]["workstreams"] == overview["workstreams"]
+        assert snap["overview"]["states"] == overview["states"]
+        assert snap["overview"]["aggregate"] == overview["aggregate"]
+        assert snap["overview"]["version_drift"] == overview["version_drift"]
+
 
 # ---------------------------------------------------------------------------
 # ClusterStateEvent protocol tests
@@ -535,6 +574,31 @@ class TestConsoleHTTPEndpoints:
             "workstreams": [],
             "aggregate": {},
         }
+        collector.get_snapshot.return_value = {
+            "nodes": [
+                {
+                    "node_id": "node-a",
+                    "server_url": "http://a:8080",
+                    "max_ws": 10,
+                    "reachable": True,
+                    "version": "0.5.0",
+                    "health": {},
+                    "aggregate": {"total_tokens": 50000, "total_tool_calls": 200},
+                    "workstreams": [
+                        {"id": "ws1", "name": "test", "state": "running", "node": "node-a"},
+                    ],
+                },
+            ],
+            "overview": {
+                "nodes": 3,
+                "workstreams": 15,
+                "states": {"running": 5, "thinking": 2, "attention": 1, "idle": 6, "error": 1},
+                "aggregate": {"total_tokens": 50000, "total_tool_calls": 200},
+                "version_drift": False,
+                "versions": ["0.5.0"],
+            },
+            "timestamp": 1234567890.0,
+        }
         return collector
 
     @pytest.fixture()
@@ -614,6 +678,16 @@ class TestConsoleHTTPEndpoints:
         assert status == 404
         assert "error" in data
 
+    def test_get_snapshot(self, client, mock_collector):
+        status, data = self._get(client, "/v1/api/cluster/snapshot")
+        assert status == 200
+        assert len(data["nodes"]) == 1
+        assert data["nodes"][0]["node_id"] == "node-a"
+        assert data["overview"]["nodes"] == 3
+        assert data["overview"]["workstreams"] == 15
+        assert data["timestamp"] == 1234567890.0
+        mock_collector.get_snapshot.assert_called_once()
+
     def test_health_endpoint(self, client, mock_collector):
         status, data = self._get(client, "/health")
         assert status == 200
@@ -1299,3 +1373,177 @@ class TestProxySharedStatic:
         resp = client.get("/node/unknown/shared/base.css")
         assert resp.status_code == 404
         client.close()
+
+
+# ---------------------------------------------------------------------------
+# SSE proxy — raw byte passthrough
+# ---------------------------------------------------------------------------
+
+
+class TestSSEProxy:
+    """Verify _proxy_sse forwards raw bytes including ping comments."""
+
+    def test_proxy_sse_preserves_pings_and_events(self):
+        """SSE proxy should forward ping comments and events verbatim."""
+        from turnstone.console.server import _proxy_sse
+
+        # Simulate an upstream SSE response with a ping comment and a real event
+        sse_payload = b': ping - 2026-03-08T12:00:00Z\n\nevent: message\ndata: {"type": "test"}\n\n'
+
+        class FakeResponse:
+            status_code = 200
+            headers = {"content-type": "text/event-stream"}
+
+            async def aiter_bytes(self):
+                yield sse_payload
+
+            async def aclose(self):
+                pass
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *args):
+                pass
+
+        class FakeClient:
+            def stream(self, method, url, **kwargs):
+                return FakeResponse()
+
+        class FakeRequest:
+            class url:  # noqa: N801
+                query = "ws_id=test123"
+
+            class app:  # noqa: N801
+                class state:  # noqa: N801
+                    proxy_sse_client = FakeClient()
+                    proxy_auth_token = ""
+
+            headers = {}
+
+            async def is_disconnected(self):
+                return False
+
+        async def _run():
+            response = await _proxy_sse(
+                FakeRequest(), "http://fake:8080", "events", api_prefix="v1/api"
+            )
+            assert response.media_type == "text/event-stream"
+            # Collect the streamed bytes
+            chunks: list[bytes] = []
+            async for chunk in response.body_iterator:
+                chunks.append(chunk if isinstance(chunk, bytes) else chunk.encode())
+            body = b"".join(chunks)
+            # Ping comment must be preserved (not filtered)
+            assert b": ping" in body
+            # Real event must be preserved
+            assert b"event: message" in body
+            assert b'"type": "test"' in body
+
+        asyncio.run(_run())
+
+    def test_proxy_sse_upstream_error_status(self):
+        """Non-200 upstream status should yield an error event."""
+
+        from turnstone.console.server import _proxy_sse
+
+        class FakeResponse:
+            status_code = 502
+
+            async def aiter_bytes(self):
+                return
+                yield  # make it an async generator
+
+            async def aclose(self):
+                pass
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *args):
+                pass
+
+        class FakeClient:
+            def stream(self, method, url, **kwargs):
+                return FakeResponse()
+
+        class FakeRequest:
+            class url:  # noqa: N801
+                query = ""
+
+            class app:  # noqa: N801
+                class state:  # noqa: N801
+                    proxy_sse_client = FakeClient()
+                    proxy_auth_token = ""
+
+            headers = {}
+
+            async def is_disconnected(self):
+                return False
+
+        async def _run():
+            response = await _proxy_sse(FakeRequest(), "http://fake:8080", "events")
+            chunks: list[bytes] = []
+            async for chunk in response.body_iterator:
+                chunks.append(chunk if isinstance(chunk, bytes) else chunk.encode())
+            body = b"".join(chunks)
+            assert b"event: error" in body
+            assert b"502" in body
+
+        asyncio.run(_run())
+
+    def test_proxy_sse_disconnect_handling(self):
+        """Proxy should stop when browser disconnects."""
+
+        from turnstone.console.server import _proxy_sse
+
+        class FakeResponse:
+            status_code = 200
+
+            async def aiter_bytes(self):
+                yield b"data: chunk1\n\n"
+                yield b"data: chunk2\n\n"  # should not be reached
+                yield b"data: chunk3\n\n"
+
+            async def aclose(self):
+                pass
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *args):
+                pass
+
+        class FakeClient:
+            def stream(self, method, url, **kwargs):
+                return FakeResponse()
+
+        call_count = 0
+
+        class FakeRequest:
+            class url:  # noqa: N801
+                query = ""
+
+            class app:  # noqa: N801
+                class state:  # noqa: N801
+                    proxy_sse_client = FakeClient()
+                    proxy_auth_token = ""
+
+            headers = {}
+
+            async def is_disconnected(self):
+                nonlocal call_count
+                call_count += 1
+                return call_count > 1  # disconnect after first chunk
+
+        async def _run():
+            response = await _proxy_sse(FakeRequest(), "http://fake:8080", "events")
+            chunks: list[bytes] = []
+            async for chunk in response.body_iterator:
+                chunks.append(chunk if isinstance(chunk, bytes) else chunk.encode())
+            body = b"".join(chunks)
+            assert b"chunk1" in body
+            # Should have stopped before chunk3
+            assert b"chunk3" not in body
+
+        asyncio.run(_run())

tests/test_providers.py

@@ -2049,3 +2049,138 @@ class TestModelCapabilitiesToolSearch:
 
         caps = ModelCapabilities()
         assert caps.supports_tool_search is False
+
+
+# ---------------------------------------------------------------------------
+# Vision support
+# ---------------------------------------------------------------------------
+
+
+class TestVisionCapabilities:
+    """Test supports_vision flag across providers."""
+
+    def test_default_is_false(self) -> None:
+        from turnstone.core.providers._protocol import ModelCapabilities
+
+        caps = ModelCapabilities()
+        assert caps.supports_vision is False
+
+    def test_openai_commercial_supports_vision(self) -> None:
+        provider = OpenAIProvider()
+        for model in ("gpt-5", "gpt-5-mini", "gpt-5.4", "o3", "o4-mini"):
+            caps = provider.get_capabilities(model)
+            assert caps.supports_vision is True, f"{model} should support vision"
+
+    def test_openai_default_no_vision(self) -> None:
+        """Unknown models (local servers) default to no vision."""
+        provider = OpenAIProvider()
+        caps = provider.get_capabilities("some-local-model")
+        assert caps.supports_vision is False
+
+    def test_anthropic_supports_vision(self) -> None:
+        from turnstone.core.providers._anthropic import AnthropicProvider
+
+        provider = AnthropicProvider()
+        for model in ("claude-opus-4-6", "claude-sonnet-4-6", "claude-haiku-4-5"):
+            caps = provider.get_capabilities(model)
+            assert caps.supports_vision is True, f"{model} should support vision"
+
+    def test_anthropic_default_supports_vision(self) -> None:
+        """Anthropic default (unknown Claude model) supports vision."""
+        from turnstone.core.providers._anthropic import AnthropicProvider
+
+        provider = AnthropicProvider()
+        caps = provider.get_capabilities("claude-unknown-9")
+        assert caps.supports_vision is True
+
+
+class TestAnthropicVisionConversion:
+    """Test image content conversion in _convert_messages."""
+
+    def setup_method(self) -> None:
+        from turnstone.core.providers._anthropic import AnthropicProvider
+
+        self.provider = AnthropicProvider()
+
+    def test_tool_result_with_image_content(self) -> None:
+        """Tool result with list content converts image_url to Anthropic image."""
+        messages = [
+            {"role": "user", "content": "Read this image"},
+            {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [
+                    {
+                        "id": "call_1",
+                        "function": {"name": "read_file", "arguments": '{"path": "img.png"}'},
+                    }
+                ],
+            },
+            {
+                "role": "tool",
+                "tool_call_id": "call_1",
+                "content": [
+                    {"type": "text", "text": "Image file: img.png (1024 bytes)"},
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": "data:image/png;base64,iVBORw0KGgo="},
+                    },
+                ],
+            },
+        ]
+        _, converted = self.provider._convert_messages(messages)
+        # Tool result should be in a user message
+        tool_user_msg = converted[2]
+        assert tool_user_msg["role"] == "user"
+        tool_result = tool_user_msg["content"][0]
+        assert tool_result["type"] == "tool_result"
+        assert tool_result["tool_use_id"] == "call_1"
+        # Content should be a list with converted image block
+        content = tool_result["content"]
+        assert isinstance(content, list)
+        assert content[0] == {"type": "text", "text": "Image file: img.png (1024 bytes)"}
+        assert content[1]["type"] == "image"
+        assert content[1]["source"]["type"] == "base64"
+        assert content[1]["source"]["media_type"] == "image/png"
+        assert content[1]["source"]["data"] == "iVBORw0KGgo="
+
+    def test_tool_result_with_string_content_unchanged(self) -> None:
+        """Tool result with plain string content is unchanged."""
+        messages = [
+            {"role": "user", "content": "Read file"},
+            {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [
+                    {
+                        "id": "call_2",
+                        "function": {"name": "read_file", "arguments": '{"path": "f.py"}'},
+                    }
+                ],
+            },
+            {
+                "role": "tool",
+                "tool_call_id": "call_2",
+                "content": "   1\tprint('hello')",
+            },
+        ]
+        _, converted = self.provider._convert_messages(messages)
+        tool_result = converted[2]["content"][0]
+        assert tool_result["content"] == "   1\tprint('hello')"
+
+    def test_convert_content_parts_static_method(self) -> None:
+        """_convert_content_parts handles both image_url and text."""
+        from turnstone.core.providers._anthropic import AnthropicProvider
+
+        parts = [
+            {"type": "text", "text": "description"},
+            {
+                "type": "image_url",
+                "image_url": {"url": "data:image/jpeg;base64,/9j/4AAQ"},
+            },
+        ]
+        result = AnthropicProvider._convert_content_parts(parts)
+        assert result[0] == {"type": "text", "text": "description"}
+        assert result[1]["type"] == "image"
+        assert result[1]["source"]["media_type"] == "image/jpeg"
+        assert result[1]["source"]["data"] == "/9j/4AAQ"

tests/test_session.py

@@ -1,9 +1,10 @@
 """Tests for turnstone.core.session — ChatSession construction."""
 
+import base64
 import json
 from unittest.mock import MagicMock, patch
 
-from turnstone.core.session import ChatSession
+from turnstone.core.session import _IMAGE_EXTENSIONS, _IMAGE_SIZE_CAP, ChatSession
 
 
 class NullUI:
@@ -265,3 +266,154 @@ class TestPlanExec:
         call_id, content, _ = self._run_plan(session, "do stuff", agent_return=agent_output)
         assert call_id == "test-call-1"
         assert content == agent_output
+
+
+# ---------------------------------------------------------------------------
+# Vision / image support
+# ---------------------------------------------------------------------------
+
+
+class TestImageExtensions:
+    """Test _IMAGE_EXTENSIONS constant and detection logic."""
+
+    def test_common_image_extensions(self):
+        for ext in (".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".tif", ".ico"):
+            assert ext in _IMAGE_EXTENSIONS, f"{ext} should be in _IMAGE_EXTENSIONS"
+
+    def test_svg_excluded(self):
+        assert ".svg" not in _IMAGE_EXTENSIONS
+
+    def test_text_extensions_excluded(self):
+        for ext in (".py", ".txt", ".json", ".md", ".rs", ".go"):
+            assert ext not in _IMAGE_EXTENSIONS
+
+
+class TestExecReadImage:
+    """Test _exec_read_image method."""
+
+    def _make_png(self, path: str, size: int = 100) -> None:
+        """Write a minimal valid-ish PNG header to a file."""
+        # 8-byte PNG signature + enough bytes to reach target size
+        header = b"\x89PNG\r\n\x1a\n"
+        with open(path, "wb") as f:
+            f.write(header + b"\x00" * max(0, size - len(header)))
+
+    def test_image_returns_content_parts(self, tmp_db, tmp_path):
+        """read_file on a PNG with vision support returns content parts."""
+        img = tmp_path / "test.png"
+        self._make_png(str(img))
+
+        session = _make_session()
+        # Mock provider to report vision support
+        mock_caps = MagicMock()
+        mock_caps.supports_vision = True
+        session._provider.get_capabilities = MagicMock(return_value=mock_caps)
+
+        item = {"call_id": "c1", "path": str(img), "offset": None, "limit": None}
+        call_id, output = session._exec_read_file(item)
+
+        assert call_id == "c1"
+        assert isinstance(output, list)
+        assert len(output) == 2
+        assert output[0]["type"] == "text"
+        assert "test.png" in output[0]["text"]
+        assert output[1]["type"] == "image_url"
+        url = output[1]["image_url"]["url"]
+        assert url.startswith("data:image/png;base64,")
+        # Verify base64 round-trip
+        b64part = url.split(",", 1)[1]
+        decoded = base64.b64decode(b64part)
+        assert decoded == img.read_bytes()
+
+    def test_no_vision_returns_text(self, tmp_db, tmp_path):
+        """read_file on image with non-vision model returns text description."""
+        img = tmp_path / "photo.jpg"
+        self._make_png(str(img), size=2048)
+
+        session = _make_session()
+        mock_caps = MagicMock()
+        mock_caps.supports_vision = False
+        session._provider.get_capabilities = MagicMock(return_value=mock_caps)
+
+        item = {"call_id": "c2", "path": str(img), "offset": None, "limit": None}
+        call_id, output = session._exec_read_file(item)
+
+        assert call_id == "c2"
+        assert isinstance(output, str)
+        assert "does not support vision" in output
+        assert "photo.jpg" in output
+
+    def test_oversized_image_returns_error(self, tmp_db, tmp_path):
+        """Images exceeding _IMAGE_SIZE_CAP return an error string."""
+        img = tmp_path / "huge.png"
+        # Write slightly over the cap
+        with open(img, "wb") as f:
+            f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * _IMAGE_SIZE_CAP)
+
+        session = _make_session()
+        mock_caps = MagicMock()
+        mock_caps.supports_vision = True
+        session._provider.get_capabilities = MagicMock(return_value=mock_caps)
+
+        item = {"call_id": "c3", "path": str(img), "offset": None, "limit": None}
+        call_id, output = session._exec_read_file(item)
+
+        assert call_id == "c3"
+        assert isinstance(output, str)
+        assert "exceeds" in output
+
+    def test_missing_image_returns_error(self, tmp_db, tmp_path):
+        """read_file on non-existent image returns error."""
+        session = _make_session()
+        mock_caps = MagicMock()
+        mock_caps.supports_vision = True
+        session._provider.get_capabilities = MagicMock(return_value=mock_caps)
+
+        item = {"call_id": "c4", "path": str(tmp_path / "nope.png"), "offset": None, "limit": None}
+        call_id, output = session._exec_read_file(item)
+        assert isinstance(output, str)
+        assert "not found" in output
+
+    def test_svg_read_as_text(self, tmp_db, tmp_path):
+        """SVG files are read as text, not as images."""
+        svg = tmp_path / "icon.svg"
+        svg.write_text('<svg xmlns="http://www.w3.org/2000/svg"><circle r="10"/></svg>')
+
+        session = _make_session()
+        item = {"call_id": "c5", "path": str(svg), "offset": None, "limit": None}
+        call_id, output = session._exec_read_file(item)
+        assert isinstance(output, str)
+        assert "<svg" in output  # Read as text
+
+
+class TestGetCapabilitiesOverride:
+    """Test _get_capabilities with config.toml overrides."""
+
+    def test_config_override_applies(self, tmp_db):
+        """capabilities dict from ModelConfig is merged onto provider caps."""
+        from turnstone.core.model_registry import ModelConfig, ModelRegistry
+        from turnstone.core.providers._protocol import ModelCapabilities
+
+        cfg = ModelConfig(
+            alias="qwen-vl",
+            base_url="http://localhost:8000/v1",
+            api_key="dummy",
+            model="qwen-3.5-vl",
+            capabilities={"supports_vision": True},
+        )
+        registry = ModelRegistry(
+            models={"qwen-vl": cfg},
+            default="qwen-vl",
+        )
+        session = _make_session(registry=registry, model_alias="qwen-vl")
+        # Ensure provider returns a real ModelCapabilities (not MagicMock)
+        session._provider.get_capabilities = MagicMock(return_value=ModelCapabilities())
+        caps = session._get_capabilities()
+        assert caps.supports_vision is True
+
+    def test_no_override_uses_provider_default(self, tmp_db):
+        """Without config override, provider defaults are used."""
+        session = _make_session()
+        caps = session._get_capabilities()
+        # Default OpenAI provider for unknown model → no vision
+        assert caps.supports_vision is False

turnstone/__init__.py

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

turnstone/api/console_schemas.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 from pydantic import BaseModel, Field
 
 # ---------------------------------------------------------------------------
@@ -48,7 +50,7 @@ class ClusterNodeInfo(BaseModel):
     total_tokens: int = 0
     started: float = 0.0
     reachable: bool = True
-    health: dict[str, str] = Field(default_factory=dict)
+    health: dict[str, Any] = Field(default_factory=dict)
     version: str = ""
 
 
@@ -91,12 +93,34 @@ class ClusterWorkstreamsResponse(BaseModel):
 class NodeDetailResponse(BaseModel):
     node_id: str
     server_url: str = ""
-    health: dict[str, str] = Field(default_factory=dict)
+    health: dict[str, Any] = Field(default_factory=dict)
     workstreams: list[ClusterWorkstreamInfo] = []
     aggregate: dict[str, int] = Field(default_factory=dict)
     reachable: bool = True
 
 
+# ---------------------------------------------------------------------------
+# Cluster snapshot
+# ---------------------------------------------------------------------------
+
+
+class ClusterSnapshotNode(BaseModel):
+    node_id: str
+    server_url: str = ""
+    max_ws: int = 10
+    reachable: bool = True
+    version: str = ""
+    health: dict[str, Any] = Field(default_factory=dict)
+    aggregate: dict[str, int] = Field(default_factory=dict)
+    workstreams: list[ClusterWorkstreamInfo] = []
+
+
+class ClusterSnapshotResponse(BaseModel):
+    nodes: list[ClusterSnapshotNode]
+    overview: ClusterOverviewResponse
+    timestamp: float = 0.0
+
+
 # ---------------------------------------------------------------------------
 # Workstream creation
 # ---------------------------------------------------------------------------

turnstone/api/console_spec.py

@@ -10,6 +10,7 @@ if TYPE_CHECKING:
 from turnstone.api.console_schemas import (
     ClusterNodesResponse,
     ClusterOverviewResponse,
+    ClusterSnapshotResponse,
     ClusterWorkstreamsResponse,
     ConsoleCreateWsRequest,
     ConsoleCreateWsResponse,
@@ -97,14 +98,23 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [
         error_codes=[400, 404, 503],
         tags=["Cluster"],
     ),
+    EndpointSpec(
+        "/v1/api/cluster/snapshot",
+        "GET",
+        "Full cluster state snapshot",
+        description="Returns the complete cluster state: all nodes with their workstreams "
+        "and overview aggregates. Used for initial load and reconnection.",
+        response_model=ClusterSnapshotResponse,
+        tags=["Cluster"],
+    ),
     # --- Streaming ---
     EndpointSpec(
         "/v1/api/cluster/events",
         "GET",
         "Cluster SSE event stream",
         description="Server-Sent Events stream for real-time cluster updates. "
-        "Returns text/event-stream with node_joined, node_lost, cluster_state, "
-        "ws_created, ws_closed, ws_rename events.",
+        "First event is a 'snapshot' with full cluster state, followed by "
+        "node_joined, node_lost, cluster_state, ws_created, ws_closed, ws_rename events.",
         tags=["Streaming"],
     ),
     # --- Auth ---
@@ -270,6 +280,7 @@ _ALL_MODELS: list[type[BaseModel]] = [
     ClusterNodesResponse,
     ClusterWorkstreamsResponse,
     NodeDetailResponse,
+    ClusterSnapshotResponse,
     ConsoleCreateWsRequest,
     ConsoleCreateWsResponse,
     ConsoleHealthResponse,

turnstone/console/collector.py

@@ -150,7 +150,7 @@ class ClusterCollector:
                         "state": "idle",
                         "node": node_id,
                         "server_url": node.server_url,
-                        "title": "",
+                        "title": data.get("title", ""),
                         "tokens": 0,
                         "context_ratio": 0.0,
                         "activity": "",
@@ -379,11 +379,11 @@ class ClusterCollector:
                 )
             total = len(items)
 
-        # Sort
+        # Sort (secondary key: node_id for stable ordering)
         if sort_by == "activity":
-            items.sort(key=lambda n: n["ws_running"] + n["ws_attention"], reverse=True)
+            items.sort(key=lambda n: (-(n["ws_running"] + n["ws_attention"]), n["node_id"]))
         elif sort_by == "tokens":
-            items.sort(key=lambda n: n["total_tokens"], reverse=True)
+            items.sort(key=lambda n: (-n["total_tokens"], n["node_id"]))
         elif sort_by == "name":
             items.sort(key=lambda n: n["node_id"])
 
@@ -455,6 +455,89 @@ class ClusterCollector:
                 "reachable": node.reachable,
             }
 
+    def get_snapshot(self) -> dict[str, Any]:
+        """Build a complete cluster snapshot under a single lock.
+
+        Returns everything the UI needs to render the full dashboard:
+        all nodes with their workstreams plus pre-computed overview aggregates.
+        """
+        with self._lock:
+            return self._build_snapshot_locked()
+
+    def get_snapshot_and_register(self, q: queue.Queue[dict[str, Any]]) -> dict[str, Any]:
+        """Build snapshot and register listener atomically.
+
+        Acquiring both locks ensures no event can be published between
+        the snapshot read and the listener registration — the client
+        receives the snapshot followed by every subsequent event with
+        no gap.
+        """
+        with self._lock:
+            snap = self._build_snapshot_locked()
+            with self._listeners_lock:
+                self._listeners.append(q)
+        return snap
+
+    def _build_snapshot_locked(self) -> dict[str, Any]:
+        """Build snapshot data — caller must hold ``_lock``."""
+        nodes_out = []
+        states: dict[str, int] = {
+            "running": 0,
+            "thinking": 0,
+            "attention": 0,
+            "idle": 0,
+            "error": 0,
+        }
+        total_tokens = 0
+        total_tool_calls = 0
+        total_ws = 0
+        versions: set[str] = set()
+
+        for node in self._nodes.values():
+            ws_list = []
+            for ws in node.workstreams.values():
+                ws_list.append(dict(ws))
+                s = ws.get("state", "idle")
+                states[s] = states.get(s, 0) + 1
+                total_ws += 1
+
+            total_tokens += node.aggregate.get("total_tokens", 0)
+            total_tool_calls += node.aggregate.get("total_tool_calls", 0)
+            ver = node.health.get("version", "")
+            if ver:
+                versions.add(ver)
+
+            nodes_out.append(
+                {
+                    "node_id": node.node_id,
+                    "server_url": node.server_url,
+                    "max_ws": node.max_ws,
+                    "reachable": node.reachable,
+                    "version": ver,
+                    "health": dict(node.health),
+                    "aggregate": dict(node.aggregate),
+                    "workstreams": ws_list,
+                }
+            )
+
+        node_count = len(self._nodes)
+
+        return {
+            "nodes": nodes_out,
+            "overview": {
+                "nodes": node_count,
+                "workstreams": total_ws,
+                "states": states,
+                "aggregate": {
+                    "total_tokens": total_tokens,
+                    "total_tool_calls": total_tool_calls,
+                },
+                "version_drift": len(versions) > 1,
+                "versions": sorted(versions),
+            },
+            "timestamp": time.time(),
+        }
+
     # -- SSE listener management ---------------------------------------------
 
     def register_listener(self, q: queue.Queue[dict[str, Any]]) -> None:

turnstone/console/server.py

@@ -30,7 +30,7 @@ import httpx
 from sse_starlette import EventSourceResponse
 from starlette.applications import Starlette
 from starlette.middleware import Middleware
-from starlette.responses import HTMLResponse, JSONResponse, Response
+from starlette.responses import HTMLResponse, JSONResponse, Response, StreamingResponse
 from starlette.routing import Mount, Route
 from starlette.staticfiles import StaticFiles
 
@@ -245,14 +245,25 @@ async def cluster_node_detail(request: Request) -> JSONResponse:
     return JSONResponse({"error": "Node not found"}, status_code=404)
 
 
+async def cluster_snapshot(request: Request) -> JSONResponse:
+    collector: ClusterCollector = request.app.state.collector
+    return JSONResponse(collector.get_snapshot())
+
+
 async def cluster_events_sse(request: Request) -> Response:
     collector: ClusterCollector = request.app.state.collector
     client_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=500)
-    collector.register_listener(client_queue)
 
     async def event_generator() -> AsyncGenerator[dict[str, str], None]:
         loop = asyncio.get_running_loop()
         try:
+            # Atomic snapshot+register — no event gap possible.
+            snap = await loop.run_in_executor(
+                None, collector.get_snapshot_and_register, client_queue
+            )
+            snap["type"] = "snapshot"
+            yield {"data": json.dumps(snap)}
+
             while True:
                 try:
                     event = await loop.run_in_executor(
@@ -568,7 +579,11 @@ async def _proxy_post(
 async def _proxy_sse(
     request: Request, server_url: str, path: str, *, api_prefix: str = "api"
 ) -> Response:
-    """Proxy an SSE stream from the target server to the browser."""
+    """Proxy an SSE stream from the target server to the browser.
+
+    Relays raw bytes verbatim so server-side ping comments, event framing,
+    and keepalives all pass through unchanged.
+    """
     target = f"{server_url}/{api_prefix}/{path}"
     if request.url.query:
         target += f"?{request.url.query}"
@@ -576,30 +591,38 @@ async def _proxy_sse(
     sse_client: httpx.AsyncClient = request.app.state.proxy_sse_client
     sse_auth = _proxy_auth_headers(request)
 
-    async def sse_generator() -> AsyncGenerator[dict[str, str], None]:
-        from httpx_sse import aconnect_sse
-
+    async def raw_stream() -> AsyncGenerator[bytes, None]:
         try:
-            async with aconnect_sse(sse_client, "GET", target, headers=sse_auth) as source:
-                if source.response.status_code != 200:
+            async with sse_client.stream(
+                "GET",
+                target,
+                headers={**sse_auth, "Accept": "text/event-stream", "Cache-Control": "no-store"},
+                timeout=httpx.Timeout(connect=10, read=None, write=5, pool=None),
+            ) as response:
+                if response.status_code != 200:
                     log.debug(
                         "SSE proxy received status %s from %s",
-                        source.response.status_code,
+                        response.status_code,
                         target,
                     )
-                    yield {
-                        "event": "error",
-                        "data": f"Upstream returned status {source.response.status_code}",
-                    }
+                    yield f"event: error\ndata: Upstream returned status {response.status_code}\n\n".encode()
                     return
-                async for sse in source.aiter_sse():
+                async for chunk in response.aiter_bytes():
                     if await request.is_disconnected():
                         return
-                    yield {"event": sse.event, "data": sse.data}
+                    yield chunk
         except httpx.HTTPError:
             log.debug("SSE proxy stream ended for %s", target)
 
-    return EventSourceResponse(sse_generator(), ping=5)
+    return StreamingResponse(
+        raw_stream(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-store",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )
 
 
 # ---------------------------------------------------------------------------
@@ -618,6 +641,7 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
     # Separate client for SSE streams — longer read timeout, shared connection pool
     app.state.proxy_sse_client = httpx.AsyncClient(
         timeout=httpx.Timeout(connect=5, read=30, write=5, pool=5),
+        limits=httpx.Limits(keepalive_expiry=30),
         headers=headers,
     )
     # Start scheduler if configured
@@ -1187,6 +1211,7 @@ def create_app(
                     Route("/api/cluster/workstreams", cluster_workstreams),
                     Route("/api/cluster/workstreams/new", create_workstream, methods=["POST"]),
                     Route("/api/cluster/node/{node_id}", cluster_node_detail),
+                    Route("/api/cluster/snapshot", cluster_snapshot),
                     Route("/api/cluster/events", cluster_events_sse),
                     Route("/api/auth/login", auth_login, methods=["POST"]),
                     Route("/api/auth/logout", auth_logout, methods=["POST"]),

turnstone/console/static/app.js

@@ -1,9 +1,6 @@
 // --- Shared hooks ---
 window.onLoginSuccess = function () {
   connectSSE();
-  if (currentView === "overview") loadOverview();
-  else if (currentView === "node") drillDownToNode(currentNodeId);
-  else if (currentView === "filtered") loadFilteredWorkstreams();
 };
 window.onLogout = function () {
   if (evtSource) {
@@ -33,6 +30,8 @@ var _lastOverviewJson = "";
 var _lastNodesJson = "";
 var evtSource = null;
 var retryDelay = 1000;
+var clusterState = null;
+var _navigatingFromPopstate = false;
 
 // --- Constants ---
 var STATE_DISPLAY = {
@@ -44,6 +43,236 @@ var STATE_DISPLAY = {
 };
 var STATE_ORDER = ["running", "thinking", "attention", "error", "idle"];
 
+// --- Cluster State Model ---
+function applySnapshot(data) {
+  clusterState = {
+    nodes: {},
+    overview: data.overview || {},
+    timestamp: data.timestamp || 0,
+  };
+  (data.nodes || []).forEach(function (n) {
+    clusterState.nodes[n.node_id] = n;
+  });
+  renderFromState();
+}
+
+function patchClusterState(data) {
+  if (!clusterState) return;
+  var t = data.type;
+  if (t === "cluster_state") {
+    var node = clusterState.nodes[data.node_id];
+    if (node) {
+      (node.workstreams || []).forEach(function (ws) {
+        if (ws.id === data.ws_id) {
+          if ("state" in data) ws.state = data.state;
+          if ("tokens" in data) ws.tokens = data.tokens;
+          if ("context_ratio" in data) ws.context_ratio = data.context_ratio;
+          if ("activity" in data) ws.activity = data.activity;
+          if ("activity_state" in data) ws.activity_state = data.activity_state;
+        }
+      });
+    }
+  } else if (t === "ws_created") {
+    var targetNode = clusterState.nodes[data.node_id];
+    if (targetNode) {
+      targetNode.workstreams = targetNode.workstreams || [];
+      targetNode.workstreams.push({
+        id: data.ws_id,
+        name: data.name || "",
+        state: "idle",
+        node: data.node_id,
+        server_url: targetNode.server_url || "",
+        title: data.title || "",
+        tokens: 0,
+        context_ratio: 0.0,
+        activity: "",
+        activity_state: "",
+        tool_calls: 0,
+      });
+    }
+  } else if (t === "ws_closed") {
+    Object.keys(clusterState.nodes).forEach(function (nid) {
+      var n = clusterState.nodes[nid];
+      n.workstreams = (n.workstreams || []).filter(function (ws) {
+        return ws.id !== data.ws_id;
+      });
+    });
+  } else if (t === "ws_rename") {
+    Object.keys(clusterState.nodes).forEach(function (nid) {
+      (clusterState.nodes[nid].workstreams || []).forEach(function (ws) {
+        if (ws.id === data.ws_id) ws.name = data.name || "";
+      });
+    });
+  } else if (t === "node_joined") {
+    if (!clusterState.nodes[data.node_id]) {
+      clusterState.nodes[data.node_id] = {
+        node_id: data.node_id,
+        server_url: "",
+        max_ws: 10,
+        reachable: true,
+        version: "",
+        health: {},
+        aggregate: {},
+        workstreams: [],
+      };
+    }
+  } else if (t === "node_lost") {
+    delete clusterState.nodes[data.node_id];
+  } else {
+    return;
+  }
+  scheduleRender();
+}
+
+var _renderTimer = null;
+function scheduleRender() {
+  if (_renderTimer) return;
+  _renderTimer = requestAnimationFrame(function () {
+    _renderTimer = null;
+    recomputeOverview();
+    renderFromState();
+  });
+}
+
+function recomputeOverview() {
+  if (!clusterState) return;
+  var states = { running: 0, thinking: 0, attention: 0, idle: 0, error: 0 };
+  var totalTokens = 0,
+    totalToolCalls = 0,
+    totalWs = 0;
+  var versions = {};
+  Object.keys(clusterState.nodes).forEach(function (nid) {
+    var node = clusterState.nodes[nid];
+    var nodeWsTokens = 0;
+    (node.workstreams || []).forEach(function (ws) {
+      var s = ws.state || "idle";
+      states[s] = (states[s] || 0) + 1;
+      totalWs++;
+      nodeWsTokens += ws.tokens || 0;
+    });
+    var aggTokens = (node.aggregate || {}).total_tokens || 0;
+    totalTokens += aggTokens || nodeWsTokens;
+    totalToolCalls += (node.aggregate || {}).total_tool_calls || 0;
+    if (node.version) versions[node.version] = true;
+  });
+  var versionList = Object.keys(versions).sort();
+  clusterState.overview = {
+    nodes: Object.keys(clusterState.nodes).length,
+    workstreams: totalWs,
+    states: states,
+    aggregate: {
+      total_tokens: totalTokens,
+      total_tool_calls: totalToolCalls,
+    },
+    version_drift: versionList.length > 1,
+    versions: versionList,
+  };
+}
+
+function buildNodeInfoFromSnapshot(node) {
+  var states = { running: 0, thinking: 0, attention: 0, idle: 0, error: 0 };
+  var ws = node.workstreams || [];
+  ws.forEach(function (w) {
+    var s = w.state || "idle";
+    states[s] = (states[s] || 0) + 1;
+  });
+  var aggTokens = (node.aggregate || {}).total_tokens || 0;
+  if (!aggTokens) {
+    ws.forEach(function (w) {
+      aggTokens += w.tokens || 0;
+    });
+  }
+  return {
+    node_id: node.node_id,
+    server_url: node.server_url || "",
+    ws_total: ws.length,
+    ws_running: states.running,
+    ws_thinking: states.thinking,
+    ws_attention: states.attention,
+    ws_idle: states.idle,
+    ws_error: states.error,
+    total_tokens: aggTokens,
+    ws_tokens: aggTokens,
+    max_ws: node.max_ws || 10,
+    started: node.started || 0,
+    reachable: node.reachable !== false,
+    health: node.health || {},
+    version: node.version || "",
+  };
+}
+
+function renderFromState() {
+  if (!clusterState) return;
+  renderStatusBar(clusterState.overview);
+  if (currentView === "overview") {
+    var nodesList = Object.keys(clusterState.nodes).map(function (nid) {
+      return buildNodeInfoFromSnapshot(clusterState.nodes[nid]);
+    });
+    nodesList.sort(function (a, b) {
+      var d = b.ws_running + b.ws_attention - (a.ws_running + a.ws_attention);
+      return d !== 0 ? d : a.node_id.localeCompare(b.node_id);
+    });
+    renderNodeGroups(nodesList, nodesList.length);
+    document.getElementById("cluster-summary").textContent =
+      clusterState.overview.nodes +
+      " nodes \u00b7 " +
+      formatCount(clusterState.overview.workstreams) +
+      " workstreams";
+  } else if (currentView === "node" && currentNodeId) {
+    var snapNode = clusterState.nodes[currentNodeId];
+    if (snapNode) {
+      var wsList = snapNode.workstreams || [];
+      var active = wsList.filter(function (w) {
+        return w.state !== "idle";
+      }).length;
+      document.getElementById("node-ws-summary").textContent =
+        active + " active \u00b7 " + wsList.length + " total";
+      renderWsTable(document.getElementById("node-ws-table"), wsList);
+    }
+  } else if (currentView === "filtered") {
+    var allWs = [];
+    Object.keys(clusterState.nodes).forEach(function (nid) {
+      (clusterState.nodes[nid].workstreams || []).forEach(function (ws) {
+        allWs.push(ws);
+      });
+    });
+    if (currentFilter.state) {
+      allWs = allWs.filter(function (ws) {
+        return ws.state === currentFilter.state;
+      });
+    }
+    if (currentFilter.node) {
+      allWs = allWs.filter(function (ws) {
+        return ws.node === currentFilter.node;
+      });
+    }
+    var stateOrder = {
+      running: 0,
+      thinking: 1,
+      attention: 2,
+      error: 3,
+      idle: 4,
+    };
+    allWs.sort(function (a, b) {
+      return (stateOrder[a.state] || 9) - (stateOrder[b.state] || 9);
+    });
+    var total = allWs.length;
+    var perPage = currentFilter.per_page || 50;
+    var pages = Math.max(1, Math.ceil(total / perPage));
+    var page = Math.min(currentFilter.page || 1, pages);
+    var start = (page - 1) * perPage;
+    var pageWs = allWs.slice(start, start + perPage);
+    document.getElementById("filtered-summary").textContent =
+      "Page " + page + " of " + pages + " (" + total + " total)";
+    renderWsTable(document.getElementById("filtered-ws-table"), pageWs);
+    renderPagination(
+      document.getElementById("filtered-pagination"),
+      page,
+      pages,
+    );
+  }
+}
+
 // --- SSE Connection ---
 function connectSSE() {
   if (evtSource) {
@@ -94,19 +323,11 @@ function connectSSE() {
   };
 }
 
-var _refreshTimer = null;
-function scheduleRefresh() {
-  if (_refreshTimer) return;
-  _refreshTimer = setTimeout(function () {
-    _refreshTimer = null;
-    if (currentView === "overview") loadOverview();
-    else if (currentView === "node" && currentNodeId)
-      loadNodeDetail(currentNodeId);
-    else if (currentView === "filtered") loadFilteredWorkstreams();
-  }, 250);
-}
-
 function handleClusterEvent(data) {
+  if (data.type === "snapshot") {
+    applySnapshot(data);
+    return;
+  }
   if (
     data.type === "cluster_state" ||
     data.type === "ws_created" ||
@@ -115,7 +336,7 @@ function handleClusterEvent(data) {
     data.type === "node_joined" ||
     data.type === "node_lost"
   ) {
-    scheduleRefresh();
+    patchClusterState(data);
   }
   if (data.type === "ws_closed" && data.reason === "evicted") {
     showToast("Evicted" + (data.name ? ": " + data.name : "") + " (capacity)");
@@ -135,28 +356,18 @@ function showOverview() {
   if (adminView) adminView.style.display = "none";
   document.getElementById("breadcrumb").style.display = "none";
   document.getElementById("main").scrollTop = 0;
-  loadOverview();
-  history.pushState({ view: "overview" }, "");
+  if (clusterState) renderFromState();
+  else loadOverview();
+  if (!_navigatingFromPopstate) history.pushState({ view: "overview" }, "");
 }
 
 function loadOverview() {
-  var overviewP = authFetch("/v1/api/cluster/overview").then(function (r) {
-    return r.json();
-  });
-  var nodesP = authFetch("/v1/api/cluster/nodes?sort=activity&limit=1000").then(
-    function (r) {
+  authFetch("/v1/api/cluster/snapshot")
+    .then(function (r) {
       return r.json();
-    },
-  );
-  Promise.all([overviewP, nodesP])
-    .then(function (res) {
-      renderStatusBar(res[0]);
-      renderNodeGroups(res[1].nodes, res[1].total);
-      document.getElementById("cluster-summary").textContent =
-        res[0].nodes +
-        " nodes \u00b7 " +
-        formatCount(res[0].workstreams) +
-        " workstreams";
+    })
+    .then(function (data) {
+      applySnapshot(data);
     })
     .catch(function () {
       document.getElementById("node-table").innerHTML =
@@ -310,7 +521,8 @@ function groupNodes(nodes) {
   });
   groupOrder.forEach(function (prefix) {
     groupMap[prefix].nodes.sort(function (a, b) {
-      return b.ws_running + b.ws_attention - (a.ws_running + a.ws_attention);
+      var d = b.ws_running + b.ws_attention - (a.ws_running + a.ws_attention);
+      return d !== 0 ? d : a.node_id.localeCompare(b.node_id);
     });
   });
   var groups = groupOrder.map(function (p) {
@@ -653,38 +865,37 @@ function drillDownToNode(nodeId, serverUrl) {
   link.href = "/node/" + encodeURIComponent(nodeId) + "/";
   link.style.display = "";
   document.getElementById("main").scrollTop = 0;
-  document.getElementById("node-ws-table").innerHTML =
-    '<div class="dashboard-empty">Loading workstreams...</div>';
-  loadNodeDetail(nodeId);
+  if (clusterState && clusterState.nodes[nodeId]) {
+    renderFromState();
+  } else {
+    document.getElementById("node-ws-table").innerHTML =
+      '<div class="dashboard-empty">Loading workstreams...</div>';
+    loadNodeDetail(nodeId);
+  }
   document.getElementById("breadcrumb-home").focus();
-  history.pushState({ view: "node", nodeId: nodeId, serverUrl: serverUrl }, "");
+  if (!_navigatingFromPopstate)
+    history.pushState(
+      { view: "node", nodeId: nodeId, serverUrl: serverUrl },
+      "",
+    );
 }
 
 function loadNodeDetail(nodeId) {
-  var detailP = authFetch(
-    "/v1/api/cluster/node/" + encodeURIComponent(nodeId),
-  ).then(function (r) {
-    return r.json();
-  });
-  var overviewP = authFetch("/v1/api/cluster/overview").then(function (r) {
-    return r.json();
-  });
-  Promise.all([detailP, overviewP]).then(function (res) {
-    var data = res[0];
-    renderStatusBar(res[1]);
-    if (data.error) {
+  authFetch("/v1/api/cluster/snapshot")
+    .then(function (r) {
+      return r.json();
+    })
+    .then(function (data) {
+      applySnapshot(data);
+      if (!clusterState || !clusterState.nodes[nodeId]) {
+        document.getElementById("node-ws-table").innerHTML =
+          '<div class="dashboard-empty">Node not found</div>';
+      }
+    })
+    .catch(function () {
       document.getElementById("node-ws-table").innerHTML =
-        '<div class="dashboard-empty">' + escapeHtml(data.error) + "</div>";
-      return;
-    }
-    var ws = data.workstreams || [];
-    var active = ws.filter(function (w) {
-      return w.state !== "idle";
-    }).length;
-    document.getElementById("node-ws-summary").textContent =
-      active + " active \u00b7 " + ws.length + " total";
-    renderWsTable(document.getElementById("node-ws-table"), ws);
-  });
+        '<div class="dashboard-empty">Failed to load</div>';
+    });
 }
 
 // --- Drill-down: Filtered ---
@@ -703,9 +914,11 @@ function drillDownByState(state) {
   document.getElementById("filtered-title").textContent =
     "WORKSTREAMS — " + sd.label.toUpperCase();
   document.getElementById("main").scrollTop = 0;
-  loadFilteredWorkstreams();
+  if (clusterState) renderFromState();
+  else loadFilteredWorkstreams();
   document.getElementById("breadcrumb-home").focus();
-  history.pushState({ view: "filtered", filter: currentFilter }, "");
+  if (!_navigatingFromPopstate)
+    history.pushState({ view: "filtered", filter: currentFilter }, "");
 }
 
 function drillDownByNode(nodeId) {
@@ -721,48 +934,20 @@ function drillDownByNode(nodeId) {
   document.getElementById("filtered-title").textContent =
     "WORKSTREAMS — " + nodeId;
   document.getElementById("main").scrollTop = 0;
-  loadFilteredWorkstreams();
+  if (clusterState) renderFromState();
+  else loadFilteredWorkstreams();
   document.getElementById("breadcrumb-home").focus();
-  history.pushState({ view: "filtered", filter: currentFilter }, "");
+  if (!_navigatingFromPopstate)
+    history.pushState({ view: "filtered", filter: currentFilter }, "");
 }
 
 function loadFilteredWorkstreams() {
-  var params =
-    "page=" + currentFilter.page + "&per_page=" + currentFilter.per_page;
-  if (currentFilter.state)
-    params += "&state=" + encodeURIComponent(currentFilter.state);
-  if (currentFilter.node)
-    params += "&node=" + encodeURIComponent(currentFilter.node);
-  var wsP = authFetch("/v1/api/cluster/workstreams?" + params).then(
-    function (r) {
+  authFetch("/v1/api/cluster/snapshot")
+    .then(function (r) {
       return r.json();
-    },
-  );
-  var overviewP = authFetch("/v1/api/cluster/overview").then(function (r) {
-    return r.json();
-  });
-  Promise.all([wsP, overviewP])
-    .then(function (res) {
-      var data = res[0];
-      renderStatusBar(res[1]);
-      document.getElementById("main").scrollTop = 0;
-      document.getElementById("filtered-summary").textContent =
-        "Page " +
-        data.page +
-        " of " +
-        data.pages +
-        " (" +
-        data.total +
-        " total)";
-      renderWsTable(
-        document.getElementById("filtered-ws-table"),
-        data.workstreams,
-      );
-      renderPagination(
-        document.getElementById("filtered-pagination"),
-        data.page,
-        data.pages,
-      );
+    })
+    .then(function (data) {
+      applySnapshot(data);
     })
     .catch(function () {
       document.getElementById("filtered-ws-table").innerHTML =
@@ -778,7 +963,8 @@ function renderPagination(container, page, pages) {
   prev.disabled = page <= 1;
   prev.onclick = function () {
     currentFilter.page--;
-    loadFilteredWorkstreams();
+    if (clusterState) renderFromState();
+    else loadFilteredWorkstreams();
   };
   container.appendChild(prev);
   var info = document.createElement("span");
@@ -789,7 +975,8 @@ function renderPagination(container, page, pages) {
   next.disabled = page >= pages;
   next.onclick = function () {
     currentFilter.page++;
-    loadFilteredWorkstreams();
+    if (clusterState) renderFromState();
+    else loadFilteredWorkstreams();
   };
   container.appendChild(next);
 }
@@ -923,19 +1110,24 @@ function renderWsTable(container, wsList) {
 window.addEventListener("popstate", function (e) {
   var overlay = document.getElementById("login-overlay");
   if (overlay && overlay.style.display !== "none") return;
-  if (!e.state) {
-    showOverview();
-    return;
-  }
-  if (e.state.view === "overview") showOverview();
-  else if (e.state.view === "admin" && typeof showAdmin === "function")
-    showAdmin();
-  else if (e.state.view === "node" && e.state.nodeId)
-    drillDownToNode(e.state.nodeId, e.state.serverUrl);
-  else if (e.state.view === "filtered" && e.state.filter) {
-    currentFilter = e.state.filter;
-    if (currentFilter.state) drillDownByState(currentFilter.state);
-    else if (currentFilter.node) drillDownByNode(currentFilter.node);
+  _navigatingFromPopstate = true;
+  try {
+    if (!e.state) {
+      showOverview();
+      return;
+    }
+    if (e.state.view === "overview") showOverview();
+    else if (e.state.view === "admin" && typeof showAdmin === "function")
+      showAdmin();
+    else if (e.state.view === "node" && e.state.nodeId)
+      drillDownToNode(e.state.nodeId, e.state.serverUrl);
+    else if (e.state.view === "filtered" && e.state.filter) {
+      currentFilter = e.state.filter;
+      if (currentFilter.state) drillDownByState(currentFilter.state);
+      else if (currentFilter.node) drillDownByNode(currentFilter.node);
+    }
+  } finally {
+    _navigatingFromPopstate = false;
   }
 });
 

turnstone/core/model_registry.py

@@ -33,6 +33,7 @@ class ModelConfig:
     model: str
     context_window: int = 131072
     provider: str = "openai"
+    capabilities: dict[str, Any] = field(default_factory=dict)
 
 
 # ---------------------------------------------------------------------------
@@ -185,6 +186,9 @@ def load_model_registry(
             model=model_name,
             context_window=entry.get("context_window", context_window),
             provider=entry.get("provider", "openai"),
+            capabilities=entry.get("capabilities", {})
+            if isinstance(entry.get("capabilities"), dict)
+            else {},
         )
 
     # Ensure a "default" entry from CLI args

turnstone/core/providers/_anthropic.py

@@ -75,6 +75,7 @@ _ANTHROPIC_DEFAULT = ModelCapabilities(
     token_param="max_tokens",
     thinking_mode="manual",
     supports_web_search=True,
+    supports_vision=True,
 )
 
 _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
@@ -87,6 +88,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
         effort_levels=("low", "medium", "high", "max"),
         supports_web_search=True,
         supports_tool_search=True,
+        supports_vision=True,
     ),
     "claude-sonnet-4-6": ModelCapabilities(
         context_window=200000,
@@ -97,6 +99,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
         effort_levels=("low", "medium", "high"),
         supports_web_search=True,
         supports_tool_search=True,
+        supports_vision=True,
     ),
     "claude-haiku-4-5": ModelCapabilities(
         context_window=200000,
@@ -104,6 +107,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
         token_param="max_tokens",
         thinking_mode="manual",
         supports_web_search=True,
+        supports_vision=True,
     ),
     "claude-sonnet-4-5": ModelCapabilities(
         context_window=200000,
@@ -111,6 +115,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
         token_param="max_tokens",
         thinking_mode="manual",
         supports_web_search=True,
+        supports_vision=True,
     ),
     "claude-opus-4-5": ModelCapabilities(
         context_window=200000,
@@ -120,6 +125,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
         supports_effort=True,
         effort_levels=("low", "medium", "high"),
         supports_web_search=True,
+        supports_vision=True,
     ),
     "claude-opus-4": ModelCapabilities(
         context_window=200000,
@@ -128,6 +134,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
         thinking_mode="manual",
         supports_web_search=True,
         supports_tool_search=True,
+        supports_vision=True,
     ),
     "claude-sonnet-4": ModelCapabilities(
         context_window=200000,
@@ -136,6 +143,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
         thinking_mode="manual",
         supports_web_search=True,
         supports_tool_search=True,
+        supports_vision=True,
     ),
 }
 
@@ -324,11 +332,15 @@ class AnthropicProvider:
                 tool_results: list[dict[str, Any]] = []
                 while i < len(messages) and messages[i]["role"] == "tool":
                     tool_msg = messages[i]
+                    content = tool_msg.get("content", "")
+                    # Convert image_url parts to Anthropic image format
+                    if isinstance(content, list):
+                        content = self._convert_content_parts(content)
                     tool_results.append(
                         {
                             "type": "tool_result",
                             "tool_use_id": tool_msg.get("tool_call_id", ""),
-                            "content": tool_msg.get("content", ""),
+                            "content": content,
                         }
                     )
                     i += 1
@@ -346,6 +358,43 @@ class AnthropicProvider:
 
         return "\n\n".join(system_parts), _merge_consecutive(converted)
 
+    @staticmethod
+    def _convert_content_parts(parts: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Convert OpenAI-format content parts to Anthropic format.
+
+        Transforms ``image_url`` parts (with ``data:`` URIs) to Anthropic's
+        ``image`` source blocks.  Text parts pass through unchanged.
+        """
+        converted: list[dict[str, Any]] = []
+        for part in parts:
+            if part.get("type") == "image_url":
+                url = part.get("image_url", {}).get("url", "")
+                if url.startswith("data:") and "," in url:
+                    # Parse "data:image/png;base64,<data>"
+                    header, _, b64data = url.partition(",")
+                    media_type = header.split(":", 1)[1].split(";", 1)[0]
+                    converted.append(
+                        {
+                            "type": "image",
+                            "source": {
+                                "type": "base64",
+                                "media_type": media_type,
+                                "data": b64data,
+                            },
+                        }
+                    )
+                else:
+                    # URL-based image — pass as Anthropic URL source
+                    converted.append(
+                        {
+                            "type": "image",
+                            "source": {"type": "url", "url": url},
+                        }
+                    )
+            else:
+                converted.append(part)
+        return converted
+
     # -- tool conversion -----------------------------------------------------
 
     def convert_tools(

turnstone/core/providers/_openai.py

@@ -30,6 +30,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         supports_temperature=False,
         reasoning_effort_values=("minimal", "low", "medium", "high"),
         default_reasoning_effort="medium",
+        supports_vision=True,
     ),
     "gpt-5-mini": ModelCapabilities(
         context_window=400000,
@@ -37,6 +38,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         supports_temperature=False,
         reasoning_effort_values=("minimal", "low", "medium", "high"),
         default_reasoning_effort="medium",
+        supports_vision=True,
     ),
     "gpt-5-nano": ModelCapabilities(
         context_window=400000,
@@ -44,6 +46,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         supports_temperature=False,
         reasoning_effort_values=("minimal", "low", "medium", "high"),
         default_reasoning_effort="medium",
+        supports_vision=True,
     ),
     # GPT-5 pro — high reasoning only, extended output
     "gpt-5-pro": ModelCapabilities(
@@ -52,6 +55,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         supports_temperature=False,
         reasoning_effort_values=("high",),
         default_reasoning_effort="high",
+        supports_vision=True,
     ),
     # GPT-5.1 — temperature OK when reasoning_effort=none (default)
     "gpt-5.1": ModelCapabilities(
@@ -59,6 +63,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         max_output_tokens=128000,
         reasoning_effort_values=("none", "low", "medium", "high"),
         default_reasoning_effort="none",
+        supports_vision=True,
     ),
     # GPT-5.2 — adds xhigh
     "gpt-5.2": ModelCapabilities(
@@ -66,6 +71,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         max_output_tokens=128000,
         reasoning_effort_values=("none", "low", "medium", "high", "xhigh"),
         default_reasoning_effort="none",
+        supports_vision=True,
     ),
     # GPT-5.2 pro — always-reasoning variant
     "gpt-5.2-pro": ModelCapabilities(
@@ -74,6 +80,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         supports_temperature=False,
         reasoning_effort_values=("medium", "high", "xhigh"),
         default_reasoning_effort="medium",
+        supports_vision=True,
     ),
     # GPT-5.3 — same capabilities as 5.2 (matches gpt-5.3-chat-latest, codex)
     "gpt-5.3": ModelCapabilities(
@@ -81,6 +88,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         max_output_tokens=128000,
         reasoning_effort_values=("none", "low", "medium", "high", "xhigh"),
         default_reasoning_effort="none",
+        supports_vision=True,
     ),
     # GPT-5.4 — 1M context window, native tool search
     "gpt-5.4": ModelCapabilities(
@@ -89,6 +97,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         reasoning_effort_values=("none", "low", "medium", "high", "xhigh"),
         default_reasoning_effort="none",
         supports_tool_search=True,
+        supports_vision=True,
     ),
     # GPT-5.4 pro — always-reasoning, 1M context, native tool search
     "gpt-5.4-pro": ModelCapabilities(
@@ -98,6 +107,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         reasoning_effort_values=("medium", "high", "xhigh"),
         default_reasoning_effort="medium",
         supports_tool_search=True,
+        supports_vision=True,
     ),
     # O-series reasoning models
     "o1": ModelCapabilities(
@@ -105,33 +115,39 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         max_output_tokens=100000,
         supports_temperature=False,
         supports_streaming=False,
+        supports_vision=True,
     ),
     "o1-mini": ModelCapabilities(
         context_window=128000,
         max_output_tokens=65536,
         supports_temperature=False,
         supports_streaming=False,
+        supports_vision=True,
     ),
     "o3": ModelCapabilities(
         context_window=200000,
         max_output_tokens=100000,
         supports_temperature=False,
+        supports_vision=True,
     ),
     "o3-mini": ModelCapabilities(
         context_window=200000,
         max_output_tokens=100000,
         supports_temperature=False,
+        supports_vision=True,
     ),
     "o3-pro": ModelCapabilities(
         context_window=200000,
         max_output_tokens=100000,
         supports_temperature=False,
         supports_streaming=False,
+        supports_vision=True,
     ),
     "o4-mini": ModelCapabilities(
         context_window=200000,
         max_output_tokens=100000,
         supports_temperature=False,
+        supports_vision=True,
     ),
     # Search models — always search on every request, no reasoning_effort
     "gpt-5-search-api": ModelCapabilities(
@@ -140,6 +156,7 @@ _OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
         supports_temperature=False,
         supports_web_search=True,
         reasoning_effort_values=(),
+        supports_vision=True,
     ),
 }
 

turnstone/core/providers/_protocol.py

@@ -77,6 +77,7 @@ class ModelCapabilities:
     default_reasoning_effort: str = "medium"
     supports_web_search: bool = False
     supports_tool_search: bool = False
+    supports_vision: bool = False
 
 
 def _lookup_capabilities(

turnstone/core/session.py

@@ -8,9 +8,12 @@ to receive events and handle approval prompts.
 
 from __future__ import annotations
 
+import base64
 import concurrent.futures
 import contextlib
+import dataclasses
 import json
+import mimetypes
 import os
 import re
 import signal
@@ -71,8 +74,21 @@ if TYPE_CHECKING:
 
     from turnstone.core.healthcheck import BackendHealthMonitor
     from turnstone.core.mcp_client import MCPClientManager
-    from turnstone.core.model_registry import ModelRegistry
-    from turnstone.core.providers import CompletionResult, LLMProvider, StreamChunk
+    from turnstone.core.model_registry import ModelConfig, ModelRegistry
+    from turnstone.core.providers import (
+        CompletionResult,
+        LLMProvider,
+        ModelCapabilities,
+        StreamChunk,
+    )
+
+# Image extensions handled as vision content (SVG excluded — it's XML text)
+_IMAGE_EXTENSIONS: frozenset[str] = frozenset(
+    {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".tif", ".ico"}
+)
+
+# 4 MB raw → ~5.3 MB base64, safely under Anthropic's per-block limit
+_IMAGE_SIZE_CAP: int = 4 * 1024 * 1024
 
 # ---------------------------------------------------------------------------
 # SessionUI protocol — the contract every frontend must implement
@@ -246,6 +262,18 @@ class ChatSession:
     def model_alias(self) -> str | None:
         return self._model_alias
 
+    def _get_capabilities(self) -> ModelCapabilities:
+        """Get model capabilities, applying config.toml overrides if present."""
+        caps = self._provider.get_capabilities(self.model)
+        if self._registry and self._model_alias:
+            cfg: ModelConfig = self._registry.get_config(self._model_alias)
+            if cfg.capabilities:
+                fields = {f.name for f in dataclasses.fields(type(caps))}
+                overrides = {k: v for k, v in cfg.capabilities.items() if k in fields}
+                if overrides:
+                    caps = dataclasses.replace(caps, **overrides)
+        return caps
+
     def _save_config(self) -> None:
         """Persist LLM-affecting config so resumed workstreams behave identically."""
         save_workstream_config(
@@ -505,7 +533,7 @@ class ChatSession:
             ]
         # Tool search hint (client-side mode only — native mode needs no hint)
         if self._tool_search:
-            caps = self._provider.get_capabilities(self.model)
+            caps = self._get_capabilities()
             if not caps.supports_tool_search:
                 dev_parts.append(
                     "\n\nAdditional tools are available via tool_search. "
@@ -564,7 +592,7 @@ class ChatSession:
         if not self._tool_search:
             return self._tools
         # Check if provider supports native tool search
-        caps = self._provider.get_capabilities(self.model)
+        caps = self._get_capabilities()
         if caps.supports_tool_search:
             # Provider handles defer_loading — send all tools
             return self._tools
@@ -576,7 +604,7 @@ class ChatSession:
         """Return names of deferred tools for native provider search, or None."""
         if not self._tool_search:
             return None
-        caps = self._provider.get_capabilities(self.model)
+        caps = self._get_capabilities()
         if not caps.supports_tool_search:
             return None  # Client-side mode — no deferred names for provider
         deferred = self._tool_search.get_deferred_tools()
@@ -746,13 +774,27 @@ class ChatSession:
                 # Map tool_call_id → tool name for logging
                 _tc_names = {c["id"]: c.get("function", {}).get("name", "") for c in tool_calls}
                 for tc_id, output in results:
-                    tool_msg = {
+                    tool_msg: dict[str, Any] = {
                         "role": "tool",
                         "tool_call_id": tc_id,
                         "content": output,
                     }
                     self.messages.append(tool_msg)
-                    self._msg_tokens.append(max(1, int(len(output) / self._chars_per_token)))
+
+                    # Token estimation — image content uses a fixed heuristic
+                    if isinstance(output, list):
+                        text_chars = sum(
+                            len(p.get("text", "")) for p in output if p.get("type") == "text"
+                        )
+                        image_count = sum(1 for p in output if p.get("type") == "image_url")
+                        tok_est = max(
+                            1,
+                            int(text_chars / self._chars_per_token) + image_count * 1000,
+                        )
+                    else:
+                        tok_est = max(1, int(len(output) / self._chars_per_token))
+                    self._msg_tokens.append(tok_est)
+
                     # Log tool result (skip memory tools to avoid noise)
                     _tname = _tc_names.get(tc_id, "")
                     if _tname not in (
@@ -760,10 +802,17 @@ class ChatSession:
                         "forget",
                         "recall",
                     ):
+                        # For image content, store text description only
+                        if isinstance(output, list):
+                            store_text = " ".join(
+                                p.get("text", "") for p in output if p.get("type") == "text"
+                            )[:2000]
+                        else:
+                            store_text = output[:2000]
                         save_message(
                             self._ws_id,
                             "tool_result",
-                            output[:2000],
+                            store_text,
                             _tname,
                             tool_call_id=tc_id,
                         )
@@ -1047,6 +1096,16 @@ class ChatSession:
             tool_calls = m.get("tool_calls")
             tc_id = m.get("tool_call_id")
 
+            # Flatten list content (image tool results) for display
+            if isinstance(content, list):
+                parts = []
+                for p in content:
+                    if p.get("type") == "text":
+                        parts.append(p.get("text", ""))
+                    elif p.get("type") == "image_url":
+                        parts.append("[image]")
+                content = " ".join(parts)
+
             # Truncate long content for readability
             if len(content) > 300:
                 display = content[:200] + f"...({len(content)} chars)..." + content[-50:]
@@ -1076,7 +1135,11 @@ class ChatSession:
 
     def _msg_char_count(self, msg: dict[str, Any]) -> int:
         """Count characters in a message, including tool call arguments."""
-        n = len(msg.get("content") or "")
+        content = msg.get("content")
+        if isinstance(content, list):
+            n = sum(len(p.get("text", "")) for p in content if p.get("type") == "text")
+        else:
+            n = len(content or "")
         for tc in msg.get("tool_calls", []):
             n += len(tc.get("function", {}).get("name", ""))
             n += len(tc.get("function", {}).get("arguments", ""))
@@ -1134,6 +1197,16 @@ class ChatSession:
             role = m["role"].upper()
             content = m.get("content") or ""
 
+            # Flatten list content (image tool results) to text for summary
+            if isinstance(content, list):
+                text_parts = []
+                for p in content:
+                    if p.get("type") == "text":
+                        text_parts.append(p["text"])
+                    elif p.get("type") == "image_url":
+                        text_parts.append("[image]")
+                content = " ".join(text_parts)
+
             if m.get("tool_calls"):
                 calls = []
                 for tc in m["tool_calls"]:
@@ -1321,7 +1394,7 @@ class ChatSession:
 
     def _execute_tools(
         self, tool_calls: list[dict[str, Any]]
-    ) -> tuple[list[tuple[str, str]], str | None]:
+    ) -> tuple[list[tuple[str, str | list[dict[str, Any]]]], str | None]:
         """Execute tool calls with batch preview and approval.
 
         Returns (results, user_feedback) where user_feedback is an optional
@@ -1343,12 +1416,14 @@ class ChatSession:
             user_feedback = None  # feedback is in the denial_msg
 
         # Phase 3: execute
-        def run_one(item: dict[str, Any]) -> tuple[str, str]:
+        def run_one(
+            item: dict[str, Any],
+        ) -> tuple[str, str | list[dict[str, Any]]]:
             if item.get("error"):
                 return item["call_id"], item["error"]
             if item.get("denied"):
                 return item["call_id"], item.get("denial_msg", "Denied by user")
-            result: tuple[str, str] = item["execute"](item)
+            result: tuple[str, str | list[dict[str, Any]]] = item["execute"](item)
             return result
 
         if len(items) == 1:
@@ -1366,6 +1441,7 @@ class ChatSession:
                 and not self.auto_approve
             ):
                 cid, output = results[i]
+                assert isinstance(output, str)  # plan always returns text
                 # Let the UI present the plan for review
                 self._emit_state("attention")
                 resp = self.ui.on_plan_review(output)
@@ -2210,13 +2286,18 @@ class ChatSession:
             self.ui.on_error(msg)
             return call_id, msg
 
-    def _exec_read_file(self, item: dict[str, Any]) -> tuple[str, str]:
-        """Read a file and return numbered lines, optionally sliced."""
+    def _exec_read_file(self, item: dict[str, Any]) -> tuple[str, str | list[dict[str, Any]]]:
+        """Read a file and return numbered lines, or image content parts."""
         call_id, path = item["call_id"], item["path"]
         offset = item.get("offset")  # 1-based, or None
         limit = item.get("limit")  # max lines, or None
         resolved = os.path.realpath(path)
 
+        # Image file detection (branch before text open)
+        ext = os.path.splitext(path)[1].lower()
+        if ext in _IMAGE_EXTENSIONS:
+            return self._exec_read_image(call_id, path, resolved)
+
         try:
             with open(path) as f:
                 all_lines = f.readlines()
@@ -2251,6 +2332,61 @@ class ChatSession:
 
         return call_id, output if output else "(empty file)"
 
+    def _exec_read_image(
+        self, call_id: str, path: str, resolved: str
+    ) -> tuple[str, str | list[dict[str, Any]]]:
+        """Read an image file and return as base64 content parts for vision."""
+        caps = self._get_capabilities()
+        if not caps.supports_vision:
+            try:
+                size = os.path.getsize(path)
+            except OSError as e:
+                self._read_files.discard(resolved)
+                return call_id, f"Error: {path}: {e}"
+            self._read_files.add(resolved)
+            desc = f"image (no vision, {size:,} bytes)"
+            self.ui.on_tool_result(call_id, "read_file", desc)
+            return call_id, (
+                f"Binary image file: {path} ({size:,} bytes). "
+                "Current model does not support vision."
+            )
+
+        try:
+            with open(path, "rb") as f:
+                raw = f.read()
+        except FileNotFoundError:
+            self._read_files.discard(resolved)
+            return call_id, f"Error: {path} not found"
+        except Exception as e:
+            self._read_files.discard(resolved)
+            return call_id, f"Error reading {path}: {e}"
+
+        if len(raw) > _IMAGE_SIZE_CAP:
+            self._read_files.discard(resolved)
+            size_mb = len(raw) / (1024 * 1024)
+            cap_mb = _IMAGE_SIZE_CAP / (1024 * 1024)
+            return call_id, (
+                f"Error: image {path} is {size_mb:.1f} MB, "
+                f"exceeds {cap_mb:.0f} MB limit for vision."
+            )
+
+        self._read_files.add(resolved)
+        b64data = base64.b64encode(raw).decode("ascii")
+        mime, _ = mimetypes.guess_type(path)
+        if not mime:
+            mime = "image/png"
+
+        content_parts: list[dict[str, Any]] = [
+            {"type": "text", "text": f"Image file: {path} ({len(raw):,} bytes)"},
+            {
+                "type": "image_url",
+                "image_url": {"url": f"data:{mime};base64,{b64data}"},
+            },
+        ]
+
+        self.ui.on_tool_result(call_id, "read_file", f"image ({len(raw):,} bytes)")
+        return call_id, content_parts
+
     def _exec_search(self, item: dict[str, Any]) -> tuple[str, str]:
         """Search file contents for a regex pattern using grep."""
         call_id = item["call_id"]

turnstone/eval.py

@@ -267,7 +267,15 @@ class HeadlessSession(ChatSession):
             with _suppress_stdout():
                 results, _ = self._execute_tools(assistant_msg["tool_calls"])
 
-            for tc, (tc_id, output) in zip(assistant_msg["tool_calls"], results, strict=False):
+            for tc, (tc_id, raw_output) in zip(assistant_msg["tool_calls"], results, strict=False):
+                # Flatten list content (image tool results) to text for logging
+                if isinstance(raw_output, list):
+                    output = " ".join(
+                        p.get("text", "[image]") if p.get("type") == "text" else "[image]"
+                        for p in raw_output
+                    )
+                else:
+                    output = raw_output
                 func_name = tc["function"]["name"]
                 args: dict[str, Any]
                 try:
@@ -302,7 +310,7 @@ class HeadlessSession(ChatSession):
                 tool_msg = {
                     "role": "tool",
                     "tool_call_id": tc_id,
-                    "content": output,
+                    "content": raw_output,
                 }
                 self.messages.append(tool_msg)
                 self._msg_tokens.append(max(1, int(len(output) / self._chars_per_token)))

turnstone/sdk/console.py

@@ -16,6 +16,7 @@ from typing import TYPE_CHECKING, Any
 from turnstone.api.console_schemas import (
     ClusterNodesResponse,
     ClusterOverviewResponse,
+    ClusterSnapshotResponse,
     ClusterWorkstreamsResponse,
     ConsoleCreateWsResponse,
     ConsoleHealthResponse,
@@ -102,6 +103,11 @@ class AsyncTurnstoneConsole(_BaseClient):
             "GET", f"/v1/api/cluster/node/{node_id}", response_model=NodeDetailResponse
         )
 
+    async def snapshot(self) -> ClusterSnapshotResponse:
+        return await self._request(
+            "GET", "/v1/api/cluster/snapshot", response_model=ClusterSnapshotResponse
+        )
+
     async def create_workstream(
         self,
         *,
@@ -342,6 +348,9 @@ class TurnstoneConsole:
     def node_detail(self, node_id: str) -> NodeDetailResponse:
         return self._runner.run(self._async.node_detail(node_id))
 
+    def snapshot(self) -> ClusterSnapshotResponse:
+        return self._runner.run(self._async.snapshot())
+
     def create_workstream(
         self,
         *,

turnstone/sdk/events.py

@@ -247,6 +247,14 @@ class ClusterWsRenameEvent(ClusterEvent):
     name: str = ""
 
 
+@dataclass
+class ClusterSnapshotEvent(ClusterEvent):
+    type: str = "snapshot"
+    nodes: list[dict[str, Any]] = field(default_factory=list)
+    overview: dict[str, Any] = field(default_factory=dict)
+    timestamp: float = 0.0
+
+
 # ---------------------------------------------------------------------------
 # Type registries (built after all classes are defined)
 # ---------------------------------------------------------------------------
@@ -296,5 +304,6 @@ _CLUSTER_REGISTRY: dict[str, type[ClusterEvent]] = {
         ClusterWsCreatedEvent,
         ClusterWsClosedEvent,
         ClusterWsRenameEvent,
+        ClusterSnapshotEvent,
     ]
 }

turnstone/tools/read_file.json

@@ -1,6 +1,6 @@
 {
   "name": "read_file",
-  "description": "Read the contents of a file. Returns numbered lines. Must be called before edit_file on the same path.",
+  "description": "Read the contents of a file. Returns numbered lines for text files. For image files (PNG, JPEG, GIF, WebP, BMP, TIFF, ICO), returns the image content if the model supports vision, or a text description otherwise. The offset and limit parameters apply to text files only.",
   "parameters": {
     "type": "object",
     "properties": {