diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 1dc7c912e7765..a8fea794d3710 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1375,6 +1375,226 @@ it.effect("SQL list older settled threads below a cursor", () => }).pipe(Effect.provide(layerTest)), ); +// regression: #25806 pending request cards outlive the history window that drops their turn +it.effect("SQL snapshot windows keep cards for requests still pending", () => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:pending-request-window"); + yield* store.apply({ + id: EventId.make("event:pending-request-window:thread"), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make("project:pending-request-window"), + title: "Pending request window", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + let ordinal = 0; + const itemBase = (runId: RunId, nodeId: NodeId, id: string) => ({ + id: TurnItemId.make(`item:pending-request-window:${id}`), + threadId, + runId, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: ++ordinal, + status: "completed" as const, + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + }); + const request = (id: string, nodeId: NodeId, kind: "user_input" | "command") => ({ + id: RuntimeRequestId.make(`request:pending-request-window:${id}`), + nodeId, + providerTurnId: null, + nativeRequestRef: null, + kind, + responseCapability: { + type: "live" as const, + providerSessionId: ProviderSessionId.make("session:pending-request-window"), + }, + createdAt: now, + }); + const turnCount = THREAD_HISTORY_PAGE_POLICY.maxUserTurns + 4; + for (let turn = 1; turn <= turnCount; turn += 1) { + const runId = RunId.make(`run:pending-request-window:${turn}`); + const nodeId = NodeId.make(`node:pending-request-window:${turn}`); + yield* store.apply({ + id: EventId.make(`event:pending-request-window:run:${turn}`), + type: "run.created", + threadId, + runId, + nodeId, + driver, + occurredAt: now, + payload: { + id: runId, + threadId, + ordinal: turn, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:pending-request-window:${turn}`), + rootNodeId: nodeId, + activeAttemptId: null, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }); + const items: Array = [ + { + ...itemBase(runId, nodeId, `message:${turn}`), + type: "user_message", + createdBy: "user", + creationSource: "web", + messageId: MessageId.make(`message:pending-request-window:${turn}`), + inputIntent: "turn_start", + text: `Turn ${turn}`, + attachments: [], + }, + ]; + if (turn === 1) { + const question = { + id: "scope", + header: "Scope", + question: "Which scope?", + options: [{ label: "All", description: "Everything" }], + }; + items.push( + { + ...itemBase(runId, nodeId, "question:pending"), + status: "running", + type: "user_input_request", + requestId: request("question:pending", nodeId, "user_input").id, + questions: [question], + }, + { + ...itemBase(runId, nodeId, "approval:pending"), + status: "running", + type: "approval_request", + requestId: request("approval:pending", nodeId, "command").id, + requestKind: "command", + prompt: "rm -rf build", + }, + { + ...itemBase(runId, nodeId, "question:resolved"), + type: "user_input_request", + requestId: request("question:resolved", nodeId, "user_input").id, + questions: [question], + }, + ); + for (const [id, kind, status] of [ + ["question:pending", "user_input", "pending"], + ["approval:pending", "command", "pending"], + ["question:resolved", "user_input", "resolved"], + ] as const) { + yield* store.apply({ + id: EventId.make(`event:pending-request-window:request:${id}`), + type: "runtime-request.updated", + threadId, + nodeId, + driver, + occurredAt: now, + payload: { + ...request(id, nodeId, kind), + status, + resolvedAt: status === "pending" ? null : now, + }, + }); + } + } + for (const item of items) { + yield* store.apply({ + id: EventId.make(`event:pending-request-window:${item.id}`), + type: "turn-item.updated", + threadId, + runId, + nodeId, + driver, + occurredAt: now, + payload: item, + }); + } + } + + const pendingCardIds = [ + TurnItemId.make("item:pending-request-window:question:pending"), + TurnItemId.make("item:pending-request-window:approval:pending"), + ]; + const { statements, tracer } = traceSqlStatements(); + const snapshot = yield* store + .getThreadSnapshotWindow(threadId, { + rowLimit: THREAD_HISTORY_PAGE_POLICY.maxItems + 2, + userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + }) + .pipe(Effect.withTracer(tracer)); + const sql = yield* SqlClient.SqlClient; + const query = statements.find( + (statement) => + statement.includes("FROM orchestration_v2_projection_runtime_requests AS request") && + statement.includes("SELECT item.payload_json"), + ); + assert.isDefined(query); + const plan = (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${query}`, + )).map((row) => row.detail); + assert.include( + plan[0], + "SEARCH request USING INDEX orchestration_v2_projection_runtime_requests_thread_status_idx", + ); + assert.include( + plan[1], + "SEARCH item USING INDEX orchestration_v2_projection_turn_items_node_ordinal_idx", + ); + const bounded = buildBoundedThreadProjection({ + projection: snapshot.projection, + snapshotSequence: snapshot.snapshotSequence, + }); + assert.isTrue(bounded.hasMoreHistory); + assert.notInclude( + bounded.projection.turnItems.map((item) => item.id), + TurnItemId.make("item:pending-request-window:question:resolved"), + ); + for (const projection of [snapshot.projection, bounded.projection]) { + const visibleIds = projection.visibleTurnItems.map((row) => row.sourceItemId); + assert.notInclude(visibleIds, TurnItemId.make("item:pending-request-window:message:1")); + assert.includeMembers( + projection.turnItems.map((item) => item.id), + pendingCardIds, + ); + for (const id of pendingCardIds) assert.notInclude(visibleIds, id); + } + }).pipe(Effect.provide(layerTest)), +); + it.effect("lists only active subagent IDs without loading shell data", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 88a9073117dd6..e523fc9d41e51 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -5698,6 +5698,33 @@ export const layer: Layer.Layer = : {}), ...(historyAnchor === undefined ? {} : { historyAnchor }), }); + // Pending requests outlive their turn's window; clients read their prompt from the card. + const retainedIds = new Set(projection.turnItems.map((item) => item.id)); + const pendingCards = (yield* decodeRows( + decodeTurnItemPayload, + threadId, + )( + yield* sql` + SELECT item.payload_json + FROM orchestration_v2_projection_runtime_requests AS request + CROSS JOIN orchestration_v2_projection_turn_items AS item + ON item.node_id = request.node_id + AND item.type IN ('approval_request', 'user_input_request') + AND json_extract(item.payload_json, '$.requestId') = request.runtime_request_id + WHERE request.thread_id = ${threadId} AND request.status = 'pending' + AND item.thread_id = ${threadId} + `, + )).filter((item) => !retainedIds.has(item.id)); + const windowProjection = + pendingCards.length === 0 + ? projection + : { + ...projection, + turnItems: [...projection.turnItems, ...pendingCards].toSorted( + (left, right) => + left.ordinal - right.ordinal || left.id.localeCompare(right.id), + ), + }; const rows = yield* sql<{ readonly snapshot_sequence: number | null }>` SELECT MAX(sequence) AS snapshot_sequence FROM orchestration_events @@ -5708,7 +5735,7 @@ export const layer: Layer.Layer = return { schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION, snapshotSequence: rows[0]?.snapshot_sequence ?? 0, - projection, + projection: windowProjection, }; }), ) diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index f4e54000222fc..3430828d2871e 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -2,6 +2,7 @@ import { MessageId, NodeId, RunId, + RuntimeRequestId, ThreadId, TurnItemId, type OrchestrationV2ProjectedTurnItem, @@ -738,6 +739,69 @@ describe("threadHistoryPaging", () => { ).toBe(2_097_152); }); + it.each([40_000, 80_000])( + "reserves pending question cards before visible history: %s bytes", + (questionBytes) => { + const rows = Array.from({ length: 20 }, (_, index) => + makeRow(index + 1, { outputBytes: 1_500 }), + ); + const requestId = RuntimeRequestId.make("pending-question"); + const card: OrchestrationV2TurnItem = { + id: TurnItemId.make("pending-question-card"), + type: "user_input_request", + threadId: THREAD, + runId: null, + nodeId: NODE, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + requestId, + responseMode: "message", + questions: [ + { id: "scope", header: "Scope", question: "q".repeat(questionBytes), options: [] }, + ], + startedAt: NOW, + completedAt: NOW, + updatedAt: NOW, + }; + const projection = { + ...makeProjection(rows), + turnItems: [card, ...rows.map((row) => row.item)], + runtimeRequests: [ + { + id: requestId, + nodeId: NODE, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input" as const, + status: "pending" as const, + responseCapability: { type: "message" as const }, + createdAt: NOW, + resolvedAt: null, + }, + ], + }; + const policy = { ...THREAD_HISTORY_PAGE_POLICY, maxEncodedBytes: 64_000 }; + + const bounded = buildBoundedThreadProjection({ projection, snapshotSequence: 9, policy }); + + expect(bounded.projection.turnItems).toContain(card); + expect(bounded.projection.visibleTurnItems.length).toBeGreaterThan(0); + expect(bounded.hasMoreHistory).toBe(true); + if (questionBytes < policy.maxEncodedBytes) { + expect(Buffer.byteLength(JSON.stringify(bounded.projection), "utf8")).toBeLessThanOrEqual( + policy.maxEncodedBytes, + ); + } else { + expect(bounded.projection.visibleTurnItems).toHaveLength(1); + } + }, + ); + it("preserves oversized actionable state", () => { const projection = makeProjection([makeRow(0)]); const actionable = { diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index 7f1ea4db95d6f..b8517270413fc 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -501,10 +501,25 @@ function isLocalProjectedRow( return row.visibility === "local" || String(row.sourceThreadId) === String(projection.thread.id); } -/** Retain small requests outside the visible window; later result pages still need them for attempt/run reducers. */ -function retainedInterruptRequestTurnItems( +/** + * Small items clients need outside the visible window: later result pages need interrupt + * requests for attempt/run reducers, and pending requests read their prompt from their card. + */ +function isWindowDependency( + item: OrchestrationV2TurnItem, + pendingRequestIds: ReadonlySet, +): boolean { + return ( + item.type === "run_interrupt_request" || + ((item.type === "approval_request" || item.type === "user_input_request") && + pendingRequestIds.has(item.requestId)) + ); +} + +function retainedDependencyTurnItems( projection: OrchestrationV2ThreadProjection, visible: ReadonlyArray, + pendingRequestIds: ReadonlySet, ): OrchestrationV2TurnItem[] { const visibleLocalIds = new Set(); for (const row of visible) { @@ -515,7 +530,7 @@ function retainedInterruptRequestTurnItems( const retained: OrchestrationV2TurnItem[] = []; for (const item of projection.turnItems) { - if (item.type !== "run_interrupt_request") { + if (!isWindowDependency(item, pendingRequestIds)) { continue; } if (visibleLocalIds.has(String(item.id))) { @@ -633,10 +648,15 @@ export function buildBoundedThreadProjection(input: { }; const latestLocalTurnOrdinal = computeLatestLocalTurnOrdinal(input.projection.turnItems); + const pendingRequestIds = new Set( + controlProjection.runtimeRequests.flatMap((request) => + request.status === "pending" ? [request.id] : [], + ), + ); const dependencyReserve = (() => { let reserve = 0; for (const item of controlProjection.turnItems) { - if (item.type === "run_interrupt_request") { + if (isWindowDependency(item, pendingRequestIds)) { reserve += bytesOfJson(item); } } @@ -648,10 +668,9 @@ export function buildBoundedThreadProjection(input: { turnItems: [], visibleTurnItems: [], }); - // Control state (checkpoint file lists, the latest task list) cannot be paged, so it may claim at most - // half the budget; past that the window would shrink to the one row every page admits. + // Keep the control-only window minimum, but required dependency cards claim their reserve first. const windowBudget = Math.max( - policy.maxEncodedBytes / 2, + dependencyReserve === 0 ? policy.maxEncodedBytes / 2 : 0, policy.maxEncodedBytes - dependencyReserve - controlBytes - 1_024, ); const windowPolicy: ThreadHistoryPagePolicy = { ...policy, maxEncodedBytes: windowBudget }; @@ -664,9 +683,10 @@ export function buildBoundedThreadProjection(input: { }); const visibleTurnItems = renumberPositions(window.items); const windowTurnItems = localTurnItemsForVisibleWindow(controlProjection, visibleTurnItems); - const dependencyTurnItems = retainedInterruptRequestTurnItems( + const dependencyTurnItems = retainedDependencyTurnItems( controlProjection, visibleTurnItems, + pendingRequestIds, ); const turnItemById = new Map(); for (const item of windowTurnItems) { diff --git a/packages/client-runtime/src/state/threadHistoryMerge.test.ts b/packages/client-runtime/src/state/threadHistoryMerge.test.ts index 59e1ea3652e52..3f2be832a2b23 100644 --- a/packages/client-runtime/src/state/threadHistoryMerge.test.ts +++ b/packages/client-runtime/src/state/threadHistoryMerge.test.ts @@ -5,6 +5,7 @@ import * as DateTime from "effect/DateTime"; import { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; import { mergeOlderHistoryIntoProjection } from "./threadHistoryMerge.ts"; import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts"; +import { derivePendingThreadRequests } from "./threadRequests.ts"; const NOW = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"); @@ -222,6 +223,97 @@ describe("threadHistoryMerge", () => { expect(next?.visibleTurnItems[3]?.item).toEqual(liveItem); }); + // regression: #25806 a pending question card retained outside the window shows when paged in + it.each(["completed", "rolled_back"] as const)( + "keeps a pending question actionable when paging a %s run", + (status) => { + const recent = row(5); + const runId = "question-run" as never; + const nodeId = "question-node" as never; + const card = { + id: "item-question" as never, + type: "user_input_request" as const, + threadId: v2ThreadId, + runId, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running" as const, + title: null, + requestId: "request-question" as never, + responseMode: "message" as const, + questions: [{ id: "scope", header: "Scope", question: "Which scope?", options: [] }], + startedAt: NOW, + completedAt: null, + updatedAt: NOW, + }; + const coldOpen = { + ...v2Projection, + runs: [ + { + id: runId, + threadId: v2ThreadId, + ordinal: 1, + providerInstanceId: v2Projection.thread.providerInstanceId, + modelSelection: v2Projection.thread.modelSelection, + providerThreadId: null, + userMessageId: "question-message" as never, + rootNodeId: nodeId, + activeAttemptId: null, + status, + requestedAt: NOW, + startedAt: NOW, + completedAt: NOW, + checkpointId: null, + contextHandoffId: null, + }, + ], + runtimeRequests: [ + { + id: card.requestId, + nodeId, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input" as const, + status: "pending" as const, + responseCapability: { type: "message" as const }, + createdAt: NOW, + resolvedAt: null, + }, + ], + turnItems: [card, recent.item], + visibleTurnItems: [{ ...recent, position: 0 }], + }; + + const merged = mergeOlderHistoryIntoProjection(coldOpen, [ + { ...row(0), position: 0 }, + { + position: 1, + visibility: "local", + sourceThreadId: v2ThreadId, + sourceItemId: card.id, + item: card, + }, + ]); + + expect(merged.visibleTurnItems.map((entry) => String(entry.sourceItemId))).toEqual( + status === "rolled_back" ? ["item-0", "item-5"] : ["item-0", "item-question", "item-5"], + ); + expect(merged.turnItems).toContain(card); + const pending = derivePendingThreadRequests(merged).userInputs; + expect(pending).toEqual(derivePendingThreadRequests(coldOpen).userInputs); + expect(pending).toHaveLength(1); + expect(pending[0]).toMatchObject({ + requestId: card.requestId, + dismissible: true, + questions: card.questions.map((question) => ({ ...question, multiSelect: false })), + }); + }, + ); + it("keeps a history-page interrupt result visible when its request was retained at cold open", () => { const runId = "run-interrupt-cold" as never; const nodeId = "node-interrupt-cold" as never; diff --git a/packages/client-runtime/src/state/threadHistoryMerge.ts b/packages/client-runtime/src/state/threadHistoryMerge.ts index 6fd678b66b129..dcec2f207b81c 100644 --- a/packages/client-runtime/src/state/threadHistoryMerge.ts +++ b/packages/client-runtime/src/state/threadHistoryMerge.ts @@ -89,9 +89,13 @@ export function mergeOlderHistoryIntoProjection( (row.visibility === "local" || row.sourceThreadId === projection.thread.id) ) { const currentItem = turnItemById.get(String(row.sourceItemId)); + // Hidden live items stay hidden. Snapshots also retain interrupt requests and + // pending request cards outside the visible window; their history page shows them. if (currentItem !== undefined) { if ( - currentItem.type !== "run_interrupt_request" || + (currentItem.type !== "run_interrupt_request" && + currentItem.type !== "approval_request" && + currentItem.type !== "user_input_request") || !isOrchestrationV2TurnItemVisible({ item: currentItem, runs: projection.runs, diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index eaf1751a1c2d2..3ba6628f5392a 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -317,6 +317,32 @@ const PENDING_QUESTION = { resolvedAt: null, }; +const QUESTION_CARD: OrchestrationV2TurnItem = { + id: TurnItemId.make("question-card"), + threadId: THREAD_ID, + runId: null, + nodeId: PENDING_QUESTION.nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: null, + startedAt: v2Now, + completedAt: null, + updatedAt: v2Now, + type: "user_input_request", + requestId: PENDING_QUESTION.id, + questions: [{ id: "scope", header: "Scope", question: "Which scope?", options: [] }], +}; + +const PROJECTION_WITH_QUESTION = { + ...BASE_PROJECTION, + runtimeRequests: [PENDING_QUESTION], + turnItems: [QUESTION_CARD], +}; + const questionResolved = (sequence: number): OrchestrationV2ThreadStreamItem => ({ kind: "event", sequence, @@ -360,6 +386,62 @@ const deleted = (sequence = 3): OrchestrationV2ThreadStreamItem => { }; describe("EnvironmentThreads", () => { + it.effect.each([ + ["disk", false], + ["retained", false], + ["disk", true], + ["retained", true], + ] as const)( + "recovers a %s cache only when its pending question card is missing: %s", + ([cacheKind, hasCard]) => + Effect.gen(function* () { + const projection = { + ...PROJECTION_WITH_QUESTION, + turnItems: hasCard ? [QUESTION_CARD] : [], + }; + const schema = Schema.fromJsonString(StoredOrchestrationThreadSnapshot); + const encoded = yield* Schema.encodeEffect(schema)({ + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + environmentId: TARGET.environmentId, + threadId: THREAD_ID, + snapshot: { snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, projection }, + }); + const stored = yield* Schema.decodeEffect(schema)(encoded); + const h = yield* makeHarness({ + loadCached: Effect.succeedSome(stored.snapshot), + ...(cacheKind === "retained" + ? { + resumeCache: { + owner: undefined, + snapshot: { + state: { + ...EMPTY_ENVIRONMENT_THREAD_STATE, + data: Option.some(projection), + status: "live", + }, + sequence: CACHED_SNAPSHOT_SEQUENCE, + persisted: true, + }, + }, + } + : {}), + httpSnapshot: { + _tag: "present", + snapshot: { + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + projection: PROJECTION_WITH_QUESTION, + }, + }, + completionMarker: true, + }); + yield* Queue.offer(h.inputs, synchronized()); + const state = yield* awaitThreadState(h.observed, (value) => value.status === "live"); + expect(yield* Ref.get(h.loaderCalls)).toBe(hasCard ? 0 : 1); + expect(Option.getOrThrow(state.data)).toEqual(PROJECTION_WITH_QUESTION); + expect(yield* Ref.get(h.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE); + }), + ); + it.effect("discards an oversized version-4 cache and requests a fresh bounded snapshot", () => Effect.gen(function* () { const item: OrchestrationV2TurnItem = { @@ -559,7 +641,7 @@ describe("EnvironmentThreads", () => { Scope.close(scope, Exit.void), ); const old = yield* makeHarness({ - cached: { ...BASE_PROJECTION, runtimeRequests: [PENDING_QUESTION] }, + cached: PROJECTION_WITH_QUESTION, resumeCache, }).pipe(Effect.provideService(Scope.Scope, oldScope)); yield* awaitThreadState(old.observed, (value) => value.status === "live"); @@ -2121,7 +2203,7 @@ describe("EnvironmentThreads", () => { (unsubscribe) => Effect.sync(unsubscribe), ); const harness = yield* makeHarness({ - cached: { ...BASE_PROJECTION, runtimeRequests: [PENDING_QUESTION] }, + cached: PROJECTION_WITH_QUESTION, completionMarker: true, }); yield* awaitThreadState( diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index c7b22d9dcdff0..63cb39adef86f 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -188,6 +188,20 @@ function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadStat }; } +function hasMissingPendingQuestionCard(projection: OrchestrationV2ThreadProjection): boolean { + const pending = new Set( + projection.runtimeRequests + .filter((request) => request.kind === "user_input" && request.status === "pending") + .map((request) => request.id), + ); + if (pending.size === 0) return false; + for (const item of projection.turnItems) { + if (item.type === "user_input_request") pending.delete(item.requestId); + if (pending.size === 0) return false; + } + return true; +} + export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* ( threadId: ThreadIdType, resumeCache?: ThreadResumeCache, @@ -205,10 +219,16 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; - const retained = resumeCache?.snapshot; + const candidate = resumeCache?.snapshot; + const retained = + candidate && + Option.isSome(candidate.state.data) && + hasMissingPendingQuestionCard(candidate.state.data.value) + ? undefined + : candidate; const owner = {}; if (resumeCache) resumeCache.owner = owner; - const cached = + const loaded = retained === undefined ? yield* cache.loadThread(environmentId, threadId).pipe( Effect.catch((error) => @@ -223,6 +243,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ), ) : Option.none(); + // Old bounded caches can omit a pending question's card; replay alone cannot restore it. + const cached = Option.filter( + loaded, + (snapshot) => !hasMissingPendingQuestionCard(snapshot.projection), + ); const cachedThread = Option.map(cached, (snapshot) => snapshot.projection); // Seed the resume cursor from the cached snapshot so a warm cache can catch up // via `afterSequence` instead of re-downloading the full thread body.