fix(server): thread reads can take the latest items and waits return on approvals

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
           ? {}
           : {

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);

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/<threadId>)` 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/<threadId>)` 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, 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,

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) => [

apps/server/src/orchestration-v2/ProjectionStore.ts

@@ -433,6 +433,8 @@ export type ProjectionRecords<K extends ProjectionRecordField> = 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<ProjectionStoreV2, never, SqlClient.SqlClient> =
                       ))
                   : 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)

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<void>();
+    let subscribed = yield* Deferred.make<void>();
     const events = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
     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<void>();
+    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");
   }),
 );
 

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<OrchestrationV2Run>,
@@ -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<ReturnType<typeof evaluate>>) => ({
@@ -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);

apps/server/src/provider/agentContextBudget.json

@@ -1,5 +1,5 @@
 {
   "claudeSystemAppendBytes": 8688,
   "codexAdditionalContextBytes": 10385,
-  "mcpToolListJsonBytes": 57867
+  "mcpToolListJsonBytes": 58034
 }

packages/contracts/src/orchestratorMcp.ts

@@ -436,6 +436,9 @@ 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: "Latest `limit` items; hasMore means older ones." }),
+  ),
   limit: Schema.optional(clampedPositiveInt(100)).annotate({
     description: "Max items, 1 to 100.",
   }),
@@ -633,7 +636,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 on a question or approval.",
+    }),
   ),
 });
 export type OrchestratorMcpThreadWaitInput = typeof OrchestratorMcpThreadWaitInput.Type;
@@ -649,6 +654,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;