feat(server): tell outside agents when a watched thread needs them

apps/server/src/mcp/toolkits/core.test.ts

@@ -44,6 +44,7 @@ import * as ProjectService from "../../project/ProjectService.ts";
 import * as ProviderRegistry from "../../provider/ProviderRegistry.ts";
 import * as SecretRequests from "../../secrets/SecretRequests.ts";
 import * as ScheduledTaskService from "../../scheduledTasks/ScheduledTaskService.ts";
+import * as ThreadWatchReactor from "../../orchestration-v2/ThreadWatchReactor.ts";
 import * as ServerSettings from "../../serverSettings.ts";
 import * as McpHttpServer from "../McpHttpServer.ts";
 import * as McpInvocationContext from "../McpInvocationContext.ts";
@@ -71,6 +72,7 @@ const decodeThreadReadResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadR
 const layerThreadToolkit = McpHttpServer.layerThreadToolkit.pipe(
   Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})),
   Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
+  Layer.provide(Layer.mock(ThreadWatchReactor.ThreadWatchReactor)({})),
 );
 
 const threadId = ThreadId.make("mcp-core-thread");
@@ -542,6 +544,7 @@ it.effect("refuses act-as-caller tools to a client caller", () =>
         Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})),
         Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})),
         Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
+        Layer.provide(Layer.mock(ThreadWatchReactor.ThreadWatchReactor)({})),
         Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
         Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
       ),
@@ -679,6 +682,7 @@ it.effect.each(["t3_thread_interrupt", "t3_workflow_stop"])(
           Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})),
           Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})),
           Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
+          Layer.provide(Layer.mock(ThreadWatchReactor.ThreadWatchReactor)({})),
           Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
           Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
         ),
@@ -771,6 +775,7 @@ it.effect("discovers workflow stop targets and requires orchestration capability
         Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})),
         Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})),
         Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
+        Layer.provide(Layer.mock(ThreadWatchReactor.ThreadWatchReactor)({})),
         Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
         Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
       ),

apps/server/src/mcp/toolkits/project/handlers.test.ts

@@ -26,6 +26,7 @@ import * as GitVcsDriver from "../../../vcs/GitVcsDriver.ts";
 import * as VcsProcess from "../../../vcs/VcsProcess.ts";
 import * as McpInvocationContext from "../../McpInvocationContext.ts";
 import * as McpToolAccess from "../../McpToolAccess.ts";
+import * as ThreadWatchReactor from "../../../orchestration-v2/ThreadWatchReactor.ts";
 import * as ProjectHandlers from "./handlers.ts";
 import { ProjectToolkit } from "./tools.ts";
 
