feat(t3,views,server,watch): t3 code orchestration protocol 2

README.md

@@ -1,6 +1,6 @@
 # t3code-mcp
 
-An MCP connector to [T3 Code](https://github.com/pingdotgg/t3code) servers. It fills the one gap T3 Code leaves for agents: they cannot create their own threads. With this connector an agent dispatches a coding thread on any machine that runs T3 Code, waits for it, answers its questions and reads the result - all through T3's HTTP API, with no shell or terminal involved.
+An MCP connector to [T3 Code](https://github.com/pingdotgg/t3code) servers. It fills the one gap T3 Code leaves for agents: they cannot create their own threads. With this connector an agent dispatches a coding thread on any machine that runs T3 Code, waits for it, answers its questions and reads the result - all through T3's own API (orchestration protocol 2: reads over HTTP, commands over the WebSocket RPC), with no shell or terminal involved.
 
 Built for the dispatcher in [beaver-agent](https://git.kotikot.com/beaver/beaver-agent); works with any MCP client.
 
@@ -10,11 +10,11 @@ Built for the dispatcher in [beaver-agent](https://git.kotikot.com/beaver/beaver
 |---|---|
 | `t3_machines()` | machines from the config, which are online now, their project allowlists |
 | `t3_projects(machine)` | allowlisted projects on a machine: path, default model, thread count |
-| `t3_dispatch(machine, project, prompt, title?, model?, thread_id?, watch?)` | creates a thread and starts a turn (or starts a turn in an existing thread); returns `thread_id` at once; `watch=false` starts it without reporting events |
-| `t3_thread(thread_id, turns?)` | the state without waiting: turn, session, last messages, tools, files, a pending question |
+| `t3_dispatch(machine, project, prompt, title?, model?, thread_id?, watch?)` | creates a thread and starts a run (or starts a run in an existing thread); returns `thread_id` at once; `watch=false` starts it without reporting events |
+| `t3_thread(thread_id, turns?)` | the state without waiting: latest run, session, messages of the last runs, tools, files, a pending question |
 | `t3_wait(thread_id, timeout?)` | blocks until completed, interrupted, failed, a question, or the timeout |
 | `t3_answer(thread_id, answers)` | answers a question the thread is waiting on (question id → label; a list for multi-select) |
-| `t3_interrupt(thread_id)` | interrupts the running turn |
+| `t3_interrupt(thread_id)` | interrupts the running run |
 | `t3_watch(thread_id?, enabled?)` | without arguments lists the watched and unwatched threads; with both turns a thread's events on or off |
 
 `t3_wait` is an ordinary long call, not an MCP Task: Claude Code moves calls over two minutes into the background by itself (`CLAUDE_CODE_MCP_AUTO_BACKGROUND_MS`). Threads run in `full-access` without approvals. A thread is looked up across all machines, so the connector may restart in between.
@@ -36,7 +36,7 @@ model = "claudeAgent/claude-opus-5"
 options = { effort = "high", contextWindow = "1m" }
 ```
 
-Tokens never live in the file: `token_env` names the variable, and the token is issued on each machine with `t3 auth session issue --token-only --ttl 365d --label beaver`. `projects` are fnmatch patterns against a project's title and workspace root; a project that matches nothing does not exist for the client. The server listens on `T3CODE_MCP_HOST` / `T3CODE_MCP_PORT` (8000), MCP at `/mcp`, health at `/healthz`.
+Tokens never live in the file: `token_env` names the variable, and the token is issued on each machine with `t3 auth session issue --scope orchestration:read --scope orchestration:operate --token-only --ttl 365d --label beaver` - nothing beyond those two scopes is used. T3 Code from 0.1.0 (orchestration protocol 2) is required; older servers do not answer. `projects` are fnmatch patterns against a project's title and workspace root; a project that matches nothing does not exist for the client. The server listens on `T3CODE_MCP_HOST` / `T3CODE_MCP_PORT` (8000), MCP at `/mcp`, health at `/healthz`.
 
 ## Develop
 

docs/PROGRESS.md

@@ -2,6 +2,14 @@
 
 > Что сделано, что нет, что проверить руками. Свежие записи сверху. Архитектурная правда - `архитектура.md` §5 (vault, симлинк в `../../архитектура.md`), журнал всей второй итерации - `../../PROGRESS.md`.
 
+## 2026-10-16 - T3 Code 0.1.0, orchestration protocol 2
+
+T3 Code 0.1.0 (desktop на маке и `t3` на dell-dev) перешёл на протокол оркестрации 2, и коннектор ослеп: WS отвечал 426, HTTP без заголовка - 400, а `POST /api/orchestration/dispatch` удалён совсем. Сделано: каждый HTTP-запрос несёт `x-t3-orchestration-protocol: 2`, WS - `orchestrationProtocol=2` в URL; команды идут через WS RPC `orchestration.dispatchCommand` (одно соединение на вызов, ответ - `Exit`). Команды v2: `thread.create` + `message.dispatch` (`dispatchMode: start_immediately`, `createdBy: agent`, `creationSource: mcp`), `run.interrupt` по id активного run, `runtime-request.respond` для `t3_answer`. Тред теперь проекция (`runs`, `messages`, `turnItems`, `runtimeRequests`, `checkpoints`, `providerTurns`): состояние - статус последнего run, вопрос - pending `runtimeRequest` с `kind: user_input` и его `user_input_request` из `turnItems`, ошибка - `error`-item упавшего run или `lastError` сессии провайдера. В выдаче `turn_id` стал `run_id`, `turns` у `t3_thread` считает run'ы. Watcher: элементы потока `thread.updated`, строка shell с `status`/`lastError`/`pendingRuntimeRequest`; `failed`/`cancelled`/`rolled_back` - терминальные; курсор хранится под ключом `shell_sequences`, старый v1-курсор игнорируется. Токены перевыпущены с минимумом скоупов `orchestration:read` + `orchestration:operate`, метка `beaver-t3code-mcp-v2`.
+
+Проверено живьём локальным коннектором: мак, `t3-smoke` - диспатч → `completed` за 10 с, продолжение треда с AskUserQuestion → `t3_wait` вернул вопрос → `t3_answer` → `completed`, долгий run → `t3_interrupt` → `interrupted` за 5 с; события `completed`/`question`/`interrupted` в логе watcher'а. dell-dev, `projects` - диспатч → `completed` за 10 с. 15 тестов на фейковом T3 v2.
+
+Не сделано: `t3_wait` тянет полную проекцию треда каждые 5 с (у длинных тредов сотни КБ) - если станет тяжело, перейти на `orchestration.subscribeThread`; аппрувы по-прежнему не отвечаются (в `full-access` их нет).
+
 ## 2026-09-01 - подписка на события треда управляется из диспетчера
 
 Сделано: у `Tracked` появился флаг `watch` (по умолчанию `true`, лежит в том же `T3CODE_MCP_STATE`); `Tracker.apply()` при `watch=false` состояние треда обновляет как раньше, но события гасит - значит после обратной подписки не прилетит задним числом переход, случившийся в тишине. Тулза `t3_watch(thread_id?, enabled?)`: без аргументов - списки `watched` / `unwatched` (тред, машина, проект, заголовок, состояние), с обоими - снять или вернуть подписку; неизвестный тред - `ToolError`. У `t3_dispatch` параметр `watch` (по умолчанию `true`) - чтобы не ловить гонку на быстрых тредах; каждый диспатч в тред флаг перезаписывает, продолжение треда через `thread_id=` тоже. Загрузка стейта теперь игнорирует незнакомые поля - откат образа на версию без `watch` не уронит `Tracker`. 15 тестов.

src/t3code_mcp/server.py

@@ -14,6 +14,7 @@ from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Annotated, Any
 
 import httpx
+import websockets
 from fastmcp import FastMCP
 from fastmcp.exceptions import ToolError
 from pydantic import Field
@@ -30,16 +31,17 @@ if TYPE_CHECKING:
 
     from starlette.requests import Request
 
+    from t3code_mcp.t3 import RpcCall
+
 TITLE_MAX = 80
 WAIT_MAX = 3600.0
-PENDING_EVERY = 6
-STALLED_POLLS = 3
 
 
 @dataclass
 class Registry:
     machines: dict[str, Machine]
     transport: httpx.AsyncBaseTransport | None = None
+    rpc: Callable[[str], RpcCall] | None = None
     clients: dict[str, T3Client] = field(default_factory=dict)
     threads: dict[str, str] = field(default_factory=dict)
     tracker: Tracker = field(default_factory=Tracker)
@@ -55,7 +57,10 @@ class Registry:
         machine = self.machine(name)
         if name not in self.clients:
             self.clients[name] = T3Client(
-                machine.url, machine.token, transport=self.transport
+                machine.url,
+                machine.token,
+                transport=self.transport,
+                rpc=self.rpc(name) if self.rpc else None,
             )
         return self.clients[name]
 
@@ -67,7 +72,7 @@ class Registry:
         )
         names = [known] if known in self.machines else list(self.machines)
         for name in names:
-            detail = await self.client(name).thread(thread_id, turn_limit=1)
+            detail = await self.client(name).thread(thread_id)
             if detail is not None:
                 self.threads[thread_id] = name
                 return name, detail
@@ -90,7 +95,7 @@ def _title(prompt: str, title: str | None) -> str:
 def _find_project(
     machine: Machine, projects: list[dict[str, Any]], key: str
 ) -> dict[str, Any]:
-    allowed = [p for p in projects if not p.get("deletedAt") and machine.allows(p)]
+    allowed = [p for p in projects if machine.allows(p)]
     for project in allowed:
         if key in (project["id"], project["title"], project["workspaceRoot"]):
             return project
@@ -107,9 +112,7 @@ def _project_view(
         for t in threads
         if t["projectId"] == project["id"] and not t.get("archivedAt")
     ]
-    running = sum(
-        1 for t in own if (t.get("latestTurn") or {}).get("state") == "running"
-    )
+    running = sum(1 for t in own if views.shell_state(t) in BUSY)
     default = project.get("defaultModelSelection")
     return {
         "id": project["id"],
@@ -201,7 +204,7 @@ def build_server(
         return [
             _project_view(p, shell["threads"])
             for p in shell["projects"]
-            if not p.get("deletedAt") and spec.allows(p)
+            if spec.allows(p)
         ]
 
     @mcp.tool
@@ -273,12 +276,12 @@ def build_server(
             if existing is None or existing["projectId"] != target["id"]:
                 msg = f"thread {thread_id} is not in {target['title']} on {machine}"
                 raise ToolError(msg)
-            if (existing.get("latestTurn") or {}).get("state") == "running":
+            if views.shell_state(existing) in BUSY:
                 msg = f"thread {thread_id} is running; t3_wait or t3_interrupt first"
                 raise ToolError(msg)
             tid = thread_id
             await _call(
-                client.dispatch(t3.turn_start(tid, text, selection.selection()))
+                client.dispatch(t3.message_dispatch(tid, text, selection.selection()))
             )
         else:
             tid = t3.new_id()
@@ -287,7 +290,9 @@ def build_server(
                     t3.thread_create(tid, target["id"], heading, selection.selection())
                 )
             )
-            await _call(client.dispatch(t3.turn_start(tid, text, title_seed=heading)))
+            await _call(
+                client.dispatch(t3.message_dispatch(tid, text, title_seed=heading))
+            )
         registry.threads[tid] = machine
         registry.tracker.track(tid, machine, target["title"], heading, watch=watch)
         return {
@@ -338,24 +343,22 @@ def build_server(
     async def t3_thread(
         thread_id: Annotated[str, Field(description="thread_id from t3_dispatch")],
         turns: Annotated[
-            int, Field(description="How many recent user turns to include", ge=1, le=20)
+            int,
+            Field(
+                description="How many recent runs (user turns) to include", ge=1, le=20
+            ),
         ] = 2,
     ) -> dict[str, Any]:
         """State of a thread without waiting.
 
         Turn state, last messages, tools run, files changed, pending questions.
         """
-        machine, _ = await registry.locate(thread_id)
-        client = registry.client(machine)
-        detail = await _call(client.thread(thread_id, turn_limit=turns))
-        shell = await _call(client.shell())
-        if detail is None:
-            msg = f"thread {thread_id} vanished from {machine}"
-            raise ToolError(msg)
+        machine, detail = await registry.locate(thread_id)
+        shell = await _call(registry.client(machine).shell())
         summary = views.summary(
-            machine, detail, project=views.project_title(shell, detail)
+            machine, detail, project=views.project_title(shell, detail), turns=turns
         )
-        summary["pending"] = views.pending_in(shell, thread_id)
+        summary["pending"] = views.pending_detail(detail["projection"])
         return summary
 
     @mcp.tool
@@ -378,23 +381,13 @@ def build_server(
         machine, detail = await registry.locate(thread_id)
         client = registry.client(machine)
         started = time.monotonic()
-        polls = 0
-        pending: dict[str, bool] | None = None
-        project: str | None = None
-        stalled = 0
+        project = views.project_title(await _call(client.shell()), detail)
         while True:
-            thread = detail["thread"]
-            state = views.state(thread)
-            session = thread.get("session") or {}
-            status = session.get("status")
-            if polls % PENDING_EVERY == 0:
-                shell = await _call(client.shell())
-                pending = views.pending_in(shell, thread_id)
-                project = views.project_title(shell, detail)
-            failed = status == "error" or bool(session.get("lastError"))
-            stalled = stalled + 1 if state == "starting" and status == "stopped" else 0
-            busy = (state in BUSY or status == "starting") and stalled < STALLED_POLLS
-            done = not busy or failed or pending is not None
+            projection = detail["projection"]
+            state = views.state(projection)
+            pending = views.pending_detail(projection)
+            stuck = views.latest_run(projection) is None and views.error(projection)
+            done = state not in BUSY or pending is not None or stuck
             timed_out = time.monotonic() - started >= timeout
             if done or timed_out:
                 summary = views.summary(machine, detail, project=project)
@@ -403,8 +396,7 @@ def build_server(
                 summary["waited"] = round(time.monotonic() - started, 1)
                 return summary
             await asyncio.sleep(poll)
-            polls += 1
-            refreshed = await _call(client.thread(thread_id, turn_limit=1))
+            refreshed = await _call(client.thread(thread_id))
             if refreshed is None:
                 msg = f"thread {thread_id} vanished from {machine}"
                 raise ToolError(msg)
@@ -425,7 +417,7 @@ def build_server(
     ) -> dict[str, Any]:
         """Answer the pending question of a thread; its turn resumes."""
         machine, detail = await registry.locate(thread_id)
-        pending = views.question(detail)
+        pending = views.question(detail["projection"])
         if pending is None or not pending.get("request_id"):
             msg = f"thread {thread_id} has no pending question"
             raise ToolError(msg)
@@ -436,7 +428,9 @@ def build_server(
             raise ToolError(msg)
         await _call(
             registry.client(machine).dispatch(
-                t3.user_input_respond(thread_id, pending["request_id"], dict(answers))
+                t3.runtime_request_respond(
+                    thread_id, pending["request_id"], dict(answers)
+                )
             )
         )
         return {
@@ -452,17 +446,15 @@ def build_server(
     ) -> dict[str, Any]:
         """Interrupt the running turn; continue later via t3_dispatch(thread_id=...)."""
         machine, detail = await registry.locate(thread_id)
-        latest = detail["thread"].get("latestTurn") or {}
-        if latest.get("state") != "running":
+        latest = views.latest_run(detail["projection"])
+        if latest is None or latest["status"] not in BUSY:
             return {
                 "thread_id": thread_id,
                 "machine": machine,
-                "state": latest.get("state", "idle"),
+                "state": latest["status"] if latest else "idle",
             }
         await _call(
-            registry.client(machine).dispatch(
-                t3.turn_interrupt(thread_id, latest.get("turnId"))
-            )
+            registry.client(machine).dispatch(t3.run_interrupt(thread_id, latest["id"]))
         )
         return {"thread_id": thread_id, "machine": machine, "state": "interrupting"}
 
@@ -474,6 +466,6 @@ async def _call(awaitable: Any) -> Any:
         return await awaitable
     except T3Error as exc:
         raise ToolError(str(exc)) from exc
-    except httpx.HTTPError as exc:
+    except (httpx.HTTPError, websockets.WebSocketException, OSError) as exc:
         msg = f"T3 unreachable: {exc}"
         raise ToolError(msg) from exc

src/t3code_mcp/t3.py

@@ -1,29 +1,45 @@
-"""HTTP client for one T3 Code server (`packages/contracts/src/environmentHttp.ts`).
+"""Client for one T3 Code server, orchestration protocol 2.
 
-Four routes are used: the unauthenticated descriptor, the shell read-model
-(projects + threads without bodies), one thread with a turn window, and
-`dispatch` with a `ClientOrchestrationCommand`. The WebSocket RPC
-(`terminal.*` and friends) is deliberately not wrapped.
+Reads go over HTTP (`packages/contracts/src/environmentHttp.ts`): the
+unauthenticated descriptor, the shell read-model (projects + thread rows)
+and one thread projection. Writes have no HTTP route since protocol 2: an
+`OrchestrationV2Command` goes through the WebSocket RPC
+`orchestration.dispatchCommand` (`packages/contracts/src/orchestrationV2.ts`).
+Every request names the protocol - a header over HTTP, a query parameter on
+the WebSocket upgrade - or the server answers 400 / 426.
+
+The token needs the `orchestration:read` and `orchestration:operate` scopes.
 """
 
 from __future__ import annotations
 
+import json
 import uuid
-from datetime import UTC, datetime
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import httpx
+import websockets
+
+if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
+    RpcCall = Callable[[str, dict[str, Any]], Awaitable[Any]]
 
 DESCRIPTOR = "/.well-known/t3/environment"
 SHELL = "/api/orchestration/shell"
 THREADS = "/api/orchestration/threads"
-DISPATCH = "/api/orchestration/dispatch"
 WS_TICKET = "/api/auth/websocket-ticket"
 WS_PATH = "/ws"
+PROTOCOL = "2"
+PROTOCOL_HEADER = "x-t3-orchestration-protocol"
+PROTOCOL_PARAM = "orchestrationProtocol"
 SUBSCRIBE_SHELL = "orchestration.subscribeShell"
+DISPATCH_COMMAND = "orchestration.dispatchCommand"
 
 RUNTIME_MODE = "full-access"
 INTERACTION_MODE = "default"
+CREATED_BY = "agent"
+CREATION_SOURCE = "mcp"
 PROBE_TIMEOUT = 5.0
 
 
@@ -32,17 +48,14 @@ class T3Error(Exception):
         self.status = status
         self.body = body
         detail = (
-            f"{body.get('code')}: {body.get('reason') or body.get('message')}"
+            f"{body.get('code') or body.get('_tag')}: "
+            f"{body.get('reason') or body.get('detail') or body.get('message')}"
             if isinstance(body, dict)
             else str(body)[:300]
         )
         super().__init__(f"T3 responded {status} - {detail}")
 
 
-def now_iso() -> str:
-    return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
-
-
 def new_id() -> str:
     return str(uuid.uuid4())
 
@@ -52,6 +65,8 @@ def thread_create(
 ) -> dict[str, Any]:
     return {
         "type": "thread.create",
+        "createdBy": CREATED_BY,
+        "creationSource": CREATION_SOURCE,
         "commandId": new_id(),
         "threadId": thread_id,
         "projectId": project_id,
@@ -61,29 +76,25 @@ def thread_create(
         "interactionMode": INTERACTION_MODE,
         "branch": None,
         "worktreePath": None,
-        "createdAt": now_iso(),
     }
 
 
-def turn_start(
+def message_dispatch(
     thread_id: str,
     text: str,
     selection: dict[str, Any] | None = None,
     title_seed: str | None = None,
 ) -> dict[str, Any]:
     command: dict[str, Any] = {
-        "type": "thread.turn.start",
+        "type": "message.dispatch",
+        "createdBy": CREATED_BY,
+        "creationSource": CREATION_SOURCE,
         "commandId": new_id(),
         "threadId": thread_id,
-        "message": {
         "messageId": new_id(),
-            "role": "user",
         "text": text,
         "attachments": [],
-        },
-        "runtimeMode": RUNTIME_MODE,
-        "interactionMode": INTERACTION_MODE,
-        "createdAt": now_iso(),
+        "dispatchMode": {"type": "start_immediately"},
     }
     if selection is not None:
         command["modelSelection"] = selection
@@ -92,31 +103,38 @@ def turn_start(
     return command
 
 
-def turn_interrupt(thread_id: str, turn_id: str | None = None) -> dict[str, Any]:
-    command: dict[str, Any] = {
-        "type": "thread.turn.interrupt",
+def run_interrupt(thread_id: str, run_id: str) -> dict[str, Any]:
+    return {
+        "type": "run.interrupt",
         "commandId": new_id(),
         "threadId": thread_id,
-        "createdAt": now_iso(),
+        "runId": run_id,
     }
-    if turn_id:
-        command["turnId"] = turn_id
-    return command
 
 
-def user_input_respond(
+def runtime_request_respond(
     thread_id: str, request_id: str, answers: dict[str, Any]
 ) -> dict[str, Any]:
     return {
-        "type": "thread.user-input.respond",
+        "type": "runtime-request.respond",
         "commandId": new_id(),
         "threadId": thread_id,
         "requestId": request_id,
         "answers": answers,
-        "createdAt": now_iso(),
     }
 
 
+def rpc_exit(message: dict[str, Any]) -> Any:
+    """The value of an Effect RPC `Exit`, or a `T3Error` with its cause."""
+    exit_ = message.get("exit") or {}
+    if exit_.get("_tag") == "Success":
+        return exit_.get("value")
+    causes = exit_.get("cause") or [{}]
+    cause = causes[0] if isinstance(causes, list) else causes
+    body = cause.get("error") or {"message": str(cause.get("defect") or cause)[:300]}
+    raise T3Error(0, body)
+
+
 class T3Client:
     def __init__(
         self,
@@ -124,12 +142,15 @@ class T3Client:
         token: str,
         *,
         transport: httpx.AsyncBaseTransport | None = None,
+        rpc: RpcCall | None = None,
         timeout: float = 30.0,
     ) -> None:
         self.url = url.rstrip("/")
+        self.timeout = timeout
+        self._rpc = rpc or self._ws_rpc
         self._http = httpx.AsyncClient(
             base_url=self.url,
-            headers={"Authorization": f"Bearer {token}"},
+            headers={"Authorization": f"Bearer {token}", PROTOCOL_HEADER: PROTOCOL},
             timeout=timeout,
             transport=transport,
         )
@@ -146,22 +167,46 @@ class T3Client:
     async def shell(self) -> dict[str, Any]:
         return self._json(await self._http.get(SHELL))
 
-    async def thread(
-        self, thread_id: str, turn_limit: int | None = None
-    ) -> dict[str, Any] | None:
-        params = {"turnLimit": turn_limit} if turn_limit else None
-        response = await self._http.get(f"{THREADS}/{thread_id}", params=params)
+    async def thread(self, thread_id: str) -> dict[str, Any] | None:
+        """`{snapshotSequence, projection}` of a thread, None when it is unknown."""
+        response = await self._http.get(f"{THREADS}/{thread_id}")
         if response.status_code == httpx.codes.NOT_FOUND:
             return None
         return self._json(response)
 
     async def dispatch(self, command: dict[str, Any]) -> dict[str, Any]:
-        return self._json(await self._http.post(DISPATCH, json=command))
+        return await self._rpc(DISPATCH_COMMAND, command)
 
     async def ws_url(self) -> str:
         ticket = self._json(await self._http.post(WS_TICKET))["ticket"]
         scheme = "wss" if self.url.startswith("https") else "ws"
-        return f"{scheme}://{self.url.split('://', 1)[1]}{WS_PATH}?wsTicket={ticket}"
+        host = self.url.split("://", 1)[1]
+        return (
+            f"{scheme}://{host}{WS_PATH}?wsTicket={ticket}&{PROTOCOL_PARAM}={PROTOCOL}"
+        )
+
+    async def _ws_rpc(self, tag: str, payload: dict[str, Any]) -> Any:
+        url = await self.ws_url()
+        async with websockets.connect(
+            url, max_size=None, open_timeout=self.timeout
+        ) as ws:
+            request = {
+                "_tag": "Request",
+                "id": "1",
+                "tag": tag,
+                "payload": payload,
+                "headers": [],
+            }
+            await ws.send(json.dumps(request))
+            async for raw in ws:
+                decoded = json.loads(raw)
+                for message in decoded if isinstance(decoded, list) else [decoded]:
+                    if message.get("_tag") == "Exit":
+                        return rpc_exit(message)
+                    if message.get("_tag") in ("Defect", "ClientProtocolError"):
+                        raise T3Error(0, str(message)[:300])
+        msg = f"{tag}: socket closed before the result"
+        raise T3Error(0, msg)
 
     @staticmethod
     def _json(response: httpx.Response) -> dict[str, Any]:

src/t3code_mcp/views.py

@@ -1,4 +1,10 @@
-"""Read-model helpers shared by the tools and the watcher."""
+"""Read-model helpers shared by the tools and the watcher.
+
+A thread detail is `{snapshotSequence, projection}`; the projection holds
+runs (one per user message), messages, turn items, runtime requests and
+checkpoints. A shell row carries the latest run's `status`, `lastError` and
+the newest `pendingRuntimeRequest`.
+"""
 
 from __future__ import annotations
 
@@ -8,48 +14,83 @@ from t3code_mcp.config import ModelSpec
 
 TEXT_MAX = 6000
 TOOLS_MAX = 12
-BUSY = frozenset({"running", "starting"})
+BUSY = frozenset({"preparing", "queued", "starting", "running", "waiting"})
+TOOL_ITEMS = frozenset(
+    {"command_execution", "dynamic_tool", "file_change", "web_search", "subagent"}
+)
+NOT_APPROVALS = frozenset({"user_input", "dynamic_tool_call", "auth_refresh"})
 
 
-def state(thread: dict[str, Any]) -> str:
-    latest = thread.get("latestTurn")
+def runs(projection: dict[str, Any]) -> list[dict[str, Any]]:
+    return sorted(projection.get("runs", []), key=lambda r: r["ordinal"])
+
+
+def latest_run(projection: dict[str, Any]) -> dict[str, Any] | None:
+    held = [r for r in runs(projection) if not r.get("queueHeld")]
+    return held[-1] if held else None
+
+
+def state(projection: dict[str, Any]) -> str:
+    latest = latest_run(projection)
     if latest:
-        return latest["state"]
-    messages = thread.get("messages", [])
+        return latest["status"]
+    messages = projection.get("messages", [])
     return "starting" if messages and messages[-1]["role"] == "user" else "idle"
 
 
 def shell_state(shell_thread: dict[str, Any]) -> str:
-    latest = shell_thread.get("latestTurn")
-    return latest["state"] if latest else "starting"
+    return shell_thread.get("status") or "idle"
 
 
-def pending(shell_thread: dict[str, Any]) -> dict[str, bool] | None:
+def _flags(request_kind: str | None, *, plan: bool) -> dict[str, bool] | None:
     flags = {
-        "approval": bool(shell_thread.get("hasPendingApprovals")),
-        "user_input": bool(shell_thread.get("hasPendingUserInput")),
-        "plan": bool(shell_thread.get("hasActionableProposedPlan")),
+        "approval": request_kind is not None and request_kind not in NOT_APPROVALS,
+        "user_input": request_kind == "user_input",
+        "plan": plan,
     }
     return flags if any(flags.values()) else None
 
 
-def pending_in(shell: dict[str, Any], thread_id: str) -> dict[str, bool] | None:
-    for thread in shell.get("threads", []):
-        if thread["id"] == thread_id:
-            return pending(thread)
-    return None
+def pending(shell_thread: dict[str, Any]) -> dict[str, bool] | None:
+    request = shell_thread.get("pendingRuntimeRequest") or {}
+    return _flags(
+        request.get("kind"), plan=bool(shell_thread.get("hasActionableProposedPlan"))
+    )
 
 
-def question(detail: dict[str, Any]) -> dict[str, Any] | None:
-    """The unanswered `user-input.requested` of a thread, newest first."""
-    for activity in reversed(detail["thread"].get("activities", [])):
-        kind = activity.get("kind")
-        if kind == "user-input.resolved":
+def pending_detail(projection: dict[str, Any]) -> dict[str, bool] | None:
+    requests = [
+        r for r in projection.get("runtimeRequests", []) if r["status"] == "pending"
+    ]
+    plan = any(
+        p.get("kind") == "proposed_plan" and p.get("status") == "active"
+        for p in projection.get("plans", [])
+    )
+    return _flags(requests[-1]["kind"] if requests else None, plan=plan)
+
+
+def question(projection: dict[str, Any]) -> dict[str, Any] | None:
+    """The pending user-input request of a thread with its questions."""
+    request = next(
+        (
+            r
+            for r in reversed(projection.get("runtimeRequests", []))
+            if r["status"] == "pending" and r["kind"] == "user_input"
+        ),
+        None,
+    )
+    if request is None:
         return None
-        if kind == "user-input.requested":
-            payload = activity.get("payload") or {}
+    item = next(
+        (
+            i
+            for i in projection.get("turnItems", [])
+            if i.get("type") == "user_input_request" and i["requestId"] == request["id"]
+        ),
+        {},
+    )
     return {
-                "request_id": payload.get("requestId"),
+        "request_id": request["id"],
         "questions": [
             {
                 "id": q["id"],
@@ -61,69 +102,100 @@ def question(detail: dict[str, Any]) -> dict[str, Any] | None:
                 ],
                 "multi_select": bool(q.get("multiSelect")),
             }
-                    for q in payload.get("questions", [])
+            for q in item.get("questions", [])
         ],
     }
-    return None
 
 
 def project_title(shell: dict[str, Any], detail: dict[str, Any]) -> str | None:
-    project_id = detail["thread"]["projectId"]
+    project_id = detail["projection"]["thread"]["projectId"]
     return next((p["title"] for p in shell["projects"] if p["id"] == project_id), None)
 
 
-def summary(
-    machine: str, detail: dict[str, Any], *, project: str | None = None
-) -> dict[str, Any]:
-    thread = detail["thread"]
-    latest = thread.get("latestTurn") or {}
-    session = thread.get("session") or {}
-    messages = [
-        {"role": m["role"], "text": m["text"][-TEXT_MAX:], "streaming": m["streaming"]}
-        for m in thread.get("messages", [])
-        if m.get("text")
-    ]
-    tools = [
-        {
-            "summary": a["summary"],
-            "detail": str((a.get("payload") or {}).get("detail", ""))[:200],
-        }
-        for a in thread.get("activities", [])
-        if a.get("kind") == "tool.completed"
-    ][-TOOLS_MAX:]
-    files = [
-        {"path": f["path"], "kind": f["kind"], "+": f["additions"], "-": f["deletions"]}
-        for c in thread.get("checkpoints", [])[-1:]
-        for f in c.get("files", [])
-    ]
-    context = next(
+def error(projection: dict[str, Any]) -> str | None:
+    latest = latest_run(projection)
+    if latest and latest["status"] == "failed":
+        failure = next(
             (
-            a["payload"]
-            for a in reversed(thread.get("activities", []))
-            if a.get("kind") == "context-window.updated"
+                i["failure"]["message"]
+                for i in reversed(projection.get("turnItems", []))
+                if i.get("type") == "error" and i.get("runId") == latest["id"]
             ),
             None,
         )
+        if failure:
+            return failure
+    sessions = projection.get("providerSessions", [])
+    return sessions[-1].get("lastError") if sessions else None
+
+
+def _tool(item: dict[str, Any]) -> dict[str, Any]:
+    kind = item["type"]
+    if kind == "command_execution":
+        detail = item.get("input", "")
+    elif kind == "file_change":
+        detail = item.get("fileName", "")
+    elif kind == "subagent":
+        detail = item.get("prompt", "")
+    else:
+        detail = item.get("toolName") or item.get("input") or ""
+    return {"summary": item.get("title") or kind, "detail": str(detail)[:200]}
+
+
+def summary(
+    machine: str, detail: dict[str, Any], *, project: str | None = None, turns: int = 1
+) -> dict[str, Any]:
+    projection = detail["projection"]
+    thread = projection["thread"]
+    window = {r["id"] for r in runs(projection)[-turns:]}
+    latest = latest_run(projection)
+    messages = [
+        {"role": m["role"], "text": m["text"][-TEXT_MAX:], "streaming": m["streaming"]}
+        for m in projection.get("messages", [])
+        if m.get("text") and m.get("runId") in window
+    ]
+    items = projection.get("turnItems", [])
+    tools = [
+        _tool(i)
+        for i in items
+        if i.get("type") in TOOL_ITEMS
+        and i.get("status") == "completed"
+        and i.get("runId") in window
+    ][-TOOLS_MAX:]
+    files = [
+        {"path": f["path"], "kind": f["kind"], "+": f["additions"], "-": f["deletions"]}
+        for c in projection.get("checkpoints", [])[-1:]
+        for f in c.get("files", [])
+    ]
+    usage = next(
+        (
+            t["tokenUsage"]
+            for t in reversed(projection.get("providerTurns", []))
+            if t.get("tokenUsage")
+        ),
+        {},
+    )
+    sessions = projection.get("providerSessions", [])
     return {
         "thread_id": thread["id"],
         "machine": machine,
         "project": project or thread["projectId"],
         "title": thread["title"],
         "model": str(ModelSpec.from_selection(thread["modelSelection"])),
-        "state": state(thread),
-        "turn_id": latest.get("turnId"),
-        "session": session.get("status"),
-        "error": session.get("lastError"),
+        "state": state(projection),
+        "run_id": latest["id"] if latest else None,
+        "session": sessions[-1]["status"] if sessions else None,
+        "error": error(projection),
         "messages": messages,
         "tools": tools,
         "files": files,
-        "context_tokens": (context or {}).get("usedTokens"),
-        "question": question(detail),
+        "context_tokens": usage.get("usedTokens"),
+        "question": question(projection),
     }
 
 
 def last_answer(detail: dict[str, Any]) -> str:
-    for m in reversed(detail["thread"].get("messages", [])):
+    for m in reversed(detail["projection"].get("messages", [])):
         if m["role"] == "assistant" and m.get("text"):
             return m["text"]
     return ""

src/t3code_mcp/watch.py

@@ -2,8 +2,9 @@
 
 One `orchestration.subscribeShell` stream per machine (Effect RPC over JSON:
 `Request` → `Chunk`s acknowledged with `Ack`, `Ping` keepalive). The shell
-read-model carries everything a transition needs - `latestTurn.state`,
-`session`, `hasPendingUserInput` - so no per-thread subscription is required.
+row carries everything a transition needs - the latest run's `status`,
+`lastError`, `pendingRuntimeRequest` - so no per-thread subscription is
+required.
 Reconnects resume with `afterSequence`; a fresh snapshot re-derives every
 tracked thread, so transitions missed while offline are still reported.
 
@@ -39,7 +40,9 @@ EventKind = Literal["completed", "interrupted", "failed", "question", "approval"
 TERMINAL: dict[str, EventKind] = {
     "completed": "completed",
     "interrupted": "interrupted",
-    "error": "failed",
+    "cancelled": "interrupted",
+    "rolled_back": "interrupted",
+    "failed": "failed",
 }
 KEEP_DAYS = 14
 PING_EVERY = 20.0
@@ -81,7 +84,7 @@ class Tracker:
                 for k, v in raw.get("threads", {}).items()
             }
             self.outbox = raw.get("outbox", [])
-            self.sequences = raw.get("sequences", {})
+            self.sequences = raw.get("shell_sequences", {})
 
     def save(self) -> None:
         if self.path is None:
@@ -92,7 +95,7 @@ class Tracker:
         payload = {
             "threads": {k: asdict(v) for k, v in self.threads.items()},
             "outbox": self.outbox,
-            "sequences": self.sequences,
+            "shell_sequences": self.sequences,
         }
         tmp = self.path.with_suffix(".tmp")
         tmp.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
@@ -124,9 +127,11 @@ class Tracker:
         if tracked is None or tracked.machine != machine:
             return []
         state = views.shell_state(shell_thread)
+        if state == "idle" and tracked.state in views.BUSY:
+            state = tracked.state
         flags = views.pending(shell_thread) or {}
         pending = next((k for k in ("user_input", "approval") if flags.get(k)), None)
-        error = (shell_thread.get("session") or {}).get("lastError")
+        error = shell_thread.get("lastError")
         events: list[Event] = []
         if pending and pending != tracked.pending:
             kind: EventKind = "question" if pending == "user_input" else "approval"
@@ -298,7 +303,7 @@ class Watcher:
             self.tracker.sequences[name] = snapshot["snapshotSequence"]
             for thread in snapshot["threads"]:
                 await self._events(self.tracker.apply(name, thread))
-        elif kind == "thread-upserted":
+        elif kind == "thread.updated":
             self.tracker.sequences[name] = item["sequence"]
             await self._events(self.tracker.apply(name, item["thread"]))
         elif "sequence" in item:
@@ -323,9 +328,7 @@ class Watcher:
             "error": tracked.error,
         }
         try:
-            detail = await self.registry.client(event.machine).thread(
-                event.thread_id, turn_limit=1
-            )
+            detail = await self.registry.client(event.machine).thread(event.thread_id)
         except (httpx.HTTPError, t3.T3Error) as exc:
             payload["detail_error"] = str(exc)[:200]
             return payload

tests/test_server.py

@@ -8,6 +8,7 @@ from fastmcp.exceptions import ToolError
 
 from t3code_mcp.config import Machine, ModelSpec, load_machines
 from t3code_mcp.server import Registry, build_server
+from t3code_mcp.t3 import T3Error
 
 
 class FakeT3:
@@ -30,7 +31,7 @@ class FakeT3:
                 200,
                 json={
                     "label": self.label,
-                    "serverVersion": "0.0.36",
+                    "serverVersion": "0.1.0",
                     "platform": {"os": "linux"},
                 },
             )
@@ -38,6 +39,8 @@ class FakeT3:
             return httpx.Response(
                 401, json={"code": "auth_invalid", "reason": "missing_credential"}
             )
+        if request.headers.get("x-t3-orchestration-protocol") != "2":
+            return httpx.Response(400, json={"code": "invalid_request"})
         if path == "/api/orchestration/shell":
             return httpx.Response(200, json=self._shell())
         if path.startswith("/api/orchestration/threads/"):
@@ -48,100 +51,127 @@ class FakeT3:
                 )
             self._tick(thread_id)
             return httpx.Response(
-                200, json={"snapshotSequence": 1, "thread": self.threads[thread_id]}
+                200, json={"snapshotSequence": 1, "projection": self.threads[thread_id]}
             )
-        if path == "/api/orchestration/dispatch":
-            command = json.loads(request.content)
-            self.commands.append(command)
-            return self._dispatch(command)
         return httpx.Response(404, json={"code": "not_found"})
 
+    async def rpc(self, tag: str, payload: dict[str, Any]) -> Any:
+        assert tag == "orchestration.dispatchCommand"
+        self.commands.append(payload)
+        return self._dispatch(payload)
+
     def _shell(self) -> dict[str, Any]:
-        threads = [
-            {
-                k: v
-                for k, v in t.items()
-                if k not in ("messages", "activities", "checkpoints")
-            }
-            | {"hasPendingUserInput": self.pending_input, "hasPendingApprovals": False}
-            for t in self.threads.values()
+        threads = []
+        for projection in self.threads.values():
+            thread = projection["thread"]
+            runs = projection["runs"]
+            pending = [
+                r for r in projection["runtimeRequests"] if r["status"] == "pending"
             ]
-        return {
-            "snapshotSequence": 1,
-            "projects": self.projects,
-            "threads": threads,
-            "updatedAt": "now",
+            threads.append(
+                {
+                    "id": thread["id"],
+                    "projectId": thread["projectId"],
+                    "title": thread["title"],
+                    "archivedAt": None,
+                    "status": runs[-1]["status"] if runs else "idle",
+                    "lastError": self.fail_with,
+                    "pendingRuntimeRequest": (
+                        {"id": pending[-1]["id"], "kind": pending[-1]["kind"]}
+                        if pending
+                        else None
+                    ),
+                    "hasActionableProposedPlan": False,
                 }
+            )
+        return {"snapshotSequence": 1, "projects": self.projects, "threads": threads}
 
     def _tick(self, thread_id: str) -> None:
-        thread = self.threads[thread_id]
-        if self.fail_with and thread.get("_pending_turn"):
-            thread["session"] = {"status": "stopped", "lastError": self.fail_with}
+        projection = self.threads[thread_id]
+        if self.pending_input and not projection["runtimeRequests"]:
+            projection["runtimeRequests"].append(
+                {"id": "req-0", "kind": "user_input", "status": "pending"}
+            )
+        if not projection["runs"]:
             return
-        if thread.get("_pending_turn"):
-            thread["latestTurn"] = {
-                "turnId": thread.pop("_pending_turn"),
-                "state": "running",
+        run = projection["runs"][-1]
+        if self.fail_with and run["status"] == "running":
+            run["status"] = "failed"
+            projection["turnItems"].append(
+                {
+                    "type": "error",
+                    "runId": run["id"],
+                    "status": "failed",
+                    "failure": {"message": self.fail_with},
                 }
+            )
             return
-        latest = thread["latestTurn"]
-        if latest and latest["state"] == "running":
-            latest["_polls"] = latest.get("_polls", 0) + 1
-            if latest["_polls"] > self.polls_until_done:
-                latest["state"] = "completed"
-                thread["messages"].append(
-                    {"role": "assistant", "text": "done: report", "streaming": False}
+        if run["status"] == "running":
+            run["_polls"] = run.get("_polls", 0) + 1
+            if run["_polls"] > self.polls_until_done:
+                run["status"] = "completed"
+                projection["messages"].append(
+                    {
+                        "role": "assistant",
+                        "text": "done: report",
+                        "streaming": False,
+                        "runId": run["id"],
+                    }
                 )
-                thread["session"]["status"] = "ready"
 
-    def _dispatch(self, command: dict[str, Any]) -> httpx.Response:
+    def _dispatch(self, command: dict[str, Any]) -> dict[str, Any]:
         kind = command["type"]
         if kind == "thread.create":
             self.threads[command["threadId"]] = {
+                "thread": {
                     "id": command["threadId"],
                     "projectId": command["projectId"],
                     "title": command["title"],
                     "modelSelection": command["modelSelection"],
                     "runtimeMode": command["runtimeMode"],
-                "latestTurn": None,
-                "session": {"status": "idle", "lastError": None},
-                "messages": [],
-                "activities": [],
-                "checkpoints": [],
                     "archivedAt": None,
-                "updatedAt": "now",
-            }
-        elif kind == "thread.turn.start":
-            thread = self.threads[command["threadId"]]
-            thread["messages"].append(
-                {"role": "user", "text": command["message"]["text"], "streaming": False}
-            )
-            thread["latestTurn"] = None
-            thread["_pending_turn"] = "turn-" + command["commandId"][:4]
-            thread["session"] = {"status": "running", "lastError": None}
-            if "modelSelection" in command:
-                thread["modelSelection"] = command["modelSelection"]
-        elif kind == "thread.turn.interrupt":
-            thread = self.threads[command["threadId"]]
-            thread.pop("_pending_turn", None)
-            thread["latestTurn"] = {"turnId": "t", "state": "interrupted"}
-        elif kind == "thread.user-input.respond":
-            self.threads[command["threadId"]]["activities"].append(
-                {
-                    "kind": "user-input.resolved",
-                    "summary": "User input submitted",
-                    "tone": "info",
-                    "payload": {
-                        "requestId": command["requestId"],
-                        "answers": command["answers"],
                 },
+                "runs": [],
+                "messages": [],
+                "turnItems": [],
+                "runtimeRequests": [],
+                "checkpoints": [],
+                "providerSessions": [],
+                "providerTurns": [],
+                "plans": [],
+            }
+        elif kind == "message.dispatch":
+            assert command["dispatchMode"] == {"type": "start_immediately"}
+            projection = self.threads[command["threadId"]]
+            run_id = f"run-{len(projection['runs']) + 1}"
+            projection["runs"].append(
+                {
+                    "id": run_id,
+                    "ordinal": len(projection["runs"]) + 1,
+                    "status": "running",
                 }
             )
-        else:
-            return httpx.Response(
-                400, json={"code": "invalid_request", "reason": "invalid_command"}
+            projection["messages"].append(
+                {
+                    "role": "user",
+                    "text": command["text"],
+                    "streaming": False,
+                    "runId": run_id,
+                }
             )
-        return httpx.Response(200, json={"sequence": len(self.commands)})
+            if "modelSelection" in command:
+                projection["thread"]["modelSelection"] = command["modelSelection"]
+        elif kind == "run.interrupt":
+            run = self.threads[command["threadId"]]["runs"][-1]
+            assert run["id"] == command["runId"]
+            run["status"] = "interrupted"
+        elif kind == "runtime-request.respond":
+            for request in self.threads[command["threadId"]]["runtimeRequests"]:
+                if request["id"] == command["requestId"]:
+                    request["status"] = "resolved"
+        else:
+            raise T3Error(0, {"_tag": "OrchestrationV2DispatchCommandError"})
+        return {"sequence": len(self.commands)}
 
 
 PROJECTS_MAC = [
@@ -209,7 +239,7 @@ def registry(fakes: dict[str, FakeT3]) -> Registry:
             projects=("/root/projects", "/root/projects/*"),
         ),
     }
-    return Registry(machines, transport=Router(fakes))
+    return Registry(machines, transport=Router(fakes), rpc=lambda name: fakes[name].rpc)
 
 
 @pytest.fixture
@@ -254,7 +284,7 @@ async def test_dispatch_wait_and_thread(
         assert started["title"] == "write README"
         assert started["model"] == "claudeAgent/claude-opus-5 effort=high"
         kinds = [c["type"] for c in fakes["mac"].commands]
-        assert kinds == ["thread.create", "thread.turn.start"]
+        assert kinds == ["thread.create", "message.dispatch"]
         assert fakes["mac"].commands[0]["modelSelection"] == {
             "instanceId": "claudeAgent",
             "model": "claude-opus-5",
@@ -303,8 +333,8 @@ async def test_follow_up_uses_same_thread(
         assert again["thread_id"] == started["thread_id"]
         assert [c["type"] for c in fakes["mac"].commands] == [
             "thread.create",
-            "thread.turn.start",
-            "thread.turn.start",
+            "message.dispatch",
+            "message.dispatch",
         ]
 
 
@@ -346,7 +376,7 @@ async def test_wait_timeout_and_interrupt(
             await client.call_tool("t3_interrupt", {"thread_id": started["thread_id"]})
         )
         assert stopped["state"] == "interrupting"
-        assert fakes["dell"].commands[-1]["type"] == "thread.turn.interrupt"
+        assert fakes["dell"].commands[-1]["type"] == "run.interrupt"
         view = data(
             await client.call_tool("t3_thread", {"thread_id": started["thread_id"]})
         )
@@ -439,7 +469,7 @@ async def test_wait_returns_on_runtime_failure(
                 "t3_wait", {"thread_id": started["thread_id"], "timeout": 5}
             )
         )
-        assert waited["state"] == "starting"
+        assert waited["state"] == "failed"
         assert waited["error"] == "Claude runtime stream failed."
         assert waited["waited"] < 1
 
@@ -449,10 +479,12 @@ def shell_thread(
 ) -> dict[str, Any]:
     return {
         "id": tid,
-        "latestTurn": {"turnId": "t", "state": state} if state else None,
-        "session": {"status": "stopped" if error else "running", "lastError": error},
-        "hasPendingUserInput": user_input,
-        "hasPendingApprovals": False,
+        "status": state or "idle",
+        "lastError": error,
+        "pendingRuntimeRequest": (
+            {"id": "req-1", "kind": "user_input"} if user_input else None
+        ),
+        "hasActionableProposedPlan": False,
     }
 
 
@@ -533,12 +565,15 @@ async def test_answer_pending_question(
             await client.call_tool(
                 "t3_answer", {"thread_id": tid, "answers": {"q1": "yes"}}
             )
-        fakes["mac"].threads[tid]["activities"].append(
+        projection = fakes["mac"].threads[tid]
+        projection["runtimeRequests"].append(
+            {"id": "req-1", "kind": "user_input", "status": "pending"}
+        )
+        projection["turnItems"].append(
             {
-                "kind": "user-input.requested",
-                "summary": "User input requested",
-                "tone": "info",
-                "payload": {
+                "type": "user_input_request",
+                "runId": "run-1",
+                "status": "waiting",
                 "requestId": "req-1",
                 "questions": [
                     {
@@ -551,7 +586,6 @@ async def test_answer_pending_question(
                         ],
                     }
                 ],
-                },
             }
         )
         view = data(await client.call_tool("t3_thread", {"thread_id": tid}))
@@ -569,7 +603,7 @@ async def test_answer_pending_question(
         assert answered["answered"] == ["q1"]
         cmd = fakes["mac"].commands[-1]
         assert (
-            cmd["type"] == "thread.user-input.respond"
+            cmd["type"] == "runtime-request.respond"
             and cmd["requestId"] == "req-1"
             and cmd["answers"] == {"q1": "yes"}
         )