From f4c54b6e16b62bc62bebc5aee2f19a53990f6f37 Mon Sep 17 00:00:00 2001 From: h Date: Fri, 16 Oct 2026 01:48:32 +0200 Subject: [PATCH] feat(t3,views,server,watch): t3 code orchestration protocol 2 --- README.md | 10 +- docs/PROGRESS.md | 8 ++ src/t3code_mcp/server.py | 90 +++++++-------- src/t3code_mcp/t3.py | 131 ++++++++++++++------- src/t3code_mcp/views.py | 238 ++++++++++++++++++++++++-------------- src/t3code_mcp/watch.py | 23 ++-- tests/test_server.py | 242 ++++++++++++++++++++++----------------- 7 files changed, 448 insertions(+), 294 deletions(-) diff --git a/README.md b/README.md index 5410d4d..5e94611 100644 --- a/README.md +++ b/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 diff --git a/docs/PROGRESS.md b/docs/PROGRESS.md index cd6506c..7b0c6a3 100644 --- a/docs/PROGRESS.md +++ b/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 тестов. diff --git a/src/t3code_mcp/server.py b/src/t3code_mcp/server.py index 6bfbdf7..877529f 100644 --- a/src/t3code_mcp/server.py +++ b/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 diff --git a/src/t3code_mcp/t3.py b/src/t3code_mcp/t3.py index d7f4452..2224700 100644 --- a/src/t3code_mcp/t3.py +++ b/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(), + "messageId": new_id(), + "text": text, + "attachments": [], + "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]: diff --git a/src/t3code_mcp/views.py b/src/t3code_mcp/views.py index 155f5bf..401f92e 100644 --- a/src/t3code_mcp/views.py +++ b/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,122 +14,188 @@ 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": - return None - if kind == "user-input.requested": - payload = activity.get("payload") or {} - return { - "request_id": payload.get("requestId"), - "questions": [ - { - "id": q["id"], - "header": q.get("header"), - "question": q["question"], - "options": [ - {"label": o["label"], "description": o.get("description")} - for o in q.get("options", []) - ], - "multi_select": bool(q.get("multiSelect")), - } - for q in payload.get("questions", []) - ], - } - return None - - -def project_title(shell: dict[str, Any], detail: dict[str, Any]) -> str | None: - project_id = detail["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") +def pending_detail(projection: dict[str, Any]) -> dict[str, bool] | None: + requests = [ + r for r in projection.get("runtimeRequests", []) if r["status"] == "pending" ] - 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( + 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( ( - a["payload"] - for a in reversed(thread.get("activities", [])) - if a.get("kind") == "context-window.updated" + r + for r in reversed(projection.get("runtimeRequests", [])) + if r["status"] == "pending" and r["kind"] == "user_input" ), None, ) + if request is None: + return None + item = next( + ( + i + for i in projection.get("turnItems", []) + if i.get("type") == "user_input_request" and i["requestId"] == request["id"] + ), + {}, + ) + return { + "request_id": request["id"], + "questions": [ + { + "id": q["id"], + "header": q.get("header"), + "question": q["question"], + "options": [ + {"label": o["label"], "description": o.get("description")} + for o in q.get("options", []) + ], + "multi_select": bool(q.get("multiSelect")), + } + for q in item.get("questions", []) + ], + } + + +def project_title(shell: dict[str, Any], detail: dict[str, Any]) -> str | None: + project_id = detail["projection"]["thread"]["projectId"] + return next((p["title"] for p in shell["projects"] if p["id"] == project_id), None) + + +def error(projection: dict[str, Any]) -> str | None: + latest = latest_run(projection) + if latest and latest["status"] == "failed": + failure = next( + ( + 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 "" diff --git a/src/t3code_mcp/watch.py b/src/t3code_mcp/watch.py index 23db3b6..1f48c0a 100644 --- a/src/t3code_mcp/watch.py +++ b/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 diff --git a/tests/test_server.py b/tests/test_server.py index a3bed20..96f36c8 100644 --- a/tests/test_server.py +++ b/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() - ] - return { - "snapshotSequence": 1, - "projects": self.projects, - "threads": threads, - "updatedAt": "now", - } + threads = [] + for projection in self.threads.values(): + thread = projection["thread"] + runs = projection["runs"] + pending = [ + r for r in projection["runtimeRequests"] if r["status"] == "pending" + ] + 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"]] = { - "id": command["threadId"], - "projectId": command["projectId"], - "title": command["title"], - "modelSelection": command["modelSelection"], - "runtimeMode": command["runtimeMode"], - "latestTurn": None, - "session": {"status": "idle", "lastError": None}, + "thread": { + "id": command["threadId"], + "projectId": command["projectId"], + "title": command["title"], + "modelSelection": command["modelSelection"], + "runtimeMode": command["runtimeMode"], + "archivedAt": None, + }, + "runs": [], "messages": [], - "activities": [], + "turnItems": [], + "runtimeRequests": [], "checkpoints": [], - "archivedAt": None, - "updatedAt": "now", + "providerSessions": [], + "providerTurns": [], + "plans": [], } - 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( + 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( { - "kind": "user-input.resolved", - "summary": "User input submitted", - "tone": "info", - "payload": { - "requestId": command["requestId"], - "answers": command["answers"], - }, + "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,25 +565,27 @@ 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": { - "requestId": "req-1", - "questions": [ - { - "id": "q1", - "header": "Deploy", - "question": "Deploy now?", - "options": [ - {"label": "yes", "description": "go"}, - {"label": "no", "description": "wait"}, - ], - } - ], - }, + "type": "user_input_request", + "runId": "run-1", + "status": "waiting", + "requestId": "req-1", + "questions": [ + { + "id": "q1", + "header": "Deploy", + "question": "Deploy now?", + "options": [ + {"label": "yes", "description": "go"}, + {"label": "no", "description": "wait"}, + ], + } + ], } ) 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"} )