@@ -44,6 +45,7 @@ const clientLaunchHarness = (input: {
     model: "claude-opus",
   };
   const layerDependencies = Layer.mergeAll(
+    Layer.mock(ThreadWatchReactor.ThreadWatchReactor)({}),
     ServerSettings.layerTest({
       ...(input.defaultRuntimeMode === undefined
         ? {}

apps/server/src/mcp/toolkits/project/handlers.ts

@@ -1,4 +1,10 @@
-import { MessageId, ThreadId, OrchestratorMcpFailure, ProjectId } from "@t3tools/contracts";
+import {
+  MessageId,
+  ThreadId,
+  OrchestratorMcpFailure,
+  ProjectId,
+  type SecretRequestError,
+} from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Option from "effect/Option";
@@ -9,6 +15,7 @@ import * as Project from "../../../project/ProjectService.ts";
 import * as ManagedProjectFolders from "../../../project/ManagedProjectFolders.ts";
 import * as Repositories from "../../../sourceControl/SourceControlRepositoryService.ts";
 import * as GitVcsDriver from "../../../vcs/GitVcsDriver.ts";
+import * as ThreadWatchReactor from "../../../orchestration-v2/ThreadWatchReactor.ts";
 import * as ServerSettings from "../../../serverSettings.ts";
 import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
 import { resolveRuntimeMode } from "../../OrchestratorMcpService.ts";
@@ -49,6 +56,21 @@ const assertProjectWorktree = Effect.fn("mcp.assertProjectWorktree")(function* (
     });
 });
 
+const watchRefused = (
+  error:
+    | ThreadWatchReactor.ThreadWatchInvalidError
+    | ThreadWatchReactor.ThreadWatchStoreError
+    | SecretRequestError,
+) =>
+  Effect.fail(
+    error._tag === "ThreadWatchStoreError"
+      ? unavailable()
+      : new OrchestratorMcpFailure({
+          code: "invalid_request",
+          message: `${error.message} No thread was created.`,
+        }),
+  );
+
 const access = Effect.gen(function* () {
   yield* readCaller();
   return yield* Project.ProjectService;
@@ -158,7 +180,7 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, {
                 Effect.orElseSucceed(() => runtimeMode),
               )
             : runtimeMode;
-        const result = yield* ThreadMessageIntake.launchThread({
+        const launch = ThreadMessageIntake.launchThread({
           commandId,
           threadId,
           projectId,
@@ -196,6 +218,20 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, {
             return unavailable();
           }),
         );
+        const delivery = input.watch;
+        const result =
+          delivery === undefined
+            ? yield* launch
+            : yield* ThreadWatchReactor.ThreadWatchReactor.pipe(
+                Effect.flatMap((watches) =>
+                  watches.watchLaunch({ threadId, projectId, delivery, launch }),
+                ),
+                Effect.catchTags({
+                  ThreadWatchInvalidError: watchRefused,
+                  ThreadWatchStoreError: watchRefused,
+                  SecretRequestError: watchRefused,
+                }),
+              );
         const thread = result.projection.thread;
         const run = result.projection.runs.find((run) => run.userMessageId === messageId);
         return {

apps/server/src/mcp/toolkits/project/tools.ts

@@ -30,6 +30,8 @@ import * as ThreadManagementService from "../../../orchestration-v2/ThreadManage
 import * as SourceControlRepositoryService from "../../../sourceControl/SourceControlRepositoryService.ts";
 import * as GitVcsDriver from "../../../vcs/GitVcsDriver.ts";
 import * as McpInvocationContext from "../../McpInvocationContext.ts";
+import * as ThreadWatchReactor from "../../../orchestration-v2/ThreadWatchReactor.ts";
+import { ThreadWatchDelivery } from "../thread/tools.ts";
 
 const shared = {
   success: Project,
@@ -130,6 +132,12 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", {
       }),
     ),
     attachments: Schema.optional(Schema.Array(McpAttachmentInput).check(Schema.isMaxLength(8))),
+    watch: Schema.optional(
+      ThreadWatchDelivery.annotate({
+        description:
+          "Watch the new thread as t3_thread_watch does, from before its first run. A delivery that cannot be used is refused and no thread is created.",
+      }),
+    ),
   }),
   success: Schema.Struct({
     threadId: ThreadId,
@@ -146,6 +154,7 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", {
     FileSystem.FileSystem,
     ServerConfig.ServerConfig,
     ServerSettings.ServerSettingsService,
+    ThreadWatchReactor.ThreadWatchReactor,
   ],
 })
   .annotate(Tool.Destructive, true)

apps/server/src/mcp/toolkits/thread/handlers.ts

@@ -20,6 +20,7 @@ import {
 } from "../../threadAccess.ts";
 import * as ThreadSearch from "../../../orchestration-v2/ThreadSearch.ts";
 import * as ScheduledTasks from "../../../scheduledTasks/ScheduledTaskService.ts";
+import * as ThreadWatchReactor from "../../../orchestration-v2/ThreadWatchReactor.ts";
 import { queuedRunsInDeliveryOrder } from "../../../orchestration-v2/QueuedRunOrder.ts";
 import { ThreadToolkit } from "./tools.ts";
 
@@ -75,11 +76,36 @@ const readQuestion = Effect.fn("mcp.readQuestion")(function* (input: {
     });
   return { ...context, request, item };
 });
+const invalidRequest = (error: { readonly message: string }) =>
+  new OrchestratorMcpFailure({ code: "invalid_request", message: error.message });
 const writesThread = <P extends { readonly threadId?: ThreadId | undefined }, A, E, R>(
   handle: (params: P) => Effect.Effect<A, E, R>,
 ) => McpToolAccess.writesThreads((params: P) => [params.threadId], handle);
 
 export const layer = McpToolAccess.toLayer(ThreadToolkit, {
+  t3_thread_watch: writesThread((input) =>
+    Effect.gen(function* () {
+      const { projection } = yield* readThread(input.threadId);
+      const { id: threadId, projectId } = projection.thread;
+      const reactor = yield* ThreadWatchReactor.ThreadWatchReactor;
+      if (input.action === "unwatch") {
+        const { wasWatching } = yield* reactor
+          .unwatch({ threadId, url: input.delivery.url })
+          .pipe(Effect.mapError(unavailable));
+        return { threadId, watching: false, wasWatching };
+      }
+      const { wasWatching } = yield* reactor
+        .watch({ threadId, projectId, delivery: input.delivery })
+        .pipe(
+          Effect.catchTags({
+            ThreadWatchInvalidError: (error) => Effect.fail(invalidRequest(error)),
+            SecretRequestError: (error) => Effect.fail(invalidRequest(error)),
+            ThreadWatchStoreError: () => Effect.fail(unavailable()),
+          }),
+        );
+      return { threadId, watching: true, wasWatching };
+    }),
+  ),
   run_scheduled_task_now: McpToolAccess.writesEnvironment((input) =>
     Effect.gen(function* () {
       const scheduler = yield* ScheduledTasks.ScheduledTaskService;

apps/server/src/mcp/toolkits/thread/tools.ts

@@ -18,13 +18,15 @@ import {
   RunId,
   NonNegativeInt,
   ProjectId,
+  SecretRef,
 } from "@t3tools/contracts";
 import * as Crypto from "effect/Crypto";
 import * as Schema from "effect/Schema";
 import { Tool, Toolkit } from "effect/ai";
 
 import * as ThreadSearch from "../../../orchestration-v2/ThreadSearch.ts";
 import * as ScheduledTaskService from "../../../scheduledTasks/ScheduledTaskService.ts";
+import * as ThreadWatchReactor from "../../../orchestration-v2/ThreadWatchReactor.ts";
 import * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts";
 import * as McpInvocationContext from "../../McpInvocationContext.ts";
 
@@ -271,6 +273,50 @@ const ScheduledTaskRunTool = Tool.make("run_scheduled_task_now", {
   .annotate(Tool.Destructive, true)
   .annotate(Tool.OpenWorld, true);
 
+export const ThreadWatchDelivery = Schema.Struct({
+  url: TrimmedNonEmptyString.annotate({
+    description: "https endpoint that receives the signed POSTs. http only for localhost.",
+  }),
+  secret: Schema.optional(
+    Schema.String.annotate({
+      description:
+        "Standard Webhooks signing secret your endpoint verifies with: whsec_ plus base64 of 24 to 64 random bytes.",
+    }),
+  ),
+  secretRef: Schema.optional(
+    SecretRef.annotate({
+      description:
+        "A secret the user entered through request_secret, used instead of secret so it never appears in this conversation. Watching consumes it.",
+    }),
+  ),
+}).annotate({
+  description:
+    "Where to deliver. watch needs exactly one of secret or secretRef; unwatch needs only url.",
+});
+
+const ThreadWatchTool = Tool.make("t3_thread_watch", {
+  ...commandTool,
+  description:
+    'Have T3 Code POST to your endpoint when a thread needs you: its run completes or fails, it asks a question, or it waits for an approval. Use this when you drive T3 Code from outside and your turn ends before the thread does, instead of holding t3_thread_wait open or polling. action watch needs delivery with a secret or secretRef; unwatch needs only delivery.url. Requests follow Standard Webhooks (webhook-id, webhook-timestamp and webhook-signature headers, HMAC-SHA256 with your secret) and carry the MCP Events body {eventId, name:"thread.watch", timestamp, data:{environmentId, threadId, title, kind, runId}, cursor:null}, where kind is completed, failed, question, approval or ended. The body is status only; read the thread for its content. A watch starts from the thread as it is now, so watch before you read the thread and nothing falls in between; a server restart starts it over from the thread as it is then. A failed request is retried a few times with the same webhook-id, so drop ids you have seen. A run the user stops sends nothing, as mobile alerts do. Settling, archiving or deleting the thread ends the watch with an ended request; a 410 from your endpoint ends it too. Settled, archived and subagent threads cannot be watched. Omit threadId for this thread.',
+
+  parameters: Schema.Struct({
+    action: Schema.Literals(["watch", "unwatch"]),
+    threadId: Schema.optional(ThreadId),
+    delivery: ThreadWatchDelivery,
+  }),
+  success: Schema.Struct({
+    threadId: ThreadId,
+    watching: Schema.Boolean.annotate({
+      description: "Whether this endpoint now watches the thread.",
+    }),
+    wasWatching: Schema.Boolean.annotate({ description: "Whether it did before the call." }),
+  }),
+  dependencies: [...commandTool.dependencies, ThreadWatchReactor.ThreadWatchReactor],
+})
+  .annotate(Tool.Title, "Watch a thread")
+  .annotate(Tool.Destructive, false)
+  .annotate(Tool.OpenWorld, true);
+
 export const ThreadToolkit = Toolkit.make(
   ScheduledTaskRunTool,
   ThreadSearchTool,
@@ -289,4 +335,5 @@ export const ThreadToolkit = Toolkit.make(
   QueueCancelTool,
   QueueReorderTool,
   QueuePromoteTool,
+  ThreadWatchTool,
 );