diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 02216cec576d..d349bae0c07c 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/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)({})), ), diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index 4c22393bf476..fbd411172f0b 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/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 ? {} diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index 380a1b52ec96..f2dbd77d0d78 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/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 { diff --git a/apps/server/src/mcp/toolkits/project/tools.ts b/apps/server/src/mcp/toolkits/project/tools.ts index 2114565d8201..ae9b37a4fe27 100644 --- a/apps/server/src/mcp/toolkits/project/tools.ts +++ b/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) diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index 4ab78b4fe07b..f8d7dd6740d8 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/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 =

( handle: (params: P) => Effect.Effect, ) => 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; diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index 21750cbb6eaa..aa020efa1702 100644 --- a/apps/server/src/mcp/toolkits/thread/tools.ts +++ b/apps/server/src/mcp/toolkits/thread/tools.ts @@ -18,6 +18,7 @@ import { RunId, NonNegativeInt, ProjectId, + SecretRef, } from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; import * as Schema from "effect/Schema"; @@ -25,6 +26,7 @@ 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, ); diff --git a/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts b/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts new file mode 100644 index 000000000000..c60f4d142771 --- /dev/null +++ b/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts @@ -0,0 +1,533 @@ +// @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect's Crypto has no createHmac. +import * as NodeCrypto from "node:crypto"; + +import { assert, describe, it } from "@effect/vitest"; +import { + EnvironmentId, + EventId, + NodeId, + type OrchestrationV2DomainEvent, + type OrchestrationV2ThreadShell, + ProjectId, + ProviderInstanceId, + RunId, + RuntimeRequestId, + SecretRef, + SecretRequestError, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; + +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as SecretRequests from "../secrets/SecretRequests.ts"; +import * as ThreadManagementService from "./ThreadManagementService.ts"; +import * as ThreadWatchReactor from "./ThreadWatchReactor.ts"; + +const THREAD_ID = ThreadId.make("watched-thread"); +const PROJECT_ID = ProjectId.make("watched-project"); +const ENVIRONMENT_ID = EnvironmentId.make("watch-environment"); +const RUN_1 = RunId.make("run-1"); +const RUN_2 = RunId.make("run-2"); +const URL = "https://receiver.example.test/hook"; +const SECRET = `whsec_${Buffer.alloc(32, 7).toString("base64")}`; +const at = (iso: string) => DateTime.makeUnsafe(iso); + +function shell(overrides: Partial = {}): OrchestrationV2ThreadShell { + return { + id: THREAD_ID, + projectId: PROJECT_ID, + title: "[bot] GM ticket desk", + providerInstanceId: ProviderInstanceId.make("codex"), + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "test-model" }, + runtimeMode: "full-access", + interactionMode: "default", + worktreePath: null, + activeProviderThreadId: null, + lineage: { rootThreadId: THREAD_ID, parentThreadId: null, relationshipToParent: null }, + forkedFrom: null, + createdBy: "agent", + creationSource: "mcp", + activeRunId: null, + latestVisibleMessage: null, + hasActionableProposedPlan: false, + itemCount: 0, + visibleItemCount: 0, + lastVisitedAt: null, + deletedAt: null, + branch: null, + linkedPullRequest: null, + status: "running", + activityRunStatus: "running", + pendingRuntimeRequest: null, + pendingBackgroundTasks: [], + latestRunId: RUN_1, + latestRunRequestedAt: null, + latestRunStartedAt: null, + latestRunCompletedAt: null, + latestUserMessageAt: null, + createdAt: at("2026-10-20T12:00:00.000Z"), + updatedAt: at("2026-10-20T12:00:00.000Z"), + archivedAt: null, + settledOverride: null, + settledAt: null, + snoozedUntil: null, + snoozedAt: null, + pinnedAt: null, + ...overrides, + }; +} + +const completed = (runId = RUN_1, completedAt = "2026-10-20T12:05:00.000Z") => + shell({ + status: "completed", + activityRunStatus: null, + latestRunId: runId, + latestRunCompletedAt: at(completedAt), + }); + +interface Received { + readonly url: string; + readonly headers: Headers; + readonly body: string; +} + +const makeHarness = Effect.fnUntraced(function* ( + options: { + readonly respond?: (attempt: number, request: Received) => Response | Promise; + readonly secretRefValue?: string; + readonly whileConsuming?: ( + current: Ref.Ref, + ) => Effect.Effect; + readonly domainEvents?: Stream.Stream; + readonly initial?: OrchestrationV2ThreadShell | null; + } = {}, +) { + const current = yield* Ref.make( + options.initial === undefined ? shell() : options.initial, + ); + const received: Array = []; + const consumed: Array<{ readonly ref: string; readonly projectId: string }> = []; + const fetch: typeof globalThis.fetch = Object.assign( + ( + input: Parameters[0], + init?: Parameters[1], + ) => { + const body = init?.body; + const request = { + url: String(input), + headers: new Headers(init?.headers), + body: body instanceof Uint8Array ? new TextDecoder().decode(body) : String(body), + }; + received.push(request); + return Promise.resolve( + options.respond?.(received.length, request) ?? new Response(null, { status: 204 }), + ); + }, + { preconnect: () => {} }, + ); + const reactor = yield* ThreadWatchReactor.make.pipe( + Effect.provide( + Layer.mergeAll( + Layer.mock(ThreadManagementService.ThreadManagementService)({ + // One shell stands in for every thread, under the id asked for. + getThreadShell: (threadId) => + Ref.get(current).pipe( + Effect.map((value) => (value === null ? null : { ...value, id: threadId })), + ), + streamDomainEvents: options.domainEvents ?? Stream.empty, + }), + Layer.mock(SecretRequests.SecretRequests)({ + consume: ({ ref, projectId }) => + Effect.suspend(() => { + consumed.push({ ref, projectId }); + const value = options.secretRefValue; + return (options.whileConsuming?.(current) ?? Effect.void).pipe( + Effect.andThen( + value === undefined + ? Effect.fail(new SecretRequestError({ reason: "ref_expired" })) + : Effect.succeed(value), + ), + ); + }), + }), + Layer.succeed(ServerEnvironment.ServerEnvironment, { + getEnvironmentId: Effect.succeed(ENVIRONMENT_ID), + getDescriptor: Effect.die("unused"), + }), + FetchHttpClient.layer, + Layer.succeed(FetchHttpClient.Fetch, fetch), + ), + ), + ); + const watch = ( + delivery: Parameters[0]["delivery"] = { url: URL, secret: SECRET }, + threadId = THREAD_ID, + ) => reactor.watch({ threadId, projectId: PROJECT_ID, delivery }); + const settle = (next: OrchestrationV2ThreadShell | null, threadId = THREAD_ID) => + Ref.set(current, next).pipe( + Effect.andThen(reactor.checkThread(threadId)), + Effect.andThen(reactor.drain), + ); + return { current, received, consumed, reactor, watch, settle }; +}); + +/** Whether the default endpoint still watches; ends the watch to find out. */ +const stillWatching = (reactor: ThreadWatchReactor.ThreadWatchReactor["Service"]) => + reactor + .unwatch({ threadId: THREAD_ID, url: URL }) + .pipe(Effect.map(({ wasWatching }) => wasWatching)); + +/** Verifies as a Standard Webhooks receiver would, independently of the sender's code. */ +function verified(request: Received | undefined) { + assert.isDefined(request); + const id = request!.headers.get("webhook-id") ?? ""; + const timestamp = request!.headers.get("webhook-timestamp") ?? ""; + const key = Buffer.from(SECRET.slice("whsec_".length), "base64"); + const expected = NodeCrypto.createHmac("sha256", key) + .update(`${id}.${timestamp}.${request!.body}`) + .digest("base64"); + assert.strictEqual(request!.headers.get("webhook-signature"), `v1,${expected}`); + assert.strictEqual(request!.headers.get("content-type"), "application/json"); + return JSON.parse(request!.body) as { + readonly eventId: string; + readonly name: string; + readonly data: Record; + readonly cursor: null; + }; +} + +describe("ThreadWatchReactor", () => { + it.effect("sends one signed MCP Events request per change, from the watch on", () => + Effect.gen(function* () { + const { received, watch, settle } = yield* makeHarness({ initial: completed() }); + assert.deepStrictEqual(yield* watch(), { wasWatching: false }); + yield* settle(completed()); + assert.strictEqual(received.length, 0, "what happened before the watch is not sent"); + + yield* settle(shell({ latestRunId: RUN_2 })); + yield* settle(completed(RUN_2, "2026-10-20T12:30:00.000Z")); + yield* settle(completed(RUN_2, "2026-10-20T12:30:00.000Z")); + assert.strictEqual(received.length, 1); + assert.strictEqual(received[0]?.url, URL); + assert.match(received[0]?.headers.get("x-mcp-subscription-id") ?? "", /^tw_[0-9a-f]{32}$/); + const body = verified(received[0]); + assert.strictEqual(body.name, "thread.watch"); + assert.strictEqual(body.eventId, received[0]?.headers.get("webhook-id")); + assert.strictEqual(body.cursor, null); + assert.deepStrictEqual(body.data, { + environmentId: ENVIRONMENT_ID, + threadId: THREAD_ID, + title: "[bot] GM ticket desk", + kind: "completed", + runId: RUN_2, + }); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("a launch is watched before its thread exists, and a launch that fails ends it", () => + Effect.gen(function* () { + const delivery = { url: URL, secret: SECRET }; + const launched = yield* makeHarness({ initial: null }); + const launch = (harness: typeof launched, effect: Effect.Effect) => + harness.reactor + .watchLaunch({ threadId: THREAD_ID, projectId: PROJECT_ID, delivery, launch: effect }) + .pipe(Effect.exit); + yield* launch(launched, Effect.void); + yield* launched.reactor.drain; + assert.strictEqual(launched.received.length, 0, "the thread is still being launched"); + yield* launched.settle(shell({ status: "failed", activityRunStatus: null })); + assert.deepInclude(verified(launched.received[0]).data, { kind: "failed", runId: RUN_1 }); + + for (const failure of [Effect.fail("no thread"), Effect.interrupt]) { + const unborn = yield* makeHarness({ initial: null }); + yield* launch(unborn, failure); + yield* unborn.reactor.drain; + assert.deepInclude(verified(unborn.received[0]).data, { kind: "ended" }); + assert.isFalse(yield* stillWatching(unborn.reactor)); + } + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("after a restart, a watch picks up from the thread as it is then", () => + Effect.gen(function* () { + const first = yield* makeHarness(); + yield* first.watch(); + const restarted = yield* makeHarness({ initial: shell() }); + assert.deepStrictEqual(yield* restarted.watch(), { wasWatching: true }); + yield* restarted.reactor.start(); + yield* restarted.reactor.drain; + yield* restarted.settle(completed()); + assert.deepInclude(verified(restarted.received[0]).data, { kind: "completed", runId: RUN_1 }); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("a change made while a watch is being registered is still sent", () => + Effect.gen(function* () { + const { received, reactor, watch } = yield* makeHarness({ + secretRefValue: SECRET, + whileConsuming: (current) => Ref.set(current, completed()), + }); + yield* watch({ + url: URL, + secretRef: SecretRef.make("secret-ref:0123456789abcdef0123456789abcdef"), + }); + yield* reactor.drain; + assert.deepInclude(verified(received[0]).data, { kind: "completed", runId: RUN_1 }); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("follows thread events, including a request_secret wait", () => + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const pulls = yield* Queue.unbounded(); + const { current, received, reactor, watch } = yield* makeHarness({ + domainEvents: Stream.fromEffectRepeat( + Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))), + ), + }); + yield* watch(); + yield* reactor.start(); + yield* Queue.take(pulls); + const asked = at("2026-10-20T12:40:00.000Z"); + const itemId = TurnItemId.make("item:secret"); + yield* Ref.set( + current, + shell({ + pendingRuntimeRequest: { + id: RuntimeRequestId.make(itemId), + kind: "user_input", + createdAt: asked, + blocking: true, + }, + }), + ); + yield* Queue.offer(events, { + id: EventId.make("event:secret"), + type: "turn-item.updated", + threadId: THREAD_ID, + runId: RUN_1, + occurredAt: asked, + payload: { + id: itemId, + threadId: THREAD_ID, + runId: RUN_1, + nodeId: NodeId.make("node:secret"), + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "waiting", + title: null, + startedAt: asked, + completedAt: null, + updatedAt: asked, + type: "secret_request", + label: "Webhook secret", + reason: "Sign deliveries", + secretStatus: "pending", + }, + }); + yield* Queue.take(pulls); + yield* reactor.drain; + assert.deepInclude(verified(received[0]).data, { kind: "question" }); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("settling sends what changed, then ended, and ends the watch", () => + Effect.gen(function* () { + const { received, reactor, settle, watch } = yield* makeHarness(); + yield* watch(); + yield* settle({ ...completed(), settledOverride: "settled" }); + assert.deepStrictEqual( + received.map((request) => verified(request).data["kind"]), + ["completed", "ended"], + ); + yield* settle(completed(RUN_2, "2026-10-20T12:30:00.000Z")); + assert.strictEqual(received.length, 2); + assert.isFalse(yield* stillWatching(reactor)); + + // Watched again after unsettling, a thread can end with nothing else changed. + const again = yield* makeHarness(); + const settled = shell({ settledOverride: "settled" }); + yield* again.watch(); + yield* again.settle(settled); + yield* again.settle(shell()); + yield* again.watch(); + yield* TestClock.adjust("1 minute"); + yield* again.settle(settled); + const [first, second] = again.received; + assert.notStrictEqual(second?.headers.get("webhook-id"), first?.headers.get("webhook-id")); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("a 410 from the receiver ends the watch", () => + Effect.gen(function* () { + const { received, reactor, settle, watch } = yield* makeHarness({ + respond: () => new Response(null, { status: 410 }), + }); + yield* watch(); + yield* settle(completed()); + yield* settle(completed(RUN_2, "2026-10-20T12:30:00.000Z")); + assert.strictEqual(received.length, 1); + assert.isFalse(yield* stillWatching(reactor)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("retries with the same event id, and unwatching drops the retry", () => + Effect.gen(function* () { + const firstTry = yield* Deferred.make(); + const retried = yield* makeHarness({ + respond: (attempt) => { + Deferred.doneUnsafe(firstTry, Effect.void); + return new Response(null, { status: attempt === 1 ? 503 : 204 }); + }, + }); + yield* retried.watch(); + yield* Ref.set(retried.current, completed()); + yield* retried.reactor.checkThread(THREAD_ID); + yield* Deferred.await(firstTry); + const draining = yield* Effect.forkChild(retried.reactor.drain); + yield* TestClock.adjust("2 seconds"); + yield* Fiber.join(draining); + const [first, retry] = retried.received; + assert.strictEqual(retried.received.length, 2); + assert.strictEqual(retry?.headers.get("webhook-id"), first?.headers.get("webhook-id")); + verified(retry); + + const firstAttempt = yield* Deferred.make(); + const dropped = yield* makeHarness({ + respond: () => { + Deferred.doneUnsafe(firstAttempt, Effect.void); + return new Response(null, { status: 503 }); + }, + }); + yield* dropped.watch({ url: "https://other.example.test/hook", secret: SECRET }); + yield* Ref.set(dropped.current, completed()); + yield* dropped.reactor.checkThread(THREAD_ID); + yield* Deferred.await(firstAttempt); + yield* dropped.reactor.unwatch({ + threadId: THREAD_ID, + url: "https://other.example.test/hook", + }); + const settling = yield* Effect.forkChild(dropped.reactor.drain); + yield* TestClock.adjust("1 minute"); + yield* Fiber.join(settling); + assert.strictEqual(dropped.received.length, 1); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("a late 410 for a replaced watch leaves the replacement in place", () => + Effect.gen(function* () { + const requested = yield* Deferred.make(); + let answer: (response: Response) => void = () => {}; + const first = yield* makeHarness({ + respond: () => { + // Made before signalling: the signal can resume the test at once. + const response = new Promise((resolve) => (answer = resolve)); + Deferred.doneUnsafe(requested, Effect.void); + return response; + }, + }); + yield* first.watch(); + yield* Ref.set(first.current, completed()); + yield* first.reactor.checkThread(THREAD_ID); + yield* Deferred.await(requested); + yield* first.watch({ url: URL, secret: `whsec_${Buffer.alloc(32, 9).toString("base64")}` }); + answer(new Response(null, { status: 410 })); + yield* first.reactor.drain; + const restarted = yield* makeHarness(); + assert.isTrue(yield* stillWatching(restarted.reactor), "the replacement's row survives"); + assert.isTrue(yield* stillWatching(first.reactor)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("a receiver that never answers does not hold up another thread", () => + Effect.gen(function* () { + const other = ThreadId.make("other-thread"); + const hanging = "https://hanging.example.test/hook"; + const otherDelivered = yield* Deferred.make(); + const { current, received, reactor, watch } = yield* makeHarness({ + respond: (_attempt, request) => { + if (request.url === hanging) return new Promise(() => {}); + Deferred.doneUnsafe(otherDelivered, Effect.void); + return new Response(null, { status: 204 }); + }, + }); + yield* watch({ url: hanging, secret: SECRET }); + yield* watch({ url: URL, secret: SECRET }, other); + yield* Ref.set(current, completed()); + yield* reactor.checkThread(THREAD_ID); + yield* reactor.checkThread(other); + yield* Deferred.await(otherDelivered); + assert.deepStrictEqual( + received.map((request) => request.url), + [hanging, URL], + ); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("refuses deliveries it could never sign or should never send", () => + Effect.gen(function* () { + const { current, watch } = yield* makeHarness(); + const refused = (delivery: Parameters[0]) => + watch(delivery).pipe( + Effect.flip, + Effect.map((error) => error.message), + ); + assert.include( + yield* refused({ url: "http://receiver.example.test/", secret: SECRET }), + "https", + ); + assert.include( + yield* refused({ url: "https://u:p@receiver.example.test/", secret: SECRET }), + "https", + ); + assert.include(yield* refused({ url: URL, secret: "shared-secret" }), "whsec_"); + assert.include(yield* refused({ url: URL }), "exactly one"); + yield* Ref.set( + current, + shell({ + lineage: { + rootThreadId: ThreadId.make("parent"), + parentThreadId: ThreadId.make("parent"), + relationshipToParent: "subagent", + }, + }), + ); + assert.include(yield* refused({ url: URL, secret: SECRET }), "parent"); + yield* Ref.set(current, shell({ settledOverride: "settled" })); + assert.include(yield* refused({ url: URL, secret: SECRET }), "Unsettle"); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); + + it.effect("stores one watch per URL, takes secrets from a secretRef, and unwatches by URL", () => + Effect.gen(function* () { + const { consumed, reactor, watch } = yield* makeHarness({ secretRefValue: SECRET }); + const ref = SecretRef.make("secret-ref:0123456789abcdef0123456789abcdef"); + yield* watch({ url: "https://Receiver.Example.Test/hook", secretRef: ref }); + assert.deepStrictEqual(consumed, [{ ref, projectId: PROJECT_ID }]); + assert.deepStrictEqual(yield* watch({ url: "http://localhost:8787/hook", secret: SECRET }), { + wasWatching: false, + }); + assert.deepStrictEqual(yield* watch(), { wasWatching: true }, "one watch per URL"); + const unwatch = (url: string) => reactor.unwatch({ threadId: THREAD_ID, url }); + assert.deepStrictEqual(yield* unwatch("https://RECEIVER.example.test/hook"), { + wasWatching: true, + }); + assert.deepStrictEqual(yield* unwatch(URL), { wasWatching: false }); + assert.deepStrictEqual(yield* unwatch("http://localhost:8787/hook"), { wasWatching: true }); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + ); +}); diff --git a/apps/server/src/orchestration-v2/ThreadWatchReactor.ts b/apps/server/src/orchestration-v2/ThreadWatchReactor.ts new file mode 100644 index 000000000000..db78b00ed83a --- /dev/null +++ b/apps/server/src/orchestration-v2/ThreadWatchReactor.ts @@ -0,0 +1,381 @@ +import { + type OrchestrationV2DomainEvent, + type ProjectId, + type SecretRef, + type SecretRequestError, + ThreadId, +} from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as FiberSet from "effect/FiberSet"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schedule from "effect/Schedule"; +import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; +import type * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; +import * as HttpBody from "effect/http/HttpBody"; +import * as HttpClient from "effect/http/HttpClient"; + +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import { + makeAgentAwarenessPublishWorker, + shouldPublishAgentAwarenessEvent, +} from "../relay/AgentAwarenessRelay.ts"; +import * as SecretRequests from "../secrets/SecretRequests.ts"; +import { forkParked } from "../serverActivation.ts"; +import * as ThreadManagementService from "./ThreadManagementService.ts"; +import { + observeThreadWatch, + standardWebhookKey, + standardWebhookSignature, + threadWatchEventBody, + threadWatchEventId, + threadWatchId, + type ThreadWatchState, + UNSEEN, +} from "./threadWatch.ts"; + +export class ThreadWatchInvalidError extends Schema.TaggedError()( + "ThreadWatchInvalidError", + { reason: Schema.Literals(["url", "secret", "secret_source", "subagent", "ended"]) }, +) { + override get message(): string { + switch (this.reason) { + case "url": + return "delivery.url must be an https URL without credentials or a fragment. Plain http is accepted only for localhost."; + case "secret": + return "The signing secret must be a Standard Webhooks secret: whsec_ followed by base64 of 24 to 64 random bytes."; + case "secret_source": + return "Pass exactly one of delivery.secret or delivery.secretRef."; + case "subagent": + return "This thread is a subagent. Watch its parent thread instead."; + case "ended": + return "This thread is settled, archived or deleted, so there is nothing to watch. Unsettle or unarchive it first."; + } + } +} + +export class ThreadWatchStoreError extends Schema.TaggedError()( + "ThreadWatchStoreError", + { cause: Schema.Defect() }, +) { + override get message(): string { + return "Could not update thread watches."; + } +} + +class ThreadWatchDeliveryError extends Schema.TaggedError()( + "ThreadWatchDeliveryError", + { status: Schema.NullOr(Schema.Number) }, +) {} + +interface ThreadWatchDelivery { + readonly url: string; + readonly secret?: string | undefined; + readonly secretRef?: SecretRef | undefined; +} + +export class ThreadWatchReactor extends Context.Service< + ThreadWatchReactor, + { + readonly start: () => Effect.Effect; + readonly checkThread: (threadId: ThreadId) => Effect.Effect; + readonly watch: (input: { + readonly threadId: ThreadId; + readonly projectId: ProjectId; + readonly delivery: ThreadWatchDelivery; + }) => Effect.Effect< + { readonly wasWatching: boolean }, + ThreadWatchInvalidError | ThreadWatchStoreError | SecretRequestError + >; + readonly unwatch: (input: { + readonly threadId: ThreadId; + readonly url: string; + }) => Effect.Effect<{ readonly wasWatching: boolean }, ThreadWatchStoreError>; + /** Watches the thread `launch` creates before it exists, so its first run cannot be missed. */ + readonly watchLaunch: (input: { + readonly threadId: ThreadId; + readonly projectId: ProjectId; + readonly delivery: ThreadWatchDelivery; + readonly launch: Effect.Effect; + }) => Effect.Effect< + A, + E | ThreadWatchInvalidError | ThreadWatchStoreError | SecretRequestError, + R + >; + readonly drain: Effect.Effect; + } +>()("t3/orchestration-v2/ThreadWatchReactor") {} + +interface Watch { + readonly id: string; + readonly threadId: ThreadId; + readonly url: string; + readonly key: Uint8Array; +} + +function threadWatchUrl(value: string): string | null { + if (!URL.canParse(value)) return null; + const url = new URL(value); + if (url.username !== "" || url.password !== "" || url.hash !== "") return null; + const loopback = + url.hostname === "localhost" || url.hostname === "127.0.0.1" || url.hostname === "[::1]"; + return url.protocol === "https:" || (url.protocol === "http:" && loopback) ? url.href : null; +} + +// A `request_secret` wait shows as pending input, but its only event is a turn +// item the mobile alert filter skips. +const wakesWatchers = (event: OrchestrationV2DomainEvent) => + shouldPublishAgentAwarenessEvent(event) || + (event.type === "turn-item.updated" && event.payload.type === "secret_request"); + +/** @public Service construction is part of the canonical Effect module API. */ +export const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const threads = yield* ThreadManagementService.ThreadManagementService; + const secretRequests = yield* SecretRequests.SecretRequests; + const environment = yield* ServerEnvironment.ServerEnvironment; + const httpClient = yield* HttpClient.HttpClient; + const deliveries = yield* FiberSet.make(); + // Registration changes take turns, so a late removal cannot delete a newer watch's row. + const registrations = yield* Semaphore.make(1); + const watchesByThread = new Map>(); + const told = new Map(); + + const add = (watch: Watch) => { + const watches = watchesByThread.get(watch.threadId) ?? new Map(); + watches.set(watch.url, watch); + watchesByThread.set(watch.threadId, watches); + }; + const isCurrent = (watch: Watch) => watchesByThread.get(watch.threadId)?.get(watch.url) === watch; + const forget = (watch: Watch) => { + if (!isCurrent(watch)) return; + watchesByThread.get(watch.threadId)?.delete(watch.url); + if (watchesByThread.get(watch.threadId)?.size === 0) watchesByThread.delete(watch.threadId); + told.delete(watch.id); + }; + + const stored = yield* sql<{ + readonly thread_id: string; + readonly url: string; + readonly secret: string; + }>`SELECT thread_id, url, secret FROM thread_watches`.pipe( + Effect.catchCause((cause) => + Effect.logWarning("thread watches could not be loaded", { cause }).pipe(Effect.as([])), + ), + ); + for (const row of stored) { + const key = standardWebhookKey(row.secret); + const threadId = ThreadId.make(row.thread_id); + if (key !== null) add({ id: threadWatchId(threadId, row.url), threadId, url: row.url, key }); + } + + const end = (watch: Watch) => + Effect.suspend(() => + isCurrent(watch) + ? sql`DELETE FROM thread_watches WHERE thread_id = ${watch.threadId} AND url = ${watch.url}`.pipe( + Effect.andThen(Effect.sync(() => forget(watch))), + ) + : Effect.void, + ).pipe( + registrations.withPermits(1), + Effect.catchCause((cause) => + Effect.logWarning("thread watch could not be removed", { watchId: watch.id, cause }), + ), + ); + + const post = Effect.fn("ThreadWatchReactor.post")(function* ( + watch: Watch, + eventId: string, + body: string, + ) { + const timestamp = Math.floor((yield* Clock.currentTimeMillis) / 1000); + // Redirects stay manual so a receiver cannot bounce the signed body to + // another host. Tracing stays off because request spans record the URL. + const response = yield* HttpClient.withScope(httpClient) + .post(watch.url, { + body: HttpBody.text(body, "application/json"), + headers: { + "webhook-id": eventId, + "webhook-timestamp": String(timestamp), + "webhook-signature": standardWebhookSignature({ + key: watch.key, + id: eventId, + timestamp, + body, + }), + "x-mcp-subscription-id": watch.id, + }, + }) + .pipe( + Effect.map(({ status }) => status), + Effect.scoped, + Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" }), + Effect.withTracerEnabled(false), + Effect.timeoutOption("10 seconds"), + Effect.option, + ); + const status = Option.getOrNull(Option.flatten(response)); + if (status === null || status < 200 || (status >= 300 && status !== 410)) + return yield* new ThreadWatchDeliveryError({ status }); + return status; + }); + + const deliver = (watch: Watch, eventId: string, body: string) => + Effect.suspend(() => (isCurrent(watch) ? post(watch, eventId, body) : Effect.succeed(0))).pipe( + Effect.retry({ schedule: Schedule.exponential("2 seconds"), times: 3 }), + // MCP Events: 410 means the receiver dropped the subscription. + Effect.flatMap((status) => (status === 410 ? end(watch) : Effect.void)), + Effect.catchTags({ + ThreadWatchDeliveryError: (error) => + Effect.logWarning("thread watch delivery failed", { + watchId: watch.id, + status: error.status, + }), + }), + ); + + // Untraced: it runs for every qualifying event on every thread. + const check = Effect.fnUntraced( + function* (threadId: ThreadId) { + const watches = watchesByThread.get(threadId); + if (watches === undefined) return; + const shell = yield* threads.getThreadShell(threadId); + const environmentId = yield* environment.getEnvironmentId; + const now = yield* DateTime.now; + for (const watch of watches.values()) { + const prior = told.get(watch.id); + const { state, kinds } = observeThreadWatch(shell, prior ?? UNSEEN); + told.set(watch.id, state); + // A first look only sets the baseline, unless the thread already ended. + const send = prior === undefined ? kinds.filter((kind) => kind === "ended") : kinds; + if (send.length === 0) continue; + yield* FiberSet.run( + deliveries, + Effect.forEach( + send, + (kind) => { + const eventId = threadWatchEventId(watch.id, kind, state, now); + const body = threadWatchEventBody({ + eventId, + environmentId, + threadId, + shell, + kind, + occurredAt: now, + }); + return deliver(watch, eventId, body); + }, + { discard: true }, + ).pipe(Effect.andThen(state.ended ? end(watch) : Effect.void)), + ); + } + }, + (effect, threadId) => + effect.pipe( + Effect.catchCause((cause) => + Effect.logWarning("thread watch check failed", { threadId, cause }), + ), + ), + ); + const worker = yield* makeAgentAwarenessPublishWorker(check); + + const watch: ThreadWatchReactor["Service"]["watch"] = Effect.fn("ThreadWatchReactor.watch")( + function* ({ threadId, projectId, delivery }) { + const url = threadWatchUrl(delivery.url); + if (url === null) return yield* new ThreadWatchInvalidError({ reason: "url" }); + if ((delivery.secret === undefined) === (delivery.secretRef === undefined)) + return yield* new ThreadWatchInvalidError({ reason: "secret_source" }); + const shell = yield* threads + .getThreadShell(threadId) + .pipe(Effect.mapError((cause) => new ThreadWatchStoreError({ cause }))); + if (shell?.lineage.relationshipToParent === "subagent") + return yield* new ThreadWatchInvalidError({ reason: "subagent" }); + const baseline = shell === null ? UNSEEN : observeThreadWatch(shell, UNSEEN).state; + if (baseline.ended) return yield* new ThreadWatchInvalidError({ reason: "ended" }); + const secret = + delivery.secretRef === undefined + ? (delivery.secret ?? "") + : yield* secretRequests.consume({ ref: delivery.secretRef, projectId }); + const key = standardWebhookKey(secret); + if (key === null) return yield* new ThreadWatchInvalidError({ reason: "secret" }); + const createdAt = DateTime.formatIso(yield* DateTime.now); + const id = threadWatchId(threadId, url); + return yield* sql` + INSERT INTO thread_watches (thread_id, url, secret, created_at) + VALUES (${threadId}, ${url}, ${secret}, ${createdAt}) + ON CONFLICT(thread_id, url) DO UPDATE SET secret = excluded.secret + `.pipe( + Effect.mapError((cause) => new ThreadWatchStoreError({ cause })), + Effect.map(() => { + const wasWatching = watchesByThread.get(threadId)?.has(url) ?? false; + if (!told.has(id)) told.set(id, baseline); + add({ id, threadId, url, key }); + return { wasWatching }; + }), + registrations.withPermits(1), + // The thread may have changed since its baseline was read. + Effect.tap(() => (shell === null ? Effect.void : worker.enqueue(threadId))), + ); + }, + ); + + const unwatch: ThreadWatchReactor["Service"]["unwatch"] = Effect.fn("ThreadWatchReactor.unwatch")( + function* ({ threadId, url }) { + const target = threadWatchUrl(url) ?? url; + return yield* sql`DELETE FROM thread_watches WHERE thread_id = ${threadId} AND url = ${target}`.pipe( + Effect.mapError((cause) => new ThreadWatchStoreError({ cause })), + Effect.map(() => { + const watch = watchesByThread.get(threadId)?.get(target); + if (watch !== undefined) forget(watch); + return { wasWatching: watch !== undefined }; + }), + registrations.withPermits(1), + ); + }, + ); + + const watchLaunch: ThreadWatchReactor["Service"]["watchLaunch"] = ({ launch, ...input }) => + Effect.uninterruptibleMask((restore) => + watch(input).pipe( + Effect.andThen( + restore(launch).pipe( + // A failed launch may still have created the thread; the check ends + // the watch only if it did not. + Effect.onExit((exit) => + Exit.isSuccess(exit) ? Effect.void : worker.enqueue(input.threadId), + ), + ), + ), + ), + ).pipe(Effect.withSpan("ThreadWatchReactor.watchLaunch")); + + const start: ThreadWatchReactor["Service"]["start"] = Effect.fn("ThreadWatchReactor.start")( + function* () { + yield* forkParked( + Stream.runForEach(threads.streamDomainEvents, (event) => + wakesWatchers(event) ? worker.enqueue(event.threadId) : Effect.void, + ), + ); + yield* Effect.forEach([...watchesByThread.keys()], worker.enqueue, { discard: true }); + }, + ); + + return ThreadWatchReactor.of({ + start, + checkThread: (threadId) => worker.enqueue(threadId).pipe(Effect.asVoid), + watch, + unwatch, + watchLaunch, + drain: worker.drain.pipe(Effect.andThen(FiberSet.awaitEmpty(deliveries))), + }); +}); + +export const layer = Layer.effect(ThreadWatchReactor, make); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index ad4c7ecda878..84163d979d3c 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -48,6 +48,7 @@ import * as ResourceCleanupService from "./ResourceCleanupService.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; import * as RuntimeRequestService from "./RuntimeRequestService.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; +import * as ThreadWatchReactor from "./ThreadWatchReactor.ts"; import * as ThreadLaunchService from "./ThreadLaunchService.ts"; import * as ThreadLifecycleService from "./ThreadLifecycleService.ts"; import * as ThreadForkService from "./ThreadForkService.ts"; @@ -292,6 +293,9 @@ const layerScheduledTaskProvided = ScheduledTaskService.layer.pipe( ), ), ); +const layerThreadWatchProvided = ThreadWatchReactor.layer.pipe( + Layer.provide(Layer.mergeAll(layerThreadManagementProvided, layerSecretRequestsProvided)), +); const layerProviderContinuationWorkerProvided = ProviderContinuationService.layer.pipe( Layer.provide( Layer.mergeAll( @@ -367,6 +371,7 @@ export const layerProduction = Layer.mergeAll( layerThreadLifecycleProvided, layerScheduledTaskProvided, layerSecretRequestsProvided, + layerThreadWatchProvided, UsageLimitRecoveryWorker.layer.pipe( Layer.provide( Layer.mergeAll( diff --git a/apps/server/src/orchestration-v2/threadWatch.test.ts b/apps/server/src/orchestration-v2/threadWatch.test.ts new file mode 100644 index 000000000000..bd5fce1aff53 --- /dev/null +++ b/apps/server/src/orchestration-v2/threadWatch.test.ts @@ -0,0 +1,166 @@ +import { assert, describe, it } from "@effect/vitest"; +import { + type OrchestrationV2ThreadShell, + ProjectId, + ProviderInstanceId, + RunId, + RuntimeRequestId, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; + +import { + observeThreadWatch, + standardWebhookKey, + standardWebhookSignature, + threadWatchEventId, + type ThreadWatchState, + UNSEEN, +} from "./threadWatch.ts"; + +const THREAD_ID = ThreadId.make("thread"); +const RUN_1 = RunId.make("run-1"); +const RUN_2 = RunId.make("run-2"); +const at = (iso: string) => DateTime.makeUnsafe(iso); + +function shell(overrides: Partial = {}): OrchestrationV2ThreadShell { + return { + id: THREAD_ID, + projectId: ProjectId.make("project"), + title: "Thread", + providerInstanceId: ProviderInstanceId.make("codex"), + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "test-model" }, + runtimeMode: "full-access", + interactionMode: "default", + worktreePath: null, + activeProviderThreadId: null, + lineage: { rootThreadId: THREAD_ID, parentThreadId: null, relationshipToParent: null }, + forkedFrom: null, + createdBy: "agent", + creationSource: "mcp", + activeRunId: null, + latestVisibleMessage: null, + hasActionableProposedPlan: false, + itemCount: 0, + visibleItemCount: 0, + lastVisitedAt: null, + deletedAt: null, + branch: null, + linkedPullRequest: null, + status: "running", + activityRunStatus: "running", + pendingRuntimeRequest: null, + pendingBackgroundTasks: [], + latestRunId: RUN_1, + latestRunRequestedAt: null, + latestRunStartedAt: null, + latestRunCompletedAt: null, + latestUserMessageAt: null, + createdAt: at("2026-10-20T12:00:00.000Z"), + updatedAt: at("2026-10-20T12:00:00.000Z"), + archivedAt: null, + settledOverride: null, + settledAt: null, + snoozedUntil: null, + snoozedAt: null, + pinnedAt: null, + ...overrides, + }; +} + +const completed = (runId = RUN_1, completedAt = "2026-10-20T12:05:00.000Z") => + shell({ + status: "completed", + activityRunStatus: null, + latestRunId: runId, + latestRunCompletedAt: at(completedAt), + }); + +function told(shells: ReadonlyArray) { + let state: ThreadWatchState = UNSEEN; + return shells.map((next) => { + const observed = observeThreadWatch(next, state); + state = observed.state; + return observed.kinds.join("+") || null; + }); +} + +describe("observeThreadWatch", () => { + it("tells each completion, question, approval and failure once", () => { + const approval = shell({ + latestRunId: RUN_2, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("request-1"), + kind: "command", + createdAt: at("2026-10-20T12:06:00.000Z"), + }, + }); + const question = shell({ + latestRunId: RUN_2, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("request-2"), + kind: "user_input", + createdAt: at("2026-10-20T12:07:00.000Z"), + }, + }); + assert.deepStrictEqual( + told([ + shell(), + completed(), + completed(), + approval, + approval, + question, + shell({ latestRunId: RUN_2, status: "failed", activityRunStatus: null }), + { ...completed(RUN_2, "2026-10-20T12:09:00.000Z"), settledOverride: "settled" }, + null, + ]), + [null, "completed", null, "approval", null, "question", "failed", "completed+ended", null], + ); + }); + + it("gives a second question in the same run its own event id", () => { + const asking = (id: string) => + shell({ + pendingRuntimeRequest: { + id: RuntimeRequestId.make(id), + kind: "user_input", + createdAt: at("2026-10-20T12:06:00.000Z"), + }, + }); + const first = observeThreadWatch(asking("request-1"), UNSEEN); + const answered = observeThreadWatch(shell(), first.state); + const second = observeThreadWatch(asking("request-2"), answered.state); + assert.deepStrictEqual(second.kinds, ["question"]); + assert.notStrictEqual( + threadWatchEventId("tw", "question", first.state, at("2026-10-20T12:06:00.000Z")), + threadWatchEventId("tw", "question", second.state, at("2026-10-20T12:06:00.000Z")), + ); + }); +}); + +describe("Standard Webhooks signing", () => { + it("matches the reference vector", () => { + const key = standardWebhookKey("whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"); + assert.isNotNull(key); + assert.strictEqual( + standardWebhookSignature({ + key: key!, + id: "msg_p5jXN8AQM9LWM0D4loKWxJek", + timestamp: 1614265330, + body: '{"test": 2432232314}', + }), + "v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE=", + ); + }); + + it("accepts only whsec_ secrets holding 24 to 64 bytes", () => { + const encode = (bytes: number) => Buffer.alloc(bytes, 1).toString("base64"); + assert.strictEqual(standardWebhookKey(`whsec_${encode(24)}`)?.length, 24); + assert.strictEqual(standardWebhookKey(`whsec_${encode(64)}`)?.length, 64); + assert.isNull(standardWebhookKey(`whsec_${encode(23)}`)); + assert.isNull(standardWebhookKey(`whsec_${encode(65)}`)); + assert.isNull(standardWebhookKey(encode(32))); + assert.isNull(standardWebhookKey("whsec_not base64!")); + }); +}); diff --git a/apps/server/src/orchestration-v2/threadWatch.ts b/apps/server/src/orchestration-v2/threadWatch.ts new file mode 100644 index 000000000000..ae7f59bd7c69 --- /dev/null +++ b/apps/server/src/orchestration-v2/threadWatch.ts @@ -0,0 +1,113 @@ +// @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect's Crypto has no createHmac or synchronous createHash. +import * as NodeCrypto from "node:crypto"; + +import type { EnvironmentId, OrchestrationV2ThreadShell, ThreadId } from "@t3tools/contracts"; +import { resolveThreadAwarenessPhaseV2 } from "@t3tools/shared/agentAwareness"; +import * as DateTime from "effect/DateTime"; + +export type ThreadWatchEventKind = "completed" | "failed" | "question" | "approval" | "ended"; + +export interface ThreadWatchState { + readonly attention: string | null; + readonly completion: string | null; + readonly ended: boolean; +} + +export const UNSEEN: ThreadWatchState = { attention: null, completion: null, ended: false }; + +/** + * Uses the phases mobile alerts use. A run that ends as its thread settles is + * reported before `ended`. + */ +export function observeThreadWatch( + shell: OrchestrationV2ThreadShell | null, + prior: ThreadWatchState, +): { readonly state: ThreadWatchState; readonly kinds: ReadonlyArray } { + if (prior.ended) return { state: prior, kinds: [] }; + const ended = + shell === null || + shell.deletedAt !== null || + shell.archivedAt !== null || + shell.settledOverride === "settled"; + const phase = shell === null ? null : resolveThreadAwarenessPhaseV2(shell); + const attentionKind = + phase === "waiting_for_approval" + ? "approval" + : phase === "waiting_for_input" + ? "question" + : phase === "failed" + ? "failed" + : null; + // Keyed by request too, so a second question in one run gets its own event id. + const attention = + attentionKind === null + ? null + : `${shell?.latestRunId ?? ""}:${attentionKind}:${shell?.pendingRuntimeRequest?.id ?? ""}`; + const completion = + phase === "completed" && shell !== null + ? DateTime.formatIso(shell.latestRunCompletedAt ?? shell.updatedAt) + : prior.completion; + const kinds: Array = []; + if (attentionKind !== null && attention !== prior.attention) kinds.push(attentionKind); + else if (completion !== null && completion !== prior.completion) kinds.push("completed"); + if (ended) kinds.push("ended"); + return { state: { attention, completion, ended }, kinds }; +} + +const digest = (value: string) => + NodeCrypto.createHash("sha256").update(value).digest("hex").slice(0, 32); + +export const threadWatchId = (threadId: ThreadId, url: string) => + `tw_${digest(`${threadId}\0${url}`)}`; + +export const threadWatchEventId = ( + watchId: string, + kind: ThreadWatchEventKind, + state: ThreadWatchState, + occurredAt: DateTime.Utc, +) => + `evt_${digest( + // A watch made again on the same URL can end with nothing else changed. + `${watchId}\0${kind}\0${kind === "ended" ? DateTime.formatIso(occurredAt) : (state.attention ?? state.completion ?? "")}`, + )}`; + +export function threadWatchEventBody(input: { + readonly eventId: string; + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly shell: OrchestrationV2ThreadShell | null; + readonly kind: ThreadWatchEventKind; + readonly occurredAt: DateTime.Utc; +}): string { + return JSON.stringify({ + eventId: input.eventId, + name: "thread.watch", + timestamp: DateTime.formatIso(input.occurredAt), + data: { + environmentId: input.environmentId, + threadId: input.threadId, + title: input.shell?.title ?? "", + kind: input.kind, + runId: input.shell?.latestRunId ?? null, + }, + cursor: null, + }); +} + +/** Standard Webhooks, which MCP Events uses for webhook delivery. */ +export function standardWebhookKey(secret: string): Uint8Array | null { + const encoded = secret.startsWith("whsec_") ? secret.slice("whsec_".length) : ""; + if (!/^[A-Za-z0-9+/]+={0,2}$/.test(encoded) || encoded.length % 4 !== 0) return null; + const key = Buffer.from(encoded, "base64"); + return key.length >= 24 && key.length <= 64 ? key : null; +} + +export const standardWebhookSignature = (input: { + readonly key: Uint8Array; + readonly id: string; + readonly timestamp: number; + readonly body: string; +}) => + `v1,${NodeCrypto.createHmac("sha256", input.key) + .update(`${input.id}.${input.timestamp}.${input.body}`) + .digest("base64")}`; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 5979eee6c18d..b031a0fb7b48 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -70,6 +70,7 @@ import Migration0064 from "./Migrations/064_CanonicalizeThreadPullRequestLinks.t import Migration0065 from "./Migrations/065_CodexUsageNativeTurnIndex.ts"; import Migration0066 from "./Migrations/066_RepairClientSettlementTimestamps.ts"; import Migration0067 from "./Migrations/067_WebhookDedupe.ts"; +import Migration0068 from "./Migrations/068_ThreadWatches.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -140,6 +141,7 @@ const migrationEntries = [ [65, "CodexUsageNativeTurnIndex", Migration0065], [66, "RepairClientSettlementTimestamps", Migration0066], [67, "WebhookDedupe", Migration0067], + [68, "ThreadWatches", Migration0068], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/068_ThreadWatches.ts b/apps/server/src/persistence/Migrations/068_ThreadWatches.ts new file mode 100644 index 000000000000..c19f756dac11 --- /dev/null +++ b/apps/server/src/persistence/Migrations/068_ThreadWatches.ts @@ -0,0 +1,15 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + // Its own table, like scheduled task webhook secrets, so the signing secret + // never reaches the event log or the thread read model clients receive. + yield* sql`CREATE TABLE thread_watches ( + thread_id TEXT NOT NULL, + url TEXT NOT NULL, + secret TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (thread_id, url) + )`; +}); diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index c6a61817f69b..6fdad0ced383 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -73,7 +73,7 @@ function isTurnItemPayload( ); } -function shouldPublishAgentAwarenessEvent( +export function shouldPublishAgentAwarenessEvent( event: Pick & { readonly payload?: unknown }, ): boolean { if ( diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 48d701103c07..d26fdd1055e0 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -5,6 +5,7 @@ import * as DatabaseStorage from "./persistence/DatabaseStorage.ts"; import * as StorageCleanup from "./storageCleanup.ts"; import * as PullRequestSyncReactor from "./orchestration-v2/PullRequestSyncReactor.ts"; import * as PullRequestWatchReactor from "./orchestration-v2/PullRequestWatchReactor.ts"; +import * as ThreadWatchReactor from "./orchestration-v2/ThreadWatchReactor.ts"; import * as DelegatedTaskQuestionReactor from "./orchestration-v2/DelegatedTaskQuestionReactor.ts"; // Effect wraps this raw Node HTTP server after its startup and write-error guards attach. // @effect-diagnostics nodeBuiltinImport:off @@ -592,6 +593,13 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( Layer.provide(layerPullRequestService), Layer.provide(ProjectionStoreV2.layer), ), + // Started from the runtime's instance, which the MCP watch tools also use. + Layer.effectDiscard( + Effect.gen(function* () { + const service = yield* ThreadWatchReactor.ThreadWatchReactor; + yield* service.start(); + }), + ), Layer.effectDiscard( Effect.gen(function* () { const service = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor; diff --git a/docs/user/outside-agents.md b/docs/user/outside-agents.md index f1503ef49b22..d0a3e64d4d5b 100644 --- a/docs/user/outside-agents.md +++ b/docs/user/outside-agents.md @@ -92,6 +92,34 @@ Any agent that supports remote MCP servers over HTTP with OAuth can connect. Giv it the MCP URL and choose OAuth, and the agent finds the sign-in settings itself. Agents without OAuth support cannot connect. +## Hear back when a thread needs you + +An agent whose turn ends before the thread it started can have T3 Code call it +back instead of waiting or polling. Pass `watch` to `t3_thread_launch`, or call +`t3_thread_watch` with `action: "watch"` on an existing thread, with an `https` +URL and a signing secret. T3 Code then sends that URL a POST when the thread's +run completes or fails, when it asks a question, or when it waits for an +approval. + +Requests follow [Standard Webhooks](https://www.standardwebhooks.com), the +format MCP Events uses, so an off-the-shelf verifier can check them. The secret +is `whsec_` followed by base64 of 24 to 64 random bytes. An agent working inside +T3 Code can pass a `secretRef` from `request_secret` instead, so the secret +stays out of its conversation. The body says which thread changed and how, not +what it produced; read the thread for that. + +A watch starts from the thread as it is when you add it, so add it before you +read the thread and nothing falls in between. After a server restart it picks up +from the thread as it is then. A failed request is retried a few times with the +same `webhook-id`, so ignore ids you have seen. A run that someone stops sends +nothing, as with mobile alerts, so keep a timeout of your own. + +Settling, archiving or deleting the thread ends the watch with one last `ended` +request, including when T3 Code settles an inactive thread on its own. A `410` from your endpoint or `t3_thread_watch` with +`action: "unwatch"` and the same URL ends it too. Settled, archived and subagent +threads cannot be watched. Watches are not tied to the agent that made them: +revoking an agent below does not end them. + ## Scripts on the host On the machine that runs the environment, `t3 thread` uses the same tools diff --git a/packages/client-runtime/src/t3ToolSummary.ts b/packages/client-runtime/src/t3ToolSummary.ts index df655c1d6c21..7c1ad00686c0 100644 --- a/packages/client-runtime/src/t3ToolSummary.ts +++ b/packages/client-runtime/src/t3ToolSummary.ts @@ -234,6 +234,9 @@ export function summarizeT3ToolCalls( case "thread-update": label = phrase("Updated", "update", quantity(countEntities(threadIds), "thread")); break; + case "thread-watch": + label = phrase("Updated", "update", `thread watches ${times}`); + break; case "queue-list": label = phrase("Listed", "list", `queued messages ${times}`); break; diff --git a/packages/client-runtime/src/work-log/presentation.ts b/packages/client-runtime/src/work-log/presentation.ts index f16b0722abe3..547c2afad65c 100644 --- a/packages/client-runtime/src/work-log/presentation.ts +++ b/packages/client-runtime/src/work-log/presentation.ts @@ -625,6 +625,7 @@ function summaryActionPriority(action: ToolGroupAction | T3McpToolSummaryAction) case "thread-merge": case "thread-organize": case "thread-update": + case "thread-watch": case "queue-edit": case "queue-cancel": case "queue-reorder": diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index df912b4bb0f1..4eafbc4aa4dd 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -94,7 +94,7 @@ export function projectThreadAwarenessV2( }; } -function resolveThreadAwarenessPhaseV2( +export function resolveThreadAwarenessPhaseV2( thread: ProjectThreadAwarenessV2Input["thread"], ): AgentAwarenessPhase | null { const runStatus = thread.activityRunStatus ?? thread.status; diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index 21f8be3941f9..9a52f2fd9f34 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -29,6 +29,7 @@ export type T3McpToolSummaryAction = | "thread-search" | "thread-transfers" | "thread-organize" + | "thread-watch" | "thread-update" | "queue-list" | "queue-read" @@ -301,6 +302,7 @@ const T3_MCP_TOOLS: Readonly> = { t3_thread_search: tool(["Search", "Searching", "Searched", "thread content"], "thread-search"), t3_thread_transfers: tool(["Read", "Reading", "Read", "thread transfers"], "thread-transfers"), t3_thread_organize: tool(["Organize", "Organizing", "Organized", "a thread"], "thread-organize"), + t3_thread_watch: tool(["Update", "Updating", "Updated", "thread watches"], "thread-watch"), t3_thread_update: tool(["Update", "Updating", "Updated", "T3 thread metadata"], "thread-update"), t3_worktree_list: tool(["List", "Listing", "Listed", "workspace branches"], "worktree-list"), t3_preview_list: tool(["List", "Listing", "Listed", "preview tabs"], "browser", "browser"),