feat(watch,server): t3_watch to turn thread events off and back on

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` без аппрувов; тред ищется по всем машинам, если коннектор перезапускался.
 

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`, трекинг при диспатче).

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

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,

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"]})