diff --git a/packages/client-runtime/src/state/threadRequests.test.ts b/packages/client-runtime/src/state/threadRequests.test.ts index d9aaf9fdca71a..c2e1024e232ae 100644 --- a/packages/client-runtime/src/state/threadRequests.test.ts +++ b/packages/client-runtime/src/state/threadRequests.test.ts @@ -2,6 +2,8 @@ import { presentThreadShell } from "./models.ts"; import { threadKey } from "./entities.ts"; import { EnvironmentId, + EventId, + MessageId, ThreadId, type ScopedThreadRef, NodeId, @@ -13,6 +15,7 @@ import { import { describe, expect, it, vi } from "vite-plus/test"; import { Atom, AtomRegistry, AsyncResult } from "effect/reactivity"; import * as Cause from "effect/Cause"; +import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts"; import { v2Now, v2Projection, v2ThreadShell } from "./orchestrationV2TestFixtures.ts"; import { createQuestionHistoryProjector, @@ -360,6 +363,210 @@ it("restores old text answers without mutating history or replacing unchanged ro expect(project({ ...answered, visibleTurnItems: [...rows] })[0]).toBe(result[0]); }); +type QuestionItem = Extract< + OrchestrationV2ThreadProjection["turnItems"][number], + { readonly type: "user_input_request" } +>; + +describe("question history through exact row replacements", () => { + const baseItem = { + threadId: v2Projection.thread.id, + runId: null, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + status: "running" as const, + title: null, + startedAt: v2Now, + completedAt: null, + updatedAt: v2Now, + }; + const assistantItem = (ordinal: number, text = "Hel") => ({ + ...baseItem, + id: TurnItemId.make(`assistant-${ordinal}`), + ordinal, + type: "assistant_message" as const, + messageId: MessageId.make(`message-${ordinal}`), + text, + streaming: true, + }); + const localRow = ( + item: OrchestrationV2ThreadProjection["turnItems"][number], + position: number, + ) => ({ + item, + position, + sourceItemId: item.id, + sourceThreadId: item.threadId, + visibility: "local" as const, + }); + const updated = ( + current: OrchestrationV2ThreadProjection, + item: OrchestrationV2ThreadProjection["turnItems"][number], + ) => + applyOrchestrationV2ProjectionEvent(current, { + id: EventId.make(`event-${item.id}`), + type: "turn-item.updated", + threadId: current.thread.id, + occurredAt: v2Now, + payload: item, + })!; + + it("reads only the replaced row and leaves the request index alone", () => { + const size = 1_000; + const counts = { items: 0, requestIds: 0 }; + const items = Array.from({ length: size }, (_, index) => assistantItem(index)); + const countedRows = items.map((item, index) => { + const row = localRow(item, index); + return Object.defineProperty(row, "item", { + enumerable: true, + get: () => { + counts.items += 1; + return item; + }, + }); + }); + const countedRequests = Array.from({ length: 400 }, (_, index) => { + const request = { + ...projection.runtimeRequests[0]!, + id: RuntimeRequestId.make(`request-${index}`), + }; + return Object.defineProperty(request, "id", { + enumerable: true, + get: () => { + counts.requestIds += 1; + return RuntimeRequestId.make(`request-${index}`); + }, + }); + }); + const initial: OrchestrationV2ThreadProjection = { + ...v2Projection, + turnItems: items, + visibleTurnItems: countedRows, + runtimeRequests: countedRequests, + }; + const project = createQuestionHistoryProjector(); + const scan = (current: OrchestrationV2ThreadProjection) => { + counts.items = 0; + counts.requestIds = 0; + const result = project(current); + return { result, ...counts }; + }; + expect(scan(initial)).toMatchObject({ items: size, requestIds: 400 }); + + const adjacent = updated(initial, assistantItem(size - 1, "Hello")); + const hinted = scan(adjacent); + expect(hinted.result).toBe(adjacent.visibleTurnItems); + expect(hinted).toMatchObject({ items: 0, requestIds: 0 }); + + // A skipped delta has no hint for this exact pair, so it scans again. + const skipped = updated( + updated(adjacent, assistantItem(size - 1, "Hello t")), + assistantItem(size - 1, "Hello there"), + ); + const fallback = scan(skipped); + expect(fallback.result).toBe(skipped.visibleTurnItems); + expect(fallback).toMatchObject({ items: size - 1, requestIds: 0 }); + }); + + it("matches a full projection through streamed rows, request changes and answers", () => { + const question = projection.turnItems[0] as QuestionItem; + const resolved = (answers: Record) => ({ + ...projection.runtimeRequests[0]!, + status: "resolved" as const, + answers, + }); + const explicitRequestId = RuntimeRequestId.make("explicit-question"); + const explicitAnswer = { + requestId: explicitRequestId, + answers: { next: "Explicit" }, + attachmentsByQuestionId: {}, + questionTextById: { next: "What should happen next?" }, + }; + const unanswered: QuestionItem = { + ...question, + id: TurnItemId.make("explicit-question-item"), + ordinal: 1, + requestId: explicitRequestId, + }; + const explicit: QuestionItem = { ...unanswered, questionAnswer: explicitAnswer }; + const explicitRequest = { + ...resolved({ next: "From request" }), + id: explicitRequestId, + }; + const stream = assistantItem(2); + const inherited = { + ...localRow({ ...question, id: TurnItemId.make("inherited-question-item") }, 3), + visibility: "inherited" as const, + sourceThreadId: ThreadId.make("parent-thread"), + }; + const turnItems = [question, explicit, stream]; + let current: OrchestrationV2ThreadProjection = { + ...v2Projection, + turnItems, + visibleTurnItems: [...turnItems.map(localRow), inherited], + runtimeRequests: [projection.runtimeRequests[0]!, explicitRequest], + }; + const project = createQuestionHistoryProjector(); + const step = (next: OrchestrationV2ThreadProjection) => { + current = next; + const result = project(next); + expect(result).toEqual(createQuestionHistoryProjector()(next)); + return result; + }; + const answerOf = (rows: ReadonlyArray<{ readonly item: unknown }>, index: number) => + (rows[index]!.item as { readonly questionAnswer?: { readonly answers: unknown } }) + .questionAnswer?.answers; + + expect(step(current)).toBe(current.visibleTurnItems); + expect(step(updated(current, assistantItem(2, "Hello")))).toBe(current.visibleTurnItems); + + // The last duplicate request wins, and every row asking it gets the answer. + const answered = step({ + ...current, + runtimeRequests: [ + projection.runtimeRequests[0]!, + resolved({ next: "Continue" }), + explicitRequest, + ], + }); + expect(answerOf(answered, 0)).toEqual({ next: "Continue" }); + expect(answerOf(answered, 1)).toEqual({ next: "Explicit" }); + expect(answerOf(answered, 3)).toEqual({ next: "Continue" }); + + const whileOverlaid = step(updated(current, assistantItem(2, "Hello there"))); + expect(whileOverlaid[0]).toBe(answered[0]); + expect(whileOverlaid[3]).toBe(answered[3]); + + const changedAnswer = step({ + ...current, + runtimeRequests: [resolved({ next: "Stop" }), explicitRequest], + }); + expect(answerOf(changedAnswer, 0)).toEqual({ next: "Stop" }); + + const reopened = step({ + ...current, + runtimeRequests: [projection.runtimeRequests[0]!, explicitRequest], + }); + expect(reopened).toBe(current.visibleTurnItems); + + // Dropping an explicit answer falls back to its resolved request. + const restored = step(updated(current, unanswered)); + expect(answerOf(restored, 1)).toEqual({ next: "From request" }); + + const retitled: QuestionItem = { + ...unanswered, + questions: [{ ...question.questions[0]!, question: "Now what?" }], + }; + const retitledRows = step(updated(current, retitled)); + expect(retitledRows[1]!.item).toMatchObject({ + questionAnswer: { questionTextById: { next: "Now what?" } }, + }); + }); +}); + it("loads only child approvals and retains their destination across environments", () => { const local = EnvironmentId.make("local"); const remote = EnvironmentId.make("remote"); diff --git a/packages/client-runtime/src/state/threadRequests.ts b/packages/client-runtime/src/state/threadRequests.ts index a010018c5051d..6cbad534cba1d 100644 --- a/packages/client-runtime/src/state/threadRequests.ts +++ b/packages/client-runtime/src/state/threadRequests.ts @@ -13,6 +13,7 @@ import type { } from "@t3tools/contracts"; import { Atom } from "effect/reactivity"; import { parseThreadKey, threadKey } from "./entities.ts"; +import { getProjectionArrayReplacementIndex } from "./orchestrationV2Projection.ts"; import type { EnvironmentThreadShell } from "./models.ts"; import * as DateTime from "effect/DateTime"; import type { AtomCommandResult } from "./runtime.ts"; @@ -292,54 +293,61 @@ export function createQuestionHistoryProjector< let previousRows: Rows | undefined; let previousRequests: Requests | undefined; let previousResult: Rows | undefined; + let byId = new Map(); const cache = new WeakMap(); - return (projection: { - readonly visibleTurnItems: Rows; - readonly runtimeRequests: Requests; - }): Rows => { - const { visibleTurnItems: rows, runtimeRequests: requests } = projection; - if (rows === previousRows && requests === previousRequests && previousResult) - return previousResult; - const byId = new Map(requests.map((request) => [request.id, request])); + const projectRow = (row: Row): Row => { + const item = row.item; + if (item.type !== "user_input_request" || item.questionAnswer) return row; + const request = byId.get(item.requestId); + if (request?.status !== "resolved" || request.answers === undefined) return row; + const cached = cache.get(row); + const next: Row = + cached?.request === request + ? cached.row + : { + ...row, + item: { + ...item, + questionAnswer: { + requestId: request.id, + answers: request.answers, + attachmentsByQuestionId: {}, + questionTextById: Object.fromEntries( + item.questions.map((question) => [question.id, question.question]), + ), + }, + }, + }; + cache.set(row, { request, row: next }); + return next; + }; + const projectRows = (rows: Rows): Rows => { let result: Row[] | undefined; for (const [index, row] of rows.entries()) { - const item = row.item; - const request = - item.type === "user_input_request" && !item.questionAnswer - ? byId.get(item.requestId) - : undefined; - let next = row; - if ( - item.type === "user_input_request" && - request?.status === "resolved" && - request.answers !== undefined - ) { - const cached = cache.get(row); - next = - cached?.request === request - ? cached.row - : { - ...row, - item: { - ...item, - questionAnswer: { - requestId: request.id, - answers: request.answers, - attachmentsByQuestionId: {}, - questionTextById: Object.fromEntries( - item.questions.map((question) => [question.id, question.question]), - ), - }, - }, - }; - cache.set(row, { request, row: next }); - } + const next = projectRow(row); if (next !== row) result ??= rows.slice(0, index); result?.push(next); } + return result ?? rows; + }; + return (projection: { + readonly visibleTurnItems: Rows; + readonly runtimeRequests: Requests; + }): Rows => { + const { visibleTurnItems: rows, runtimeRequests: requests } = projection; + const sameRequests = requests === previousRequests; + if (rows === previousRows && sameRequests && previousResult) return previousResult; + if (!sameRequests) byId = new Map(requests.map((request) => [request.id, request])); + // Nothing was overlaid and requests are unchanged, so only an exactly replaced row can differ. + const replacedIndex = + sameRequests && previousRows !== undefined && previousResult === previousRows + ? getProjectionArrayReplacementIndex(previousRows, rows) + : undefined; + const unchanged = + replacedIndex !== undefined && projectRow(rows[replacedIndex]!) === rows[replacedIndex]; previousRows = rows; previousRequests = requests; - previousResult = result ?? rows; + previousResult = unchanged ? rows : projectRows(rows); return previousResult; }; }