From 741322326921a4a76709a74c2f335f5e32ddf35f Mon Sep 17 00:00:00 2001 From: h Date: Fri, 23 Oct 2026 22:24:45 +0200 Subject: [PATCH] feat(server,watch): threads remember the dispatch origin header and report it back --- README.md | 2 +- docs/PROGRESS.md | 4 ++++ src/t3code_mcp/server.py | 11 ++++++++++- src/t3code_mcp/watch.py | 5 ++++- tests/test_server.py | 23 +++++++++++++++++++++++ 5 files changed, 42 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 5e94611..498b171 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ Built for the dispatcher in [beaver-agent](https://git.kotikot.com/beaver/beaver ## Events -The connector keeps a websocket subscription to every machine (`orchestration.subscribeShell`, Effect RPC over JSON, reconnect with replay). Threads started through `t3_dispatch` are tracked in a state file; the transitions running → completed / interrupted / failed and a new pending question become events, `POST`ed to `T3CODE_MCP_HOOK_URL` with a bearer through a persisted outbox with retries. In the beaver setup that URL is the gateway's `/hooks/t3code`, and the event lands in the master conversation as an urgent inject. Every `t3_dispatch` subscribes its thread, a continuation included; `t3_watch(thread_id, enabled=false)` or `t3_dispatch(..., watch=false)` lets a thread run silently. +The connector keeps a websocket subscription to every machine (`orchestration.subscribeShell`, Effect RPC over JSON, reconnect with replay). Threads started through `t3_dispatch` are tracked in a state file; the transitions running → completed / interrupted / failed and a new pending question become events, `POST`ed to `T3CODE_MCP_HOOK_URL` with a bearer through a persisted outbox with retries. In the beaver setup that URL is the gateway's `/hooks/t3code`. `t3_dispatch` remembers the `X-Conversation-Origin` request header as the thread's `origin` and every event carries it back; the beaver gateway puts the calling conversation's id there, so the event lands as an urgent inject in the conversation that started the thread (the master when it is gone or the header was absent). Every `t3_dispatch` subscribes its thread, a continuation included; `t3_watch(thread_id, enabled=false)` or `t3_dispatch(..., watch=false)` lets a thread run silently. ## Configuration diff --git a/docs/PROGRESS.md b/docs/PROGRESS.md index 7b0c6a3..49a07da 100644 --- a/docs/PROGRESS.md +++ b/docs/PROGRESS.md @@ -2,6 +2,10 @@ > Что сделано, что нет, что проверить руками. Свежие записи сверху. Архитектурная правда - `архитектура.md` §5 (vault, симлинк в `../../архитектура.md`), журнал всей второй итерации - `../../PROGRESS.md`. +## 2026-10-23 - событие треда помнит, кто его запустил + +`t3_dispatch` читает заголовок `x-conversation-origin` (гейтвей ставит туда id разговора на каждый MCP-запрос сессии) и хранит его в `Tracked.origin`; хук несёт поле `origin`, джоб `t3code` в beaver-agent кладёт событие в этот разговор, а не всегда в мастер. Продолжение треда из другого разговора переписывает владельца. Клиент без заголовка - `origin: null`, событие уходит в мастер как раньше. + ## 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`. diff --git a/src/t3code_mcp/server.py b/src/t3code_mcp/server.py index 877529f..4b6260b 100644 --- a/src/t3code_mcp/server.py +++ b/src/t3code_mcp/server.py @@ -17,6 +17,7 @@ import httpx import websockets from fastmcp import FastMCP from fastmcp.exceptions import ToolError +from fastmcp.server.dependencies import get_http_headers from pydantic import Field from starlette.responses import JSONResponse @@ -34,6 +35,7 @@ if TYPE_CHECKING: from t3code_mcp.t3 import RpcCall TITLE_MAX = 80 +ORIGIN_HEADER = "x-conversation-origin" WAIT_MAX = 3600.0 @@ -294,7 +296,14 @@ def build_server( client.dispatch(t3.message_dispatch(tid, text, title_seed=heading)) ) registry.threads[tid] = machine - registry.tracker.track(tid, machine, target["title"], heading, watch=watch) + registry.tracker.track( + tid, + machine, + target["title"], + heading, + watch=watch, + origin=get_http_headers().get(ORIGIN_HEADER), + ) return { "thread_id": tid, "machine": machine, diff --git a/src/t3code_mcp/watch.py b/src/t3code_mcp/watch.py index 1f48c0a..6289c80 100644 --- a/src/t3code_mcp/watch.py +++ b/src/t3code_mcp/watch.py @@ -60,6 +60,7 @@ class Tracked: pending: str | None = None error: str | None = None watch: bool = True + origin: str | None = None updated: float = field(default_factory=time.time) @@ -109,9 +110,10 @@ class Tracker: title: str, *, watch: bool = True, + origin: str | None = None, ) -> None: self.threads[thread_id] = Tracked( - machine=machine, project=project, title=title, watch=watch + machine=machine, project=project, title=title, watch=watch, origin=origin ) self.save() @@ -326,6 +328,7 @@ class Watcher: "title": tracked.title, "state": tracked.state, "error": tracked.error, + "origin": tracked.origin, } try: detail = await self.registry.client(event.machine).thread(event.thread_id) diff --git a/tests/test_server.py b/tests/test_server.py index 96f36c8..8c4049f 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -625,6 +625,29 @@ async def test_dispatch_tracks_thread(client: Client, registry: Registry) -> Non ) +async def test_dispatch_remembers_its_origin( + client: Client, registry: Registry, monkeypatch: pytest.MonkeyPatch +) -> None: + from t3code_mcp import server + from t3code_mcp.watch import Event, Watcher + + monkeypatch.setattr( + server, "get_http_headers", lambda: {"x-conversation-origin": "conv-7"} + ) + async with client: + started = data( + await client.call_tool( + "t3_dispatch", + {"machine": "mac", "project": "t3-smoke", "prompt": "from a branch"}, + ) + ) + tid = started["thread_id"] + assert registry.tracker.threads[tid].origin == "conv-7" + watcher = Watcher(registry, registry.tracker, None) + payload = await watcher.payload(Event("completed", tid, "mac")) + assert payload["origin"] == "conv-7" + + def test_tracker_watch_flag(tmp_path) -> None: from t3code_mcp.watch import Tracker