From a29c526ee9ff87c2352ae7551ae11b6ee4e8042e Mon Sep 17 00:00:00 2001 From: h Date: Tue, 1 Sep 2026 13:37:04 +0000 Subject: [PATCH] feat(watch,server): t3_watch to turn thread events off and back on --- README.md | 5 ++-- docs/PROGRESS.md | 6 ++++ src/t3code_mcp/server.py | 60 +++++++++++++++++++++++++++++++++++++-- src/t3code_mcp/watch.py | 38 +++++++++++++++++++++---- tests/test_server.py | 61 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 159 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index bbd75bd..25ef740 100644 --- a/README.md +++ b/README.md @@ -8,15 +8,16 @@ MCP-коннектор к серверам [T3 Code](https://github.com/pingdotg |---|---| | `t3_machines()` | машины из конфига, кто сейчас online (`/.well-known/t3/environment`), allowlist проектов | | `t3_projects(machine)` | проекты машины, прошедшие allowlist: путь, модель по умолчанию, число тредов | -| `t3_dispatch(machine, project, prompt, title?, model?, thread_id?)` | `thread.create` + `thread.turn.start` (или только `turn.start` в существующий тред) → `thread_id` сразу | +| `t3_dispatch(machine, project, prompt, title?, model?, thread_id?, watch?)` | `thread.create` + `thread.turn.start` (или только `turn.start` в существующий тред) → `thread_id` сразу; `watch=false` - не слать по треду события | | `t3_thread(thread_id, turns?)` | состояние без ожидания: тёрн, сессия, последние сообщения, тулзы, файлы, `pending` (вопрос/аппрув) | | `t3_wait(thread_id, timeout?)` | блокируется до `completed` / `interrupted` / `error`, вопроса или таймаута; та же вьюха | | `t3_answer(thread_id, answers)` | `thread.user-input.respond` на висящий вопрос треда (`answers`: id вопроса → label, список для multi_select) | | `t3_interrupt(thread_id)` | `thread.turn.interrupt` | +| `t3_watch(thread_id?, enabled?)` | без аргументов - списки `watched` / `unwatched`; с обоими - включить или выключить события по треду | ## События -Коннектор держит WS-подписку `orchestration.subscribeShell` на каждую машину (Effect RPC поверх JSON: `Request` → `Chunk` + `Ack`, `Ping` раз в 20 с, реконнект с `afterSequence`, снапшот при реконнекте переигрывает пропущенное). Треды, запущенные через `t3_dispatch`, отслеживаются (`T3CODE_MCP_STATE`, json на volume); переходы `running → completed/interrupted/error` и появление вопроса (`hasPendingUserInput`) превращаются в события `completed` / `interrupted` / `failed` / `question` и уходят `POST` в `T3CODE_MCP_HOOK_URL` (gateway `/hooks/t3code`, bearer `T3CODE_MCP_GATEWAY_TOKEN`, scope `api`) через persisted outbox с ретраями. Gateway делает из события инжект в мастер (`beaver-agent/config.py`, job `t3code`). +Коннектор держит WS-подписку `orchestration.subscribeShell` на каждую машину (Effect RPC поверх JSON: `Request` → `Chunk` + `Ack`, `Ping` раз в 20 с, реконнект с `afterSequence`, снапшот при реконнекте переигрывает пропущенное). Треды, запущенные через `t3_dispatch`, отслеживаются (`T3CODE_MCP_STATE`, json на volume); переходы `running → completed/interrupted/error` и появление вопроса (`hasPendingUserInput`) превращаются в события `completed` / `interrupted` / `failed` / `question` и уходят `POST` в `T3CODE_MCP_HOOK_URL` (gateway `/hooks/t3code`, bearer `T3CODE_MCP_GATEWAY_TOKEN`, scope `api`) через persisted outbox с ретраями. Gateway делает из события инжект в мастер (`beaver-agent/config.py`, job `t3code`). Подписка включается на каждый `t3_dispatch` (в том числе на продолжение треда); снять её - `t3_watch(thread_id, enabled=false)` или сразу `t3_dispatch(..., watch=false)`: тред продолжает работать, просто молча. `t3_wait` - обычный долгий вызов, не MCP Task: Claude Code расширение Tasks не поддерживает, зато сам уводит вызов дольше двух минут в фоновую задачу (`CLAUDE_CODE_MCP_AUTO_BACKGROUND_MS`). Треды бегут в `full-access` без аппрувов; тред ищется по всем машинам, если коннектор перезапускался. diff --git a/docs/PROGRESS.md b/docs/PROGRESS.md index c04cb18..cd6506c 100644 --- a/docs/PROGRESS.md +++ b/docs/PROGRESS.md @@ -2,6 +2,12 @@ > Что сделано, что нет, что проверить руками. Свежие записи сверху. Архитектурная правда - `архитектура.md` §5 (vault, симлинк в `../../архитектура.md`), журнал всей второй итерации - `../../PROGRESS.md`. +## 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 тестов. + +Не сделано: подписаться на тред, который не запускали через `t3_dispatch` (в трекере его нет - `t3_watch` ответит ошибкой); гранулярности по типу события (`question` отдельно от `completed`) нет - флаг один на тред. Проверить руками после рестарта `beaver-t3code-mcp`: `t3_watch()` из диспетчера отдаёт список, `t3_watch("ba48cd69-7c55-461f-9a15-8058425032a0", enabled=false)` - тред фейбла на маке, по которому инжекты больше не нужны (на старом образе снять было нечем: трекер держит состояние в памяти и переписывает json, правка файла не живёт до рестарта). + ## 2026-08-29 - S10b: события тредов и ответы на вопросы Сделано: `watch.py` - WS-подписка `orchestration.subscribeShell` на каждую машину (протокол Effect RPC проверен живьём: тикет `POST /api/auth/websocket-ticket` → `ws://…/ws?wsTicket=`, `Request{id, tag, payload, headers: []}` → `Chunk{values}` + обязательный `Ack`, `Ping` keepalive, `Exit`; реконнект с `afterSequence`, при снапшоте состояние трекаемых тредов переигрывается - переходы, пропущенные офлайн, не теряются). `Tracker` держит треды из `t3_dispatch` в json (`T3CODE_MCP_STATE`, volume `t3code-state`), `apply()` превращает shell-апдейты в события `question` / `approval` / `completed` / `interrupted` / `failed` (`session.lastError`). `Hook` - persisted outbox с ретраями в `T3CODE_MCP_HOOK_URL` (bearer `T3CODE_MCP_GATEWAY_TOKEN`). Тулза `t3_answer(thread_id, answers)` → `thread.user-input.respond` (`requestId` из activity `user-input.requested`, `answers` - id вопроса → label; у Claude id = текст вопроса). `t3_thread`/`t3_wait` отдают `question` (request_id, вопросы, варианты). Общие read-model хелперы вынесены в `views.py`. Gateway: `Job(dedupe=False)` (`beaver-gateway` `0365ab5`), в `beaver-agent/config.py` job `t3code` (webhook) → `inject_master(urgency="urgent", origin="t3code")` с текстом события и подсказкой про `t3_answer`/`say`; compose: env `T3CODE_MCP_HOOK_URL=http://gateway:62990/hooks/t3code`, `T3CODE_MCP_STATE`, volume; токен gateway `t3code-mcp-hooks` (scope `api`) выпущен и лежит в `/root/beaver-agent/.env` как `T3CODE_MCP_GATEWAY_TOKEN`. Живой прогон: диспатч на мак с AskUserQuestion → событие `question` через 10 с в приёмник → `t3_answer(…, {id: "beta"})` → тред дописал файл → событие `completed` с ответом; 13 тестов (трекер, outbox с ретраем, `t3_answer`, трекинг при диспатче). diff --git a/src/t3code_mcp/server.py b/src/t3code_mcp/server.py index c52c371..6bfbdf7 100644 --- a/src/t3code_mcp/server.py +++ b/src/t3code_mcp/server.py @@ -23,7 +23,7 @@ from t3code_mcp import t3, views from t3code_mcp.config import Machine, ModelSpec from t3code_mcp.t3 import T3Client, T3Error from t3code_mcp.views import BUSY -from t3code_mcp.watch import Tracker +from t3code_mcp.watch import Tracked, Tracker if TYPE_CHECKING: from collections.abc import AsyncIterator, Callable, Coroutine @@ -121,6 +121,16 @@ def _project_view( } +def _watch_view(thread_id: str, tracked: Tracked) -> dict[str, Any]: + return { + "thread_id": thread_id, + "machine": tracked.machine, + "project": tracked.project, + "title": tracked.title, + "state": tracked.state, + } + + def build_server( registry: Registry, *, @@ -145,7 +155,8 @@ def build_server( "Coding threads on Бобёр's machines through T3 Code. Flow: t3_machines → " "t3_projects(machine) → t3_dispatch, then either t3_wait (block) or just " "carry on: the gateway injects an event when a dispatched thread finishes, " - "fails or asks a question (answer with t3_answer). Prompts run " + "fails or asks a question (answer with t3_answer; t3_watch turns those " + "events off for a thread). Prompts run " "autonomously (full-access, no approvals): say what to change, what to " "check, what not to touch, and that the answer should end with a report." ), @@ -222,6 +233,15 @@ def build_server( str | None, Field(description="Continue this existing thread instead of creating one"), ] = None, + watch: Annotated[ # noqa: FBT002 - MCP arguments are always named + bool, + Field( + description=( + "Report this thread back as an event; false for a thread you " + "read yourself. Every dispatch resets the flag - see t3_watch" + ) + ), + ] = True, ) -> dict[str, Any]: """Start a coding thread on a machine (or follow up in one); returns at once. @@ -269,7 +289,7 @@ def build_server( ) await _call(client.dispatch(t3.turn_start(tid, text, title_seed=heading))) registry.threads[tid] = machine - registry.tracker.track(tid, machine, target["title"], heading) + registry.tracker.track(tid, machine, target["title"], heading, watch=watch) return { "thread_id": tid, "machine": machine, @@ -278,6 +298,40 @@ def build_server( "title": heading, "model": str(selection), "state": "running", + "watch": watch, + } + + @mcp.tool + async def t3_watch( + thread_id: Annotated[ + str | None, + Field(description="Thread to subscribe or unsubscribe; omit to only list"), + ] = None, + enabled: Annotated[ + bool | None, + Field(description="True - report the thread back again, false - stop"), + ] = None, + ) -> dict[str, Any]: + """Which dispatched threads report back, and turn that on or off per thread. + + t3_dispatch subscribes its thread, so its completion, failure and + questions arrive here as events. Unsubscribe a thread someone reads + directly: it keeps running, it just stops reporting. Called with no + arguments this only lists. + """ + if thread_id is not None: + if enabled is None: + msg = "pass enabled=true or enabled=false along with thread_id" + raise ToolError(msg) + try: + registry.tracker.set_watch(thread_id, enabled=enabled) + except KeyError: + msg = f"thread {thread_id} was not dispatched from here; nothing to " + raise ToolError(msg + ("watch" if enabled else "unwatch")) from None + threads = registry.tracker.threads + return { + "watched": [_watch_view(k, v) for k, v in threads.items() if v.watch], + "unwatched": [_watch_view(k, v) for k, v in threads.items() if not v.watch], } @mcp.tool diff --git a/src/t3code_mcp/watch.py b/src/t3code_mcp/watch.py index d8ccddf..23db3b6 100644 --- a/src/t3code_mcp/watch.py +++ b/src/t3code_mcp/watch.py @@ -7,8 +7,10 @@ read-model carries everything a transition needs - `latestTurn.state`, Reconnects resume with `afterSequence`; a fresh snapshot re-derives every tracked thread, so transitions missed while offline are still reported. -Only threads started through `t3_dispatch` are tracked. Events go to the -gateway webhook through a persisted outbox that retries until accepted. +Only threads started through `t3_dispatch` are tracked, and only while +their `watch` flag is on - `t3_watch` turns it off for a thread someone +reads directly. Events go to the gateway webhook through a persisted +outbox that retries until accepted. """ from __future__ import annotations @@ -18,7 +20,7 @@ import contextlib import json import logging import time -from dataclasses import asdict, dataclass, field +from dataclasses import asdict, dataclass, field, fields from typing import TYPE_CHECKING, Any, Literal import httpx @@ -54,6 +56,7 @@ class Tracked: state: str = "starting" pending: str | None = None error: str | None = None + watch: bool = True updated: float = field(default_factory=time.time) @@ -72,7 +75,11 @@ class Tracker: self.sequences: dict[str, int] = {} if path is not None and path.exists(): raw = json.loads(path.read_text(encoding="utf-8")) - self.threads = {k: Tracked(**v) for k, v in raw.get("threads", {}).items()} + known = {f.name for f in fields(Tracked)} + self.threads = { + k: Tracked(**{n: x for n, x in v.items() if n in known}) + for k, v in raw.get("threads", {}).items() + } self.outbox = raw.get("outbox", []) self.sequences = raw.get("sequences", {}) @@ -91,10 +98,27 @@ class Tracker: tmp.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") tmp.replace(self.path) - def track(self, thread_id: str, machine: str, project: str, title: str) -> None: - self.threads[thread_id] = Tracked(machine=machine, project=project, title=title) + def track( + self, + thread_id: str, + machine: str, + project: str, + title: str, + *, + watch: bool = True, + ) -> None: + self.threads[thread_id] = Tracked( + machine=machine, project=project, title=title, watch=watch + ) self.save() + def set_watch(self, thread_id: str, *, enabled: bool) -> Tracked: + tracked = self.threads[thread_id] + tracked.watch = enabled + tracked.updated = time.time() + self.save() + return tracked + def apply(self, machine: str, shell_thread: dict[str, Any]) -> list[Event]: tracked = self.threads.get(shell_thread["id"]) if tracked is None or tracked.machine != machine: @@ -111,6 +135,8 @@ class Tracker: events.append(Event(TERMINAL[state], shell_thread["id"], machine)) elif tracked.state in views.BUSY and error and not tracked.error: events.append(Event("failed", shell_thread["id"], machine)) + if not tracked.watch: + events.clear() changed = (state, pending, error) != ( tracked.state, tracked.pending, diff --git a/tests/test_server.py b/tests/test_server.py index 981cbe9..a3bed20 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -589,3 +589,64 @@ async def test_dispatch_tracks_thread(client: Client, registry: Registry) -> Non "t3-smoke", "track me", ) + + +def test_tracker_watch_flag(tmp_path) -> None: + from t3code_mcp.watch import Tracker + + tracker = Tracker(tmp_path / "state.json") + tracker.track("t1", "dell", "projects", "quiet", watch=False) + tracker.apply("dell", shell_thread("t1", "running")) + assert tracker.apply("dell", shell_thread("t1", "completed")) == [] + assert tracker.threads["t1"].state == "completed" + tracker.set_watch("t1", enabled=True) + tracker.apply("dell", shell_thread("t1", "running")) + assert [e.kind for e in tracker.apply("dell", shell_thread("t1", "completed"))] == [ + "completed" + ] + assert Tracker(tmp_path / "state.json").threads["t1"].watch is True + + +async def test_watch_tool_lists_and_toggles(client: Client) -> None: + async with client: + loud = data( + await client.call_tool( + "t3_dispatch", + {"machine": "mac", "project": "t3-smoke", "prompt": "tell me"}, + ) + ) + quiet = data( + await client.call_tool( + "t3_dispatch", + { + "machine": "mac", + "project": "t3-smoke", + "prompt": "leave me alone", + "watch": False, + }, + ) + ) + assert (loud["watch"], quiet["watch"]) == (True, False) + view = data(await client.call_tool("t3_watch")) + assert [t["thread_id"] for t in view["watched"]] == [loud["thread_id"]] + assert [t["thread_id"] for t in view["unwatched"]] == [quiet["thread_id"]] + view = data( + await client.call_tool( + "t3_watch", {"thread_id": loud["thread_id"], "enabled": False} + ) + ) + assert view["watched"] == [] + assert {t["thread_id"] for t in view["unwatched"]} == { + loud["thread_id"], + quiet["thread_id"], + } + view = data( + await client.call_tool( + "t3_watch", {"thread_id": quiet["thread_id"], "enabled": True} + ) + ) + assert [t["thread_id"] for t in view["watched"]] == [quiet["thread_id"]] + with pytest.raises(ToolError, match="nothing to unwatch"): + await client.call_tool("t3_watch", {"thread_id": "nope", "enabled": False}) + with pytest.raises(ToolError, match="enabled=true or enabled=false"): + await client.call_tool("t3_watch", {"thread_id": loud["thread_id"]})