From 16878476ac837c4a57a900a38d30f5a7d4fa0217 Mon Sep 17 00:00:00 2001 From: JJ Eaton <58495321+oldbettie@users.noreply.github.com> Date: Sat, 31 Oct 2026 09:55:36 +0700 Subject: [PATCH 1/2] fix(server): thread reads can take the latest items and waits return on approvals t3_thread_read with limit:1 returned the first item, so listeners got a child's opening brief instead of its final reply. A new tail option takes the last `limit` matching items, still in position order; only those payloads are decoded. hasMore then means older items exist. t3_thread_wait with returnOnQuestion only returned on live questions, so callers polled t3_get_pending_questions for approvals. returnOnQuestion now also returns on a live approval and reports the oldest as pendingApproval ({requestId, kind}). It reports only; it never approves. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/mcp/OrchestratorMcpService.ts | 4 ++ apps/server/src/mcp/agents/remoteThreads.ts | 4 +- .../src/mcp/toolkits/orchestrator/tools.ts | 4 +- .../orchestration-v2/ProjectionStore.test.ts | 14 +++++ .../src/orchestration-v2/ProjectionStore.ts | 6 +- .../ThreadManagementService.test.ts | 36 +++++++++++- .../ThreadManagementService.ts | 57 ++++++++++++++----- packages/contracts/src/orchestratorMcp.ts | 13 ++++- 8 files changed, 117 insertions(+), 21 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 943303ba7930e..d5a887a2482c5 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -3003,6 +3003,7 @@ const make = Effect.gen(function* () { afterPosition, limit, view, + ...(input.tail === true ? { tail: true } : {}), ...(input.itemId === undefined ? {} : { itemId: input.itemId }), }) .pipe(Effect.mapError(threadManagementFailure)); @@ -3195,6 +3196,9 @@ const make = Effect.gen(function* () { ...(result.pendingQuestion === undefined ? {} : { hasPendingQuestion: true, pendingRequestId: result.pendingQuestion }), + ...(result.pendingApproval === undefined + ? {} + : { pendingApproval: result.pendingApproval }), ...(result.resultRun === undefined ? {} : { diff --git a/apps/server/src/mcp/agents/remoteThreads.ts b/apps/server/src/mcp/agents/remoteThreads.ts index 12c995b2cccd1..9000f85714eab 100644 --- a/apps/server/src/mcp/agents/remoteThreads.ts +++ b/apps/server/src/mcp/agents/remoteThreads.ts @@ -233,7 +233,9 @@ export const readThread = Effect.fn("remoteThreads.readThread")(function* ( const afterPosition = input.afterPosition ?? -1; const limit = input.limit ?? DEFAULT_READ_LIMIT; const after = messages.filter(({ position }) => position > afterPosition); - items = after.slice(0, limit).map((entry) => item(entry, 0)); + items = (input.tail === true ? after.slice(-limit) : after.slice(0, limit)).map((entry) => + item(entry, 0), + ); hasMore = after.length > limit; } const latestRun = thread.runs.at(-1); diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 67a6d6e1383ee..ea9223299f0e2 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -246,7 +246,7 @@ const ThreadListTool = Tool.make("t3_thread_list", { const ThreadReadTool = Tool.make("t3_thread_read", { description: - "Read durable state and a paginated timeline from any T3 thread in this environment, or pass environmentId for a thread on another environment (gateway connections support messages view over the latest 500 messages; linked environments support the full read; the result carries environmentId). The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this thread's direct app-owned helper acknowledges its automatic completion delivery. Continue with afterPosition=nextPosition. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units. The thread also reports its snooze state and, in pullRequests, every linked pull request; linkedPullRequest is a legacy single link and can be null while pullRequests still has entries. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title.", + "Read durable state and a paginated timeline from any T3 thread in this environment, or pass environmentId for a thread on another environment (gateway connections support messages view over the latest 500 messages; linked environments support the full read; the result carries environmentId). The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this thread's direct app-owned helper acknowledges its automatic completion delivery. Continue with afterPosition=nextPosition; pass tail=true with limit=1 for the latest item, such as a child's final reply. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units. The thread also reports its snooze state and, in pullRequests, every linked pull request; linkedPullRequest is a legacy single link and can be null while pullRequests still has entries. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title.", parameters: OrchestratorMcpThreadReadInput, success: OrchestratorMcpThreadReadResult, failure: OrchestratorMcpFailure, @@ -293,7 +293,7 @@ const ThreadSendTool = Tool.make("t3_thread_send", { const ThreadWaitTool = Tool.make("t3_thread_wait", { description: - "Wait for a T3 thread run to reach a terminal durable state. Pass environmentId for a thread on another environment. Without runId, the latest run at call time is selected; an idle thread returns immediately. A thread blocked on a question returns early with hasPendingQuestion and pendingRequestId for t3_pending_request. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. Waiting reports status only and does not acknowledge a delegated result.", + "Wait for a T3 thread run to reach a terminal durable state. Pass environmentId for a thread on another environment. Without runId, the latest run at call time is selected; an idle thread returns immediately. A thread blocked on a question returns early with hasPendingQuestion and pendingRequestId for t3_pending_request; one blocked on an approval returns early with pendingApproval, which grants no approval rights. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. Waiting reports status only and does not acknowledge a delegated result.", parameters: OrchestratorMcpThreadWaitInput, success: OrchestratorMcpThreadWaitResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index db1f2e5369db6..bc343c83a5982 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6243,6 +6243,20 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }); assert.deepEqual(followingPage.items, targetAfterRollback.visibleTurnItems.slice(2)); assert.isFalse(followingPage.hasMore); + const latestReply = yield* projectionStore.getTimelinePage(targetThreadId, { + limit: 1, + tail: true, + }); + assert.deepEqual(latestReply.items, targetAfterRollback.visibleTurnItems.slice(3, 4)); + assert.isTrue(latestReply.hasMore); + const latestActivity = yield* projectionStore.getTimelinePage(targetThreadId, { + view: "activity", + limit: 2, + tail: true, + afterPosition: 1, + }); + assert.deepEqual(latestActivity.items, targetAfterRollback.visibleTurnItems.slice(3)); + assert.isTrue(latestActivity.hasMore); assert.deepEqual( targetAfterRollback.visibleTurnItems.map((row) => [ diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 4732fc2fc87a8..c24663c9d4155 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -433,6 +433,8 @@ export type ProjectionRecords = Pick< export interface ProjectionTimelinePageOptions { readonly afterPosition?: number; + /** Take the last `limit` matches instead of the first; `hasMore` then means older ones exist. */ + readonly tail?: boolean; readonly itemId?: TurnItemId; readonly view?: "messages" | "activity"; readonly limit: number; @@ -5874,7 +5876,9 @@ export const layer: Layer.Layer = )) : row.sourceItemId === options.itemId, ); - const page = matching.slice(0, options.limit); + const page = options.tail + ? matching.slice(-options.limit) + : matching.slice(0, options.limit); const items = yield* Effect.forEach(page, (row) => Effect.gen(function* () { if (row.synthetic !== undefined) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 1b5651e02e34e..8200da586bf01 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -276,7 +276,7 @@ it.effect("waitForThread returns early once the thread blocks on a live question const projectId = ProjectId.make("project:thread-management:wait-question"); const threadId = ThreadId.make("thread:thread-management:wait-question"); const runId = RunId.make("run:thread-management:wait-question"); - const subscribed = yield* Deferred.make(); + let subscribed = yield* Deferred.make(); const events = yield* Queue.unbounded(); const question = (id: string, type: "live" | "message") => ({ id: RuntimeRequestId.make(id), @@ -361,6 +361,40 @@ it.effect("waitForThread returns early once the thread blocks on a live question run: { id: runId, status: "running" }, pendingQuestion: "request:wait-question:later", }); + + // An approval blocks the run on a person too, so it also ends the wait. + const approval = (id: string, createdAt: number) => ({ + id: RuntimeRequestId.make(id), + kind: "command", + status: "pending", + responseCapability: { type: "live" }, + createdAt: DateTime.makeUnsafe(createdAt), + }); + runtimeRequests = []; + subscribed = yield* Deferred.make(); + const approvalFiber = yield* service + .waitForThread({ + projectId, + threadId, + runId, + timeoutMs: 60 * 60 * 1_000, + untilQuestion: true, + }) + .pipe(Effect.forkChild); + yield* Effect.raceFirst(Deferred.await(subscribed), Fiber.await(approvalFiber)); + runtimeRequests = [ + approval("request:wait-approval:newer", 2_000), + approval("request:wait-approval:older", 1_000), + ] as unknown as typeof runtimeRequests; + yield* Queue.offer(events, stored(3)); + + const approvalResult = yield* Fiber.join(approvalFiber); + expect(approvalResult).toMatchObject({ + timedOut: false, + run: { id: runId, status: "running" }, + pendingApproval: { requestId: "request:wait-approval:older", kind: "command" }, + }); + expect(approvalResult).not.toHaveProperty("pendingQuestion"); }), ); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index f4ff79f5897d9..4384766e18166 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -19,6 +19,7 @@ import { OrchestrationV2StopWorkflowError, type OrchestrationV2StopWorkflowInput, type ModelSelection, + ProviderRequestKind, type OrchestrationV2Actor, type OrchestrationV2Command, type OrchestrationV2GetTurnItemResult, @@ -176,7 +177,10 @@ export interface ThreadManagementWaitInput { * completion delivery will wake it again, as a delegated task's result does. */ readonly untilDelegatedWorkSettled?: boolean; - /** Also return while the thread is blocked on a live question, which no run end will follow. */ + /** + * Also return while the thread is blocked on a live question or approval, + * which no run end will follow. + */ readonly untilQuestion?: boolean; } @@ -188,6 +192,11 @@ export interface ThreadManagementWaitResult { readonly resultRun?: OrchestrationV2Run; /** With `untilQuestion`: the live question the thread is blocked on. */ readonly pendingQuestion?: RuntimeRequestId; + /** With `untilQuestion`: the oldest live approval the thread is blocked on. */ + readonly pendingApproval?: { + readonly requestId: RuntimeRequestId; + readonly kind: ProviderRequestKind; + }; } export interface ThreadManagementInterruptInput { @@ -557,6 +566,7 @@ function latestSteerableRun( } const SETTLE_AFTER_RUN_WAIT_MS = 24 * 60 * 60 * 1_000; +const isApprovalRequestKind = Schema.is(ProviderRequestKind); const make = Effect.gen(function* () { const threadDispatch = yield* ThreadCommandExecutor.ThreadCommandExecutor; @@ -1074,16 +1084,25 @@ const make = Effect.gen(function* () { return pending ? undefined : progress.resultRun; }); const untilQuestion = input.untilQuestion === true; - const blockingQuestion = getProjectThreadRecords(input, ["runtimeRequests"]).pipe( - Effect.map( - ({ runtimeRequests }) => - runtimeRequests.find( - (request) => - request.kind === "user_input" && - request.status === "pending" && - request.responseCapability.type === "live", - )?.id, - ), + const blockingRequests = getProjectThreadRecords(input, ["runtimeRequests"]).pipe( + Effect.map(({ runtimeRequests }) => { + const live = runtimeRequests.filter( + (request) => request.status === "pending" && request.responseCapability.type === "live", + ); + let pendingApproval: ThreadManagementWaitResult["pendingApproval"]; + let oldestCreatedAt = Infinity; + for (const request of live) { + if (!isApprovalRequestKind(request.kind)) continue; + const createdAt = DateTime.toEpochMillis(request.createdAt); + if (createdAt >= oldestCreatedAt) continue; + oldestCreatedAt = createdAt; + pendingApproval = { requestId: request.id, kind: request.kind }; + } + return { + pendingQuestion: live.find((request) => request.kind === "user_input")?.id, + pendingApproval, + }; + }), ); const evaluate = Effect.fn("orchestrationV2.threadManagement.evaluate")(function* ( runs: ReadonlyArray, @@ -1099,12 +1118,17 @@ const make = Effect.gen(function* () { })), ) : { ...outcomeNow, resultRun: undefined }; - const pendingQuestion = - current.settled || !untilQuestion ? undefined : yield* blockingQuestion; + const blocked = + current.settled || !untilQuestion + ? { pendingQuestion: undefined, pendingApproval: undefined } + : yield* blockingRequests; return { ...current, - pendingQuestion, - done: current.settled || pendingQuestion !== undefined, + ...blocked, + done: + current.settled || + blocked.pendingQuestion !== undefined || + blocked.pendingApproval !== undefined, }; }); const settledResult = (current: Effect.Success>) => ({ @@ -1115,6 +1139,9 @@ const make = Effect.gen(function* () { ...(current.pendingQuestion === undefined ? {} : { pendingQuestion: current.pendingQuestion }), + ...(current.pendingApproval === undefined + ? {} + : { pendingApproval: current.pendingApproval }), }); const initial = yield* evaluate(target.runs); if (initial.done) return settledResult(initial); diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 5b6b72df9a40d..9b95afea0057b 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -436,6 +436,12 @@ export const OrchestratorMcpThreadReadInput = Schema.Struct({ textOffset: Schema.optional(NonNegativeInt), view: Schema.optional(Schema.Literals(["messages", "activity"])), afterPosition: Schema.optional(NonNegativeInt), + tail: Schema.optional( + Schema.Boolean.annotate({ + description: + "Return the latest `limit` items (after afterPosition, if set), still in position order. hasMore then means older items exist.", + }), + ), limit: Schema.optional(clampedPositiveInt(100)).annotate({ description: "Max items, 1 to 100.", }), @@ -633,7 +639,9 @@ export const OrchestratorMcpThreadWaitInput = Schema.Struct({ }), ), returnOnQuestion: Schema.optional( - Schema.Boolean.annotate({ description: "Default true: return when it blocks on a question." }), + Schema.Boolean.annotate({ + description: "Default true: return when it blocks on a question or an approval.", + }), ), }); export type OrchestratorMcpThreadWaitInput = typeof OrchestratorMcpThreadWaitInput.Type; @@ -649,6 +657,9 @@ export const OrchestratorMcpThreadWaitResult = Schema.Struct({ resultRun: Schema.optional(Schema.Struct({ runId: RunId, status: OrchestrationV2RunStatus })), hasPendingQuestion: Schema.optional(Schema.Boolean), pendingRequestId: Schema.optional(RuntimeRequestId), + pendingApproval: Schema.optional( + Schema.Struct({ requestId: RuntimeRequestId, kind: ProviderRequestKind }), + ).annotate({ description: "Oldest pending approval. Grants no approval rights." }), }); export type OrchestratorMcpThreadWaitResult = typeof OrchestratorMcpThreadWaitResult.Type; From f6cf9df442d2c52ce83a0f49303385cf2363b03a Mon Sep 17 00:00:00 2001 From: JJ Eaton <58495321+oldbettie@users.noreply.github.com> Date: Sat, 31 Oct 2026 10:02:22 +0700 Subject: [PATCH 2/2] fix(server): trim new thread tool wording and record the context budget Co-Authored-By: Claude Opus 5.5 --- apps/server/src/mcp/toolkits/orchestrator/tools.ts | 4 ++-- apps/server/src/provider/agentContextBudget.json | 2 +- packages/contracts/src/orchestratorMcp.ts | 7 ++----- 3 files changed, 5 insertions(+), 8 deletions(-) diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index ea9223299f0e2..3ebb0dc3cc15e 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -246,7 +246,7 @@ const ThreadListTool = Tool.make("t3_thread_list", { const ThreadReadTool = Tool.make("t3_thread_read", { description: - "Read durable state and a paginated timeline from any T3 thread in this environment, or pass environmentId for a thread on another environment (gateway connections support messages view over the latest 500 messages; linked environments support the full read; the result carries environmentId). The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this thread's direct app-owned helper acknowledges its automatic completion delivery. Continue with afterPosition=nextPosition; pass tail=true with limit=1 for the latest item, such as a child's final reply. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units. The thread also reports its snooze state and, in pullRequests, every linked pull request; linkedPullRequest is a legacy single link and can be null while pullRequests still has entries. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title.", + "Read durable state and a paginated timeline from any T3 thread in this environment, or pass environmentId for a thread on another environment (gateway connections support messages view over the latest 500 messages; linked environments support the full read; the result carries environmentId). The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this thread's direct app-owned helper acknowledges its automatic completion delivery. Continue with afterPosition=nextPosition, or pass tail=true for the latest items. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units. The thread also reports its snooze state and, in pullRequests, every linked pull request; linkedPullRequest is a legacy single link and can be null while pullRequests still has entries. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title.", parameters: OrchestratorMcpThreadReadInput, success: OrchestratorMcpThreadReadResult, failure: OrchestratorMcpFailure, @@ -293,7 +293,7 @@ const ThreadSendTool = Tool.make("t3_thread_send", { const ThreadWaitTool = Tool.make("t3_thread_wait", { description: - "Wait for a T3 thread run to reach a terminal durable state. Pass environmentId for a thread on another environment. Without runId, the latest run at call time is selected; an idle thread returns immediately. A thread blocked on a question returns early with hasPendingQuestion and pendingRequestId for t3_pending_request; one blocked on an approval returns early with pendingApproval, which grants no approval rights. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. Waiting reports status only and does not acknowledge a delegated result.", + "Wait for a T3 thread run to reach a terminal durable state. Pass environmentId for a thread on another environment. Without runId, the latest run at call time is selected; an idle thread returns immediately. A thread blocked on a question returns early with hasPendingQuestion and pendingRequestId for t3_pending_request, or on an approval with pendingApproval. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. Waiting reports status only and does not acknowledge a delegated result.", parameters: OrchestratorMcpThreadWaitInput, success: OrchestratorMcpThreadWaitResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index d30afe749da23..1b9d5d06e9fd1 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 8688, "codexAdditionalContextBytes": 10385, - "mcpToolListJsonBytes": 57867 + "mcpToolListJsonBytes": 58034 } diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 9b95afea0057b..42befc37d5bf0 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -437,10 +437,7 @@ export const OrchestratorMcpThreadReadInput = Schema.Struct({ view: Schema.optional(Schema.Literals(["messages", "activity"])), afterPosition: Schema.optional(NonNegativeInt), tail: Schema.optional( - Schema.Boolean.annotate({ - description: - "Return the latest `limit` items (after afterPosition, if set), still in position order. hasMore then means older items exist.", - }), + Schema.Boolean.annotate({ description: "Latest `limit` items; hasMore means older ones." }), ), limit: Schema.optional(clampedPositiveInt(100)).annotate({ description: "Max items, 1 to 100.", @@ -640,7 +637,7 @@ export const OrchestratorMcpThreadWaitInput = Schema.Struct({ ), returnOnQuestion: Schema.optional( Schema.Boolean.annotate({ - description: "Default true: return when it blocks on a question or an approval.", + description: "Default true: return on a question or approval.", }), ), });