From 885b1b4860f806578163ac12474552ef871a7694 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Wed, 28 Oct 2026 22:17:04 +0100 Subject: [PATCH 1/6] perf(server): trim discarded text from remote failure scans --- apps/server/src/peer/RemoteDelegation.test.ts | 224 +++++++++++++----- apps/server/src/peer/RemoteDelegation.ts | 6 +- 2 files changed, 172 insertions(+), 58 deletions(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index eeef7bf05a36e..6ddd893f48447 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -9,6 +9,7 @@ import { type ModelSelection, NodeId, type OrchestrationV2Run, + type OrchestratorMcpThreadReadInput, ProjectId, ProviderDriverKind, ProviderInstanceId, @@ -34,6 +35,7 @@ import * as Schedule from "effect/Schedule"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as Tracer from "effect/Tracer"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; import * as TestClock from "effect/testing/TestClock"; import type * as SqlClient from "effect/sql/SqlClient"; import { McpSchema, McpServer } from "effect/ai"; @@ -88,7 +90,17 @@ type LaunchedWith = Pick< >; /** The thread the box launches for the task, which the test finishes. */ +interface TimelineEntry { + readonly id: TurnItemId; + readonly runId: RunId; + readonly text: string; + readonly type: "assistant_message" | "error"; +} + interface BoxThread { + readonly timeline: Ref.Ref | null>; + readonly pageSize: Ref.Ref; + readonly holdRefill: Ref.Ref; readonly launched: Ref.Ref; /** The link session that launched it, as the box stamps it. */ readonly linkOrigin: Ref.Ref<{ readonly sessionId: string; readonly label: string } | null>; @@ -145,6 +157,9 @@ interface BoxThread { const makeBoxThread = Effect.gen(function* () { return { + timeline: yield* Ref.make | null>(null), + pageSize: yield* Ref.make(1), + holdRefill: yield* Ref.make(false), launched: yield* Ref.make(null), linkOrigin: yield* Ref.make<{ readonly sessionId: string; readonly label: string } | null>( null, @@ -359,7 +374,10 @@ const boxToolkitLayer = (thread: BoxThread) => { }), getTimelinePage: (threadId, options) => Effect.gen(function* () { - if (yield* Ref.get(thread.holdRead)) { + if ( + (yield* Ref.get(thread.holdRead)) || + (options.itemId !== undefined && (yield* Ref.get(thread.holdRefill))) + ) { yield* Deferred.succeed(thread.readStarted, undefined); yield* Deferred.await(thread.releaseRead); } @@ -370,61 +388,63 @@ const boxToolkitLayer = (thread: BoxThread) => { const laterError = yield* Ref.get(thread.laterError); const restart = yield* Ref.get(thread.restart); // One item per page, with errors available only in the activity view. - const replies = [ - ...(reply === null - ? [] - : [ - { - id: TurnItemId.make("item:reply"), - runId, - text: reply, - type: "assistant_message", - }, - ]), - ...(error === null - ? [] - : [{ id: TurnItemId.make("item:error"), runId, text: error, type: "error" }]), - ...(later === null - ? [] - : [ - { - id: TurnItemId.make("item:later-reply"), - runId: boxLaterRunId, - text: later, - type: "assistant_message", - }, - ]), - ...(laterError === null - ? [] - : [ - { - id: TurnItemId.make("item:later-error"), - runId: boxLaterRunId, - text: laterError, - type: "error", - }, - ]), - ...(queued === null - ? [] - : [ - { - id: TurnItemId.make("item:queued-reply"), - runId: boxQueuedRunId, - text: queued, - type: "assistant_message", - }, - ]), - ...(restart === null - ? [] - : [ - { - id: TurnItemId.make("item:restarted-reply"), - runId: continuationOf(restart.of), - text: restart.reply, - type: "assistant_message", - }, - ]), - ].map((entry, position) => ({ ...entry, position })); + const replies = ( + (yield* Ref.get(thread.timeline)) ?? [ + ...(reply === null + ? [] + : [ + { + id: TurnItemId.make("item:reply"), + runId, + text: reply, + type: "assistant_message", + }, + ]), + ...(error === null + ? [] + : [{ id: TurnItemId.make("item:error"), runId, text: error, type: "error" }]), + ...(later === null + ? [] + : [ + { + id: TurnItemId.make("item:later-reply"), + runId: boxLaterRunId, + text: later, + type: "assistant_message", + }, + ]), + ...(laterError === null + ? [] + : [ + { + id: TurnItemId.make("item:later-error"), + runId: boxLaterRunId, + text: laterError, + type: "error", + }, + ]), + ...(queued === null + ? [] + : [ + { + id: TurnItemId.make("item:queued-reply"), + runId: boxQueuedRunId, + text: queued, + type: "assistant_message", + }, + ]), + ...(restart === null + ? [] + : [ + { + id: TurnItemId.make("item:restarted-reply"), + runId: continuationOf(restart.of), + text: restart.reply, + type: "assistant_message", + }, + ]), + ] + ).map((entry, position) => ({ ...entry, position })); const visible = options.itemId === undefined ? replies.filter( @@ -433,7 +453,7 @@ const boxToolkitLayer = (thread: BoxThread) => { : replies.filter((entry) => entry.id === options.itemId); const matching = visible .filter((entry) => entry.position > (options.afterPosition ?? -1)) - .slice(0, 1); + .slice(0, yield* Ref.get(thread.pageSize)); return { items: matching.map(({ id, runId: itemRunId, text, position, type }) => ({ position, @@ -1592,6 +1612,96 @@ it.effect.each([false, true])( ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); +it.effect.each(["tail-error", "middle-error", "reply-fallback", "cancel-refill"] as const)( + "reads a failed task through full activity pages (%s)", + (scenario) => + Effect.gen(function* () { + const reads: Array = []; + const nativeFetch = yield* FetchHttpClient.Fetch; + const fetch: typeof globalThis.fetch = (input, init) => { + const body = init?.body instanceof Uint8Array ? new TextDecoder().decode(init.body) : null; + if (body?.startsWith("{")) { + const rpc: { + method?: string; + params?: { name?: string; arguments?: OrchestratorMcpThreadReadInput }; + } = JSON.parse(body); + if ( + rpc.method === "tools/call" && + rpc.params?.name === "t3_thread_read" && + rpc.params.arguments !== undefined + ) { + reads.push(rpc.params.arguments); + } + } + return nativeFetch(input, init); + }; + return yield* Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const a = yield* makeLaptop(database, offers); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); + const delegated = yield* delegateToBox(a); + const result = `Provider details: ${"detail ".repeat(8_000)}end.`; + const middle = scenario !== "tail-error"; + const ownType = scenario === "reply-fallback" ? "assistant_message" : "error"; + const history = (count: number, runId: RunId, prefix: string): Array => + Array.from({ length: count }, (_, index) => ({ + id: TurnItemId.make(`${prefix}:${index}`), + runId, + text: "discarded history ".repeat(120), + type: "assistant_message", + })); + yield* Ref.set(thread.timeline, [ + ...history(middle ? 150 : 300, boxRunId, "before"), + { id: TurnItemId.make("selected"), runId: boxRunId, text: result, type: ownType }, + ...(middle ? history(200, boxLaterRunId, "later") : []), + { + id: TurnItemId.make("other-error"), + runId: boxLaterRunId, + text: "Wrong run", + type: "error", + }, + ]); + yield* Ref.set(thread.pageSize, 100); + yield* Ref.set(thread.status, "failed"); + if (scenario === "cancel-refill") yield* Ref.set(thread.holdRefill, true); + yield* Deferred.succeed(thread.finished, undefined); + if (scenario === "cancel-refill") { + yield* Deferred.await(thread.readStarted); + yield* a.threads.stopDelegatedTasks({ + threadId: parentThreadId, + commandId: CommandId.make("command:stop:mid-refill"), + }); + yield* Deferred.succeed(thread.releaseRead, undefined); + expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ status: "interrupted" }); + yield* Deferred.await(thread.interruptedOnce); + expect(yield* Ref.get(thread.stops)).toContain("thread.stop"); + } else { + expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ + status: "failed", + result, + }); + } + const pages = reads.filter((read) => read.itemId === undefined); + expect(pages.map((read) => read.afterPosition)).toEqual([undefined, 99, 199, 299]); + expect(pages.map((read) => read.maxCharsPerItem)).toEqual([undefined, 1, 1, undefined]); + const refills = reads.filter((read) => read.itemId !== undefined); + expect(refills[0]).toMatchObject({ itemId: "selected", textOffset: middle ? 1 : 20_000 }); + if (scenario !== "cancel-refill") expect(refills).toHaveLength(middle ? 2 : 1); + }), + ).pipe( + Effect.provide(NodeHttpServer.layerTest), + Effect.provideService(FetchHttpClient.Fetch, fetch), + ); + }), +); + it.effect("keeps a successful remote reply despite an error item in the run's activity", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 932b012969e36..470e196e4efde 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -618,13 +618,16 @@ const make = Effect.gen(function* () { let error: OrchestratorMcpThreadTimelineItem | undefined; let afterPosition: number | undefined; let run: OrchestratorMcpThreadRun | undefined; + let remainingItems = 0; while (true) { const read = yield* forwarding.readThread(scope, remote.environmentId, { threadId: remote.threadId, view: failed ? "activity" : "messages", limit: 100, runLimit: 50, - maxCharsPerItem: 20_000, + // Keep the first and expected final activity pages whole. A result on a + // middle page then needs an extra text read after the scan. + maxCharsPerItem: failed && remainingItems > 100 ? 1 : 20_000, ...(afterPosition === undefined ? {} : { afterPosition }), }); run ??= read.recentRuns.find((candidate) => candidate.runId === runId); @@ -640,6 +643,7 @@ const make = Effect.gen(function* () { } if (!read.hasMore || read.nextPosition === null) break; afterPosition = read.nextPosition; + remainingItems = read.thread.itemCount - afterPosition - 1; } reply = error ?? reply; if (reply === undefined) return undefined; From 3c4478abd4c8048020cf070be176db1e54d95dfe Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 29 Oct 2026 07:40:58 +0100 Subject: [PATCH 2/6] perf(server): read terminal successful task replies without scanning middle pages --- apps/server/src/peer/RemoteDelegation.test.ts | 142 ++++++++++++++++++ apps/server/src/peer/RemoteDelegation.ts | 29 +++- 2 files changed, 170 insertions(+), 1 deletion(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 6ddd893f48447..19eabb54f9727 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -101,6 +101,7 @@ interface BoxThread { readonly timeline: Ref.Ref | null>; readonly pageSize: Ref.Ref; readonly holdRefill: Ref.Ref; + readonly beforeTail: Ref.Ref>; readonly launched: Ref.Ref; /** The link session that launched it, as the box stamps it. */ readonly linkOrigin: Ref.Ref<{ readonly sessionId: string; readonly label: string } | null>; @@ -160,6 +161,7 @@ const makeBoxThread = Effect.gen(function* () { timeline: yield* Ref.make | null>(null), pageSize: yield* Ref.make(1), holdRefill: yield* Ref.make(false), + beforeTail: yield* Ref.make>(Effect.void), launched: yield* Ref.make(null), linkOrigin: yield* Ref.make<{ readonly sessionId: string; readonly label: string } | null>( null, @@ -374,6 +376,9 @@ const boxToolkitLayer = (thread: BoxThread) => { }), getTimelinePage: (threadId, options) => Effect.gen(function* () { + if (options.afterPosition === 299 && options.itemId === undefined) { + yield* Ref.getAndSet(thread.beforeTail, Effect.void).pipe(Effect.flatten); + } if ( (yield* Ref.get(thread.holdRead)) || (options.itemId !== undefined && (yield* Ref.get(thread.holdRefill))) @@ -1702,6 +1707,143 @@ it.effect.each(["tail-error", "middle-error", "reply-fallback", "cancel-refill"] }), ); +it.effect.each([ + "terminal", + "long", + "later-run", + "sparse", + "growing", + "shrinking", + "cancel", + "no-run", + "cancelled", + "interrupted", +] as const)("reads a successful task with an aligned tail probe (%s)", (scenario) => + Effect.gen(function* () { + const reads: Array = []; + const nativeFetch = yield* FetchHttpClient.Fetch; + const fetch: typeof globalThis.fetch = (input, init) => { + const body = init?.body instanceof Uint8Array ? new TextDecoder().decode(init.body) : null; + if (body?.startsWith("{")) { + const rpc: { + method?: string; + params?: { name?: string; arguments?: OrchestratorMcpThreadReadInput }; + } = JSON.parse(body); + if ( + rpc.method === "tools/call" && + rpc.params?.name === "t3_thread_read" && + rpc.params.arguments !== undefined + ) { + reads.push(rpc.params.arguments); + } + } + return nativeFetch(input, init); + }; + return yield* Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const a = yield* makeLaptop(database, offers); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); + const result = + scenario === "long" + ? `Complete ${"detail ".repeat(12_000)}end.` + : "Authoritative result."; + const middle = scenario === "later-run" || scenario === "shrinking"; + const history = (count: number, runId: RunId, prefix: string): Array => + Array.from({ length: count }, (_, index) => ({ + id: TurnItemId.make(`${prefix}:${index}`), + runId, + text: "History", + type: scenario === "sparse" && index % 2 === 0 ? "error" : "assistant_message", + })); + const timeline: Array = [ + ...history(middle ? 150 : 300, boxRunId, "before"), + { + id: TurnItemId.make("selected"), + runId: scenario === "no-run" ? boxLaterRunId : boxRunId, + text: result, + type: "assistant_message", + }, + ...(middle ? history(200, boxLaterRunId, "later") : []), + ]; + yield* Ref.set(thread.timeline, timeline); + yield* Ref.set(thread.pageSize, 100); + if (scenario === "growing") { + yield* Ref.set( + thread.beforeTail, + Ref.set(thread.timeline, [...timeline, ...history(101, boxLaterRunId, "growth")]), + ); + } + if (scenario === "shrinking") { + yield* Ref.set(thread.beforeTail, Ref.set(thread.timeline, timeline.slice(0, 151))); + } + if (scenario === "cancel") { + yield* Ref.set( + thread.beforeTail, + Effect.gen(function* () { + yield* Deferred.succeed(thread.readStarted, undefined); + yield* Deferred.await(thread.releaseRead); + }), + ); + } + if (scenario === "no-run") yield* Ref.set(thread.runless, true); + const delegated = yield* delegateToBox(a); + const status = + scenario === "cancelled" || scenario === "interrupted" ? scenario : "completed"; + yield* Ref.set(thread.status, status); + yield* Deferred.succeed(thread.finished, undefined); + if (scenario === "cancel") { + yield* Deferred.await(thread.readStarted); + yield* a.threads.stopDelegatedTasks({ + threadId: parentThreadId, + commandId: CommandId.make("command:stop:tail-probe"), + }); + yield* Deferred.succeed(thread.releaseRead, undefined); + expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ status: "interrupted" }); + yield* Deferred.await(thread.interruptedOnce); + expect(yield* Ref.get(thread.stops)).toContain("thread.stop"); + } else { + expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ + status, + result, + }); + } + const pages = reads.filter((read) => read.itemId === undefined); + const positions = + scenario === "cancelled" || scenario === "interrupted" + ? [undefined, 99, 199, 299] + : scenario === "later-run" + ? [undefined, 299, 99, 199, 299] + : scenario === "shrinking" + ? [undefined, 299, 99] + : scenario === "growing" + ? [undefined, 299, 99, 199, 299, 399] + : [undefined, 299]; + expect(pages.map((read) => read.afterPosition)).toEqual(positions); + expect( + pages.every( + (read) => + read.view === "messages" && read.limit === 100 && read.maxCharsPerItem === undefined, + ), + ).toBe(true); + const refills = reads.filter((read) => read.itemId !== undefined); + expect(refills).toHaveLength(scenario === "long" ? 2 : 0); + if (scenario === "long") + expect(refills.map((read) => read.textOffset)).toEqual([20_000, 70_000]); + }), + ).pipe( + Effect.provide(NodeHttpServer.layerTest), + Effect.provideService(FetchHttpClient.Fetch, fetch), + ); + }), +); + it.effect("keeps a successful remote reply despite an error item in the run's activity", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 470e196e4efde..762ca87e0e1d6 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -613,12 +613,14 @@ const make = Effect.gen(function* () { remote: OrchestrationV2RemoteTaskChild, runId: RunId | null, failed = false, + probeSuccess = true, ) { let reply: OrchestratorMcpThreadTimelineItem | undefined; let error: OrchestratorMcpThreadTimelineItem | undefined; let afterPosition: number | undefined; let run: OrchestratorMcpThreadRun | undefined; let remainingItems = 0; + let tailProbed = false; while (true) { const read = yield* forwarding.readThread(scope, remote.environmentId, { threadId: remote.threadId, @@ -642,6 +644,25 @@ const make = Effect.gen(function* () { ) ?? error; } if (!read.hasMore || read.nextPosition === null) break; + if (!failed && probeSuccess && !tailProbed) { + tailProbed = true; + const tailPosition = Math.floor((read.thread.itemCount - 1) / 100) * 100 - 1; + if (tailPosition > read.nextPosition) { + const tail = yield* forwarding.readThread(scope, remote.environmentId, { + threadId: remote.threadId, + view: "messages", + limit: 100, + afterPosition: tailPosition, + }); + const tailReply = tail.items.findLast( + (item) => item.type === "assistant_message" && (runId === null || item.runId === runId), + ); + if (!tail.hasMore && tailReply !== undefined) { + reply = tailReply; + break; + } + } + } afterPosition = read.nextPosition; remainingItems = read.thread.itemCount - afterPosition - 1; } @@ -859,7 +880,13 @@ const make = Effect.gen(function* () { waited.resultRun !== undefined && !sharedThread ? waited.resultRun : { runId: waited.runId, status: waited.status }; - const result = yield* lastReply(scope, remote, resultRun.runId, resultRun.status === "failed"); + const result = yield* lastReply( + scope, + remote, + resultRun.runId, + resultRun.status === "failed", + resultRun.status === "completed", + ); yield* complete( parentThreadId, taskId, From 576320467e53c75c6e32c51728f0110f6ee5f912 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sun, 8 Nov 2026 23:30:12 +0100 Subject: [PATCH 3/6] test(server): share remote thread read capture --- apps/server/src/peer/RemoteDelegation.test.ts | 62 +++++++------------ 1 file changed, 24 insertions(+), 38 deletions(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 19eabb54f9727..1962fe1fe94f4 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -941,6 +941,28 @@ const taskResult = (laptopEnv: Effect.Success>, ta ), ); +const captureThreadReads = (nativeFetch: typeof globalThis.fetch) => { + const reads: Array = []; + const fetch: typeof globalThis.fetch = (input, init) => { + const body = init?.body instanceof Uint8Array ? new TextDecoder().decode(init.body) : null; + if (body?.startsWith("{")) { + const rpc: { + method?: string; + params?: { name?: string; arguments?: OrchestratorMcpThreadReadInput }; + } = JSON.parse(body); + if ( + rpc.method === "tools/call" && + rpc.params?.name === "t3_thread_read" && + rpc.params.arguments !== undefined + ) { + reads.push(rpc.params.arguments); + } + } + return nativeFetch(input, init); + }; + return { reads, fetch }; +}; + it.effect("a task delegated to the box wakes the laptop's parent when it ends there", () => Effect.scoped( Effect.gen(function* () { @@ -1621,25 +1643,7 @@ it.effect.each(["tail-error", "middle-error", "reply-fallback", "cancel-refill"] "reads a failed task through full activity pages (%s)", (scenario) => Effect.gen(function* () { - const reads: Array = []; - const nativeFetch = yield* FetchHttpClient.Fetch; - const fetch: typeof globalThis.fetch = (input, init) => { - const body = init?.body instanceof Uint8Array ? new TextDecoder().decode(init.body) : null; - if (body?.startsWith("{")) { - const rpc: { - method?: string; - params?: { name?: string; arguments?: OrchestratorMcpThreadReadInput }; - } = JSON.parse(body); - if ( - rpc.method === "tools/call" && - rpc.params?.name === "t3_thread_read" && - rpc.params.arguments !== undefined - ) { - reads.push(rpc.params.arguments); - } - } - return nativeFetch(input, init); - }; + const { reads, fetch } = captureThreadReads(yield* FetchHttpClient.Fetch); return yield* Effect.scoped( Effect.gen(function* () { const thread = yield* makeBoxThread; @@ -1720,25 +1724,7 @@ it.effect.each([ "interrupted", ] as const)("reads a successful task with an aligned tail probe (%s)", (scenario) => Effect.gen(function* () { - const reads: Array = []; - const nativeFetch = yield* FetchHttpClient.Fetch; - const fetch: typeof globalThis.fetch = (input, init) => { - const body = init?.body instanceof Uint8Array ? new TextDecoder().decode(init.body) : null; - if (body?.startsWith("{")) { - const rpc: { - method?: string; - params?: { name?: string; arguments?: OrchestratorMcpThreadReadInput }; - } = JSON.parse(body); - if ( - rpc.method === "tools/call" && - rpc.params?.name === "t3_thread_read" && - rpc.params.arguments !== undefined - ) { - reads.push(rpc.params.arguments); - } - } - return nativeFetch(input, init); - }; + const { reads, fetch } = captureThreadReads(yield* FetchHttpClient.Fetch); return yield* Effect.scoped( Effect.gen(function* () { const thread = yield* makeBoxThread; From 99849c6a1e2001d6cd23a217e01dec31302a1a4c Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Mon, 9 Nov 2026 22:14:09 +0100 Subject: [PATCH 4/6] test(server): retain executed model notes in tail result reads --- apps/server/src/peer/RemoteDelegation.test.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 1962fe1fe94f4..e500d1cb9fdd9 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -1782,6 +1782,8 @@ it.effect.each([ const delegated = yield* delegateToBox(a); const status = scenario === "cancelled" || scenario === "interrupted" ? scenario : "completed"; + const modelNote = scenario === "terminal" || scenario === "long"; + if (modelNote) yield* Ref.set(thread.executedModel, "gpt-5.4-mini"); yield* Ref.set(thread.status, status); yield* Deferred.succeed(thread.finished, undefined); if (scenario === "cancel") { @@ -1797,7 +1799,9 @@ it.effect.each([ } else { expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ status, - result, + result: modelNote + ? `Requested model gpt-5.4; the provider reported running gpt-5.4-mini.\n\n${result}` + : result, }); } const pages = reads.filter((read) => read.itemId === undefined); From beba1f466758ba1a2e394bf6cfce217c270b9782 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sat, 14 Nov 2026 12:12:57 +0100 Subject: [PATCH 5/6] perf(server): cap the remote tail probe to whole items Keep the probe's pages from carrying text the scan discards, matching the capped pages other remote reads already use. --- apps/server/src/peer/RemoteDelegation.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 762ca87e0e1d6..f6834dcc1b3f6 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -652,6 +652,7 @@ const make = Effect.gen(function* () { threadId: remote.threadId, view: "messages", limit: 100, + maxCharsPerItem: 20_000, afterPosition: tailPosition, }); const tailReply = tail.items.findLast( From 90eb8ec0d16a7d8798237a2fe72ae81da9b8c443 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sat, 14 Nov 2026 12:12:58 +0100 Subject: [PATCH 6/6] test(server): assert remote task reads by response budget The far side now caps each read at 30 KB, so fixed item positions no longer describe its pages. Assert the tail probe, the discard-page trimming, and the forward fallback instead of exact offsets. --- apps/server/src/peer/RemoteDelegation.test.ts | 67 +++++++++++++------ 1 file changed, 46 insertions(+), 21 deletions(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index e500d1cb9fdd9..5163ee67add1a 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -1666,7 +1666,7 @@ it.effect.each(["tail-error", "middle-error", "reply-fallback", "cancel-refill"] text: "discarded history ".repeat(120), type: "assistant_message", })); - yield* Ref.set(thread.timeline, [ + const timeline: Array = [ ...history(middle ? 150 : 300, boxRunId, "before"), { id: TurnItemId.make("selected"), runId: boxRunId, text: result, type: ownType }, ...(middle ? history(200, boxLaterRunId, "later") : []), @@ -1676,7 +1676,8 @@ it.effect.each(["tail-error", "middle-error", "reply-fallback", "cancel-refill"] text: "Wrong run", type: "error", }, - ]); + ]; + yield* Ref.set(thread.timeline, timeline); yield* Ref.set(thread.pageSize, 100); yield* Ref.set(thread.status, "failed"); if (scenario === "cancel-refill") yield* Ref.set(thread.holdRefill, true); @@ -1697,12 +1698,24 @@ it.effect.each(["tail-error", "middle-error", "reply-fallback", "cancel-refill"] result, }); } - const pages = reads.filter((read) => read.itemId === undefined); - expect(pages.map((read) => read.afterPosition)).toEqual([undefined, 99, 199, 299]); - expect(pages.map((read) => read.maxCharsPerItem)).toEqual([undefined, 1, 1, undefined]); + const pages = reads.filter((read) => read.itemId === undefined && read.limit === 100); + // A page moves on by however many items the response budget left it. + const remaining = (read: OrchestratorMcpThreadReadInput) => + timeline.length - 1 - (read.afterPosition ?? -1); + expect(pages[0]).toMatchObject({ + view: "activity", + limit: 100, + maxCharsPerItem: 20_000, + }); + expect(pages.slice(1).some((read) => remaining(read) > 100)).toBe(true); + for (const read of pages.slice(1)) + expect(read.maxCharsPerItem).toBe(remaining(read) > 100 ? 1 : 20_000); + const forwards = pages.slice(1).map((read) => read.afterPosition ?? -1); + expect(forwards).toEqual([...forwards].sort((left, right) => left - right)); + expect(new Set(forwards).size).toBe(forwards.length); const refills = reads.filter((read) => read.itemId !== undefined); expect(refills[0]).toMatchObject({ itemId: "selected", textOffset: middle ? 1 : 20_000 }); - if (scenario !== "cancel-refill") expect(refills).toHaveLength(middle ? 2 : 1); + if (scenario !== "cancel-refill") expect(refills).toHaveLength(middle ? 3 : 2); }), ).pipe( Effect.provide(NodeHttpServer.layerTest), @@ -1804,28 +1817,40 @@ it.effect.each([ : result, }); } - const pages = reads.filter((read) => read.itemId === undefined); - const positions = - scenario === "cancelled" || scenario === "interrupted" - ? [undefined, 99, 199, 299] - : scenario === "later-run" - ? [undefined, 299, 99, 199, 299] - : scenario === "shrinking" - ? [undefined, 299, 99] - : scenario === "growing" - ? [undefined, 299, 99, 199, 299, 399] - : [undefined, 299]; - expect(pages.map((read) => read.afterPosition)).toEqual(positions); + const pages = reads.filter((read) => read.itemId === undefined && read.limit === 100); + // The thread's final page starts at its canonical 100-item boundary. + const tailPosition = Math.floor((timeline.length - 1) / 100) * 100 - 1; + expect(pages.at(0)?.afterPosition).toBeUndefined(); expect( pages.every( (read) => - read.view === "messages" && read.limit === 100 && read.maxCharsPerItem === undefined, + read.view === "messages" && read.limit === 100 && read.maxCharsPerItem === 20_000, ), ).toBe(true); + if (scenario === "cancelled" || scenario === "interrupted") { + // A run that did not complete keeps the established scan, without the probe. + expect(pages.map((read) => read.afterPosition)).not.toContain(tailPosition); + const forwards = pages.slice(1).map((read) => read.afterPosition ?? -1); + expect(forwards.length).toBeGreaterThan(0); + expect(forwards).toEqual([...forwards].sort((left, right) => left - right)); + } else { + expect(pages.at(1)?.afterPosition).toBe(tailPosition); + const rest = pages.slice(2); + if (scenario === "later-run" || scenario === "shrinking" || scenario === "growing") { + // An inconclusive tail falls back to scanning from where the first page ended. + expect(rest.length).toBeGreaterThan(0); + const resumed = rest.map((read) => read.afterPosition ?? -1); + expect(resumed[0]).toBeLessThan(tailPosition); + expect(resumed).toEqual([...resumed].sort((left, right) => left - right)); + } else { + // A terminal page holding the run's last reply ends the read there. + expect(rest).toHaveLength(0); + } + } const refills = reads.filter((read) => read.itemId !== undefined); - expect(refills).toHaveLength(scenario === "long" ? 2 : 0); + expect(refills).toHaveLength(scenario === "long" ? 4 : 0); if (scenario === "long") - expect(refills.map((read) => read.textOffset)).toEqual([20_000, 70_000]); + expect(refills.map((read) => read.textOffset)).toEqual([20_000, 40_000, 60_000, 80_000]); }), ).pipe( Effect.provide(NodeHttpServer.layerTest),