From 35ffff2092089a60017f99ce5e6629f1b75e3fcf Mon Sep 17 00:00:00 2001 From: feldrok Date: Tue, 20 Oct 2026 21:21:10 -0300 Subject: [PATCH 1/3] feat(server): tell outside agents when a watched thread needs them Agents that drive T3 Code through /mcp end their turn before the threads they start, and today learn that a thread finished or is waiting only by holding t3_thread_wait open or polling. Add t3_thread_watch (watch/unwatch) and a watch option on t3_thread_launch: T3 Code POSTs a Standard Webhooks request in the MCP Events body shape when a watched thread's run completes or fails, asks a question, waits for an approval, or ends. A reactor beside the awareness relay decides from the same phases mobile alerts use. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/mcp/toolkits/core.test.ts | 5 + .../src/mcp/toolkits/project/handlers.test.ts | 2 + .../src/mcp/toolkits/project/handlers.ts | 46 +- apps/server/src/mcp/toolkits/project/tools.ts | 9 + .../src/mcp/toolkits/thread/handlers.ts | 26 + apps/server/src/mcp/toolkits/thread/tools.ts | 47 ++ .../ThreadWatchReactor.test.ts | 492 ++++++++++++++++++ .../orchestration-v2/ThreadWatchReactor.ts | 365 +++++++++++++ .../src/orchestration-v2/runtimeLayer.ts | 5 + .../src/orchestration-v2/threadWatch.test.ts | 166 ++++++ .../src/orchestration-v2/threadWatch.ts | 119 +++++ apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/068_ThreadWatches.ts | 15 + apps/server/src/relay/AgentAwarenessRelay.ts | 2 +- apps/server/src/server.ts | 8 + docs/user/outside-agents.md | 28 + packages/client-runtime/src/t3ToolSummary.ts | 3 + .../src/work-log/presentation.ts | 1 + packages/shared/src/agentAwareness.ts | 2 +- packages/shared/src/t3McpToolPresentation.ts | 2 + 20 files changed, 1342 insertions(+), 3 deletions(-) create mode 100644 apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts create mode 100644 apps/server/src/orchestration-v2/ThreadWatchReactor.ts create mode 100644 apps/server/src/orchestration-v2/threadWatch.test.ts create mode 100644 apps/server/src/orchestration-v2/threadWatch.ts create mode 100644 apps/server/src/persistence/Migrations/068_ThreadWatches.ts diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 806f3360e9fd..df65f583b3d3 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -43,6 +43,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"; @@ -70,6 +71,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"); @@ -462,6 +464,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)({})), ), @@ -599,6 +602,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)({})), ), @@ -691,6 +695,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 aecc99f7dda2..fe40408a18ee 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -24,6 +24,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"; @@ -39,6 +40,7 @@ const clientLaunchHarness = (input: { model: "claude-opus", }; const layerDependencies = Layer.mergeAll( + Layer.mock(ThreadWatchReactor.ThreadWatchReactor)({}), ServerSettings.layerTest(), NodeCrypto.layer, Layer.succeed(McpInvocationContext.McpInvocationContext, { diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index 912d6ad127fc..f18e573cd7c1 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 McpToolAccess from "../../McpToolAccess.ts"; import { newCommandId, readCaller, resolveProjectId, unavailable } from "../../threadAccess.ts"; import { ProjectToolkit } from "./tools.ts"; @@ -46,6 +53,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; @@ -133,6 +155,19 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { message: "Pass modelSelection: the project has no default model. orchestrator_capabilities lists providers and models.", }); + // Watched before the thread exists, so its first run cannot be missed. + const watchDelivery = input.watch; + if (watchDelivery !== undefined) + yield* ThreadWatchReactor.ThreadWatchReactor.pipe( + Effect.flatMap((watches) => + watches.watch({ threadId, projectId, delivery: watchDelivery }), + ), + Effect.catchTags({ + ThreadWatchInvalidError: watchRefused, + ThreadWatchStoreError: watchRefused, + SecretRequestError: watchRefused, + }), + ); const result = yield* ThreadMessageIntake.launchThread({ commandId, threadId, @@ -170,6 +205,15 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { } return unavailable(); }), + // A launch that failed may still have created the thread, so the watch + // looks at it: one that never came to be ends its watch. + Effect.onError(() => + watchDelivery === undefined + ? Effect.void + : ThreadWatchReactor.ThreadWatchReactor.pipe( + Effect.flatMap((watches) => watches.checkThread(threadId)), + ), + ), ); const thread = result.projection.thread; const run = result.projection.runs.find((run) => run.userMessageId === messageId); 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 fec17deae27c..4b81c3ca6e82 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"; @@ -70,11 +71,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..7ceb6018cd14 --- /dev/null +++ b/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts @@ -0,0 +1,492 @@ +// @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; +} + +/** A real reactor on the caller's SQLite, with the thread, secrets and receiver faked. */ +const makeHarness = Effect.fnUntraced(function* ( + options: { + readonly respond?: (attempt: number, request: Received) => Response | Promise; + readonly secretRefValue?: string; + 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 }); + return options.secretRefValue === undefined + ? Effect.fail(new SecretRequestError({ reason: "ref_expired" })) + : Effect.succeed(options.secretRefValue); + }), + }), + 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 watch made before its thread exists hears its first run, or ends if it never comes", + () => + Effect.gen(function* () { + const { received, watch, settle } = yield* makeHarness({ initial: null }); + yield* watch(); + yield* settle(shell({ status: "failed", activityRunStatus: null })); + assert.deepInclude(verified(received[0]).data, { kind: "failed", runId: RUN_1 }); + + // A launch that failed before creating its thread: the next look ends the watch. + const unborn = yield* makeHarness({ initial: null }); + yield* unborn.watch(); + yield* unborn.settle(null); + 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(); + // The run is still going when the server comes back, then it completes. + 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("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, + }, + }), + ); + // request_secret records only a turn item, which the mobile alert filter skips. + 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 the thread sends ended and ends the watch", () => + Effect.gen(function* () { + const { received, reactor, settle, watch } = yield* makeHarness(); + yield* watch(); + yield* settle(shell({ settledOverride: "settled" })); + assert.deepInclude(verified(received[0]).data, { kind: "ended" }); + yield* settle(completed()); + assert.strictEqual(received.length, 1); + assert.isFalse(yield* stillWatching(reactor)); + }).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..17bfdda3c1d7 --- /dev/null +++ b/apps/server/src/orchestration-v2/ThreadWatchReactor.ts @@ -0,0 +1,365 @@ +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 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 { FetchHttpClient, HttpBody, HttpClient } from "effect/http"; + +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) }, +) {} + +export class ThreadWatchReactor extends Context.Service< + ThreadWatchReactor, + { + readonly start: () => Effect.Effect; + /** Queues a check of the thread's watches, as a thread event would. */ + readonly checkThread: (threadId: ThreadId) => Effect.Effect; + /** + * Watching the same URL again replaces its secret. A watch made before its + * thread exists, as a launch does, hears the first run; any other watch + * starts from the thread as it is now. + */ + readonly watch: (input: { + readonly threadId: ThreadId; + readonly projectId: ProjectId; + readonly delivery: { + readonly url: string; + readonly secret?: string | undefined; + readonly secretRef?: SecretRef | undefined; + }; + }) => Effect.Effect< + { readonly wasWatching: boolean }, + ThreadWatchInvalidError | ThreadWatchStoreError | SecretRequestError + >; + readonly unwatch: (input: { + readonly threadId: ThreadId; + readonly url: string; + }) => Effect.Effect<{ readonly wasWatching: boolean }, ThreadWatchStoreError>; + /** Resolves once queued threads are checked and their deliveries have settled. */ + readonly drain: Effect.Effect; + } +>()("t3/orchestration-v2/ThreadWatchReactor") {} + +interface Watch { + readonly id: string; + readonly threadId: ThreadId; + readonly url: string; + readonly key: Uint8Array; +} + +/** The URL as it is stored and compared, or null when a delivery must not go there. */ +function threadWatchUrl(value: string): string | null { + if (!URL.canParse(value)) return null; + const url = new URL(value); + // Credentials or a fragment would be stored and sent along with every delivery. + 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; +} + +/** + * The mobile alerts' events, plus a `request_secret` wait: it shows as pending + * input, but its only event is a turn item those alerts skip. + */ +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); + // Watches are few, so they live in memory and the table only keeps them across restarts. + const watchesByThread = new Map>(); + // What each watch was last told. A watch with no entry takes its first look as + // a baseline, so after a restart watches start over from what threads look like. + 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(() => + // A watch ended or replaced meanwhile gets nothing more. + 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, and returns at once for the + // many threads nobody watches. + 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, kind } = observeThreadWatch(shell, prior ?? UNSEEN); + told.set(watch.id, state); + // A first look only sets a baseline, unless the thread has already ended. + if (kind === null || (prior === undefined && kind !== "ended")) continue; + const eventId = threadWatchEventId(watch.id, kind, state); + const body = threadWatchEventBody({ + eventId, + environmentId, + threadId, + shell, + kind, + occurredAt: now, + }); + // The watch ends once its ended event has been sent or given up on. + yield* FiberSet.run( + deliveries, + kind === "ended" + ? deliver(watch, eventId, body).pipe(Effect.andThen(end(watch))) + : deliver(watch, eventId, body), + ); + } + }, + (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" }); + // A thread that does not exist yet is being launched, so its first run is news. + const baseline = shell === null ? UNSEEN : observeThreadWatch(shell, UNSEEN).state; + if (baseline.ended) return yield* new ThreadWatchInvalidError({ reason: "ended" }); + // A secretRef keeps the secret out of the calling agent's transcript, as + // it does for scheduled task webhooks. Using it consumes it. + 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), + ); + }, + ); + + 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 start: ThreadWatchReactor["Service"]["start"] = Effect.fn("ThreadWatchReactor.start")( + function* () { + yield* forkParked( + Stream.runForEach(threads.streamDomainEvents, (event) => + wakesWatchers(event) ? worker.enqueue(event.threadId) : Effect.void, + ), + ); + // Watches from before this start take their baseline now, so the next change is news. + yield* Effect.forEach([...watchesByThread.keys()], worker.enqueue, { discard: true }); + }, + ); + + return ThreadWatchReactor.of({ + start, + checkThread: (threadId) => worker.enqueue(threadId).pipe(Effect.asVoid), + watch, + unwatch, + 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..6332da511a18 --- /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), + }); + +/** Feeds shells through in order and collects what a watcher would be told. */ +function told(shells: ReadonlyArray) { + let state: ThreadWatchState = UNSEEN; + return shells.map((next) => { + const observed = observeThreadWatch(next, state); + state = observed.state; + return observed.kind; + }); +} + +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"), + ]), + [null, "completed", null, "approval", null, "question", "failed", "completed"], + ); + }); + + 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.strictEqual(second.kind, "question"); + assert.notStrictEqual( + threadWatchEventId("tw", "question", first.state), + threadWatchEventId("tw", "question", second.state), + ); + }); +}); + +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..36e13676c3d0 --- /dev/null +++ b/apps/server/src/orchestration-v2/threadWatch.ts @@ -0,0 +1,119 @@ +// @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"; + +/** What a watcher was last told about a thread. */ +export interface ThreadWatchState { + readonly attention: string | null; + readonly completion: string | null; + readonly ended: boolean; +} + +export const UNSEEN: ThreadWatchState = { attention: null, completion: null, ended: false }; + +/** + * The thread's state now and what changed since `prior`. Uses the phases mobile + * alerts use, so a watcher hears what a phone would. Settling, archiving or + * deleting the thread ends the watch. + */ +export function observeThreadWatch( + shell: OrchestrationV2ThreadShell | null, + prior: ThreadWatchState, +): { readonly state: ThreadWatchState; readonly kind: ThreadWatchEventKind | null } { + if ( + shell === null || + shell.deletedAt !== null || + shell.archivedAt !== null || + shell.settledOverride === "settled" + ) { + return { state: { ...prior, ended: true }, kind: prior.ended ? null : "ended" }; + } + const phase = 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 is news with its own event id. + const attention = + attentionKind === null + ? null + : `${shell.latestRunId ?? ""}:${attentionKind}:${shell.pendingRuntimeRequest?.id ?? ""}`; + const completion = + phase === "completed" + ? DateTime.formatIso(shell.latestRunCompletedAt ?? shell.updatedAt) + : prior.completion; + const state = { attention, completion, ended: false }; + if (attentionKind !== null && attention !== prior.attention) + return { state, kind: attentionKind }; + if (completion !== null && completion !== prior.completion) return { state, kind: "completed" }; + return { state, kind: null }; +} + +const digest = (value: string) => + NodeCrypto.createHash("sha256").update(value).digest("hex").slice(0, 32); + +/** Deterministic, like an MCP Events subscription id, and sent as `x-mcp-subscription-id`. */ +export const threadWatchId = (threadId: ThreadId, url: string) => + `tw_${digest(`${threadId}\0${url}`)}`; + +/** The same change keeps its id across retries, so receivers can drop repeats. */ +export const threadWatchEventId = ( + watchId: string, + kind: ThreadWatchEventKind, + state: ThreadWatchState, +) => `evt_${digest(`${watchId}\0${kind}\0${state.attention ?? state.completion ?? ""}`)}`; + +/** The MCP Events webhook body. Status only: the receiver reads the thread itself. */ +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 secrets (`whsec_` and base64 of 24 to 64 bytes), the form + * MCP Events requires. Returns the signing key, or null for any other secret. + */ +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; +} + +/** The `webhook-signature` value: HMAC-SHA256 over `id.timestamp.body`. */ +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 7cc6692e38f6..d9ec380514ca 100644 --- a/docs/user/outside-agents.md +++ b/docs/user/outside-agents.md @@ -88,6 +88,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. + ## Remove an agent Approved agents appear under **Settings → Connections** like other clients. 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 1c22deb9ff59..10c8f6522562 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -92,7 +92,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"), From 5d67de3511ca92770a62806076d090da97ae0201 Mon Sep 17 00:00:00 2001 From: feldrok Date: Tue, 20 Oct 2026 22:31:43 -0300 Subject: [PATCH 2/3] fix(server): send a change made while a watch is being registered Watching reads the thread's baseline, then consumes the secretRef and stores the row before the watch is live, so a run that ended in between was never sent if no later event followed. Check the thread once the watch is live, except for a watch made before its thread exists, which has no baseline to miss and would otherwise end itself. Also cut comments that restated names or narrated code, to match the comment density of the neighbouring reactors. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/toolkits/project/handlers.ts | 4 +- .../ThreadWatchReactor.test.ts | 38 +++++++++++++++---- .../orchestration-v2/ThreadWatchReactor.ts | 37 +++++------------- .../src/orchestration-v2/threadWatch.test.ts | 1 - .../src/orchestration-v2/threadWatch.ts | 20 +++------- 5 files changed, 46 insertions(+), 54 deletions(-) diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index f18e573cd7c1..df9715a04b59 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -205,8 +205,8 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { } return unavailable(); }), - // A launch that failed may still have created the thread, so the watch - // looks at it: one that never came to be ends its watch. + // A failed launch may still have created the thread; a check ends the + // watch only if it did not. Effect.onError(() => watchDelivery === undefined ? Effect.void diff --git a/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts b/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts index 7ceb6018cd14..d98649d24758 100644 --- a/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts +++ b/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts @@ -102,11 +102,13 @@ interface Received { readonly body: string; } -/** A real reactor on the caller's SQLite, with the thread, secrets and receiver faked. */ 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; } = {}, @@ -149,9 +151,14 @@ const makeHarness = Effect.fnUntraced(function* ( consume: ({ ref, projectId }) => Effect.suspend(() => { consumed.push({ ref, projectId }); - return options.secretRefValue === undefined - ? Effect.fail(new SecretRequestError({ reason: "ref_expired" })) - : Effect.succeed(options.secretRefValue); + 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, { @@ -232,12 +239,14 @@ describe("ThreadWatchReactor", () => { "a watch made before its thread exists hears its first run, or ends if it never comes", () => Effect.gen(function* () { - const { received, watch, settle } = yield* makeHarness({ initial: null }); + const { received, reactor, watch, settle } = yield* makeHarness({ initial: null }); yield* watch(); + yield* reactor.drain; + assert.strictEqual(received.length, 0, "the thread is still being launched"); yield* settle(shell({ status: "failed", activityRunStatus: null })); assert.deepInclude(verified(received[0]).data, { kind: "failed", runId: RUN_1 }); - // A launch that failed before creating its thread: the next look ends the watch. + // A launch that failed before creating its thread. const unborn = yield* makeHarness({ initial: null }); yield* unborn.watch(); yield* unborn.settle(null); @@ -250,7 +259,6 @@ describe("ThreadWatchReactor", () => { Effect.gen(function* () { const first = yield* makeHarness(); yield* first.watch(); - // The run is still going when the server comes back, then it completes. const restarted = yield* makeHarness({ initial: shell() }); assert.deepStrictEqual(yield* restarted.watch(), { wasWatching: true }); yield* restarted.reactor.start(); @@ -260,6 +268,21 @@ describe("ThreadWatchReactor", () => { }).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(); @@ -285,7 +308,6 @@ describe("ThreadWatchReactor", () => { }, }), ); - // request_secret records only a turn item, which the mobile alert filter skips. yield* Queue.offer(events, { id: EventId.make("event:secret"), type: "turn-item.updated", diff --git a/apps/server/src/orchestration-v2/ThreadWatchReactor.ts b/apps/server/src/orchestration-v2/ThreadWatchReactor.ts index 17bfdda3c1d7..b9e3eea554f3 100644 --- a/apps/server/src/orchestration-v2/ThreadWatchReactor.ts +++ b/apps/server/src/orchestration-v2/ThreadWatchReactor.ts @@ -77,13 +77,7 @@ export class ThreadWatchReactor extends Context.Service< ThreadWatchReactor, { readonly start: () => Effect.Effect; - /** Queues a check of the thread's watches, as a thread event would. */ readonly checkThread: (threadId: ThreadId) => Effect.Effect; - /** - * Watching the same URL again replaces its secret. A watch made before its - * thread exists, as a launch does, hears the first run; any other watch - * starts from the thread as it is now. - */ readonly watch: (input: { readonly threadId: ThreadId; readonly projectId: ProjectId; @@ -100,7 +94,6 @@ export class ThreadWatchReactor extends Context.Service< readonly threadId: ThreadId; readonly url: string; }) => Effect.Effect<{ readonly wasWatching: boolean }, ThreadWatchStoreError>; - /** Resolves once queued threads are checked and their deliveries have settled. */ readonly drain: Effect.Effect; } >()("t3/orchestration-v2/ThreadWatchReactor") {} @@ -112,21 +105,17 @@ interface Watch { readonly key: Uint8Array; } -/** The URL as it is stored and compared, or null when a delivery must not go there. */ function threadWatchUrl(value: string): string | null { if (!URL.canParse(value)) return null; const url = new URL(value); - // Credentials or a fragment would be stored and sent along with every delivery. 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; } -/** - * The mobile alerts' events, plus a `request_secret` wait: it shows as pending - * input, but its only event is a turn item those alerts skip. - */ +// 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"); @@ -141,10 +130,7 @@ export const make = Effect.gen(function* () { 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); - // Watches are few, so they live in memory and the table only keeps them across restarts. const watchesByThread = new Map>(); - // What each watch was last told. A watch with no entry takes its first look as - // a baseline, so after a restart watches start over from what threads look like. const told = new Map(); const add = (watch: Watch) => { @@ -227,10 +213,7 @@ export const make = Effect.gen(function* () { }); const deliver = (watch: Watch, eventId: string, body: string) => - Effect.suspend(() => - // A watch ended or replaced meanwhile gets nothing more. - isCurrent(watch) ? post(watch, eventId, body) : Effect.succeed(0), - ).pipe( + 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)), @@ -243,8 +226,7 @@ export const make = Effect.gen(function* () { }), ); - // Untraced: it runs for every qualifying event, and returns at once for the - // many threads nobody watches. + // Untraced: it runs for every qualifying event on every thread. const check = Effect.fnUntraced( function* (threadId: ThreadId) { const watches = watchesByThread.get(threadId); @@ -256,7 +238,7 @@ export const make = Effect.gen(function* () { const prior = told.get(watch.id); const { state, kind } = observeThreadWatch(shell, prior ?? UNSEEN); told.set(watch.id, state); - // A first look only sets a baseline, unless the thread has already ended. + // A first look only sets the baseline, unless the thread already ended. if (kind === null || (prior === undefined && kind !== "ended")) continue; const eventId = threadWatchEventId(watch.id, kind, state); const body = threadWatchEventBody({ @@ -267,7 +249,6 @@ export const make = Effect.gen(function* () { kind, occurredAt: now, }); - // The watch ends once its ended event has been sent or given up on. yield* FiberSet.run( deliveries, kind === "ended" @@ -296,11 +277,9 @@ export const make = Effect.gen(function* () { .pipe(Effect.mapError((cause) => new ThreadWatchStoreError({ cause }))); if (shell?.lineage.relationshipToParent === "subagent") return yield* new ThreadWatchInvalidError({ reason: "subagent" }); - // A thread that does not exist yet is being launched, so its first run is news. + // A thread that does not exist yet is being launched: its first run is news. const baseline = shell === null ? UNSEEN : observeThreadWatch(shell, UNSEEN).state; if (baseline.ended) return yield* new ThreadWatchInvalidError({ reason: "ended" }); - // A secretRef keeps the secret out of the calling agent's transcript, as - // it does for scheduled task webhooks. Using it consumes it. const secret = delivery.secretRef === undefined ? (delivery.secret ?? "") @@ -322,6 +301,8 @@ export const make = Effect.gen(function* () { return { wasWatching }; }), registrations.withPermits(1), + // The thread may have changed since its baseline was read. + Effect.tap(() => (shell === null ? Effect.void : worker.enqueue(threadId))), ); }, ); @@ -348,7 +329,7 @@ export const make = Effect.gen(function* () { wakesWatchers(event) ? worker.enqueue(event.threadId) : Effect.void, ), ); - // Watches from before this start take their baseline now, so the next change is news. + // Watches from before this start take their baseline now. yield* Effect.forEach([...watchesByThread.keys()], worker.enqueue, { discard: true }); }, ); diff --git a/apps/server/src/orchestration-v2/threadWatch.test.ts b/apps/server/src/orchestration-v2/threadWatch.test.ts index 6332da511a18..352414906d8d 100644 --- a/apps/server/src/orchestration-v2/threadWatch.test.ts +++ b/apps/server/src/orchestration-v2/threadWatch.test.ts @@ -76,7 +76,6 @@ const completed = (runId = RUN_1, completedAt = "2026-10-20T12:05:00.000Z") => latestRunCompletedAt: at(completedAt), }); -/** Feeds shells through in order and collects what a watcher would be told. */ function told(shells: ReadonlyArray) { let state: ThreadWatchState = UNSEEN; return shells.map((next) => { diff --git a/apps/server/src/orchestration-v2/threadWatch.ts b/apps/server/src/orchestration-v2/threadWatch.ts index 36e13676c3d0..5ae68bcbe955 100644 --- a/apps/server/src/orchestration-v2/threadWatch.ts +++ b/apps/server/src/orchestration-v2/threadWatch.ts @@ -7,7 +7,6 @@ import * as DateTime from "effect/DateTime"; export type ThreadWatchEventKind = "completed" | "failed" | "question" | "approval" | "ended"; -/** What a watcher was last told about a thread. */ export interface ThreadWatchState { readonly attention: string | null; readonly completion: string | null; @@ -16,11 +15,7 @@ export interface ThreadWatchState { export const UNSEEN: ThreadWatchState = { attention: null, completion: null, ended: false }; -/** - * The thread's state now and what changed since `prior`. Uses the phases mobile - * alerts use, so a watcher hears what a phone would. Settling, archiving or - * deleting the thread ends the watch. - */ +/** Uses the phases mobile alerts use, so a watcher hears what a phone would. */ export function observeThreadWatch( shell: OrchestrationV2ThreadShell | null, prior: ThreadWatchState, @@ -42,7 +37,7 @@ export function observeThreadWatch( : phase === "failed" ? "failed" : null; - // Keyed by request too, so a second question in one run is news with its own event id. + // Keyed by request too, so a second question in one run gets its own event id. const attention = attentionKind === null ? null @@ -61,18 +56,17 @@ export function observeThreadWatch( const digest = (value: string) => NodeCrypto.createHash("sha256").update(value).digest("hex").slice(0, 32); -/** Deterministic, like an MCP Events subscription id, and sent as `x-mcp-subscription-id`. */ +/** Deterministic, as MCP Events subscription ids are. */ export const threadWatchId = (threadId: ThreadId, url: string) => `tw_${digest(`${threadId}\0${url}`)}`; -/** The same change keeps its id across retries, so receivers can drop repeats. */ +/** Stable across retries, so receivers can drop repeats. */ export const threadWatchEventId = ( watchId: string, kind: ThreadWatchEventKind, state: ThreadWatchState, ) => `evt_${digest(`${watchId}\0${kind}\0${state.attention ?? state.completion ?? ""}`)}`; -/** The MCP Events webhook body. Status only: the receiver reads the thread itself. */ export function threadWatchEventBody(input: { readonly eventId: string; readonly environmentId: EnvironmentId; @@ -96,10 +90,7 @@ export function threadWatchEventBody(input: { }); } -/** - * Standard Webhooks secrets (`whsec_` and base64 of 24 to 64 bytes), the form - * MCP Events requires. Returns the signing key, or null for any other secret. - */ +/** 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; @@ -107,7 +98,6 @@ export function standardWebhookKey(secret: string): Uint8Array | null { return key.length >= 24 && key.length <= 64 ? key : null; } -/** The `webhook-signature` value: HMAC-SHA256 over `id.timestamp.body`. */ export const standardWebhookSignature = (input: { readonly key: Uint8Array; readonly id: string; From d9a75ab428faec5483ce256d07ae4468f5da89d5 Mon Sep 17 00:00:00 2001 From: feldrok Date: Tue, 20 Oct 2026 22:52:03 -0300 Subject: [PATCH 3/3] fix(server): deliver a run that ends as its thread settles, and own launch watches in the reactor - A run that completes or fails as its thread is settled was reported only as ended, so the completion was lost; this is how an agent that settles its own thread ends. The rules now report what changed, then ended, and the reactor sends them in order before ending the watch. - An ended event's id now includes when it was seen, so a thread watched again after unsettling can end again without receivers dropping it as a repeat. - The launch lifecycle moves out of the MCP handler into ThreadWatchReactor.watchLaunch, which registers the watch and runs the launch as one step and, when the launch fails or is interrupted, checks the thread so a watch for a thread that never came to be ends. - Import effect/http modules from their subpaths, and drop comments that restated the code. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/toolkits/project/handlers.ts | 38 ++++----- .../ThreadWatchReactor.test.ts | 57 ++++++++----- .../orchestration-v2/ThreadWatchReactor.ts | 79 +++++++++++++------ .../src/orchestration-v2/threadWatch.test.ts | 13 +-- .../src/orchestration-v2/threadWatch.ts | 40 +++++----- 5 files changed, 139 insertions(+), 88 deletions(-) diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index df9715a04b59..603cd3174656 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -155,20 +155,7 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { message: "Pass modelSelection: the project has no default model. orchestrator_capabilities lists providers and models.", }); - // Watched before the thread exists, so its first run cannot be missed. - const watchDelivery = input.watch; - if (watchDelivery !== undefined) - yield* ThreadWatchReactor.ThreadWatchReactor.pipe( - Effect.flatMap((watches) => - watches.watch({ threadId, projectId, delivery: watchDelivery }), - ), - Effect.catchTags({ - ThreadWatchInvalidError: watchRefused, - ThreadWatchStoreError: watchRefused, - SecretRequestError: watchRefused, - }), - ); - const result = yield* ThreadMessageIntake.launchThread({ + const launch = ThreadMessageIntake.launchThread({ commandId, threadId, projectId, @@ -205,16 +192,21 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { } return unavailable(); }), - // A failed launch may still have created the thread; a check ends the - // watch only if it did not. - Effect.onError(() => - watchDelivery === undefined - ? Effect.void - : ThreadWatchReactor.ThreadWatchReactor.pipe( - Effect.flatMap((watches) => watches.checkThread(threadId)), - ), - ), ); + 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/orchestration-v2/ThreadWatchReactor.test.ts b/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts index d98649d24758..c60f4d142771 100644 --- a/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts +++ b/apps/server/src/orchestration-v2/ThreadWatchReactor.test.ts @@ -235,24 +235,28 @@ describe("ThreadWatchReactor", () => { }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); - it.effect( - "a watch made before its thread exists hears its first run, or ends if it never comes", - () => - Effect.gen(function* () { - const { received, reactor, watch, settle } = yield* makeHarness({ initial: null }); - yield* watch(); - yield* reactor.drain; - assert.strictEqual(received.length, 0, "the thread is still being launched"); - yield* settle(shell({ status: "failed", activityRunStatus: null })); - assert.deepInclude(verified(received[0]).data, { kind: "failed", runId: RUN_1 }); + 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 }); - // A launch that failed before creating its thread. + for (const failure of [Effect.fail("no thread"), Effect.interrupt]) { const unborn = yield* makeHarness({ initial: null }); - yield* unborn.watch(); - yield* unborn.settle(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)), + } + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); it.effect("after a restart, a watch picks up from the thread as it is then", () => @@ -341,15 +345,30 @@ describe("ThreadWatchReactor", () => { }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); - it.effect("settling the thread sends ended and ends the watch", () => + 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(shell({ settledOverride: "settled" })); - assert.deepInclude(verified(received[0]).data, { kind: "ended" }); - yield* settle(completed()); - assert.strictEqual(received.length, 1); + 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)), ); diff --git a/apps/server/src/orchestration-v2/ThreadWatchReactor.ts b/apps/server/src/orchestration-v2/ThreadWatchReactor.ts index b9e3eea554f3..db78b00ed83a 100644 --- a/apps/server/src/orchestration-v2/ThreadWatchReactor.ts +++ b/apps/server/src/orchestration-v2/ThreadWatchReactor.ts @@ -9,6 +9,7 @@ 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"; @@ -18,7 +19,9 @@ 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 { FetchHttpClient, HttpBody, HttpClient } from "effect/http"; +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 { @@ -73,6 +76,12 @@ class ThreadWatchDeliveryError extends Schema.TaggedError Effect.Effect< { readonly wasWatching: boolean }, ThreadWatchInvalidError | ThreadWatchStoreError | SecretRequestError @@ -94,6 +99,17 @@ export class ThreadWatchReactor extends Context.Service< 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") {} @@ -236,24 +252,29 @@ export const make = Effect.gen(function* () { const now = yield* DateTime.now; for (const watch of watches.values()) { const prior = told.get(watch.id); - const { state, kind } = observeThreadWatch(shell, prior ?? UNSEEN); + const { state, kinds } = observeThreadWatch(shell, prior ?? UNSEEN); told.set(watch.id, state); // A first look only sets the baseline, unless the thread already ended. - if (kind === null || (prior === undefined && kind !== "ended")) continue; - const eventId = threadWatchEventId(watch.id, kind, state); - const body = threadWatchEventBody({ - eventId, - environmentId, - threadId, - shell, - kind, - occurredAt: now, - }); + const send = prior === undefined ? kinds.filter((kind) => kind === "ended") : kinds; + if (send.length === 0) continue; yield* FiberSet.run( deliveries, - kind === "ended" - ? deliver(watch, eventId, body).pipe(Effect.andThen(end(watch))) - : deliver(watch, eventId, body), + 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)), ); } }, @@ -277,7 +298,6 @@ export const make = Effect.gen(function* () { .pipe(Effect.mapError((cause) => new ThreadWatchStoreError({ cause }))); if (shell?.lineage.relationshipToParent === "subagent") return yield* new ThreadWatchInvalidError({ reason: "subagent" }); - // A thread that does not exist yet is being launched: its first run is news. const baseline = shell === null ? UNSEEN : observeThreadWatch(shell, UNSEEN).state; if (baseline.ended) return yield* new ThreadWatchInvalidError({ reason: "ended" }); const secret = @@ -322,6 +342,21 @@ export const make = Effect.gen(function* () { }, ); + 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( @@ -329,7 +364,6 @@ export const make = Effect.gen(function* () { wakesWatchers(event) ? worker.enqueue(event.threadId) : Effect.void, ), ); - // Watches from before this start take their baseline now. yield* Effect.forEach([...watchesByThread.keys()], worker.enqueue, { discard: true }); }, ); @@ -339,6 +373,7 @@ export const make = Effect.gen(function* () { checkThread: (threadId) => worker.enqueue(threadId).pipe(Effect.asVoid), watch, unwatch, + watchLaunch, drain: worker.drain.pipe(Effect.andThen(FiberSet.awaitEmpty(deliveries))), }); }); diff --git a/apps/server/src/orchestration-v2/threadWatch.test.ts b/apps/server/src/orchestration-v2/threadWatch.test.ts index 352414906d8d..bd5fce1aff53 100644 --- a/apps/server/src/orchestration-v2/threadWatch.test.ts +++ b/apps/server/src/orchestration-v2/threadWatch.test.ts @@ -81,7 +81,7 @@ function told(shells: ReadonlyArray) { return shells.map((next) => { const observed = observeThreadWatch(next, state); state = observed.state; - return observed.kind; + return observed.kinds.join("+") || null; }); } @@ -112,9 +112,10 @@ describe("observeThreadWatch", () => { approval, question, shell({ latestRunId: RUN_2, status: "failed", activityRunStatus: null }), - completed(RUN_2, "2026-10-20T12:09:00.000Z"), + { ...completed(RUN_2, "2026-10-20T12:09:00.000Z"), settledOverride: "settled" }, + null, ]), - [null, "completed", null, "approval", null, "question", "failed", "completed"], + [null, "completed", null, "approval", null, "question", "failed", "completed+ended", null], ); }); @@ -130,10 +131,10 @@ describe("observeThreadWatch", () => { const first = observeThreadWatch(asking("request-1"), UNSEEN); const answered = observeThreadWatch(shell(), first.state); const second = observeThreadWatch(asking("request-2"), answered.state); - assert.strictEqual(second.kind, "question"); + assert.deepStrictEqual(second.kinds, ["question"]); assert.notStrictEqual( - threadWatchEventId("tw", "question", first.state), - threadWatchEventId("tw", "question", second.state), + threadWatchEventId("tw", "question", first.state, at("2026-10-20T12:06:00.000Z")), + threadWatchEventId("tw", "question", second.state, at("2026-10-20T12:06:00.000Z")), ); }); }); diff --git a/apps/server/src/orchestration-v2/threadWatch.ts b/apps/server/src/orchestration-v2/threadWatch.ts index 5ae68bcbe955..ae7f59bd7c69 100644 --- a/apps/server/src/orchestration-v2/threadWatch.ts +++ b/apps/server/src/orchestration-v2/threadWatch.ts @@ -15,20 +15,21 @@ export interface ThreadWatchState { export const UNSEEN: ThreadWatchState = { attention: null, completion: null, ended: false }; -/** Uses the phases mobile alerts use, so a watcher hears what a phone would. */ +/** + * 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 kind: ThreadWatchEventKind | null } { - if ( +): { 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" - ) { - return { state: { ...prior, ended: true }, kind: prior.ended ? null : "ended" }; - } - const phase = resolveThreadAwarenessPhaseV2(shell); + shell.settledOverride === "settled"; + const phase = shell === null ? null : resolveThreadAwarenessPhaseV2(shell); const attentionKind = phase === "waiting_for_approval" ? "approval" @@ -41,31 +42,34 @@ export function observeThreadWatch( const attention = attentionKind === null ? null - : `${shell.latestRunId ?? ""}:${attentionKind}:${shell.pendingRuntimeRequest?.id ?? ""}`; + : `${shell?.latestRunId ?? ""}:${attentionKind}:${shell?.pendingRuntimeRequest?.id ?? ""}`; const completion = - phase === "completed" + phase === "completed" && shell !== null ? DateTime.formatIso(shell.latestRunCompletedAt ?? shell.updatedAt) : prior.completion; - const state = { attention, completion, ended: false }; - if (attentionKind !== null && attention !== prior.attention) - return { state, kind: attentionKind }; - if (completion !== null && completion !== prior.completion) return { state, kind: "completed" }; - return { state, kind: null }; + 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); -/** Deterministic, as MCP Events subscription ids are. */ export const threadWatchId = (threadId: ThreadId, url: string) => `tw_${digest(`${threadId}\0${url}`)}`; -/** Stable across retries, so receivers can drop repeats. */ export const threadWatchEventId = ( watchId: string, kind: ThreadWatchEventKind, state: ThreadWatchState, -) => `evt_${digest(`${watchId}\0${kind}\0${state.attention ?? state.completion ?? ""}`)}`; + 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;