From f6292407e3206429e23e978a0dba01cd8d364001 Mon Sep 17 00:00:00 2001 From: Yuruzuu Date: Mon, 2 Nov 2026 14:02:16 +0800 Subject: [PATCH] perf(client): streamed updates no longer rescan history for Stop state Co-Authored-By: Claude Opus 5.5 Made with GPT-6.1 Sol via Codex in T3 Code. --- .../src/state/threadDetail.test.ts | 178 ++++++++++++++++++ .../client-runtime/src/state/threadDetail.ts | 28 ++- 2 files changed, 201 insertions(+), 5 deletions(-) diff --git a/packages/client-runtime/src/state/threadDetail.test.ts b/packages/client-runtime/src/state/threadDetail.test.ts index cde705bdfb157..1d1dc7608000f 100644 --- a/packages/client-runtime/src/state/threadDetail.test.ts +++ b/packages/client-runtime/src/state/threadDetail.test.ts @@ -20,6 +20,7 @@ import { deriveWorkspaceMutationId, sameThreadStructure, } from "./threadDetail.ts"; +import { presentRunStop } from "./threadExecution.ts"; import { mergeOlderHistoryIntoProjection } from "./threadHistoryMerge.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE } from "./threadState.ts"; @@ -386,3 +387,180 @@ describe("deriveWorkspaceMutationId", () => { ); }); }); + +describe("runStopAtom", () => { + const ref = { environmentId: EnvironmentId.make("environment-1"), threadId: v2ThreadId }; + const running: OrchestrationV2ThreadProjection["runs"][number] = { + id: RunId.make("run-1"), + threadId: v2ThreadId, + ordinal: 1, + providerInstanceId: v2Projection.thread.providerInstanceId, + modelSelection: v2Projection.thread.modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message-user"), + rootNodeId: null, + activeAttemptId: null, + status: "running", + requestedAt: v2Now, + startedAt: v2Now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + const request: OrchestrationV2TurnItem = { + id: TurnItemId.make("item-request"), + threadId: v2ThreadId, + runId: assistant.runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "completed", + title: null, + startedAt: v2Now, + completedAt: null, + updatedAt: v2Now, + type: "run_interrupt_request", + message: "Stopped by user", + }; + const result: OrchestrationV2TurnItem = { + ...request, + id: TurnItemId.make("item-result"), + ordinal: 3, + status: "interrupted", + type: "run_interrupt_result", + }; + const unresponsive = { ...request, unresponsiveAt: later }; + const text = (value: string) => ({ ...assistant, text: value, updatedAt: later }); + + function setup(initial: OrchestrationV2ThreadProjection) { + const stateAtom = Atom.make( + AsyncResult.success({ ...EMPTY_ENVIRONMENT_THREAD_STATE, data: Option.some(initial) }), + ); + const stopAtom = createEnvironmentThreadDetailAtoms(() => stateAtom).runStopAtom(ref); + const registry = AtomRegistry.make(); + registry.mount(stopAtom); + return { + read: () => registry.get(stopAtom), + [Symbol.dispose]: () => registry.dispose(), + push: (projection: OrchestrationV2ThreadProjection | null) => + registry.set( + stateAtom, + AsyncResult.success({ + ...EMPTY_ENVIRONMENT_THREAD_STATE, + data: Option.fromNullishOr(projection), + }), + ), + }; + } + + it("does not rescan history for consecutive replacements unrelated to Stop", () => { + let runIdReads = 0; + const history = Array.from({ length: 99 }, (_, index) => + Object.defineProperty({ ...assistant, id: TurnItemId.make(`history-${index}`) }, "runId", { + enumerable: true, + get: () => { + runIdReads++; + return assistant.runId; + }, + }), + ); + const initial = { + ...withItem(assistant), + runs: [running], + turnItems: [...history, assistant], + }; + using thread = setup(initial); + thread.read(); + const first = replaceItem(initial, text("Hello")); + const second = replaceItem(first, text("Hello there")); + runIdReads = 0; + + thread.push(first); + expect(thread.read()).toBeNull(); + thread.push(second); + expect(thread.read()).toBeNull(); + expect(runIdReads).toBe(0); + }); + + it("matches a full derivation across every kind of change", () => { + type Projection = OrchestrationV2ThreadProjection; + const setRun = (projection: Projection, status: "running" | "completed") => ({ + ...projection, + runs: [{ ...running, status }], + }); + const steps: ReadonlyArray<{ + readonly name: string; + readonly apply: (projection: Projection) => Projection; + readonly state: "stopping" | "unresponsive" | null; + }> = [ + { name: "text", apply: (p) => replaceItem(p, text("Hello")), state: null }, + { name: "request inserted", apply: (p) => replaceItem(p, request), state: "stopping" }, + { name: "text while stopping", apply: (p) => replaceItem(p, text("A")), state: "stopping" }, + { + name: "request turned unresponsive", + apply: (p) => replaceItem(p, unresponsive), + state: "unresponsive", + }, + { + name: "text while unresponsive", + apply: (p) => replaceItem(p, text("B")), + state: "unresponsive", + }, + { + name: "request rewritten to text", + apply: (p) => replaceItem(p, { ...assistant, id: request.id, text: "C" }), + state: null, + }, + { + name: "text rewritten to request", + apply: (p) => replaceItem(p, request), + state: "stopping", + }, + { + name: "skipped request update with a later text update", + apply: (p) => replaceItem(replaceItem(p, unresponsive), text("D")), + state: "unresponsive", + }, + { + name: "skipped request reset with a later text update", + apply: (p) => replaceItem(replaceItem(p, request), text("E")), + state: "stopping", + }, + { name: "run settled", apply: (p) => setRun(p, "completed"), state: null }, + { name: "run resumed", apply: (p) => setRun(p, "running"), state: "stopping" }, + { name: "result inserted", apply: (p) => replaceItem(p, result), state: null }, + { name: "text after result", apply: (p) => replaceItem(p, text("F")), state: null }, + { + name: "result removed by a snapshot", + apply: (p) => ({ ...p, turnItems: p.turnItems.filter((item) => item.id !== result.id) }), + state: "stopping", + }, + { + name: "reconnect snapshot", + apply: (p) => ({ + ...p, + turnItems: p.turnItems.map((item) => (item.id === request.id ? unresponsive : item)), + }), + state: "unresponsive", + }, + ]; + + let projection: Projection = { ...withItem(assistant), runs: [running] }; + using thread = setup(projection); + thread.read(); + for (const { name, apply, state } of steps) { + projection = apply(projection); + thread.push(projection); + expect({ name, state: thread.read()?.state ?? null }).toEqual({ name, state }); + expect(thread.read()).toEqual(presentRunStop(projection)); + } + + thread.push(null); + expect(thread.read()).toBeNull(); + thread.push(projection); + expect(thread.read()?.state).toBe("unresponsive"); + }); +}); diff --git a/packages/client-runtime/src/state/threadDetail.ts b/packages/client-runtime/src/state/threadDetail.ts index 7a17e7326b802..b39baefd04ad8 100644 --- a/packages/client-runtime/src/state/threadDetail.ts +++ b/packages/client-runtime/src/state/threadDetail.ts @@ -243,6 +243,24 @@ export function sameThreadStructure( ); } +type RunStopInputs = Pick; + +function isRunInterruptItem(item: TurnItem): boolean { + return item.type === "run_interrupt_request" || item.type === "run_interrupt_result"; +} + +/** Whether presentRunStop would return the same value: it only reads runs and interrupt items. */ +function sameRunStopInputs(previous: RunStopInputs, next: RunStopInputs): boolean { + if (previous.runs !== next.runs) return false; + if (previous.turnItems === next.turnItems) return true; + const replacedIndex = getProjectionArrayReplacementIndex(previous.turnItems, next.turnItems); + return ( + replacedIndex !== undefined && + !isRunInterruptItem(previous.turnItems[replacedIndex]!) && + !isRunInterruptItem(next.turnItems[replacedIndex]!) + ); +} + export function createEnvironmentThreadDetailAtoms( threadStateAtom: ( environmentId: ScopedThreadRef["environmentId"], @@ -374,17 +392,14 @@ export function createEnvironmentThreadDetailAtoms( }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-queue:${key}`)); }); const runStopAtomFamily = Atom.family((key: string) => { - let previous: Pick | null = null; + let previous: RunStopInputs | null = null; let stop: RunStopPresentation | null = null; return Atom.make((get) => { const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data); if (projection === null) { previous = null; stop = null; - } else if ( - projection.runs !== previous?.runs || - projection.turnItems !== previous?.turnItems - ) { + } else if (previous === null || !sameRunStopInputs(previous, projection)) { const { runs, turnItems } = projection; previous = { runs, turnItems }; const next = presentRunStop(previous); @@ -395,6 +410,9 @@ export function createEnvironmentThreadDetailAtoms( next?.background !== stop?.background ) stop = next; + } else { + // Replacement hints only link consecutive arrays. + previous = { runs: projection.runs, turnItems: projection.turnItems }; } return stop; }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-run-stop:${key}`));