diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c7cd7a9ff288e..5e5d0c16365f7 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1973,13 +1973,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const projection = yield* projectionStore.getThreadRecords( threadId, ["runs", "nodes", "attempts", "providerThreads", "turnItems", "messages"], - { turnItemTypes: [], messageRoles: ["user"] }, + { turnItemTypes: ["user_message"], messageRoles: ["user"] }, ); const queuedRun = queuedRunId === undefined ? nextQueuedRun(projection) : projection.runs.find((run) => run.id === queuedRunId && run.status === "queued"); if (queuedRun === undefined) return; + // A steer that fell back to the queue keeps its user item, hidden while the + // run is queued. Republish it before the failure so clients whose bounded + // snapshot omitted it can show the prompt beside the error. + const retainedUserItems = projection.turnItems.filter( + (item) => item.type === "user_message" && item.runId === queuedRun.id, + ); const now = yield* DateTime.now; const rootNode = projection.nodes.find((node) => node.id === queuedRun.rootNodeId); const attempt = projection.attempts.find((entry) => entry.id === queuedRun.activeAttemptId); @@ -2005,6 +2011,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio class: handoffUnsupported ? "validation_error" : "unknown", }); yield* writeSystemEvents([ + ...retainedUserItems.map((item) => ({ + type: "turn-item.updated" as const, + threadId, + runId: queuedRun.id, + ...(rootNode === undefined ? {} : { nodeId: rootNode.id }), + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...item, status: "failed" as const, completedAt: now, updatedAt: now }, + })), ...(attempt !== undefined && rootNode !== undefined ? [ { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 511c252689cc8..68862a6db3916 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -5018,26 +5018,36 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { })} ) `; - const cancelledSuffixSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { - rowLimit: sqlPageLimit, - }); - const cancelledSuffixPage = buildBoundedThreadProjection({ - policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined }, - projection: cancelledSuffixSnapshot.projection, - snapshotSequence: cancelledSuffixSnapshot.snapshotSequence, - }); - assert.strictEqual(cancelledSuffixPage.latestLocalTurnOrdinal, 1_202); - assert.lengthOf(cancelledSuffixPage.projection.visibleTurnItems, 75); - assert.isFalse( - cancelledSuffixPage.projection.visibleTurnItems.some((row) => - String(row.sourceItemId).includes(":cancelled:"), - ), - ); - assert.isTrue( - cancelledSuffixPage.projection.visibleTurnItems.some( - (row) => row.sourceItemId === runlessQueuedId, - ), - ); + // A steer that fell back to the queue keeps its queued-turn item while its + // run waits. End on cancelled for the checks below. + for (const status of ["queued", "cancelled"] as const) { + yield* sql` + UPDATE orchestration_v2_projection_runs + SET status = ${status}, payload_json = json_set(payload_json, '$.status', ${status}) + WHERE run_id = ${cancelledRunId} + `; + const cancelledSuffixSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: sqlPageLimit, + }); + const cancelledSuffixPage = buildBoundedThreadProjection({ + policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined }, + projection: cancelledSuffixSnapshot.projection, + snapshotSequence: cancelledSuffixSnapshot.snapshotSequence, + }); + assert.strictEqual(cancelledSuffixPage.latestLocalTurnOrdinal, 1_202); + assert.lengthOf(cancelledSuffixPage.projection.visibleTurnItems, 75); + assert.isFalse( + cancelledSuffixPage.projection.visibleTurnItems.some((row) => + String(row.sourceItemId).includes(":cancelled:"), + ), + status, + ); + assert.isTrue( + cancelledSuffixPage.projection.visibleTurnItems.some( + (row) => row.sourceItemId === runlessQueuedId, + ), + ); + } // A steer cancels the queued run before it starts while the steered run keeps writing, so // the queued message leaves the window but its run is still the shell's latest run. @@ -5830,6 +5840,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { : [ "rolled back", "cancelled queued turn", + "held queued turn", "replaced attempt", "stopped", "stopped", @@ -5860,9 +5871,11 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ? "rolled_back" : kind === "cancelled queued turn" ? "cancelled" - : kind === "short" && chance(0.5) - ? "interrupted" - : "completed"; + : kind === "held queued turn" + ? "queued" + : kind === "short" && chance(0.5) + ? "interrupted" + : "completed"; runs.push({ id: runId, threadId, @@ -5941,7 +5954,12 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ? run % 10 !== 0 : kind !== "late wake" && (agentDriven ? run > 1 : chance(0.2)); if (wakes) notification(); - else userMessage(kind === "cancelled queued turn" ? "queued_turn" : "turn_start"); + else + userMessage( + kind === "cancelled queued turn" || kind === "held queued turn" + ? "queued_turn" + : "turn_start", + ); // A stopped run never folds, so it is long enough for page ends to fall inside it. const work = kind === "late wake" @@ -6573,7 +6591,13 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { assert.isNotEmpty(countQueries); for (const countQuery of countQueries) { for (const detail of yield* planOf(countQuery)) { - assert.notInclude(detail, "orchestration_v2_projection_turn_items"); + if (detail.includes("orchestration_v2_projection_turn_items")) { + // Only held input is read from the sparse index; transcript history stays unread. + assert.include( + detail, + "SEARCH held USING INDEX orchestration_v2_projection_turn_items_queued_input_idx (run_id=?)", + ); + } } } }), @@ -10416,6 +10440,236 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("keeps undelivered queued input out of forks taken after a later run", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const projectId = ProjectId.make("project:projection-fork-held-queue"); + + for (const heldStatus of ["queued", "cancelled"] as const) { + const prefix = `projection-fork-held-queue:${heldStatus}`; + const sourceThreadId = ThreadId.make(`thread:${prefix}:source`); + const targetThreadId = ThreadId.make(`thread:${prefix}:target`); + const heldRunId = RunId.make(`run:${prefix}:held`); + const heldAttemptId = RunAttemptId.make(`attempt:${prefix}:held`); + const answeredRunId = RunId.make(`run:${prefix}:answered`); + + for (const [threadId, forkedFrom] of [ + [sourceThreadId, null], + [targetThreadId, { type: "run", threadId: sourceThreadId, runId: answeredRunId }], + ] as const) { + yield* projectionStore.apply({ + id: EventId.make(`event:${threadId}:created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId, + title: threadId, + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: forkedFrom === null ? null : sourceThreadId, + relationshipToParent: forkedFrom === null ? null : "fork", + rootThreadId: sourceThreadId, + }, + forkedFrom, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + } + + const applyRun = ( + runId: RunId, + ordinal: number, + status: typeof heldStatus | "completed" | "rolled_back", + ) => + projectionStore.apply({ + id: EventId.make(`event:${runId}:${status}`), + type: "run.updated", + threadId: sourceThreadId, + runId, + nodeId: NodeId.make(`node:${runId}`), + driver, + occurredAt: now, + payload: { + id: runId, + threadId: sourceThreadId, + ordinal, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:${runId}`), + rootNodeId: NodeId.make(`node:${runId}`), + activeAttemptId: runId === heldRunId ? heldAttemptId : null, + status, + requestedAt: now, + startedAt: runId === heldRunId ? null : now, + completedAt: status === "queued" ? null : now, + checkpointId: null, + contextHandoffId: null, + }, + }); + const applyUserItem = (input: { + readonly threadId: ThreadId; + readonly runId: RunId | null; + readonly ordinal: number; + readonly text: string; + readonly inputIntent: "turn_start" | "queued_turn"; + }) => + projectionStore.apply({ + id: EventId.make(`event:${input.threadId}:item:${input.ordinal}`), + type: "turn-item.updated", + threadId: input.threadId, + ...(input.runId === null ? {} : { runId: input.runId }), + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: TurnItemId.make(`turn-item:${input.threadId}:${input.ordinal}`), + threadId: input.threadId, + runId: input.runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: input.ordinal, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "user_message", + messageId: MessageId.make(`message:${input.threadId}:${input.ordinal}`), + inputIntent: input.inputIntent, + text: input.text, + attachments: [], + }, + }); + + yield* applyRun(heldRunId, 1, heldStatus); + yield* projectionStore.apply({ + id: EventId.make(`event:${heldAttemptId}:created`), + type: "run-attempt.created", + threadId: sourceThreadId, + runId: heldRunId, + nodeId: NodeId.make(`node:${heldRunId}`), + driver, + occurredAt: now, + payload: { + id: heldAttemptId, + runId: heldRunId, + attemptOrdinal: 1, + rootNodeId: NodeId.make(`node:${heldRunId}`), + providerInstanceId, + providerThreadId: ProviderThreadId.make(`provider-thread:${prefix}`), + providerTurnId: null, + reason: "initial", + status: heldStatus === "queued" ? "pending" : "cancelled", + startedAt: null, + completedAt: heldStatus === "queued" ? null : now, + }, + }); + yield* applyRun(answeredRunId, 2, "completed"); + yield* applyUserItem({ + threadId: sourceThreadId, + runId: heldRunId, + ordinal: 100, + text: "held", + inputIntent: "queued_turn", + }); + yield* applyUserItem({ + threadId: sourceThreadId, + runId: answeredRunId, + ordinal: 200, + text: "answered", + inputIntent: "turn_start", + }); + yield* applyUserItem({ + threadId: sourceThreadId, + runId: answeredRunId, + ordinal: 201, + text: "delivered steer", + inputIntent: "queued_turn", + }); + // Enough local turns for a turn-limited window to count inherited rows in SQL. + for (let index = 0; index < THREAD_HISTORY_PAGE_POLICY.maxUserTurns; index += 1) { + yield* applyUserItem({ + threadId: targetThreadId, + runId: null, + ordinal: index, + text: `local ${index}`, + inputIntent: "turn_start", + }); + } + + const assertForkHidesHeldInput = Effect.fn("assertForkHidesHeldInput")(function* () { + const target = yield* projectionStore.getThreadProjection(targetThreadId); + assert.deepEqual( + target.visibleTurnItems + .slice(0, 3) + .map((row) => (row.item.type === "user_message" ? row.item.text : row.item.type)), + ["answered", "delivered steer", "fork"], + ); + const page = yield* projectionStore.getTimelinePage(targetThreadId, { + view: "activity", + limit: target.visibleTurnItems.length + 1, + }); + assert.deepEqual(page.items, target.visibleTurnItems); + const bounded = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, { + rowLimit: target.visibleTurnItems.length + 1, + }); + assert.deepEqual(bounded.projection.visibleTurnItems, target.visibleTurnItems); + const shellSnapshot = yield* projectionStore.getShellSnapshot(); + for (const projection of [ + yield* projectionStore.getThreadProjection(sourceThreadId), + target, + ]) { + for (const shell of [ + yield* projectionStore.getThreadShell(projection.thread.id), + shellSnapshot.threads.find((thread) => thread.id === projection.thread.id), + ]) { + assert.equal(shell?.visibleItemCount, projection.visibleTurnItems.length); + } + } + const filledLocalPage = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, { + rowLimit: target.visibleTurnItems.length, + userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + }); + assert.deepEqual( + filledLocalPage.projection.visibleTurnItems.filter((row) => row.visibility === "local"), + target.visibleTurnItems.filter((row) => row.visibility === "local"), + ); + }); + + yield* assertForkHidesHeldInput(); + // Forks keep rolled-back source history, but still not undelivered input. + yield* applyRun(answeredRunId, 2, "rolled_back"); + yield* assertForkHidesHeldInput(); + // Rollback also takes cancelled runs, and overwrites the status that hid their input. + if (heldStatus === "cancelled") { + yield* applyRun(heldRunId, 1, "rolled_back"); + yield* assertForkHidesHeldInput(); + } + } + }), + ); + it.effect("hides source turns rewound before the fork from a fork's visible items", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index d3c55b6d85715..3aeac72d533a1 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -78,10 +78,7 @@ import { ProjectId, ProviderInstanceId, } from "@t3tools/contracts"; -import { - createOrchestrationV2TurnItemVisibility, - isOrchestrationV2SupersededInterrupt, -} from "@t3tools/shared/orchestrationV2Timeline"; +import { createOrchestrationV2TurnItemVisibility } from "@t3tools/shared/orchestrationV2Timeline"; import { deriveBlockedOnThreadWait, derivePendingBackgroundWork, @@ -1461,6 +1458,28 @@ function isTurnItemAtOrBeforeRun(input: { return ordinal !== undefined && ordinal <= input.sourceRunOrdinal; } +// Run ordinals enforce the fork cutoff; this keeps later rollbacks from hiding +// inherited history while retaining cancellation evidence for undelivered input. +function createInheritedTurnItemVisibility( + source: Parameters[0], +) { + const cancelledRunIds = new Set( + source.attempts + .filter((attempt) => attempt.status === "cancelled") + .map((attempt) => attempt.runId), + ); + return createOrchestrationV2TurnItemVisibility({ + ...source, + runs: source.runs.flatMap((run) => + run.status !== "rolled_back" + ? [run] + : cancelledRunIds.has(run.id) + ? [{ ...run, status: "cancelled" as const }] + : [], + ), + }); +} + // Rollback stamps completedAt; a fork made after the rewind never received that run. function runRolledBackAt(run: Pick) { return run.status === "rolled_back" ? run.completedAt : null; @@ -1495,6 +1514,11 @@ function visibleTurnItemsThroughRun(input: { } const runOrdinalById = forkInheritedRunOrdinals(input.sourceProjection.runs, input.forkCreatedAt); + const isVisible = createInheritedTurnItemVisibility({ + runs: input.sourceProjection.runs, + attempts: input.sourceProjection.attempts, + items: input.sourceProjection.turnItems, + }); const inheritedPrefix = input.sourceProjection.visibleTurnItems .filter( (row) => row.item.threadId !== input.sourceProjection.thread.id || row.item.type === "fork", @@ -1506,23 +1530,16 @@ function visibleTurnItemsThroughRun(input: { item: row.item, })); const localPrefix = inheritedVisibleTurnItemsFromLocalItems( - input.sourceProjection.turnItems.filter((item) => { - if ( - isOrchestrationV2SupersededInterrupt({ - item, - attempts: input.sourceProjection.attempts, - items: input.sourceProjection.turnItems, - }) - ) { - return false; - } - return isTurnItemAtOrBeforeRun({ - historyOrigin: input.sourceProjection.thread.historyOrigin, - itemRunId: item.runId, - runOrdinalById, - sourceRunOrdinal: sourceRun.ordinal, - }); - }), + input.sourceProjection.turnItems.filter( + (item) => + isVisible(item) && + isTurnItemAtOrBeforeRun({ + historyOrigin: input.sourceProjection.thread.historyOrigin, + itemRunId: item.runId, + runOrdinalById, + sourceRunOrdinal: sourceRun.ordinal, + }), + ), ); return [...inheritedPrefix, ...localPrefix]; @@ -3121,16 +3138,27 @@ export const layer: Layer.Layer = ), 9223372036854775807 ) + -- The fork cutoff below filters rewound runs; undelivered queued input stays hidden. AND ( ${window.requiredRunId ?? null} IS NOT NULL - OR ( - (run.status IS NULL OR run.status <> 'rolled_back') - AND NOT ( - item.type = 'user_message' - AND json_extract(item.payload_json, '$.inputIntent') = 'queued_turn' - AND run.status IS 'cancelled' + OR run.status IS NULL + OR run.status <> 'rolled_back' + ) + AND NOT ( + item.type = 'user_message' + AND ( + COALESCE(run.status IN ('queued', 'cancelled'), 0) + OR ( + run.status = 'rolled_back' + AND EXISTS ( + SELECT 1 + FROM orchestration_v2_projection_run_attempts AS attempt + WHERE attempt.run_id = run.run_id + AND attempt.status = 'cancelled' + ) ) ) + AND json_extract(item.payload_json, '$.inputIntent') = 'queued_turn' ) AND ( ${window.requiredRunId ?? null} IS NULL @@ -6367,18 +6395,18 @@ export const layer: Layer.Layer = let inherited: Array = []; if (sourceRun !== undefined) { const ordinals = forkInheritedRunOrdinals(source.records.runs, records.thread.createdAt); - const sourceItems = source.local.map((row) => row.item); + const isInheritedVisible = createInheritedTurnItemVisibility({ + runs: source.records.runs, + attempts: source.records.attempts, + items: source.local.map((row) => row.item), + }); inherited = [ ...source.visible.filter( (row) => row.item.threadId !== fork.threadId || row.item.type === "fork", ), ...source.local.filter( (row) => - !isOrchestrationV2SupersededInterrupt({ - item: row.item, - attempts: source.records.attempts, - items: sourceItems, - }) && + isInheritedVisible(row.item) && isTurnItemAtOrBeforeRun({ historyOrigin: source.records.thread.historyOrigin, itemRunId: row.item.runId, @@ -6575,13 +6603,13 @@ export const layer: Layer.Layer = })), }); // Only these runs can hide rows: every row of a rolled-back run, the queued turn of a - // cancelled run, and the stop result of a replaced attempt. So the hidden rows are found - // without reading the thread. + // queued or cancelled run, and the stop result of a replaced attempt. So the hidden rows + // are found without reading the thread. const stoppedRuns = new Set(stopRequests.map((row) => row.run_id)); - const runsWithStatus = (status: "rolled_back" | "cancelled") => - records.runs.flatMap((run) => (run.status === status ? [String(run.id)] : [])); + const runsWithStatus = (...statuses: ReadonlyArray) => + records.runs.flatMap((run) => (statuses.includes(run.status) ? [String(run.id)] : [])); const rolledBack = runsWithStatus("rolled_back"); - const cancelled = runsWithStatus("cancelled"); + const heldInput = runsWithStatus("queued", "cancelled"); const steered = [...new Set(superseded.map((attempt) => String(attempt.runId)))].filter( (runId) => !stoppedRuns.has(runId) && !rolledBack.includes(runId), ); @@ -6606,8 +6634,8 @@ export const layer: Layer.Layer = INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx ON item.thread_id = ${threadId} AND item.run_id = wanted.value`, // The conversation index holds the type and run of these rows, so only the matches are - // fetched. Through the run index, every row of each cancelled run would be. - cancelled.length + steered.length === 0 + // fetched. Through the run index, every row of each such run would be. + heldInput.length + steered.length === 0 ? Effect.succeed([]) : sql` SELECT ${candidateColumns} @@ -6616,7 +6644,7 @@ export const layer: Layer.Layer = WHERE item.thread_id = ${threadId} AND item.type NOT IN ${CONVERSATION_INDEX_EXCLUDED_TYPES} AND ((item.type = 'user_message' - AND item.run_id IN (SELECT value FROM json_each(${encodeIdList(cancelled)}))) + AND item.run_id IN (SELECT value FROM json_each(${encodeIdList(heldInput)}))) OR (item.type = 'run_interrupt_result' AND item.run_id IN (SELECT value FROM json_each(${encodeIdList(steered)}))))`, ]); @@ -7705,18 +7733,58 @@ export const layer: Layer.Layer = WHERE thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) `; + // Forks do not inherit undelivered queued input, including from cancelled runs + // rolled back later. Checking the run first limits sparse-index seeks to those few runs. const selectShellRunItemCounts = (threadIds: ReadonlyArray) => sql` - SELECT thread_id, run_key AS run_id, item_count - FROM orchestration_v2_projection_turn_item_counts - WHERE run_key <> '' AND thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) + SELECT per_run.thread_id, per_run.run_key AS run_id, + per_run.item_count - CASE + WHEN EXISTS ( + SELECT 1 + FROM orchestration_v2_projection_runs AS run + WHERE run.run_id = per_run.run_key + AND ( + run.status IN ('queued', 'cancelled') + OR ( + run.status = 'rolled_back' + AND EXISTS ( + SELECT 1 + FROM orchestration_v2_projection_run_attempts AS attempt + WHERE attempt.run_id = run.run_id + AND attempt.status = 'cancelled' + ) + ) + ) + ) THEN ( + SELECT COUNT(*) + FROM orchestration_v2_projection_turn_items AS held + INDEXED BY orchestration_v2_projection_turn_items_queued_input_idx + WHERE held.run_id = per_run.run_key + AND held.type = 'user_message' + AND json_extract(held.payload_json, '$.inputIntent') = 'queued_turn' + ) + ELSE 0 + END AS item_count + FROM orchestration_v2_projection_turn_item_counts AS per_run + WHERE per_run.run_key <> '' AND per_run.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) `; const selectShellItemCounts = (threadIds: ReadonlyArray) => sql` SELECT counts.thread_id, SUM(CASE WHEN counts.run_key = '' OR r.status <> 'rolled_back' - THEN counts.item_count ELSE 0 END) AS item_count, + THEN counts.item_count - CASE + WHEN r.status IN ('queued', 'cancelled') THEN ( + SELECT COUNT(*) + FROM orchestration_v2_projection_turn_items held + INDEXED BY orchestration_v2_projection_turn_items_queued_input_idx + WHERE held.run_id = counts.run_key + AND held.type = 'user_message' + AND json_extract(held.payload_json, '$.inputIntent') = 'queued_turn' + ) + ELSE 0 + END + ELSE 0 END) AS item_count, SUM(CASE WHEN counts.run_key = '' THEN counts.item_count ELSE 0 END) AS runless_item_count FROM orchestration_v2_projection_turn_item_counts counts diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index ad0c179d783fe..f4f2e09d24c28 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -379,6 +379,13 @@ it.effect.each( } if (delivery !== null && !question) { const requeued = yield* orchestrator.getThreadProjection(threadId); + assert.isTrue(requeued.runs.some((run) => run.status === "queued")); + assert.isFalse( + requeued.visibleTurnItems.some( + ({ item }) => item.type === "user_message" && item.messageId === messageId, + ), + "a steer that fell back to the queue must not appear delivered", + ); const item = requeued.turnItems.find( (item) => item.type === "user_message" && item.messageId === messageId, ); @@ -462,6 +469,14 @@ it.effect.each( ).length, 1, ); + if (!mailbox && !question) { + assert.equal( + final.visibleTurnItems.filter( + ({ item }) => item.type === "user_message" && item.messageId === messageId, + ).length, + 1, + ); + } yield* worker.drain(); assert.equal(started.length, 2); }).pipe( diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 4ca516284a3b9..c56a321aaee13 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -3771,6 +3771,11 @@ describe("orchestration v2 provider switching", () => { ); assert.equal(queuedItem?.runId, queued.runs[1]?.id); assert.equal(queuedItem?.providerThreadId, queued.runs[1]?.providerThreadId); + assert.equal(queuedItem?.status, "pending"); + assert.isNull(queuedItem?.startedAt); + assert.isNull(queuedItem?.completedAt); + assert.isFalse(queued.visibleTurnItems.some((row) => row.item.id === queuedItem?.id)); + const queuedSequence = yield* eventSink.latestSequence({ threadId: rejectedThreadId }); yield* eventSink.write({ events: [ { @@ -3795,6 +3800,40 @@ describe("orchestration v2 provider switching", () => { Stream.runHead, ); yield* worker.drain(); + // Clients whose bounded snapshot omitted the hidden queued item must + // receive it while the run is still queued, before it fails. + const throughSequence = yield* eventSink.latestSequence({ threadId: rejectedThreadId }); + const failureEvents = yield* eventSink + .stream({ threadId: rejectedThreadId, afterSequence: queuedSequence }) + .pipe( + Stream.takeUntil((stored) => stored.sequence === throughSequence), + Stream.map(({ event }) => event), + Stream.filter((event) => event.runId === queued.runs[1]!.id), + Stream.runCollect, + ); + const republished = failureEvents.findIndex( + (event) => + event.type === "turn-item.updated" && + event.payload.type === "user_message" && + event.payload.messageId === rejectedMessageId, + ); + assert.isAtLeast(republished, 0); + const restored = failureEvents[republished]; + assert.equal(restored?.type, "turn-item.updated"); + if (restored?.type === "turn-item.updated") { + assert.deepEqual(restored.payload, { + ...queuedItem, + status: "failed", + completedAt: restored.occurredAt, + updatedAt: restored.occurredAt, + }); + } + assert.isBelow( + republished, + failureEvents.findIndex( + (event) => event.type === "run.updated" && event.payload.status === "failed", + ), + ); return yield* orchestrator.getThreadProjection(rejectedThreadId); }).pipe( Effect.provide( @@ -3838,6 +3877,11 @@ describe("orchestration v2 provider switching", () => { failureItem?.type === "error" ? failureItem.failure.code : null, "context_handoff_unsupported", ); + assert.isTrue( + projection.visibleTurnItems.some( + ({ item }) => item.type === "user_message" && item.messageId === rejectedMessageId, + ), + ); assert.deepEqual( (yield* Ref.get(capturedTurns)).map((turn) => turn.driver), [CODEX_DRIVER, CODEX_DRIVER], diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 8b33ba60f8e9a..9b6e2b3e0fc76 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -88,6 +88,7 @@ import Migration0082 from "./Migrations/082_TurnItemConversationIndex.ts"; import Migration0083 from "./Migrations/083_CommandReadIndexes.ts"; import Migration0084 from "./Migrations/084_LastExecutedRunIndex.ts"; import Migration0085 from "./Migrations/085_TurnItemCounts.ts"; +import Migration0086 from "./Migrations/086_QueuedTurnInputIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -176,6 +177,7 @@ const migrationEntries = [ [83, "CommandReadIndexes", Migration0083], [84, "LastExecutedRunIndex", Migration0084], [85, "TurnItemCounts", Migration0085], + [86, "QueuedTurnInputIndex", Migration0086], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/086_QueuedTurnInputIndex.ts b/apps/server/src/persistence/Migrations/086_QueuedTurnInputIndex.ts new file mode 100644 index 0000000000000..55480532949ce --- /dev/null +++ b/apps/server/src/persistence/Migrations/086_QueuedTurnInputIndex.ts @@ -0,0 +1,13 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Shell counts subtract held input without reading a run's tool history or item payloads. + yield* sql` + CREATE INDEX orchestration_v2_projection_turn_items_queued_input_idx + ON orchestration_v2_projection_turn_items(run_id) + WHERE type = 'user_message' AND json_extract(payload_json, '$.inputIntent') = 'queued_turn' + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 3c2cb075cfab4..98a79e7a7370d 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -65,6 +65,7 @@ describe("V2 preview upgrade", () => { [83, "CommandReadIndexes"], [84, "LastExecutedRunIndex"], [85, "TurnItemCounts"], + [86, "QueuedTurnInputIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -173,6 +174,7 @@ describe("V2 preview upgrade", () => { [83, "CommandReadIndexes"], [84, "LastExecutedRunIndex"], [85, "TurnItemCounts"], + [86, "QueuedTurnInputIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts index f6ba3e9f0c1a7..9284b6f9b82d0 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts @@ -483,6 +483,105 @@ describe("applyOrchestrationV2ProjectionEvent", () => { }); }); + it("hides a steer that fell back to the queue until its run starts", () => { + const queuedRunId = RunId.make("run-requeued-steer"); + const queuedRun = { ...run, id: queuedRunId, ordinal: 2, status: "queued" } as const; + const reply = commandItem("item-reply", "reply", 102); + const steer: OrchestrationV2TurnItem = { + id: TurnItemId.make("item-steer"), + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 101, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "user_message", + messageId: MessageId.make("message-steer"), + inputIntent: "steer", + text: "steer", + attachments: [], + createdBy: "user", + creationSource: "web", + }; + const event = (type: "run.updated" | "turn-item.updated", payload: unknown) => + ({ + id: `event-${type}`, + type, + threadId, + occurredAt: now, + payload, + }) as OrchestrationV2DomainEvent; + const apply = ( + projection: OrchestrationV2ThreadProjection | null, + events: ReadonlyArray, + options?: Parameters[2], + ) => + events.reduce( + (next, item) => applyOrchestrationV2ProjectionEvent(next, item, options), + projection, + ); + const visibleIds = (projection: OrchestrationV2ThreadProjection | null) => + projection?.visibleTurnItems.map((row) => row.item.id); + + const accepted = apply(emptyProjection, [ + event("run.updated", { ...run, status: "running" }), + event("turn-item.updated", steer), + event("turn-item.updated", reply), + ]); + expect(visibleIds(accepted)).toEqual([steer.id, reply.id]); + + const queuedSteer = event("turn-item.updated", { + ...steer, + runId: queuedRunId, + inputIntent: "queued_turn", + status: "pending", + startedAt: null, + completedAt: null, + }); + const requeued = apply(accepted, [event("run.updated", queuedRun), queuedSteer]); + expect(visibleIds(requeued)).toEqual([reply.id]); + + // Servers publish the start's user item and run update in either order. + const starting = event("run.updated", { ...queuedRun, status: "starting" }); + const dispatchedSteer = event("turn-item.updated", { + ...steer, + runId: queuedRunId, + inputIntent: "queued_turn", + }); + expect(visibleIds(apply(requeued, [dispatchedSteer, starting]))).toEqual([steer.id, reply.id]); + expect(visibleIds(apply(requeued, [starting, dispatchedSteer]))).toEqual([steer.id, reply.id]); + const cancelled = event("run.updated", { ...queuedRun, status: "cancelled" }); + expect(visibleIds(apply(requeued, [cancelled]))).toEqual([reply.id]); + + // A bounded snapshot taken while queued omits the hidden item, and the + // newer reply puts it below the watermark. + const bounded = { + ...requeued!, + turnItems: requeued!.turnItems.filter((item) => item.id !== steer.id), + }; + const partial = { partialTimeline: true, latestLocalTurnOrdinal: reply.ordinal }; + const failed = event("run.updated", { ...queuedRun, status: "failed" }); + const failedSteer = event("turn-item.updated", { + ...steer, + runId: queuedRunId, + inputIntent: "queued_turn", + status: "failed", + startedAt: null, + }); + const retained = apply(bounded, [failedSteer], partial); + expect(visibleIds(retained)).toEqual([reply.id]); + expect(visibleIds(apply(retained, [failed], partial))).toEqual([steer.id, reply.id]); + // Leaving the queue never invents an item the client did not receive. + expect(visibleIds(apply(bounded, [failed], partial))).toEqual([reply.id]); + }); + it("removes only hidden local items while preserving inherited rows", () => { const inherited = commandItem("item-inherited"); const local = commandItem("item-local"); diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index d9368ef4058a4..1bc8397fc2bd7 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -256,6 +256,29 @@ function upsertVisibleTurnItem( return updated; } +// A user item retained for a queued run stays hidden until the run leaves the +// queue, and its update can arrive before or after that run update. Items in +// `turnItems` were loaded or streamed, so they skip the partial-window check; +// items never received stay absent. +function restoreQueuedUserItems( + projection: OrchestrationV2ThreadProjection, + runId: OrchestrationV2Run["id"], +): OrchestrationV2ThreadProjection["visibleTurnItems"] { + let visibleTurnItems = projection.visibleTurnItems; + for (const item of projection.turnItems) { + if (item.runId !== runId || item.type !== "user_message" || item.inputIntent !== "queued_turn") + continue; + if (!shouldShowLocalTurnItem(projection, item)) continue; + visibleTurnItems = upsertVisibleTurnItem( + { ...projection, visibleTurnItems }, + item, + false, + null, + ); + } + return visibleTurnItems; +} + export function applyOrchestrationV2ProjectionEvent( projection: OrchestrationV2ThreadProjection | null, event: OrchestrationV2DomainEvent, @@ -301,7 +324,13 @@ export function applyOrchestrationV2ProjectionEvent( case "run.created": case "run.updated": { const next = { ...base, runs: upsertEntity(base.runs, event.payload) }; - return { ...next, visibleTurnItems: activeVisibleTurnItems(next) }; + const visible = { ...next, visibleTurnItems: activeVisibleTurnItems(next) }; + const wasQueued = base.runs.some( + (run) => run.id === event.payload.id && run.status === "queued", + ); + return wasQueued && event.payload.status !== "queued" + ? { ...visible, visibleTurnItems: restoreQueuedUserItems(visible, event.payload.id) } + : visible; } case "run.background-work-cancelled": return { diff --git a/packages/shared/src/orchestrationV2Timeline.test.ts b/packages/shared/src/orchestrationV2Timeline.test.ts index 7d2b7b7cbee45..c2b7a0c920159 100644 --- a/packages/shared/src/orchestrationV2Timeline.test.ts +++ b/packages/shared/src/orchestrationV2Timeline.test.ts @@ -14,7 +14,6 @@ type VisibilityInput = Parameters ({ status, hidden: [] })), + { status: "queued", hidden: ["queued"] }, { status: "cancelled", hidden: ["queued"] }, { status: "rolled_back", @@ -217,11 +217,15 @@ describe.each([ describe("indexed timeline visibility", () => { it.each([ - ...ordinaryStatuses.flatMap( + ...[...ordinaryStatuses, "queued" as const].flatMap( (status) => [ - { statuses: ["rolled_back", status], visible: true, ordinaryVisible: true }, - { statuses: ["cancelled", status], visible: true, ordinaryVisible: true }, + { + statuses: ["rolled_back", status], + visible: status !== "queued", + ordinaryVisible: true, + }, + { statuses: ["cancelled", status], visible: status !== "queued", ordinaryVisible: true }, { statuses: [status, "rolled_back"], visible: false, ordinaryVisible: false }, { statuses: [status, "cancelled"], visible: false, ordinaryVisible: true }, ] as const, diff --git a/packages/shared/src/orchestrationV2Timeline.ts b/packages/shared/src/orchestrationV2Timeline.ts index 05f13557138ba..e651dbfb01653 100644 --- a/packages/shared/src/orchestrationV2Timeline.ts +++ b/packages/shared/src/orchestrationV2Timeline.ts @@ -11,7 +11,7 @@ type TimelineTurnItem = Pick; readonly items: ReadonlyArray; @@ -52,12 +52,15 @@ export function isOrchestrationV2TurnItemVisible(input: { return false; } - // Cancelled queued input never reached the provider and must stay out of the transcript. + // A steer can return to the queue before delivery. Keep its retained item out + // of conversation history until its run starts; cancelled queued inputs stay hidden. if ( item.type === "user_message" && item.inputIntent === "queued_turn" && item.runId !== null && - input.runs.some((run) => run.id === item.runId && run.status === "cancelled") + input.runs.some( + (run) => run.id === item.runId && (run.status === "queued" || run.status === "cancelled"), + ) ) { return false; } @@ -76,7 +79,7 @@ export function createOrchestrationV2TurnItemVisibility(input: { }): (item: TimelineTurnItem) => boolean { const statuses = new Map(); for (const run of input.runs) { - if (run.status === "rolled_back" || run.status === "cancelled") + if (run.status === "rolled_back" || run.status === "cancelled" || run.status === "queued") statuses.set(run.id, run.status); else statuses.delete(run.id); } @@ -101,7 +104,7 @@ export function createOrchestrationV2TurnItemVisibility(input: { const status = item.runId === null ? undefined : statuses.get(item.runId); if (status === "rolled_back") return false; if ( - status === "cancelled" && + (status === "queued" || status === "cancelled") && item.type === "user_message" && item.inputIntent === "queued_turn" )