From 712349860793856f620bea8a4f52415a9c9ade19 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Wed, 11 Nov 2026 20:33:53 -0300 Subject: [PATCH 1/8] perf(server): sending a prompt in a compacted thread reads only what follows the compaction A message sent to a thread that was compacted read the thread's whole history before the Working status could appear. The command read now keeps only the runs since the last prompt before the latest completed compaction, plus live runs, fresh-context runs and the runs the command names. A fresh session or another provider still reads everything, as before. Migration 083 adds the indexes these reads need; it is idempotent for databases that already have them from earlier builds. Co-Authored-By: Claude Sonnet 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 46 +++- .../orchestration-v2/ProjectionStore.test.ts | 200 ++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 255 ++++++++++++------ .../ProviderTurnControlService.test.ts | 2 +- .../SteeringCompletion.integration.test.ts | 133 +++++++++ apps/server/src/persistence/Migrations.ts | 4 +- .../Migrations/083_CommandWindowIndexes.ts | 31 +++ .../reconcileV2PreviewMigration.test.ts | 4 +- 8 files changed, 579 insertions(+), 96 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index ac23af5324dae..909400da04084 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1528,7 +1528,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); // Runs in every command transaction; avoid reading thousands of finished leaf nodes. - const readCommandProjection = (threadId: ThreadId) => + const readCommandProjection = ( + threadId: ThreadId, + recent?: { keepRunIds: ReadonlyArray }, + ) => projectionStore .getThreadRecords( threadId, @@ -1553,8 +1556,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Commands here read only undelivered or active tasks; a dispatcher thread keeps // thousands of delivered ones, each carrying its full prompt. omitSettledSubagents: true, - // Commands act on pending forks and merge-backs only; consumed ones stay unread. - contextTransferStatuses: ["pending"], + ...(recent === undefined ? {} : { sinceLastCompaction: recent }), }, ) .pipe( @@ -1643,9 +1645,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.fnUntraced(function* ( threadId: ThreadId, events: Ref.Ref>, + recent?: { keepRunIds: ReadonlyArray }, ) { const pending = (yield* Ref.get(events)).filter((event) => event.threadId === threadId); - const stored = yield* Effect.option(readCommandProjection(threadId)); + const stored = yield* Effect.option(readCommandProjection(threadId, recent)); let projection: OrchestrationV2ThreadProjection; if (Option.isSome(stored)) { projection = stored.value; @@ -6305,7 +6308,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: Ref.Ref>, effects: Ref.Ref>, ) { - let projection = yield* getProjectionWithPendingEvents(command.threadId, events); + // A compacted thread needs only its recent runs, plus any older run the command names. + let projection = yield* getProjectionWithPendingEvents(command.threadId, events, { + keepRunIds: [ + command.manualContinuationOfRunId, + command.restartContinuationOfRunId, + command.usageLimitContinuationOfRunId, + command.delegatedCompletion?.parentRunId, + command.steerFollowUpOfRunId, + ...(command.dispatchMode.type === "steer_active" || + command.dispatchMode.type === "restart_active" + ? [command.dispatchMode.targetRunId] + : []), + ].flatMap((runId) => (runId === undefined ? [] : [runId])), + }); if (command.steerFollowUpOfRunId !== undefined) { const source = projection.runs.find((run) => run.id === command.steerFollowUpOfRunId); if ( @@ -6342,6 +6358,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); return; } + const activeProviderInstanceId = projection.providerThreads.find( + (candidate) => candidate.id === projection.thread.activeProviderThreadId, + )?.providerInstanceId; + // A fresh session or another provider is handed the whole history, so it reads all of it. + if ( + command.freshContext === true || + (activeProviderInstanceId !== undefined && + activeProviderInstanceId !== + (command.modelSelection ?? projection.thread.modelSelection).instanceId) + ) { + projection = yield* getProjectionWithPendingEvents(command.threadId, events); + } if (command.manualContinuationOfRunId !== undefined) { const source = projection.runs.find((run) => run.id === command.manualContinuationOfRunId); const limited = latestRootProviderFailure(source ?? null, projection.turnItems); @@ -6544,16 +6572,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); projection = applyToProjection(projection, event); } - const userMessages = projection.messages.filter((message) => message.role === "user"); - const onlyMaintenanceHistory = - userMessages.length > 0 && userMessages.every(isNativeMaintenanceCommand); if ( !isNativeMaintenanceCommand(command) && ((command.titleSeed !== undefined && (yield* projectionStore .getMessageCount(command.threadId) .pipe(mapDispatchError(command))) === 0) || - (command.createdBy === "user" && onlyMaintenanceHistory)) + (command.createdBy === "user" && + (yield* projectionStore + .hasOnlyMaintenanceUserMessages(command.threadId) + .pipe(mapDispatchError(command))))) ) { const now = yield* DateTime.now; const thread: OrchestrationV2AppThread = { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 53cd77a88f852..ef06ecc8ce986 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -795,6 +795,206 @@ it.effect("idle compaction filters keep the latest run, live work and the newest }).pipe(Effect.provide(layerTest)), ); +it.effect("a compacted thread reads its recent runs, live runs, fresh starts and named runs", () => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const threadId = yield* addRolledBackRecoveryCandidate("compacted"); + const now = yield* DateTime.now; + const nodeId = NodeId.make("node:compacted"); + const runId = (ordinal: number) => RunId.make(`run:compacted:${ordinal}`); + const read = (keepRunIds: ReadonlyArray) => + store.getThreadRecords(threadId, ["runs", "messages"], { + messageRoles: ["user"], + sinceLastCompaction: { keepRunIds }, + }); + // Run 1 is the helper's rolled-back run. Run 2 is live and run 3 starts a fresh session. + // Runs 6 and 9 are `/compact` runs that hold a compaction; run 8 is one that does not. + for (const [ordinal, status] of [ + [2, "running"], + [3, "completed"], + [4, "completed"], + [5, "completed"], + [6, "completed"], + [7, "completed"], + [8, "completed"], + [9, "completed"], + ] as const) { + yield* store.apply({ + id: EventId.make(`event:compacted:run-${ordinal}`), + type: "run.created", + threadId, + runId: runId(ordinal), + nodeId, + driver, + providerInstanceId, + occurredAt: now, + payload: { + id: runId(ordinal), + threadId, + ordinal, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:compacted:${ordinal}`), + rootNodeId: nodeId, + activeAttemptId: null, + status, + requestedAt: now, + startedAt: now, + completedAt: status === "running" ? null : now, + checkpointId: null, + contextHandoffId: null, + ...(ordinal === 3 ? { freshContext: true } : {}), + }, + }); + yield* store.apply({ + id: EventId.make(`event:compacted:message-${ordinal}`), + type: "message.updated", + threadId, + runId: runId(ordinal), + occurredAt: now, + payload: { + id: MessageId.make(`message:compacted:${ordinal}`), + threadId, + runId: runId(ordinal), + nodeId, + createdBy: "user", + creationSource: "web", + role: "user", + text: [6, 8, 9].includes(ordinal) ? "/compact" : `Prompt ${ordinal}`, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + } + const rolledBackId = RunId.make("run:compacted:rolled-back"); + const ids = (rows: ReadonlyArray) => + rows.map((row) => row.id); + const compact = (ordinal: number) => + store.apply({ + id: EventId.make(`event:compacted:compaction-${ordinal}`), + type: "turn-item.updated", + threadId, + runId: runId(ordinal), + nodeId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make(`item:compacted:compaction-${ordinal}`), + threadId, + runId: runId(ordinal), + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "compaction", + driver, + }, + }); + // Nothing is compacted yet, so the whole history is current. + assert.lengthOf((yield* read([])).runs, 9); + + yield* compact(6); + // Legacy rows can hold unreadable JSON; the fresh-context index must not reject them. + const sql = yield* SqlClient.SqlClient; + yield* sql` + INSERT INTO orchestration_v2_projection_runs + (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) + VALUES ('run:compacted:legacy', ${threadId}, 0, 'codex', 'completed', '', '{') + `; + const runIdsOf = (ordinals: ReadonlyArray) => ordinals.map(runId); + // The window starts at run 5, the prompt before the compaction. + const recent = yield* read([rolledBackId]); + assert.deepEqual(ids(recent.runs), [rolledBackId, ...runIdsOf([2, 3, 5, 6, 7, 8, 9])]); + assert.deepEqual( + ids(recent.messages), + [2, 3, 5, 6, 7, 8, 9].map((ordinal) => `message:compacted:${ordinal}`), + ); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9])); + + // Run 8 is a `/compact` run, so the model-switch baseline is the prompt before it, run 7. + yield* compact(9); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 7, 8, 9])); + + // The search for a prompt before the compaction may walk past an unreadable run. + const legacyThreadId = yield* addRolledBackRecoveryCandidate("compacted-legacy"); + const legacyNodeId = NodeId.make("node:compacted-legacy"); + const compactionRunId = RunId.make("run:compacted-legacy:2"); + yield* sql` + INSERT INTO orchestration_v2_projection_runs + (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) + VALUES ('run:compacted-legacy:0', ${legacyThreadId}, 0, 'codex', 'completed', '', '{') + `; + yield* store.apply({ + id: EventId.make("event:compacted-legacy:run"), + type: "run.created", + threadId: legacyThreadId, + runId: compactionRunId, + nodeId: legacyNodeId, + driver, + providerInstanceId, + occurredAt: now, + payload: { + id: compactionRunId, + threadId: legacyThreadId, + ordinal: 2, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:compacted-legacy:2"), + rootNodeId: legacyNodeId, + activeAttemptId: null, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }); + yield* store.apply({ + id: EventId.make("event:compacted-legacy:compaction"), + type: "turn-item.updated", + threadId: legacyThreadId, + runId: compactionRunId, + nodeId: legacyNodeId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make("item:compacted-legacy:compaction"), + threadId: legacyThreadId, + runId: compactionRunId, + nodeId: legacyNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "compaction", + driver, + }, + }); + const legacy = yield* store.getThreadRecords(legacyThreadId, ["runs"], { + sinceLastCompaction: { keepRunIds: [] }, + }); + assert.deepEqual(ids(legacy.runs), [compactionRunId]); + }).pipe(Effect.provide(layerTest)), +); + it.effect("record reads filter notification sources without dropping matching message bodies", () => 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 292aa61ecdd6f..d45327c2ffb3c 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -372,6 +372,13 @@ export interface ProjectionRecordFilter { * run is active or `except` names them. */ readonly omitSettledSubagents?: true | { readonly except: ReadonlyArray }; + /** + * Reads what a new message needs from a thread that was compacted: the runs from the last + * prompt before the latest completed compaction, plus live runs, fresh-context runs and `keepRunIds`. Run- + * scoped records follow their runs, provider threads are the thread's own, and context transfers + * are the open ones that target it. A thread without a completed compaction reads in full. + */ + readonly sinceLastCompaction?: { readonly keepRunIds: ReadonlyArray }; } // A fork inherits its source's history before it has messages of its own, @@ -472,11 +479,10 @@ export interface ProjectionStoreV2Shape { options: ProjectionTimelinePageOptions, ) => Effect.Effect; readonly getMessageCount: (threadId: ThreadId) => Effect.Effect; - /** Bytes of tool output a transcript export is sure to hold, or 0 when it can fit in `limit`. */ - readonly getTranscriptByteFloor: ( + /** True when the thread has user messages and every one is a native maintenance command. */ + readonly hasOnlyMaintenanceUserMessages: ( threadId: ThreadId, - limit: number, - ) => Effect.Effect; + ) => Effect.Effect; readonly getSubagentCount: (threadId: ThreadId) => Effect.Effect; /** Child threads of the thread's app-owned tasks, in task order, without decoding the tasks. */ readonly listAppOwnedChildThreadIds: ( @@ -3417,6 +3423,67 @@ export const layer: Layer.Layer = const threadColumn = (ids: ReadonlyArray | undefined, column = "thread_id") => sql.literal(ids === undefined ? column : `+${column}`); + // Mirrors JavaScript's `trim().toLowerCase()` of a message's text. + const trimmedLowerText = (payload: Statement.Fragment) => sql`lower(trim( + json_extract(${payload}, '$.text'), + char(9, 10, 11, 12, 13, 32, 160, 5760, 8192, 8193, 8194, 8195, + 8196, 8197, 8198, 8199, 8200, 8201, 8202, 8232, 8233, 8239, 8287, 12288, 65279) + ))`; + + /** Null when nothing was compacted, so the whole history is current. */ + const runsSinceLastCompaction = Effect.fn("ProjectionStore.runsSinceLastCompaction")(function* ( + threadId: ThreadId, + keepRunIds: ReadonlyArray, + ) { + // Starts at the last prompt before the compaction: a model switch is measured from the + // last run whose prompt the timeline shows, and `/compact` runs are not shown. + const [compaction] = yield* sql<{ readonly from_ordinal: number }>` + SELECT COALESCE( + (SELECT previous.ordinal FROM orchestration_v2_projection_runs AS previous + WHERE previous.thread_id = ${threadId} AND previous.ordinal < latest.ordinal + AND previous.status NOT IN ('queued', 'rolled_back') + AND CASE WHEN json_valid(previous.payload_json) THEN NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages AS prompt + WHERE prompt.message_id = json_extract(previous.payload_json, '$.userMessageId') + AND prompt.role = 'user' + AND coalesce(json_array_length(prompt.payload_json, '$.attachments'), 0) = 0 + AND ${trimmedLowerText(sql.literal("prompt.payload_json"))} = ${IDLE_COMPACTION_TEXT} + ) ELSE 0 END + ORDER BY previous.ordinal DESC + LIMIT 1), + latest.ordinal + ) AS from_ordinal + FROM ( + SELECT run.ordinal + FROM orchestration_v2_projection_turn_items AS item + INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx + INNER JOIN orchestration_v2_projection_runs AS run ON run.run_id = item.run_id + WHERE (item.thread_id || '') = ${threadId} + AND (item.type || '') = 'compaction' AND item.status = 'completed' + ORDER BY item.ordinal DESC + LIMIT 1 + ) AS latest + `; + return compaction === undefined + ? null + : sql` + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} AND ordinal >= ${compaction.from_ordinal} + UNION + SELECT run_id FROM orchestration_v2_projection_runs + INDEXED BY orchestration_v2_projection_runs_thread_status_idx + WHERE thread_id = ${threadId} + AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + UNION + SELECT run_id FROM orchestration_v2_projection_runs + INDEXED BY orchestration_v2_projection_runs_fresh_context_idx + WHERE thread_id = ${threadId} AND CASE WHEN json_valid(payload_json) + THEN json_extract(payload_json, '$.freshContext') = 1 ELSE 0 END + UNION + SELECT value FROM json_each(${encodeIdList(keepRunIds)}) + `; + }); + const readCanonicalProjection = Effect.fn("ProjectionStore.readCanonicalProjection")( function* ( threadId: ThreadId, @@ -3434,6 +3501,10 @@ export const layer: Layer.Layer = if (!threadRow) { return yield* new ProjectionStoreThreadNotFoundError({ threadId }); } + const recentRuns = + window === undefined && filter?.sinceLastCompaction !== undefined + ? yield* runsSinceLastCompaction(threadId, filter.sinceLastCompaction.keepRunIds) + : null; type WindowPayloadRow = PayloadRow & { readonly work_last_id?: string | null; @@ -3455,18 +3526,13 @@ export const layer: Layer.Layer = ? yield* sql` SELECT payload_json FROM orchestration_v2_projection_turn_items - ${ - // A type list of conversation items reads them from the conversation index. - // Any other plan fetches every tool call of the thread to check its type. - readsConversationItemsOnly(filter) - ? sql`INDEXED BY orchestration_v2_projection_turn_items_conversation_idx - WHERE thread_id = ${threadId} AND type NOT IN ${CONVERSATION_INDEX_EXCLUDED_TYPES}` - : sql`WHERE thread_id = ${threadId}` - } + ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx`} + WHERE thread_id = ${threadId} ${filter?.turnItemRequestIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.requestId') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRequestIds)}))`} ${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`} + ${recentRuns === null ? sql`` : sql`AND run_id IN (${recentRuns})`} ${filter?.turnItemNodeIds === undefined ? sql`` : sql`AND node_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemNodeIds)}))`} ${filter?.turnItemRunIds === undefined ? sql`` : sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR (${filter.turnItemRunIds.includes(null) ? 1 : 0} = 1 AND run_id IS NULL))`} ORDER BY ordinal ASC, turn_item_id ASC @@ -3657,7 +3723,14 @@ export const layer: Layer.Layer = : window === undefined ? sql` SELECT payload_json - FROM orchestration_v2_projection_runs + FROM ${ + // Looking runs up by key beats filtering the thread's whole run index. + recentRuns === null + ? sql`orchestration_v2_projection_runs` + : sql`(SELECT run_id AS recent_run_id FROM (${recentRuns})) AS recent + CROSS JOIN orchestration_v2_projection_runs + ON orchestration_v2_projection_runs.run_id = recent.recent_run_id` + } WHERE ${threadColumn(filter?.runIds)} = ${threadId} ${filter?.runIds === undefined ? sql`` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.runIds)}))`} ${ @@ -3711,6 +3784,7 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_run_attempts WHERE thread_id = ${threadId} + ${recentRuns === null ? sql`` : sql`AND run_id IN (${recentRuns})`} ORDER BY run_id ASC, attempt_ordinal ASC ` : sql` @@ -3727,24 +3801,21 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_nodes AS node - WHERE ${ - filter?.omitSettledLeafNodes === true - ? // Seeks the kept nodes. One filter over the thread reads every settled leaf to - // drop it. The first predicate must match migration 083's index expression. A - // node's run is always a run of its own thread. - sql`node.rowid IN ( - SELECT rowid FROM orchestration_v2_projection_nodes - WHERE thread_id = ${threadId} - AND (kind IN ('tool_call', 'assistant_message', 'reasoning') - AND status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') - AND run_id IS NOT NULL) = 0 - UNION ALL - SELECT rowid FROM orchestration_v2_projection_nodes - WHERE thread_id = ${threadId} - AND run_id IN ( - SELECT run_id FROM orchestration_v2_projection_runs - WHERE thread_id = ${threadId} - AND status IN ('preparing', 'queued', 'starting', 'running', 'waiting') + ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_nodes_thread_run_idx`} + WHERE ${recentRuns === null ? threadColumn(filter?.nodeIds, "node.thread_id") : sql.literal("node.thread_id")} = ${threadId} + ${recentRuns === null ? sql`` : sql`AND node.run_id IN (${recentRuns})`} + ${filter?.nodeIds === undefined ? sql`` : sql`AND node.node_id IN (SELECT value FROM json_each(${encodeIdList(filter.nodeIds)}))`} + ${ + filter?.omitSettledLeafNodes === true + ? sql`AND NOT ( + node.kind IN ('tool_call', 'assistant_message', 'reasoning') + AND node.status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') + AND node.run_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs AS run + INDEXED BY orchestration_v2_projection_runs_id_status_idx + WHERE (run.run_id || '') = (node.run_id || '') + AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') ) AND kind IN ('tool_call', 'assistant_message', 'reasoning') AND status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') @@ -3775,26 +3846,33 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_subagents AS subagent - WHERE ${threadColumn(filter?.subagentIds)} = ${threadId} + ${ + filter?.omitSettledSubagents === undefined + ? sql`` + : // Its columns decide settled rows, so only the open ones read their payload. + sql`INDEXED BY orchestration_v2_projection_subagents_settled_idx` + } + WHERE ${filter?.omitSettledSubagents === undefined ? threadColumn(filter?.subagentIds) : sql.literal("thread_id")} = ${threadId} ${filter?.subagentIds === undefined ? sql`` : sql`AND subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentIds)}))`} ${filter?.subagentRunIds === undefined ? sql`` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentRunIds)}))`} ${ - filter?.omitSettledSubagents === true - ? // The settled index holds every column this check reads, so the settled - // tasks are dropped there and their rows, each with a full prompt, stay unread. - sql`AND subagent.rowid IN ( - SELECT candidate.rowid - FROM orchestration_v2_projection_subagents AS candidate - INDEXED BY orchestration_v2_projection_subagents_settled_idx - WHERE candidate.thread_id = ${threadId} - AND NOT ${settledSubagent(sql`candidate`)} - )` - : filter?.omitSettledSubagents !== undefined - ? sql`AND ( - subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.omitSettledSubagents.except)})) - OR NOT ${settledSubagent(sql`subagent`)} - )` - : sql`` + filter?.omitSettledSubagents !== undefined + ? sql`AND (${ + filter.omitSettledSubagents === true + ? sql`0` + : sql`subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.omitSettledSubagents.except)}))` + } OR NOT ( + subagent.status IN ('completed', 'failed', 'cancelled', 'interrupted') + AND COALESCE(subagent.completion_delivery_state, '') + IN ('acknowledged', 'delivered', 'disposed') + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs AS run + INDEXED BY orchestration_v2_projection_runs_id_status_idx + WHERE (run.run_id || '') = (subagent.run_id || '') + AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') + ) + ))` + : sql`` } ORDER BY COALESCE(started_at, ''), subagent_id ASC ` @@ -3837,13 +3915,17 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_provider_threads WHERE (thread_id = ${threadId} - OR owner_node_id IN (${nodesOwningProviderThreads(threadId)}) + ${ + recentRuns === null + ? sql`OR owner_node_id IN (${nodesOwningProviderThreads(threadId)}) OR provider_thread_id IN ( SELECT provider_thread_id FROM orchestration_v2_projection_subagents WHERE thread_id = ${threadId} AND provider_thread_id IS NOT NULL - )) + )` + : sql`` + }) ${filter?.providerThreadIds === undefined ? sql`` : sql`AND provider_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerThreadIds)}))`} ORDER BY COALESCE(first_run_ordinal, 0), provider_thread_id ASC ` @@ -3864,7 +3946,15 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_provider_turns + ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_provider_turns_attempt_idx`} WHERE thread_id = ${threadId} + ${ + recentRuns === null + ? sql`` + : sql`AND run_attempt_id IN ( + SELECT attempt_id FROM orchestration_v2_projection_run_attempts + WHERE thread_id = ${threadId} AND run_id IN (${recentRuns}))` + } ${filter?.providerTurnAttemptIds === undefined ? sql`` : sql`AND run_attempt_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnAttemptIds)}))`} ${filter?.providerTurnThreadIds === undefined ? sql`` : sql`AND provider_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnThreadIds)}))`} ${ @@ -3910,7 +4000,9 @@ export const layer: Layer.Layer = : window === undefined ? sql` SELECT payload_json FROM orchestration_v2_projection_messages - WHERE ${threadColumn(filter?.messageIds)} = ${threadId} + ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_messages_run_idx`} + WHERE ${recentRuns === null ? threadColumn(filter?.messageIds) : sql.literal("thread_id")} = ${threadId} + ${recentRuns === null ? sql`` : sql`AND run_id IN (${recentRuns})`} ${filter?.messageIds === undefined ? sql`` : sql`AND message_id IN (SELECT value FROM json_each(${encodeIdList(filter.messageIds)}))`} ${filter?.messageRoles === undefined ? sql`` : sql`AND role IN (SELECT value FROM json_each(${encodeIdList(filter.messageRoles)}))`} ${filter?.monitorNotificationsOnly === true ? sql`AND json_extract(payload_json, '$.notification.source.kind') = 'monitor'` : sql``} @@ -4029,20 +4121,11 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_context_transfers WHERE ${ - filter?.contextTransferStatuses === undefined + recentRuns === null ? sql`(source_thread_id = ${threadId} OR target_thread_id = ${threadId})` - : // Each side seeks its (thread, status) index. One OR over both sides checks the - // status on every row the thread ever sent or received. - sql`rowid IN ( - SELECT rowid FROM orchestration_v2_projection_context_transfers - WHERE source_thread_id = ${threadId} - AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)})) - UNION - SELECT rowid FROM orchestration_v2_projection_context_transfers - WHERE target_thread_id = ${threadId} - AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)})) - )` + : sql`target_thread_id = ${threadId} AND status IN ('pending', 'running', 'waiting')` } + ${filter?.contextTransferStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)}))`} ORDER BY rowid ASC ` : sql` @@ -4088,23 +4171,12 @@ export const layer: Layer.Layer = decodeRows(decodeContextTransferPayload, threadId)(contextTransferRows), ]); const messageOrdinals = new Map(); - // The literal `type = 'user_message'` lets SQLite pick the partial user-message index. - const userOnly = messages.every((message) => message.role === "user"); - // The loaded items already order the messages when they hold every item the lookup reads. - const loadsEveryItemOf = (type: OrchestrationV2TurnItem["type"]) => - window === undefined && - (fields === undefined || fields.includes("turnItems")) && - filter?.turnItemRequestIds === undefined && - filter?.turnItemStatuses === undefined && - filter?.turnItemRunId === undefined && - filter?.turnItemRunIds === undefined && - filter?.turnItemNodeIds === undefined && - (filter?.turnItemTypes === undefined || filter.turnItemTypes.includes(type)); + // The recent user turns, loaded above, already order their own messages. if ( + recentRuns === null && fields !== undefined && fields.includes("messages") && - messages.length > 1 && - !(loadsEveryItemOf("user_message") && (userOnly || loadsEveryItemOf("assistant_message"))) + messages.length > 1 ) { // Broad role reads keep the original query to avoid per-ID lookup overhead. const loadedMessageIds = @@ -6621,15 +6693,34 @@ export const layer: Layer.Layer = ), ); + // Mirrors isNativeMaintenanceCommand over the thread's user messages. + const hasOnlyMaintenanceUserMessages: ProjectionStoreV2Shape["hasOnlyMaintenanceUserMessages"] = + (threadId) => + sql<{ readonly any_user: number; readonly other_user: number }>` + SELECT + EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages + WHERE thread_id = ${threadId} AND role = 'user' + ) AS any_user, + EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages + WHERE thread_id = ${threadId} AND role = 'user' + AND NOT ( + coalesce(json_array_length(payload_json, '$.attachments'), 0) = 0 + AND ${trimmedLowerText(sql.literal("payload_json"))} IN ('/compact', '/logout') + ) + ) AS other_user + `.pipe( + Effect.map(([row]) => row?.any_user === 1 && row.other_user === 0), + Effect.mapError(controlReadError(threadId)), + ); + // Mirror idleCompactionOf, including JavaScript trim whitespace. const isIdleCompactionMessage = (alias: string) => { const payload = sql.literal(`${alias}.payload_json`); return sql`( ${sql.literal(`${alias}.role`)} = 'user' - AND lower(trim(json_extract(${payload}, '$.text'), - char(9, 10, 11, 12, 13, 32, 160, 5760, 8192, 8193, 8194, 8195, - 8196, 8197, 8198, 8199, 8200, 8201, 8202, 8232, 8233, 8239, 8287, 12288, 65279) - )) = ${IDLE_COMPACTION_TEXT} + AND ${trimmedLowerText(payload)} = ${IDLE_COMPACTION_TEXT} AND coalesce(json_array_length(${payload}, '$.attachments'), 0) = 0 AND ( json_extract(${payload}, '$.idleCompaction') IS NOT NULL @@ -8038,7 +8129,7 @@ export const layer: Layer.Layer = hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest), isNodeStored: readOnReader(isNodeStored), getMessageCount: readOnReader(getMessageCount), - getTranscriptByteFloor: readOnReader(getTranscriptByteFloor), + hasOnlyMaintenanceUserMessages: readOnReader(hasOnlyMaintenanceUserMessages), getSubagentCount: readOnReader(getSubagentCount), listAppOwnedChildThreadIds: readOnReader(listAppOwnedChildThreadIds), hasPendingDelegatedCompletion: readOnReader(hasPendingDelegatedCompletion), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 34ff5a7631105..b4a9764ba04c1 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -239,7 +239,7 @@ it.effect( getThreadHistoryPage: () => Effect.die("unused"), getTimelinePage: () => Effect.die("Unused timeline read"), getMessageCount: () => Effect.die("unused message count"), - getTranscriptByteFloor: () => Effect.die("unused transcript byte floor"), + hasOnlyMaintenanceUserMessages: () => Effect.die("unused maintenance history"), hasPendingDelegatedCompletion: () => Effect.die("unused hasPendingDelegatedCompletion"), getSubagentCount: () => Effect.die("unused subagent count"), listAppOwnedChildThreadIds: () => Effect.die("unused listAppOwnedChildThreadIds"), diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 8d784a91f59cb..0c655b83f6b41 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1379,6 +1379,139 @@ it.effect("starts a steer that missed the turn on the saved next-turn selection" ), ); +it.effect("starts a steer of a finished run from before the last compaction as a new turn", () => + Effect.scoped( + Effect.gen(function* () { + const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + "steering-before-compaction", + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = yield* startFirstTurn; + const finish = (turn: (typeof started)[number]) => + Effect.gen(function* () { + const settled = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === turn.runId && + event.payload.status === "waiting", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const providerTurn = (yield* orchestrator.getThreadProjection( + threadId, + )).providerTurns.find((candidate) => candidate.runAttemptId === turn.attemptId)!; + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + ...providerTurn, + status: "completed", + completedAt: yield* DateTime.now, + }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: providerTurn.providerThreadId, + providerTurnId: providerTurn.id, + runOrdinal: turn.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + }); + const first = started[0]!; + yield* finish(first); + for (const id of ["second", "third"]) { + const running = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(id), + threadId, + messageId: MessageId.make(`message:${id}`), + text: id, + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(running); + yield* finish(started.at(-1)!); + } + // The window starts at the second run, the prompt before the compaction in the third. + const third = started[2]!; + const now = yield* DateTime.now; + yield* sink.write({ + events: [ + { + id: EventId.make("event:compaction"), + type: "turn-item.updated" as const, + threadId, + runId: third.runId, + nodeId: third.rootNodeId, + occurredAt: now, + payload: { + id: TurnItemId.make("item:compaction"), + threadId, + runId: third.runId, + nodeId: third.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1_000, + status: "completed" as const, + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "compaction" as const, + driver, + }, + }, + ], + }); + + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("steer"), + threadId, + messageId: MessageId.make("message:steer"), + text: "late steer", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "steer_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + + assert.deepEqual(steered, []); + assert.equal(started.length, 4); + assert.equal(started[3]?.message.messageId, MessageId.make("message:steer")); + }).pipe(Effect.provide(layer)); + }), + ), +); + it.effect.each(["complete", "stop-before-decline", "stop-after-decline", "queued-model"] as const)( "persists a declined steer without blocking the worker: %s", (timing) => diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 422d2122d03bd..ab414e4aac16c 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -85,7 +85,7 @@ import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; import Migration0081 from "./Migrations/081_ProjectionMessageSearchText.ts"; import Migration0082 from "./Migrations/082_TurnItemConversationIndex.ts"; -import Migration0083 from "./Migrations/083_CommandReadIndexes.ts"; +import Migration0083 from "./Migrations/083_CommandWindowIndexes.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -171,7 +171,7 @@ const migrationEntries = [ [80, "ScheduledTaskThreadMode", Migration0080], [81, "ProjectionMessageSearchText", Migration0081], [82, "TurnItemConversationIndex", Migration0082], - [83, "CommandReadIndexes", Migration0083], + [83, "CommandWindowIndexes", Migration0083], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts b/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts new file mode 100644 index 0000000000000..c22f3682cc1bd --- /dev/null +++ b/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts @@ -0,0 +1,31 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Reading the runs a new message needs from a compacted thread must not touch the older ones. + // Every column a lookup could match is wrapped in `|| ''`, so only a lookup that writes the same + // expression can use these indexes; plain ones were chosen for unrelated statements instead. + // IF NOT EXISTS: earlier builds created some of these under migrations that were renumbered. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_thread_type_idx + ON orchestration_v2_projection_turn_items((thread_id || ''), (type || ''), ordinal, turn_item_id) + `; + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_id_status_idx + ON orchestration_v2_projection_runs((run_id || ''), status) + `; + // Fresh-context starts are rare, so their partial index stays tiny. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_fresh_context_idx + ON orchestration_v2_projection_runs(thread_id, ordinal) + WHERE CASE WHEN json_valid(payload_json) + THEN json_extract(payload_json, '$.freshContext') = 1 + ELSE 0 END + `; + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_provider_turns_attempt_idx + ON orchestration_v2_projection_provider_turns(run_attempt_id) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 05f09f8e16850..2ebef1789f091 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -62,7 +62,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], - [83, "CommandReadIndexes"], + [83, "CommandWindowIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -168,7 +168,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], - [83, "CommandReadIndexes"], + [83, "CommandWindowIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From 2378fecc25334a4dd3852c96ab6bc4f447085cb3 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Wed, 11 Nov 2026 20:57:10 -0300 Subject: [PATCH 2/8] fix(server): a compacted thread keeps the baseline a model switch needs and the parent of a restarted wake The window skipped /compact prompts but still counted async-answer runs as the prompt a model switch is measured from; it now applies the same rules as modelSwitchSince. A steer or restart of a delegated-completion wake also keeps the run that wake belongs to, so replacing the wake still settles it. Co-Authored-By: Claude Sonnet 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 37 +++++--- .../orchestration-v2/ProjectionStore.test.ts | 16 ++-- .../src/orchestration-v2/ProjectionStore.ts | 21 +++-- .../SteeringCompletion.integration.test.ts | 85 ++++++++++++++++++- 4 files changed, 129 insertions(+), 30 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 909400da04084..032f4d78c4067 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6309,19 +6309,34 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio effects: Ref.Ref>, ) { // A compacted thread needs only its recent runs, plus any older run the command names. + const targetRunId = + command.dispatchMode.type === "steer_active" || command.dispatchMode.type === "restart_active" + ? command.dispatchMode.targetRunId + : undefined; + const keepRunIds = [ + command.manualContinuationOfRunId, + command.restartContinuationOfRunId, + command.usageLimitContinuationOfRunId, + command.delegatedCompletion?.parentRunId, + targetRunId, + command.steerFollowUpOfRunId, + ].flatMap((runId) => (runId === undefined ? [] : [runId])); let projection = yield* getProjectionWithPendingEvents(command.threadId, events, { - keepRunIds: [ - command.manualContinuationOfRunId, - command.restartContinuationOfRunId, - command.usageLimitContinuationOfRunId, - command.delegatedCompletion?.parentRunId, - command.steerFollowUpOfRunId, - ...(command.dispatchMode.type === "steer_active" || - command.dispatchMode.type === "restart_active" - ? [command.dispatchMode.targetRunId] - : []), - ].flatMap((runId) => (runId === undefined ? [] : [runId])), + keepRunIds, }); + // Replacing the target's input settles the wake it carried, which needs that wake's parent run. + const targetWakeParentRunId = projection.messages.find( + (message) => + message.id === projection.runs.find((run) => run.id === targetRunId)?.userMessageId, + )?.delegatedCompletion?.parentRunId; + if ( + targetWakeParentRunId !== undefined && + !projection.runs.some((run) => run.id === targetWakeParentRunId) + ) { + projection = yield* getProjectionWithPendingEvents(command.threadId, events, { + keepRunIds: [...keepRunIds, targetWakeParentRunId], + }); + } if (command.steerFollowUpOfRunId !== undefined) { const source = projection.runs.find((run) => run.id === command.steerFollowUpOfRunId); if ( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index ef06ecc8ce986..186f1a1f24b77 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -802,6 +802,9 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and const now = yield* DateTime.now; const nodeId = NodeId.make("node:compacted"); const runId = (ordinal: number) => RunId.make(`run:compacted:${ordinal}`); + // Run 7 answers an async question, which the timeline folds into its question. + const messageId = (ordinal: number) => + MessageId.make(ordinal === 7 ? "async-answer:compacted:7" : `message:compacted:${ordinal}`); const read = (keepRunIds: ReadonlyArray) => store.getThreadRecords(threadId, ["runs", "messages"], { messageRoles: ["user"], @@ -835,7 +838,7 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and providerInstanceId, modelSelection, providerThreadId: null, - userMessageId: MessageId.make(`message:compacted:${ordinal}`), + userMessageId: messageId(ordinal), rootNodeId: nodeId, activeAttemptId: null, status, @@ -854,7 +857,7 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and runId: runId(ordinal), occurredAt: now, payload: { - id: MessageId.make(`message:compacted:${ordinal}`), + id: messageId(ordinal), threadId, runId: runId(ordinal), nodeId, @@ -915,15 +918,12 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and // The window starts at run 5, the prompt before the compaction. const recent = yield* read([rolledBackId]); assert.deepEqual(ids(recent.runs), [rolledBackId, ...runIdsOf([2, 3, 5, 6, 7, 8, 9])]); - assert.deepEqual( - ids(recent.messages), - [2, 3, 5, 6, 7, 8, 9].map((ordinal) => `message:compacted:${ordinal}`), - ); + assert.deepEqual(ids(recent.messages), [2, 3, 5, 6, 7, 8, 9].map(messageId)); assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9])); - // Run 8 is a `/compact` run, so the model-switch baseline is the prompt before it, run 7. + // Runs 8 and 7 are not shown (`/compact`, async answer), so the baseline is run 5. yield* compact(9); - assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 7, 8, 9])); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9])); // The search for a prompt before the compaction may walk past an unreadable run. const legacyThreadId = yield* addRolledBackRecoveryCandidate("compacted-legacy"); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index d45327c2ffb3c..e7c77817b74bb 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3436,18 +3436,25 @@ export const layer: Layer.Layer = keepRunIds: ReadonlyArray, ) { // Starts at the last prompt before the compaction: a model switch is measured from the - // last run whose prompt the timeline shows, and `/compact` runs are not shown. + // last run `modelSwitchSince` counts, so `/compact` and async-answer runs are skipped + // unless a context handoff shows them. const [compaction] = yield* sql<{ readonly from_ordinal: number }>` SELECT COALESCE( (SELECT previous.ordinal FROM orchestration_v2_projection_runs AS previous WHERE previous.thread_id = ${threadId} AND previous.ordinal < latest.ordinal AND previous.status NOT IN ('queued', 'rolled_back') - AND CASE WHEN json_valid(previous.payload_json) THEN NOT EXISTS ( - SELECT 1 FROM orchestration_v2_projection_messages AS prompt - WHERE prompt.message_id = json_extract(previous.payload_json, '$.userMessageId') - AND prompt.role = 'user' - AND coalesce(json_array_length(prompt.payload_json, '$.attachments'), 0) = 0 - AND ${trimmedLowerText(sql.literal("prompt.payload_json"))} = ${IDLE_COMPACTION_TEXT} + AND CASE WHEN json_valid(previous.payload_json) THEN ( + json_extract(previous.payload_json, '$.contextHandoffId') IS NOT NULL + OR NOT ( + json_extract(previous.payload_json, '$.userMessageId') LIKE 'async-answer:%' + OR EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages AS prompt + WHERE prompt.message_id = json_extract(previous.payload_json, '$.userMessageId') + AND prompt.role = 'user' + AND coalesce(json_array_length(prompt.payload_json, '$.attachments'), 0) = 0 + AND ${trimmedLowerText(sql.literal("prompt.payload_json"))} = ${IDLE_COMPACTION_TEXT} + ) + ) ) ELSE 0 END ORDER BY previous.ordinal DESC LIMIT 1), diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 0c655b83f6b41..06aad0cd261ca 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1768,6 +1768,7 @@ const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( "events" | "started" | "startFirstTurn" >, siblingFinishesFirst: boolean, + userRunBeforeWake = false, ) { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; @@ -1869,6 +1870,24 @@ const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( })), }); yield* finishTurn(origin); + // The wake queues behind a user run that started after the origin run finished. + if (userRunBeforeWake) { + const userRunning = yield* watchTurnRunning(2); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:user-run"), + threadId, + messageId: MessageId.make("message:user-run"), + text: "unrelated question", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(userRunning); + } const completeTask = (taskId: NodeId) => orchestrator.dispatch({ @@ -1902,19 +1921,55 @@ const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( createdBy: "agent", creationSource: "server", }); - const wakeRunning = yield* watchTurnRunning(2); + const wakeOrdinal = userRunBeforeWake ? 3 : 2; + const wakeRunning = yield* watchTurnRunning(wakeOrdinal); + if (userRunBeforeWake) yield* finishTurn(harness.started[1]!); yield* orchestrator.resumeQueuedRuns; yield* worker.drain(); yield* Fiber.join(wakeRunning); - const wake = harness.started[1]!; + const wake = harness.started[wakeOrdinal - 1]!; assert.equal(wake.message.messageId, firstDelivery.messageId); + if (userRunBeforeWake) { + // The wake's run holds a compaction, so the window starts at the user run and drops the origin. + const compactedAt = yield* DateTime.now; + yield* sink.write({ + events: [ + { + id: EventId.make("event:wake-compaction"), + type: "turn-item.updated" as const, + threadId, + runId: wake.runId, + nodeId: wake.rootNodeId, + occurredAt: compactedAt, + payload: { + id: TurnItemId.make("item:wake-compaction"), + threadId, + runId: wake.runId, + nodeId: wake.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1_000, + status: "completed" as const, + title: null, + startedAt: compactedAt, + completedAt: compactedAt, + updatedAt: compactedAt, + type: "compaction" as const, + driver, + }, + }, + ], + }); + } if (siblingFinishesFirst) { yield* completeTask(siblingTaskId); } // The provider cannot steer natively, so the steer restarts the wake run. - const steeredRunning = yield* watchTurnRunning(3); + const steeredRunning = yield* watchTurnRunning(wakeOrdinal + 1); yield* orchestrator.dispatch({ type: "message.dispatch", commandId: CommandId.make("command:steer-wake"), @@ -1929,7 +1984,7 @@ const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( }); yield* worker.drain(); yield* Fiber.join(steeredRunning); - const steered = harness.started[2]!; + const steered = harness.started[wakeOrdinal]!; assert.equal(steered.runId, wake.runId); assert.equal( (yield* orchestrator.getThreadProjection(threadId)).attempts.find( @@ -2016,6 +2071,28 @@ it.effect("wakes the parent for a sibling result after a steer restarts the comp ), ); +it.effect("settles a steered completion wake whose parent run is before the last compaction", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* nextTurnSelectionHarness("steered-wake-after-compaction", true, false); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { threadId, steered, firstTaskId, finishTurn } = yield* restartCompletionWake( + harness, + false, + true, + ); + yield* finishTurn(steered); + const tasks = (yield* orchestrator.getThreadProjection(threadId)).subagents; + assert.equal( + tasks.find((task) => task.id === firstTaskId)?.completionDelivery?.state, + "delivered", + ); + }).pipe(Effect.provide(harness.layer)); + }), + ), +); + it.effect( "offers the next completion wake when a steer restarts a wake a sibling finished behind", () => From fc98972e777367813fea1e32802d956afa86ee49 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Wed, 11 Nov 2026 20:59:03 -0300 Subject: [PATCH 3/8] fix(server): a rolled-back run no longer sets where a compacted thread's read starts Rollback leaves the run's compaction item completed, so the window could start at a discarded compaction and drop surviving runs. The latest-compaction lookup now skips rolled-back runs, and the unread-prompt check gets a named span. Co-Authored-By: Claude Sonnet 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 17 ++++++++++++----- .../src/orchestration-v2/ProjectionStore.ts | 6 ++++-- 2 files changed, 16 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 186f1a1f24b77..98bc55822a604 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -812,6 +812,7 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and }); // Run 1 is the helper's rolled-back run. Run 2 is live and run 3 starts a fresh session. // Runs 6 and 9 are `/compact` runs that hold a compaction; run 8 is one that does not. + // Run 11 was rolled back after it compacted, so that compaction no longer counts. for (const [ordinal, status] of [ [2, "running"], [3, "completed"], @@ -821,6 +822,8 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and [7, "completed"], [8, "completed"], [9, "completed"], + [10, "completed"], + [11, "rolled_back"], ] as const) { yield* store.apply({ id: EventId.make(`event:compacted:run-${ordinal}`), @@ -904,7 +907,7 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and }, }); // Nothing is compacted yet, so the whole history is current. - assert.lengthOf((yield* read([])).runs, 9); + assert.lengthOf((yield* read([])).runs, 11); yield* compact(6); // Legacy rows can hold unreadable JSON; the fresh-context index must not reject them. @@ -917,13 +920,17 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and const runIdsOf = (ordinals: ReadonlyArray) => ordinals.map(runId); // The window starts at run 5, the prompt before the compaction. const recent = yield* read([rolledBackId]); - assert.deepEqual(ids(recent.runs), [rolledBackId, ...runIdsOf([2, 3, 5, 6, 7, 8, 9])]); - assert.deepEqual(ids(recent.messages), [2, 3, 5, 6, 7, 8, 9].map(messageId)); - assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9])); + assert.deepEqual(ids(recent.runs), [rolledBackId, ...runIdsOf([2, 3, 5, 6, 7, 8, 9, 10, 11])]); + assert.deepEqual(ids(recent.messages), [2, 3, 5, 6, 7, 8, 9, 10, 11].map(messageId).toSorted()); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9, 10, 11])); // Runs 8 and 7 are not shown (`/compact`, async answer), so the baseline is run 5. yield* compact(9); - assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9])); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9, 10, 11])); + + // The rolled-back run's later compaction is ignored. + yield* compact(11); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9, 10, 11])); // The search for a prompt before the compaction may walk past an unreadable run. const legacyThreadId = yield* addRolledBackRecoveryCandidate("compacted-legacy"); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index e7c77817b74bb..fd0f4c6325256 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3467,6 +3467,7 @@ export const layer: Layer.Layer = INNER JOIN orchestration_v2_projection_runs AS run ON run.run_id = item.run_id WHERE (item.thread_id || '') = ${threadId} AND (item.type || '') = 'compaction' AND item.status = 'completed' + AND run.status <> 'rolled_back' ORDER BY item.ordinal DESC LIMIT 1 ) AS latest @@ -6702,7 +6703,7 @@ export const layer: Layer.Layer = // Mirrors isNativeMaintenanceCommand over the thread's user messages. const hasOnlyMaintenanceUserMessages: ProjectionStoreV2Shape["hasOnlyMaintenanceUserMessages"] = - (threadId) => + Effect.fn("ProjectionStore.hasOnlyMaintenanceUserMessages")((threadId: ThreadId) => sql<{ readonly any_user: number; readonly other_user: number }>` SELECT EXISTS ( @@ -6720,7 +6721,8 @@ export const layer: Layer.Layer = `.pipe( Effect.map(([row]) => row?.any_user === 1 && row.other_user === 0), Effect.mapError(controlReadError(threadId)), - ); + ), + ); // Mirror idleCompactionOf, including JavaScript trim whitespace. const isIdleCompactionMessage = (alias: string) => { From 34038fd98d134f645cb706ce760d04d6ecdaa0b4 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Wed, 11 Nov 2026 21:14:39 -0300 Subject: [PATCH 4/8] fix(server): a restart chosen by delivery intent also keeps its wake's parent run The wake-parent run was kept only for a target the command named. A restart or steer resolved from the delivery intent targets the active run, so its wake's parent is now looked up after the intent resolves as well. Co-Authored-By: Claude Sonnet 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 27 ++++++----- .../SteeringCompletion.integration.test.ts | 48 +++++++++++-------- 2 files changed, 43 insertions(+), 32 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 032f4d78c4067..f6e7340d417e5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6324,19 +6324,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio let projection = yield* getProjectionWithPendingEvents(command.threadId, events, { keepRunIds, }); - // Replacing the target's input settles the wake it carried, which needs that wake's parent run. - const targetWakeParentRunId = projection.messages.find( - (message) => - message.id === projection.runs.find((run) => run.id === targetRunId)?.userMessageId, - )?.delegatedCompletion?.parentRunId; - if ( - targetWakeParentRunId !== undefined && - !projection.runs.some((run) => run.id === targetWakeParentRunId) - ) { - projection = yield* getProjectionWithPendingEvents(command.threadId, events, { - keepRunIds: [...keepRunIds, targetWakeParentRunId], - }); - } + // Replacing a run's input settles the wake it carried, which needs that wake's parent run. + const keepWakeParentOf = Effect.fnUntraced(function* (runId: RunId | undefined) { + const parentRunId = projection.messages.find( + (message) => message.id === projection.runs.find((run) => run.id === runId)?.userMessageId, + )?.delegatedCompletion?.parentRunId; + if (parentRunId === undefined || projection.runs.some((run) => run.id === parentRunId)) + return; + keepRunIds.push(parentRunId); + projection = yield* getProjectionWithPendingEvents(command.threadId, events, { keepRunIds }); + }); + yield* keepWakeParentOf(targetRunId); if (command.steerFollowUpOfRunId !== undefined) { const source = projection.runs.find((run) => run.id === command.steerFollowUpOfRunId); if ( @@ -6630,6 +6628,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.dispatchMode, command.deliveryIntent, ); + if (dispatchMode.type === "steer_active" || dispatchMode.type === "restart_active") { + yield* keepWakeParentOf(dispatchMode.targetRunId); + } if (dispatchMode.type === "steer_active" || dispatchMode.type === "restart_active") { const targetRunId = dispatchMode.targetRunId; const target = projection.runs.find((run) => run.id === targetRunId); diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 06aad0cd261ca..bb34ba5327693 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1769,6 +1769,7 @@ const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( >, siblingFinishesFirst: boolean, userRunBeforeWake = false, + intentNamesTarget = false, ) { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; @@ -1978,7 +1979,9 @@ const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( text: "also check the logs", attachments: [], modelSelection: runSelection, - dispatchMode: { type: "steer_active", targetRunId: wake.runId }, + ...(intentNamesTarget + ? { dispatchMode: { type: "start_immediately" as const }, deliveryIntent: "restart" as const } + : { dispatchMode: { type: "steer_active" as const, targetRunId: wake.runId } }), createdBy: "user", creationSource: "web", }); @@ -2071,26 +2074,33 @@ it.effect("wakes the parent for a sibling result after a steer restarts the comp ), ); -it.effect("settles a steered completion wake whose parent run is before the last compaction", () => - Effect.scoped( - Effect.gen(function* () { - const harness = yield* nextTurnSelectionHarness("steered-wake-after-compaction", true, false); - yield* Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const { threadId, steered, firstTaskId, finishTurn } = yield* restartCompletionWake( - harness, - false, +it.effect.each([false, true])( + "settles a steered completion wake whose parent run is before the last compaction (restart by intent: %s)", + (intentNamesTarget) => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* nextTurnSelectionHarness( + `steered-wake-after-compaction-${intentNamesTarget}`, true, + false, ); - yield* finishTurn(steered); - const tasks = (yield* orchestrator.getThreadProjection(threadId)).subagents; - assert.equal( - tasks.find((task) => task.id === firstTaskId)?.completionDelivery?.state, - "delivered", - ); - }).pipe(Effect.provide(harness.layer)); - }), - ), + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { threadId, steered, firstTaskId, finishTurn } = yield* restartCompletionWake( + harness, + false, + true, + intentNamesTarget, + ); + yield* finishTurn(steered); + const tasks = (yield* orchestrator.getThreadProjection(threadId)).subagents; + assert.equal( + tasks.find((task) => task.id === firstTaskId)?.completionDelivery?.state, + "delivered", + ); + }).pipe(Effect.provide(harness.layer)); + }), + ), ); it.effect( From d109133f8b9ea80db6ed4990e2fc561b02c240b9 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Thu, 12 Nov 2026 00:19:56 -0300 Subject: [PATCH 5/8] fix(server): a wake before a compaction is not the baseline for a model switch A notification or wake run is not shown as a prompt, so modelSwitchSince skips it; the window now skips it too when it looks for the last prompt before the compaction. Also adds store tests for the maintenance-only title predicate. Co-Authored-By: Claude Sonnet 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 77 +++++++++++++++++-- .../src/orchestration-v2/ProjectionStore.ts | 7 +- 2 files changed, 75 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 98bc55822a604..c2ff995d5e5f1 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -795,6 +795,52 @@ it.effect("idle compaction filters keep the latest run, live work and the newest }).pipe(Effect.provide(layerTest)), ); +it.effect.each([ + { name: "no user messages", messages: [], expected: false }, + { name: "assistant messages only", messages: [["assistant", "/compact"]], expected: false }, + { + name: "maintenance commands in any case and spacing", + messages: [ + ["user", " /Compact\n"], + ["user", "/LOGOUT"], + ], + expected: true, + }, + { + name: "a maintenance command followed by a prompt", + messages: [ + ["user", "/compact"], + ["user", "hello"], + ], + expected: false, + }, + { + name: "a maintenance command with an attachment", + messages: [["user", "/compact", [{ type: "image" }]]], + expected: false, + }, +] as const)( + "hasOnlyMaintenanceUserMessages is $expected for $name", + ({ messages, expected }, ctx) => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make(`thread:maintenance:${ctx.task.id}`); + yield* Effect.forEach(messages, ([role, text, attachments], index) => { + const messageId = `message:maintenance:${ctx.task.id}:${index}`; + return sql` + INSERT INTO orchestration_v2_projection_messages ( + message_id, thread_id, role, streaming, created_at, updated_at, payload_json + ) VALUES ( + ${messageId}, ${threadId}, ${role}, 0, '', '', + ${JSON.stringify({ text, attachments: attachments ?? [] })} + ) + `; + }); + assert.equal(yield* store.hasOnlyMaintenanceUserMessages(threadId), expected); + }).pipe(Effect.provide(layerTest)), +); + it.effect("a compacted thread reads its recent runs, live runs, fresh starts and named runs", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; @@ -802,7 +848,7 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and const now = yield* DateTime.now; const nodeId = NodeId.make("node:compacted"); const runId = (ordinal: number) => RunId.make(`run:compacted:${ordinal}`); - // Run 7 answers an async question, which the timeline folds into its question. + // Run 7 answers an async question, which the timeline folds into its question; run 5 is a wake. const messageId = (ordinal: number) => MessageId.make(ordinal === 7 ? "async-answer:compacted:7" : `message:compacted:${ordinal}`); const read = (keepRunIds: ReadonlyArray) => @@ -868,6 +914,15 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and creationSource: "web", role: "user", text: [6, 8, 9].includes(ordinal) ? "/compact" : `Prompt ${ordinal}`, + ...(ordinal === 5 + ? { + notification: { + source: { kind: "monitor" as const }, + outcome: "updated" as const, + summary: "Background notification", + }, + } + : {}), attachments: [], streaming: false, createdAt: now, @@ -918,19 +973,25 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and VALUES ('run:compacted:legacy', ${threadId}, 0, 'codex', 'completed', '', '{') `; const runIdsOf = (ordinals: ReadonlyArray) => ordinals.map(runId); - // The window starts at run 5, the prompt before the compaction. + // The window starts at run 4: run 5 is a wake, which the timeline does not show as a prompt. const recent = yield* read([rolledBackId]); - assert.deepEqual(ids(recent.runs), [rolledBackId, ...runIdsOf([2, 3, 5, 6, 7, 8, 9, 10, 11])]); - assert.deepEqual(ids(recent.messages), [2, 3, 5, 6, 7, 8, 9, 10, 11].map(messageId).toSorted()); - assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9, 10, 11])); + assert.deepEqual(ids(recent.runs), [ + rolledBackId, + ...runIdsOf([2, 3, 4, 5, 6, 7, 8, 9, 10, 11]), + ]); + assert.deepEqual( + ids(recent.messages), + [2, 3, 4, 5, 6, 7, 8, 9, 10, 11].map(messageId).toSorted(), + ); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 4, 5, 6, 7, 8, 9, 10, 11])); - // Runs 8 and 7 are not shown (`/compact`, async answer), so the baseline is run 5. + // Runs 8 and 7 are not shown (`/compact`, async answer), and run 5 is a wake, so the baseline stays run 4. yield* compact(9); - assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9, 10, 11])); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 4, 5, 6, 7, 8, 9, 10, 11])); // The rolled-back run's later compaction is ignored. yield* compact(11); - assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9, 10, 11])); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 4, 5, 6, 7, 8, 9, 10, 11])); // The search for a prompt before the compaction may walk past an unreadable run. const legacyThreadId = yield* addRolledBackRecoveryCandidate("compacted-legacy"); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index fd0f4c6325256..216ac9f3f8077 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3436,7 +3436,7 @@ export const layer: Layer.Layer = keepRunIds: ReadonlyArray, ) { // Starts at the last prompt before the compaction: a model switch is measured from the - // last run `modelSwitchSince` counts, so `/compact` and async-answer runs are skipped + // last run `modelSwitchSince` counts, so `/compact`, async-answer and notification runs are skipped // unless a context handoff shows them. const [compaction] = yield* sql<{ readonly from_ordinal: number }>` SELECT COALESCE( @@ -3447,6 +3447,11 @@ export const layer: Layer.Layer = json_extract(previous.payload_json, '$.contextHandoffId') IS NOT NULL OR NOT ( json_extract(previous.payload_json, '$.userMessageId') LIKE 'async-answer:%' + OR EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages AS wake + WHERE wake.message_id = json_extract(previous.payload_json, '$.userMessageId') + AND json_extract(wake.payload_json, '$.notification') IS NOT NULL + ) OR EXISTS ( SELECT 1 FROM orchestration_v2_projection_messages AS prompt WHERE prompt.message_id = json_extract(previous.payload_json, '$.userMessageId') From 87ac734106aec23a8edd6237ed55421bbe48a1e2 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Thu, 12 Nov 2026 07:53:13 -0300 Subject: [PATCH 6/8] perf(server): say why the wake-parent helper is untraced Co-Authored-By: Claude Sonnet 5.5 --- apps/server/src/orchestration-v2/Orchestrator.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index f6e7340d417e5..9a5810995c042 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6325,6 +6325,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio keepRunIds, }); // Replacing a run's input settles the wake it carried, which needs that wake's parent run. + // Untraced: the enclosing dispatchMessage span already covers both reloads. const keepWakeParentOf = Effect.fnUntraced(function* (runId: RunId | undefined) { const parentRunId = projection.messages.find( (message) => message.id === projection.runs.find((run) => run.id === runId)?.userMessageId, From c466357025be6e5fb3affc97c278c93e2c33d001 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 02:56:51 -0800 Subject: [PATCH 7/8] fix(server): a compacted thread's send still reads every run and every prompt The window dropped runs older than the compaction. The dispatch decision reads old runs in several ways the window did not keep: a held queue run that ended after the compaction, the worktree setup check, the idle-compaction busy check and the legacy-import check. On real data it also lost the model-switch divider when the run before the compaction was a delegated-result wake. Every run and every turn item the command asks for is still read. Attempts, nodes, messages and provider turns are read only for the runs from the compaction's run on, runs that had not ended when that run was requested, the runs the command names and the run that holds a re-sent message. An idle compaction and a provider switch read in full. This needs no migration, no new store method and no second read for a wake's parent run, so those are removed. Turn items and context transfers keep the plans main already gives them. Co-authored-by: Braulio Oliveira Co-Authored-By: Claude Fable 5.2 --- .../src/orchestration-v2/Orchestrator.ts | 89 ++-- .../orchestration-v2/ProjectionStore.test.ts | 403 +++++++++--------- .../src/orchestration-v2/ProjectionStore.ts | 318 +++++++------- .../ProviderTurnControlService.test.ts | 2 +- .../SteeringCompletion.integration.test.ts | 135 +----- apps/server/src/persistence/Migrations.ts | 4 +- .../Migrations/083_CommandWindowIndexes.ts | 31 -- .../reconcileV2PreviewMigration.test.ts | 4 +- 8 files changed, 400 insertions(+), 586 deletions(-) delete mode 100644 apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 9a5810995c042..bca66fff791f8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1530,7 +1530,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Runs in every command transaction; avoid reading thousands of finished leaf nodes. const readCommandProjection = ( threadId: ThreadId, - recent?: { keepRunIds: ReadonlyArray }, + recent?: ProjectionRecordFilter["sinceLastCompaction"], ) => projectionStore .getThreadRecords( @@ -1556,6 +1556,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Commands here read only undelivered or active tasks; a dispatcher thread keeps // thousands of delivered ones, each carrying its full prompt. omitSettledSubagents: true, + // Commands act on pending forks and merge-backs only; consumed ones stay unread. + contextTransferStatuses: ["pending"], ...(recent === undefined ? {} : { sinceLastCompaction: recent }), }, ) @@ -1645,7 +1647,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.fnUntraced(function* ( threadId: ThreadId, events: Ref.Ref>, - recent?: { keepRunIds: ReadonlyArray }, + recent?: ProjectionRecordFilter["sinceLastCompaction"], ) { const pending = (yield* Ref.get(events)).filter((event) => event.threadId === threadId); const stored = yield* Effect.option(readCommandProjection(threadId, recent)); @@ -6308,34 +6310,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: Ref.Ref>, effects: Ref.Ref>, ) { - // A compacted thread needs only its recent runs, plus any older run the command names. - const targetRunId = - command.dispatchMode.type === "steer_active" || command.dispatchMode.type === "restart_active" - ? command.dispatchMode.targetRunId - : undefined; - const keepRunIds = [ - command.manualContinuationOfRunId, - command.restartContinuationOfRunId, - command.usageLimitContinuationOfRunId, - command.delegatedCompletion?.parentRunId, - targetRunId, - command.steerFollowUpOfRunId, - ].flatMap((runId) => (runId === undefined ? [] : [runId])); - let projection = yield* getProjectionWithPendingEvents(command.threadId, events, { - keepRunIds, - }); - // Replacing a run's input settles the wake it carried, which needs that wake's parent run. - // Untraced: the enclosing dispatchMessage span already covers both reloads. - const keepWakeParentOf = Effect.fnUntraced(function* (runId: RunId | undefined) { - const parentRunId = projection.messages.find( - (message) => message.id === projection.runs.find((run) => run.id === runId)?.userMessageId, - )?.delegatedCompletion?.parentRunId; - if (parentRunId === undefined || projection.runs.some((run) => run.id === parentRunId)) - return; - keepRunIds.push(parentRunId); - projection = yield* getProjectionWithPendingEvents(command.threadId, events, { keepRunIds }); - }); - yield* keepWakeParentOf(targetRunId); + // A compacted thread reads every run and turn item, but attempts, nodes, messages and provider + // turns only for its recent runs and the runs this command names. An idle compaction reads + // in full: its busy check below looks for a delegated result's message under any old run. + let projection = yield* getProjectionWithPendingEvents( + command.threadId, + events, + command.idleCompaction === undefined + ? { + keepRunIds: [ + command.manualContinuationOfRunId, + command.restartContinuationOfRunId, + command.usageLimitContinuationOfRunId, + command.steerFollowUpOfRunId, + ...(command.dispatchMode.type === "steer_active" || + command.dispatchMode.type === "restart_active" + ? [command.dispatchMode.targetRunId] + : []), + ].flatMap((runId) => (runId === undefined ? [] : [runId])), + keepMessageId: command.messageId, + } + : undefined, + ); if (command.steerFollowUpOfRunId !== undefined) { const source = projection.runs.find((run) => run.id === command.steerFollowUpOfRunId); if ( @@ -6375,12 +6371,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const activeProviderInstanceId = projection.providerThreads.find( (candidate) => candidate.id === projection.thread.activeProviderThreadId, )?.providerInstanceId; - // A fresh session or another provider is handed the whole history, so it reads all of it. + // Another provider is handed what ran since it last saw the thread, which can be older than + // the compaction, so a provider switch reads the whole history. if ( - command.freshContext === true || - (activeProviderInstanceId !== undefined && - activeProviderInstanceId !== - (command.modelSelection ?? projection.thread.modelSelection).instanceId) + activeProviderInstanceId !== undefined && + activeProviderInstanceId !== + (command.modelSelection ?? projection.thread.modelSelection).instanceId ) { projection = yield* getProjectionWithPendingEvents(command.threadId, events); } @@ -6586,16 +6582,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); projection = applyToProjection(projection, event); } + const userMessages = projection.messages.filter((message) => message.role === "user"); + // The recent runs of a compacted thread can hold only a `/compact` while an older prompt + // exists. Every prompt's turn item is loaded; without one, every user message is checked. + const onlyMaintenanceHistory = [ + ...userMessages, + ...projection.turnItems.flatMap((item) => (item.type === "user_message" ? [item] : [])), + ].every(isNativeMaintenanceCommand) + ? projectionStore + .getThreadRecords(command.threadId, ["messages"], { messageRoles: ["user"] }) + .pipe( + Effect.map(({ messages }) => [...messages, ...userMessages]), + Effect.map((all) => all.length > 0 && all.every(isNativeMaintenanceCommand)), + mapDispatchError(command), + ) + : Effect.succeed(false); if ( !isNativeMaintenanceCommand(command) && ((command.titleSeed !== undefined && (yield* projectionStore .getMessageCount(command.threadId) .pipe(mapDispatchError(command))) === 0) || - (command.createdBy === "user" && - (yield* projectionStore - .hasOnlyMaintenanceUserMessages(command.threadId) - .pipe(mapDispatchError(command))))) + (command.createdBy === "user" && (yield* onlyMaintenanceHistory))) ) { const now = yield* DateTime.now; const thread: OrchestrationV2AppThread = { @@ -6629,9 +6637,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.dispatchMode, command.deliveryIntent, ); - if (dispatchMode.type === "steer_active" || dispatchMode.type === "restart_active") { - yield* keepWakeParentOf(dispatchMode.targetRunId); - } if (dispatchMode.type === "steer_active" || dispatchMode.type === "restart_active") { const targetRunId = dispatchMode.targetRunId; const target = projection.runs.find((run) => run.id === targetRunId); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index c2ff995d5e5f1..e1f723995797b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -795,82 +795,75 @@ it.effect("idle compaction filters keep the latest run, live work and the newest }).pipe(Effect.provide(layerTest)), ); -it.effect.each([ - { name: "no user messages", messages: [], expected: false }, - { name: "assistant messages only", messages: [["assistant", "/compact"]], expected: false }, - { - name: "maintenance commands in any case and spacing", - messages: [ - ["user", " /Compact\n"], - ["user", "/LOGOUT"], - ], - expected: true, - }, - { - name: "a maintenance command followed by a prompt", - messages: [ - ["user", "/compact"], - ["user", "hello"], - ], - expected: false, - }, - { - name: "a maintenance command with an attachment", - messages: [["user", "/compact", [{ type: "image" }]]], - expected: false, - }, -] as const)( - "hasOnlyMaintenanceUserMessages is $expected for $name", - ({ messages, expected }, ctx) => - Effect.gen(function* () { - const store = yield* ProjectionStore.ProjectionStoreV2; - const sql = yield* SqlClient.SqlClient; - const threadId = ThreadId.make(`thread:maintenance:${ctx.task.id}`); - yield* Effect.forEach(messages, ([role, text, attachments], index) => { - const messageId = `message:maintenance:${ctx.task.id}:${index}`; - return sql` - INSERT INTO orchestration_v2_projection_messages ( - message_id, thread_id, role, streaming, created_at, updated_at, payload_json - ) VALUES ( - ${messageId}, ${threadId}, ${role}, 0, '', '', - ${JSON.stringify({ text, attachments: attachments ?? [] })} - ) - `; - }); - assert.equal(yield* store.hasOnlyMaintenanceUserMessages(threadId), expected); - }).pipe(Effect.provide(layerTest)), -); - -it.effect("a compacted thread reads its recent runs, live runs, fresh starts and named runs", () => +// One thread holds every way a run can still matter after a compaction. A run is requested at +// its ordinal in minutes and ends 30 seconds later, unless the table says otherwise. +it.effect("a compacted thread reads every run, and run records only for its recent runs", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; const threadId = yield* addRolledBackRecoveryCandidate("compacted"); - const now = yield* DateTime.now; - const nodeId = NodeId.make("node:compacted"); + const created = yield* DateTime.now; + const at = (minutes: number) => DateTime.add(created, { minutes }); const runId = (ordinal: number) => RunId.make(`run:compacted:${ordinal}`); - // Run 7 answers an async question, which the timeline folds into its question; run 5 is a wake. - const messageId = (ordinal: number) => - MessageId.make(ordinal === 7 ? "async-answer:compacted:7" : `message:compacted:${ordinal}`); - const read = (keepRunIds: ReadonlyArray) => - store.getThreadRecords(threadId, ["runs", "messages"], { - messageRoles: ["user"], - sinceLastCompaction: { keepRunIds }, + const rootNodeId = (ordinal: number) => NodeId.make(`node:compacted:${ordinal}`); + const attemptId = (ordinal: number) => RunAttemptId.make(`attempt:compacted:${ordinal}`); + const providerThreadId = ProviderThreadId.make("provider-thread:compacted"); + const messageId = (ordinal: number) => MessageId.make(`message:compacted:${ordinal}`); + const item = ( + ordinal: number, + itemOrdinal: number, + status: "completed" | "failed" | "running", + ) => ({ + threadId, + runId: runId(ordinal), + nodeId: rootNodeId(ordinal), + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: itemOrdinal, + status, + title: null, + startedAt: at(ordinal), + completedAt: status === "running" ? null : at(ordinal), + updatedAt: at(ordinal), + }); + const compaction = (ordinal: number, status: "completed" | "failed" | "running") => + store.apply({ + id: EventId.make(`event:compacted:compaction-${ordinal}`), + type: "turn-item.updated", + threadId, + runId: runId(ordinal), + nodeId: rootNodeId(ordinal), + driver, + occurredAt: at(ordinal), + payload: { + ...item(ordinal, ordinal * 10 + 1, status), + id: TurnItemId.make(`item:compacted:compaction-${ordinal}`), + type: "compaction", + driver, + }, }); - // Run 1 is the helper's rolled-back run. Run 2 is live and run 3 starts a fresh session. - // Runs 6 and 9 are `/compact` runs that hold a compaction; run 8 is one that does not. - // Run 11 was rolled back after it compacted, so that compaction no longer counts. - for (const [ordinal, status] of [ - [2, "running"], + for (const [ordinal, status, endedAt] of [ + [2, "queued", null], [3, "completed"], - [4, "completed"], + // A held queue ran this old run after the compactions. + [4, "completed", at(11.5)], [5, "completed"], [6, "completed"], [7, "completed"], [8, "completed"], [9, "completed"], [10, "completed"], - [11, "rolled_back"], + [11, "completed"], + [12, "completed"], + [13, "rolled_back"], + [14, "completed"], + [15, "running", null], ] as const) { + const text = `Prompt ${ordinal}`; + const authored = { createdBy: "user", creationSource: "web" } as const; + const nodeId = rootNodeId(ordinal); + const ended = endedAt === undefined ? at(ordinal + 0.5) : endedAt; yield* store.apply({ id: EventId.make(`event:compacted:run-${ordinal}`), type: "run.created", @@ -879,7 +872,7 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and nodeId, driver, providerInstanceId, - occurredAt: now, + occurredAt: at(ordinal), payload: { id: runId(ordinal), threadId, @@ -889,14 +882,80 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and providerThreadId: null, userMessageId: messageId(ordinal), rootNodeId: nodeId, - activeAttemptId: null, + activeAttemptId: attemptId(ordinal), status, - requestedAt: now, - startedAt: now, - completedAt: status === "running" ? null : now, + requestedAt: at(ordinal), + startedAt: status === "queued" ? null : at(ordinal), + completedAt: ended, checkpointId: null, contextHandoffId: null, - ...(ordinal === 3 ? { freshContext: true } : {}), + }, + }); + yield* store.apply({ + id: EventId.make(`event:compacted:attempt-${ordinal}`), + type: "run-attempt.created", + threadId, + runId: runId(ordinal), + nodeId, + driver, + occurredAt: at(ordinal), + payload: { + id: attemptId(ordinal), + runId: runId(ordinal), + attemptOrdinal: 1, + rootNodeId: nodeId, + providerInstanceId, + providerThreadId, + providerTurnId: null, + reason: "initial", + status: ended === null ? "running" : "completed", + startedAt: at(ordinal), + completedAt: ended, + }, + }); + yield* store.apply({ + id: EventId.make(`event:compacted:node-${ordinal}`), + type: "node.updated", + threadId, + runId: runId(ordinal), + nodeId, + driver, + occurredAt: at(ordinal), + payload: { + id: nodeId, + threadId, + runId: runId(ordinal), + rootNodeId: nodeId, + parentNodeId: null, + kind: "root_turn", + status: ended === null ? "running" : "completed", + countsForRun: true, + providerThreadId, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: at(ordinal), + completedAt: ended, + }, + }); + yield* store.apply({ + id: EventId.make(`event:compacted:turn-${ordinal}`), + type: "provider-turn.updated", + threadId, + nodeId, + driver, + occurredAt: at(ordinal), + payload: { + id: ProviderTurnId.make(`turn:compacted:${ordinal}`), + providerThreadId, + nodeId, + runAttemptId: attemptId(ordinal), + nativeTurnRef: null, + ordinal, + status: ended === null ? "running" : "completed", + startedAt: at(ordinal), + completedAt: ended, }, }); yield* store.apply({ @@ -904,162 +963,102 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and type: "message.updated", threadId, runId: runId(ordinal), - occurredAt: now, + occurredAt: at(ordinal), payload: { + ...authored, id: messageId(ordinal), threadId, runId: runId(ordinal), nodeId, - createdBy: "user", - creationSource: "web", role: "user", - text: [6, 8, 9].includes(ordinal) ? "/compact" : `Prompt ${ordinal}`, - ...(ordinal === 5 - ? { - notification: { - source: { kind: "monitor" as const }, - outcome: "updated" as const, - summary: "Background notification", - }, - } - : {}), + text, attachments: [], streaming: false, - createdAt: now, - updatedAt: now, + createdAt: at(ordinal), + updatedAt: at(ordinal), }, }); - } - const rolledBackId = RunId.make("run:compacted:rolled-back"); - const ids = (rows: ReadonlyArray) => - rows.map((row) => row.id); - const compact = (ordinal: number) => - store.apply({ - id: EventId.make(`event:compacted:compaction-${ordinal}`), + yield* store.apply({ + id: EventId.make(`event:compacted:prompt-${ordinal}`), type: "turn-item.updated", threadId, runId: runId(ordinal), nodeId, driver, - occurredAt: now, + occurredAt: at(ordinal), payload: { - id: TurnItemId.make(`item:compacted:compaction-${ordinal}`), - threadId, - runId: runId(ordinal), - nodeId, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal, - status: "completed", - title: null, - startedAt: now, - completedAt: now, - updatedAt: now, - type: "compaction", - driver, + ...authored, + ...item(ordinal, ordinal * 10, status === "queued" ? "running" : "completed"), + ...(status === "queued" ? { status: "pending" as const } : {}), + id: TurnItemId.make(`item:compacted:prompt-${ordinal}`), + type: "user_message", + messageId: messageId(ordinal), + inputIntent: status === "queued" ? "queued_turn" : "turn_start", + text, + attachments: [], }, }); + } + // Each collection as the ordinals of the runs it was read for. Run 1 is the helper's. + const ordinals = (ids: ReadonlyArray) => + [...new Set(ids.map((id) => Number(id?.split(":").at(-1)) || 1))].toSorted((a, b) => a - b); + const read = (keepRunIds: ReadonlyArray = [], keepMessageId = messageId(0)) => + store + .getThreadRecords( + threadId, + ["runs", "attempts", "nodes", "providerTurns", "messages", "turnItems"], + { + messageRoles: ["user"], + omitSettledLeafNodes: true, + sinceLastCompaction: { keepRunIds, keepMessageId }, + }, + ) + .pipe( + Effect.map((records) => ({ + runs: records.runs.length, + attempts: ordinals(records.attempts.map((attempt) => attempt.runId)), + nodes: ordinals(records.nodes.map((node) => node.runId)), + providerTurns: ordinals(records.providerTurns.map((turn) => turn.runAttemptId)), + messages: ordinals(records.messages.map((message) => message.runId)), + turnItems: ordinals(records.turnItems.map((row) => row.runId)), + })), + ); + const all = [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]; + // Every run and every turn item is read, so a model switch is still measured from the last + // prompt. Run 1, the helper's, holds only a command item. + const expected = (recent: ReadonlyArray) => ({ + runs: 15, + attempts: recent, + nodes: recent, + providerTurns: recent, + messages: recent, + turnItems: [1, ...all], + }); + // Nothing is compacted yet, so the whole history is current. - assert.lengthOf((yield* read([])).runs, 11); + assert.deepEqual(yield* read(), expected(all)); - yield* compact(6); - // Legacy rows can hold unreadable JSON; the fresh-context index must not reject them. - const sql = yield* SqlClient.SqlClient; - yield* sql` - INSERT INTO orchestration_v2_projection_runs - (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) - VALUES ('run:compacted:legacy', ${threadId}, 0, 'codex', 'completed', '', '{') - `; - const runIdsOf = (ordinals: ReadonlyArray) => ordinals.map(runId); - // The window starts at run 4: run 5 is a wake, which the timeline does not show as a prompt. - const recent = yield* read([rolledBackId]); - assert.deepEqual(ids(recent.runs), [ - rolledBackId, - ...runIdsOf([2, 3, 4, 5, 6, 7, 8, 9, 10, 11]), - ]); - assert.deepEqual( - ids(recent.messages), - [2, 3, 4, 5, 6, 7, 8, 9, 10, 11].map(messageId).toSorted(), - ); - assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 4, 5, 6, 7, 8, 9, 10, 11])); + // The window starts at the compaction's run. Queued run 2 and held run 4 are older but had + // not ended by then. + yield* compaction(7, "completed"); + assert.deepEqual(yield* read(), expected([2, 4, 7, 8, 9, 10, 11, 12, 13, 14, 15])); - // Runs 8 and 7 are not shown (`/compact`, async answer), and run 5 is a wake, so the baseline stays run 4. - yield* compact(9); - assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 4, 5, 6, 7, 8, 9, 10, 11])); + // The latest compaction wins. + yield* compaction(11, "completed"); + const latest = expected([2, 4, 11, 12, 13, 14, 15]); + assert.deepEqual(yield* read(), latest); - // The rolled-back run's later compaction is ignored. - yield* compact(11); - assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 4, 5, 6, 7, 8, 9, 10, 11])); + // A compaction that was rolled back, failed or is still running does not move the window. + yield* compaction(13, "completed"); + yield* compaction(14, "failed"); + yield* compaction(15, "running"); + assert.deepEqual(yield* read(), latest); - // The search for a prompt before the compaction may walk past an unreadable run. - const legacyThreadId = yield* addRolledBackRecoveryCandidate("compacted-legacy"); - const legacyNodeId = NodeId.make("node:compacted-legacy"); - const compactionRunId = RunId.make("run:compacted-legacy:2"); - yield* sql` - INSERT INTO orchestration_v2_projection_runs - (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) - VALUES ('run:compacted-legacy:0', ${legacyThreadId}, 0, 'codex', 'completed', '', '{') - `; - yield* store.apply({ - id: EventId.make("event:compacted-legacy:run"), - type: "run.created", - threadId: legacyThreadId, - runId: compactionRunId, - nodeId: legacyNodeId, - driver, - providerInstanceId, - occurredAt: now, - payload: { - id: compactionRunId, - threadId: legacyThreadId, - ordinal: 2, - providerInstanceId, - modelSelection, - providerThreadId: null, - userMessageId: MessageId.make("message:compacted-legacy:2"), - rootNodeId: legacyNodeId, - activeAttemptId: null, - status: "completed", - requestedAt: now, - startedAt: now, - completedAt: now, - checkpointId: null, - contextHandoffId: null, - }, - }); - yield* store.apply({ - id: EventId.make("event:compacted-legacy:compaction"), - type: "turn-item.updated", - threadId: legacyThreadId, - runId: compactionRunId, - nodeId: legacyNodeId, - driver, - occurredAt: now, - payload: { - id: TurnItemId.make("item:compacted-legacy:compaction"), - threadId: legacyThreadId, - runId: compactionRunId, - nodeId: legacyNodeId, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 2, - status: "completed", - title: null, - startedAt: now, - completedAt: now, - updatedAt: now, - type: "compaction", - driver, - }, - }); - const legacy = yield* store.getThreadRecords(legacyThreadId, ["runs"], { - sinceLastCompaction: { keepRunIds: [] }, - }); - assert.deepEqual(ids(legacy.runs), [compactionRunId]); + // A named run and the run that holds a re-sent message are read too. + assert.deepEqual( + yield* read([runId(3)], messageId(6)), + expected([2, 3, 4, 6, 11, 12, 13, 14, 15]), + ); }).pipe(Effect.provide(layerTest)), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 216ac9f3f8077..9d5b621030441 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -373,12 +373,15 @@ export interface ProjectionRecordFilter { */ readonly omitSettledSubagents?: true | { readonly except: ReadonlyArray }; /** - * Reads what a new message needs from a thread that was compacted: the runs from the last - * prompt before the latest completed compaction, plus live runs, fresh-context runs and `keepRunIds`. Run- - * scoped records follow their runs, provider threads are the thread's own, and context transfers - * are the open ones that target it. A thread without a completed compaction reads in full. + * For a compacted thread: reads attempts, nodes, messages and provider turns only for runs from + * the latest completed compaction's run on, runs that had not ended when that run was + * requested, and the runs named here. Runs and turn items are not narrowed. */ - readonly sinceLastCompaction?: { readonly keepRunIds: ReadonlyArray }; + readonly sinceLastCompaction?: { + readonly keepRunIds: ReadonlyArray; + /** Keeps the run that already holds this message, for a message that is sent again. */ + readonly keepMessageId: MessageId; + }; } // A fork inherits its source's history before it has messages of its own, @@ -479,10 +482,11 @@ export interface ProjectionStoreV2Shape { options: ProjectionTimelinePageOptions, ) => Effect.Effect; readonly getMessageCount: (threadId: ThreadId) => Effect.Effect; - /** True when the thread has user messages and every one is a native maintenance command. */ - readonly hasOnlyMaintenanceUserMessages: ( + /** Bytes of tool output a transcript export is sure to hold, or 0 when it can fit in `limit`. */ + readonly getTranscriptByteFloor: ( threadId: ThreadId, - ) => Effect.Effect; + limit: number, + ) => Effect.Effect; readonly getSubagentCount: (threadId: ThreadId) => Effect.Effect; /** Child threads of the thread's app-owned tasks, in task order, without decoding the tasks. */ readonly listAppOwnedChildThreadIds: ( @@ -3423,79 +3427,48 @@ export const layer: Layer.Layer = const threadColumn = (ids: ReadonlyArray | undefined, column = "thread_id") => sql.literal(ids === undefined ? column : `+${column}`); - // Mirrors JavaScript's `trim().toLowerCase()` of a message's text. - const trimmedLowerText = (payload: Statement.Fragment) => sql`lower(trim( - json_extract(${payload}, '$.text'), - char(9, 10, 11, 12, 13, 32, 160, 5760, 8192, 8193, 8194, 8195, - 8196, 8197, 8198, 8199, 8200, 8201, 8202, 8232, 8233, 8239, 8287, 12288, 65279) - ))`; - - /** Null when nothing was compacted, so the whole history is current. */ - const runsSinceLastCompaction = Effect.fn("ProjectionStore.runsSinceLastCompaction")(function* ( - threadId: ThreadId, - keepRunIds: ReadonlyArray, - ) { - // Starts at the last prompt before the compaction: a model switch is measured from the - // last run `modelSwitchSince` counts, so `/compact`, async-answer and notification runs are skipped - // unless a context handoff shows them. - const [compaction] = yield* sql<{ readonly from_ordinal: number }>` - SELECT COALESCE( - (SELECT previous.ordinal FROM orchestration_v2_projection_runs AS previous - WHERE previous.thread_id = ${threadId} AND previous.ordinal < latest.ordinal - AND previous.status NOT IN ('queued', 'rolled_back') - AND CASE WHEN json_valid(previous.payload_json) THEN ( - json_extract(previous.payload_json, '$.contextHandoffId') IS NOT NULL - OR NOT ( - json_extract(previous.payload_json, '$.userMessageId') LIKE 'async-answer:%' - OR EXISTS ( - SELECT 1 FROM orchestration_v2_projection_messages AS wake - WHERE wake.message_id = json_extract(previous.payload_json, '$.userMessageId') - AND json_extract(wake.payload_json, '$.notification') IS NOT NULL - ) - OR EXISTS ( - SELECT 1 FROM orchestration_v2_projection_messages AS prompt - WHERE prompt.message_id = json_extract(previous.payload_json, '$.userMessageId') - AND prompt.role = 'user' - AND coalesce(json_array_length(prompt.payload_json, '$.attachments'), 0) = 0 - AND ${trimmedLowerText(sql.literal("prompt.payload_json"))} = ${IDLE_COMPACTION_TEXT} - ) - ) - ) ELSE 0 END - ORDER BY previous.ordinal DESC - LIMIT 1), - latest.ordinal - ) AS from_ordinal - FROM ( - SELECT run.ordinal + /** The runs whose records a new message reads. Null when nothing was compacted. */ + const runIdsSinceLastCompaction = Effect.fn("ProjectionStore.runIdsSinceLastCompaction")( + function* ( + threadId: ThreadId, + keep: NonNullable, + ) { + // A held queue can run out of ordinal order, so a run that ended after the compaction's + // run was requested is recent too. Every run row is read anyway, so this scan reads no + // extra pages. + const recent = yield* sql<{ readonly run_id: RunId }>` + WITH compaction AS ( + SELECT run.ordinal, run.requested_at FROM orchestration_v2_projection_turn_items AS item - INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx + INDEXED BY orchestration_v2_projection_turn_items_conversation_idx INNER JOIN orchestration_v2_projection_runs AS run ON run.run_id = item.run_id - WHERE (item.thread_id || '') = ${threadId} - AND (item.type || '') = 'compaction' AND item.status = 'completed' + WHERE item.thread_id = ${threadId} + AND item.type = 'compaction' AND item.status = 'completed' + AND item.type NOT IN ${CONVERSATION_INDEX_EXCLUDED_TYPES} AND run.status <> 'rolled_back' ORDER BY item.ordinal DESC LIMIT 1 - ) AS latest + ) + SELECT run.run_id + FROM compaction + INNER JOIN orchestration_v2_projection_runs AS run ON run.thread_id = ${threadId} + WHERE run.ordinal >= compaction.ordinal + OR run.completed_at IS NULL + OR run.completed_at >= compaction.requested_at + OR run.status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + UNION + SELECT message.run_id + FROM compaction + INNER JOIN orchestration_v2_projection_messages AS message + ON message.message_id = ${keep.keepMessageId} + WHERE message.thread_id = ${threadId} AND message.run_id IS NOT NULL `; - return compaction === undefined - ? null - : sql` - SELECT run_id FROM orchestration_v2_projection_runs - WHERE thread_id = ${threadId} AND ordinal >= ${compaction.from_ordinal} - UNION - SELECT run_id FROM orchestration_v2_projection_runs - INDEXED BY orchestration_v2_projection_runs_thread_status_idx - WHERE thread_id = ${threadId} - AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') - UNION - SELECT run_id FROM orchestration_v2_projection_runs - INDEXED BY orchestration_v2_projection_runs_fresh_context_idx - WHERE thread_id = ${threadId} AND CASE WHEN json_valid(payload_json) - THEN json_extract(payload_json, '$.freshContext') = 1 ELSE 0 END - UNION - SELECT value FROM json_each(${encodeIdList(keepRunIds)}) - `; - }); + // The compaction's own run is always recent, so no rows means no compaction. + return recent.length === 0 + ? null + : encodeIdList([...recent.map((row) => row.run_id), ...keep.keepRunIds]); + }, + ); const readCanonicalProjection = Effect.fn("ProjectionStore.readCanonicalProjection")( function* ( @@ -3514,10 +3487,19 @@ export const layer: Layer.Layer = if (!threadRow) { return yield* new ProjectionStoreThreadNotFoundError({ threadId }); } - const recentRuns = + const recentRunIds = window === undefined && filter?.sinceLastCompaction !== undefined - ? yield* runsSinceLastCompaction(threadId, filter.sinceLastCompaction.keepRunIds) + ? yield* runIdsSinceLastCompaction(threadId, filter.sinceLastCompaction) : null; + const inRecentRuns = (column: string) => + recentRunIds === null + ? sql`` + : sql`AND ${sql.literal(column)} IN (SELECT value FROM json_each(${recentRunIds}))`; + // Without the hint SQLite reads every row of the thread to find the recent runs' rows. + const recentRunIndex = (index: string) => + recentRunIds === null + ? sql`` + : sql`INDEXED BY ${sql.literal(`orchestration_v2_projection_${index}`)}`; type WindowPayloadRow = PayloadRow & { readonly work_last_id?: string | null; @@ -3539,13 +3521,18 @@ export const layer: Layer.Layer = ? yield* sql` SELECT payload_json FROM orchestration_v2_projection_turn_items - ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx`} - WHERE thread_id = ${threadId} + ${ + // A type list of conversation items reads them from the conversation index. + // Any other plan fetches every tool call of the thread to check its type. + readsConversationItemsOnly(filter) + ? sql`INDEXED BY orchestration_v2_projection_turn_items_conversation_idx + WHERE thread_id = ${threadId} AND type NOT IN ${CONVERSATION_INDEX_EXCLUDED_TYPES}` + : sql`WHERE thread_id = ${threadId}` + } ${filter?.turnItemRequestIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.requestId') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRequestIds)}))`} ${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`} - ${recentRuns === null ? sql`` : sql`AND run_id IN (${recentRuns})`} ${filter?.turnItemNodeIds === undefined ? sql`` : sql`AND node_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemNodeIds)}))`} ${filter?.turnItemRunIds === undefined ? sql`` : sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR (${filter.turnItemRunIds.includes(null) ? 1 : 0} = 1 AND run_id IS NULL))`} ORDER BY ordinal ASC, turn_item_id ASC @@ -3736,14 +3723,7 @@ export const layer: Layer.Layer = : window === undefined ? sql` SELECT payload_json - FROM ${ - // Looking runs up by key beats filtering the thread's whole run index. - recentRuns === null - ? sql`orchestration_v2_projection_runs` - : sql`(SELECT run_id AS recent_run_id FROM (${recentRuns})) AS recent - CROSS JOIN orchestration_v2_projection_runs - ON orchestration_v2_projection_runs.run_id = recent.recent_run_id` - } + FROM orchestration_v2_projection_runs WHERE ${threadColumn(filter?.runIds)} = ${threadId} ${filter?.runIds === undefined ? sql`` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.runIds)}))`} ${ @@ -3797,7 +3777,7 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_run_attempts WHERE thread_id = ${threadId} - ${recentRuns === null ? sql`` : sql`AND run_id IN (${recentRuns})`} + ${inRecentRuns("run_id")} ORDER BY run_id ASC, attempt_ordinal ASC ` : sql` @@ -3814,26 +3794,32 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_nodes AS node - ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_nodes_thread_run_idx`} - WHERE ${recentRuns === null ? threadColumn(filter?.nodeIds, "node.thread_id") : sql.literal("node.thread_id")} = ${threadId} - ${recentRuns === null ? sql`` : sql`AND node.run_id IN (${recentRuns})`} - ${filter?.nodeIds === undefined ? sql`` : sql`AND node.node_id IN (SELECT value FROM json_each(${encodeIdList(filter.nodeIds)}))`} - ${ - filter?.omitSettledLeafNodes === true - ? sql`AND NOT ( - node.kind IN ('tool_call', 'assistant_message', 'reasoning') - AND node.status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') - AND node.run_id IS NOT NULL - AND NOT EXISTS ( - SELECT 1 FROM orchestration_v2_projection_runs AS run - INDEXED BY orchestration_v2_projection_runs_id_status_idx - WHERE (run.run_id || '') = (node.run_id || '') - AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') + WHERE ${ + filter?.omitSettledLeafNodes === true + ? // Seeks the kept nodes. One filter over the thread reads every settled leaf to + // drop it. The first predicate must match migration 083's index expression. A + // node's run is always a run of its own thread, and an active run is recent. + sql`node.rowid IN ( + SELECT rowid FROM orchestration_v2_projection_nodes + ${recentRunIndex("nodes_thread_run_idx")} + WHERE thread_id = ${threadId} + AND (kind IN ('tool_call', 'assistant_message', 'reasoning') + AND status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') + AND run_id IS NOT NULL) = 0 + ${inRecentRuns("run_id")} + UNION ALL + SELECT rowid FROM orchestration_v2_projection_nodes + WHERE thread_id = ${threadId} + AND run_id IN ( + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} + AND status IN ('preparing', 'queued', 'starting', 'running', 'waiting') ) AND kind IN ('tool_call', 'assistant_message', 'reasoning') AND status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') )` - : sql`${threadColumn(filter?.nodeIds, "node.thread_id")} = ${threadId}` + : sql`${threadColumn(filter?.nodeIds, "node.thread_id")} = ${threadId} + ${inRecentRuns("node.run_id")}` } ${filter?.nodeIds === undefined ? sql`` : sql`AND node.node_id IN (SELECT value FROM json_each(${encodeIdList(filter.nodeIds)}))`} ORDER BY COALESCE(node.started_at, ''), node.node_id ASC @@ -3859,33 +3845,26 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_subagents AS subagent - ${ - filter?.omitSettledSubagents === undefined - ? sql`` - : // Its columns decide settled rows, so only the open ones read their payload. - sql`INDEXED BY orchestration_v2_projection_subagents_settled_idx` - } - WHERE ${filter?.omitSettledSubagents === undefined ? threadColumn(filter?.subagentIds) : sql.literal("thread_id")} = ${threadId} + WHERE ${threadColumn(filter?.subagentIds)} = ${threadId} ${filter?.subagentIds === undefined ? sql`` : sql`AND subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentIds)}))`} ${filter?.subagentRunIds === undefined ? sql`` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentRunIds)}))`} ${ - filter?.omitSettledSubagents !== undefined - ? sql`AND (${ - filter.omitSettledSubagents === true - ? sql`0` - : sql`subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.omitSettledSubagents.except)}))` - } OR NOT ( - subagent.status IN ('completed', 'failed', 'cancelled', 'interrupted') - AND COALESCE(subagent.completion_delivery_state, '') - IN ('acknowledged', 'delivered', 'disposed') - AND NOT EXISTS ( - SELECT 1 FROM orchestration_v2_projection_runs AS run - INDEXED BY orchestration_v2_projection_runs_id_status_idx - WHERE (run.run_id || '') = (subagent.run_id || '') - AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') - ) - ))` - : sql`` + filter?.omitSettledSubagents === true + ? // The settled index holds every column this check reads, so the settled + // tasks are dropped there and their rows, each with a full prompt, stay unread. + sql`AND subagent.rowid IN ( + SELECT candidate.rowid + FROM orchestration_v2_projection_subagents AS candidate + INDEXED BY orchestration_v2_projection_subagents_settled_idx + WHERE candidate.thread_id = ${threadId} + AND NOT ${settledSubagent(sql`candidate`)} + )` + : filter?.omitSettledSubagents !== undefined + ? sql`AND ( + subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.omitSettledSubagents.except)})) + OR NOT ${settledSubagent(sql`subagent`)} + )` + : sql`` } ORDER BY COALESCE(started_at, ''), subagent_id ASC ` @@ -3928,17 +3907,13 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_provider_threads WHERE (thread_id = ${threadId} - ${ - recentRuns === null - ? sql`OR owner_node_id IN (${nodesOwningProviderThreads(threadId)}) + OR owner_node_id IN (${nodesOwningProviderThreads(threadId)}) OR provider_thread_id IN ( SELECT provider_thread_id FROM orchestration_v2_projection_subagents WHERE thread_id = ${threadId} AND provider_thread_id IS NOT NULL - )` - : sql`` - }) + )) ${filter?.providerThreadIds === undefined ? sql`` : sql`AND provider_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerThreadIds)}))`} ORDER BY COALESCE(first_run_ordinal, 0), provider_thread_id ASC ` @@ -3959,14 +3934,13 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_provider_turns - ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_provider_turns_attempt_idx`} WHERE thread_id = ${threadId} ${ - recentRuns === null + recentRunIds === null ? sql`` : sql`AND run_attempt_id IN ( SELECT attempt_id FROM orchestration_v2_projection_run_attempts - WHERE thread_id = ${threadId} AND run_id IN (${recentRuns}))` + WHERE thread_id = ${threadId} ${inRecentRuns("run_id")})` } ${filter?.providerTurnAttemptIds === undefined ? sql`` : sql`AND run_attempt_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnAttemptIds)}))`} ${filter?.providerTurnThreadIds === undefined ? sql`` : sql`AND provider_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnThreadIds)}))`} @@ -4013,9 +3987,9 @@ export const layer: Layer.Layer = : window === undefined ? sql` SELECT payload_json FROM orchestration_v2_projection_messages - ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_messages_run_idx`} - WHERE ${recentRuns === null ? threadColumn(filter?.messageIds) : sql.literal("thread_id")} = ${threadId} - ${recentRuns === null ? sql`` : sql`AND run_id IN (${recentRuns})`} + ${recentRunIndex("messages_run_idx")} + WHERE ${recentRunIds === null ? threadColumn(filter?.messageIds) : sql.literal("thread_id")} = ${threadId} + ${inRecentRuns("run_id")} ${filter?.messageIds === undefined ? sql`` : sql`AND message_id IN (SELECT value FROM json_each(${encodeIdList(filter.messageIds)}))`} ${filter?.messageRoles === undefined ? sql`` : sql`AND role IN (SELECT value FROM json_each(${encodeIdList(filter.messageRoles)}))`} ${filter?.monitorNotificationsOnly === true ? sql`AND json_extract(payload_json, '$.notification.source.kind') = 'monitor'` : sql``} @@ -4134,11 +4108,20 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_context_transfers WHERE ${ - recentRuns === null + filter?.contextTransferStatuses === undefined ? sql`(source_thread_id = ${threadId} OR target_thread_id = ${threadId})` - : sql`target_thread_id = ${threadId} AND status IN ('pending', 'running', 'waiting')` + : // Each side seeks its (thread, status) index. One OR over both sides checks the + // status on every row the thread ever sent or received. + sql`rowid IN ( + SELECT rowid FROM orchestration_v2_projection_context_transfers + WHERE source_thread_id = ${threadId} + AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)})) + UNION + SELECT rowid FROM orchestration_v2_projection_context_transfers + WHERE target_thread_id = ${threadId} + AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)})) + )` } - ${filter?.contextTransferStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)}))`} ORDER BY rowid ASC ` : sql` @@ -4184,12 +4167,23 @@ export const layer: Layer.Layer = decodeRows(decodeContextTransferPayload, threadId)(contextTransferRows), ]); const messageOrdinals = new Map(); - // The recent user turns, loaded above, already order their own messages. + // The literal `type = 'user_message'` lets SQLite pick the partial user-message index. + const userOnly = messages.every((message) => message.role === "user"); + // The loaded items already order the messages when they hold every item the lookup reads. + const loadsEveryItemOf = (type: OrchestrationV2TurnItem["type"]) => + window === undefined && + (fields === undefined || fields.includes("turnItems")) && + filter?.turnItemRequestIds === undefined && + filter?.turnItemStatuses === undefined && + filter?.turnItemRunId === undefined && + filter?.turnItemRunIds === undefined && + filter?.turnItemNodeIds === undefined && + (filter?.turnItemTypes === undefined || filter.turnItemTypes.includes(type)); if ( - recentRuns === null && fields !== undefined && fields.includes("messages") && - messages.length > 1 + messages.length > 1 && + !(loadsEveryItemOf("user_message") && (userOnly || loadsEveryItemOf("assistant_message"))) ) { // Broad role reads keep the original query to avoid per-ID lookup overhead. const loadedMessageIds = @@ -6706,35 +6700,15 @@ export const layer: Layer.Layer = ), ); - // Mirrors isNativeMaintenanceCommand over the thread's user messages. - const hasOnlyMaintenanceUserMessages: ProjectionStoreV2Shape["hasOnlyMaintenanceUserMessages"] = - Effect.fn("ProjectionStore.hasOnlyMaintenanceUserMessages")((threadId: ThreadId) => - sql<{ readonly any_user: number; readonly other_user: number }>` - SELECT - EXISTS ( - SELECT 1 FROM orchestration_v2_projection_messages - WHERE thread_id = ${threadId} AND role = 'user' - ) AS any_user, - EXISTS ( - SELECT 1 FROM orchestration_v2_projection_messages - WHERE thread_id = ${threadId} AND role = 'user' - AND NOT ( - coalesce(json_array_length(payload_json, '$.attachments'), 0) = 0 - AND ${trimmedLowerText(sql.literal("payload_json"))} IN ('/compact', '/logout') - ) - ) AS other_user - `.pipe( - Effect.map(([row]) => row?.any_user === 1 && row.other_user === 0), - Effect.mapError(controlReadError(threadId)), - ), - ); - // Mirror idleCompactionOf, including JavaScript trim whitespace. const isIdleCompactionMessage = (alias: string) => { const payload = sql.literal(`${alias}.payload_json`); return sql`( ${sql.literal(`${alias}.role`)} = 'user' - AND ${trimmedLowerText(payload)} = ${IDLE_COMPACTION_TEXT} + AND lower(trim(json_extract(${payload}, '$.text'), + char(9, 10, 11, 12, 13, 32, 160, 5760, 8192, 8193, 8194, 8195, + 8196, 8197, 8198, 8199, 8200, 8201, 8202, 8232, 8233, 8239, 8287, 12288, 65279) + )) = ${IDLE_COMPACTION_TEXT} AND coalesce(json_array_length(${payload}, '$.attachments'), 0) = 0 AND ( json_extract(${payload}, '$.idleCompaction') IS NOT NULL @@ -8143,7 +8117,7 @@ export const layer: Layer.Layer = hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest), isNodeStored: readOnReader(isNodeStored), getMessageCount: readOnReader(getMessageCount), - hasOnlyMaintenanceUserMessages: readOnReader(hasOnlyMaintenanceUserMessages), + getTranscriptByteFloor: readOnReader(getTranscriptByteFloor), getSubagentCount: readOnReader(getSubagentCount), listAppOwnedChildThreadIds: readOnReader(listAppOwnedChildThreadIds), hasPendingDelegatedCompletion: readOnReader(hasPendingDelegatedCompletion), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index b4a9764ba04c1..34ff5a7631105 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -239,7 +239,7 @@ it.effect( getThreadHistoryPage: () => Effect.die("unused"), getTimelinePage: () => Effect.die("Unused timeline read"), getMessageCount: () => Effect.die("unused message count"), - hasOnlyMaintenanceUserMessages: () => Effect.die("unused maintenance history"), + getTranscriptByteFloor: () => Effect.die("unused transcript byte floor"), hasPendingDelegatedCompletion: () => Effect.die("unused hasPendingDelegatedCompletion"), getSubagentCount: () => Effect.die("unused subagent count"), listAppOwnedChildThreadIds: () => Effect.die("unused listAppOwnedChildThreadIds"), diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index bb34ba5327693..ad0c179d783fe 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1379,139 +1379,6 @@ it.effect("starts a steer that missed the turn on the saved next-turn selection" ), ); -it.effect("starts a steer of a finished run from before the last compaction as a new turn", () => - Effect.scoped( - Effect.gen(function* () { - const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( - "steering-before-compaction", - ); - yield* Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const sink = yield* EventSink.EventSinkV2; - const threadId = yield* startFirstTurn; - const finish = (turn: (typeof started)[number]) => - Effect.gen(function* () { - const settled = yield* orchestrator.streamDomainEvents.pipe( - Stream.filter( - (event) => - event.type === "run.updated" && - event.payload.id === turn.runId && - event.payload.status === "waiting", - ), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, - ); - const providerTurn = (yield* orchestrator.getThreadProjection( - threadId, - )).providerTurns.find((candidate) => candidate.runAttemptId === turn.attemptId)!; - yield* Queue.offer(events, { - type: "provider_turn.updated", - driver, - providerTurn: { - ...providerTurn, - status: "completed", - completedAt: yield* DateTime.now, - }, - }); - yield* Queue.offer(events, { - type: "turn.terminal", - driver, - providerThreadId: providerTurn.providerThreadId, - providerTurnId: providerTurn.id, - runOrdinal: turn.runOrdinal, - status: "completed", - failure: null, - threadDisposition: "reusable", - }); - yield* Fiber.join(settled); - yield* worker.drain(); - }); - const first = started[0]!; - yield* finish(first); - for (const id of ["second", "third"]) { - const running = yield* orchestrator.streamDomainEvents.pipe( - Stream.filter( - (event) => - event.type === "provider-turn.updated" && event.payload.status === "running", - ), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, - ); - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make(id), - threadId, - messageId: MessageId.make(`message:${id}`), - text: id, - attachments: [], - modelSelection: runSelection, - dispatchMode: { type: "start_immediately" }, - createdBy: "user", - creationSource: "web", - }); - yield* worker.drain(); - yield* Fiber.join(running); - yield* finish(started.at(-1)!); - } - // The window starts at the second run, the prompt before the compaction in the third. - const third = started[2]!; - const now = yield* DateTime.now; - yield* sink.write({ - events: [ - { - id: EventId.make("event:compaction"), - type: "turn-item.updated" as const, - threadId, - runId: third.runId, - nodeId: third.rootNodeId, - occurredAt: now, - payload: { - id: TurnItemId.make("item:compaction"), - threadId, - runId: third.runId, - nodeId: third.rootNodeId, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1_000, - status: "completed" as const, - title: null, - startedAt: now, - completedAt: now, - updatedAt: now, - type: "compaction" as const, - driver, - }, - }, - ], - }); - - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make("steer"), - threadId, - messageId: MessageId.make("message:steer"), - text: "late steer", - attachments: [], - modelSelection: runSelection, - dispatchMode: { type: "steer_active", targetRunId: first.runId }, - createdBy: "user", - creationSource: "web", - }); - yield* worker.drain(); - - assert.deepEqual(steered, []); - assert.equal(started.length, 4); - assert.equal(started[3]?.message.messageId, MessageId.make("message:steer")); - }).pipe(Effect.provide(layer)); - }), - ), -); - it.effect.each(["complete", "stop-before-decline", "stop-after-decline", "queued-model"] as const)( "persists a declined steer without blocking the worker: %s", (timing) => @@ -1931,7 +1798,7 @@ const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( const wake = harness.started[wakeOrdinal - 1]!; assert.equal(wake.message.messageId, firstDelivery.messageId); if (userRunBeforeWake) { - // The wake's run holds a compaction, so the window starts at the user run and drops the origin. + // The wake's run holds a compaction, so the origin run's records are older than the window. const compactedAt = yield* DateTime.now; yield* sink.write({ events: [ diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index ab414e4aac16c..422d2122d03bd 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -85,7 +85,7 @@ import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; import Migration0081 from "./Migrations/081_ProjectionMessageSearchText.ts"; import Migration0082 from "./Migrations/082_TurnItemConversationIndex.ts"; -import Migration0083 from "./Migrations/083_CommandWindowIndexes.ts"; +import Migration0083 from "./Migrations/083_CommandReadIndexes.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -171,7 +171,7 @@ const migrationEntries = [ [80, "ScheduledTaskThreadMode", Migration0080], [81, "ProjectionMessageSearchText", Migration0081], [82, "TurnItemConversationIndex", Migration0082], - [83, "CommandWindowIndexes", Migration0083], + [83, "CommandReadIndexes", Migration0083], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts b/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts deleted file mode 100644 index c22f3682cc1bd..0000000000000 --- a/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts +++ /dev/null @@ -1,31 +0,0 @@ -import * as Effect from "effect/Effect"; -import * as SqlClient from "effect/sql/SqlClient"; - -export default Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - - // Reading the runs a new message needs from a compacted thread must not touch the older ones. - // Every column a lookup could match is wrapped in `|| ''`, so only a lookup that writes the same - // expression can use these indexes; plain ones were chosen for unrelated statements instead. - // IF NOT EXISTS: earlier builds created some of these under migrations that were renumbered. - yield* sql` - CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_thread_type_idx - ON orchestration_v2_projection_turn_items((thread_id || ''), (type || ''), ordinal, turn_item_id) - `; - yield* sql` - CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_id_status_idx - ON orchestration_v2_projection_runs((run_id || ''), status) - `; - // Fresh-context starts are rare, so their partial index stays tiny. - yield* sql` - CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_fresh_context_idx - ON orchestration_v2_projection_runs(thread_id, ordinal) - WHERE CASE WHEN json_valid(payload_json) - THEN json_extract(payload_json, '$.freshContext') = 1 - ELSE 0 END - `; - yield* sql` - CREATE INDEX IF NOT EXISTS orchestration_v2_projection_provider_turns_attempt_idx - ON orchestration_v2_projection_provider_turns(run_attempt_id) - `; -}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 2ebef1789f091..05f09f8e16850 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -62,7 +62,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], - [83, "CommandWindowIndexes"], + [83, "CommandReadIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -168,7 +168,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], - [83, "CommandWindowIndexes"], + [83, "CommandReadIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From a91dfd397d64efcfc1e9dbae6e870ae6cea22370 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 04:23:33 -0800 Subject: [PATCH 8/8] test(server): a compacted thread's send also skips its settled tasks' records The command read now applies two filters at once: the window after the latest completed compaction, and the settled subagent filter from #35178. The compacted store test gets tasks that ended on both sides of the compaction and checks that the combined read is the intersection of the two filters. It fails when either filter is dropped from the combined read. Co-Authored-By: Claude Fable 5.2 --- .../orchestration-v2/ProjectionStore.test.ts | 168 ++++++++++++++++++ 1 file changed, 168 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 67844aff0a754..8ab27060636fe 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1059,6 +1059,174 @@ it.effect("a compacted thread reads every run, and run records only for its rece yield* read([runId(3)], messageId(6)), expected([2, 3, 4, 6, 11, 12, 13, 14, 15]), ); + + // A command also skips the node and the provider thread of each settled task. Tasks ended on + // both sides of the compaction; one ran in the provider thread the thread's own runs use. + const addProviderThread = (id: ProviderThreadId, appThreadId: ThreadId) => + store.apply({ + id: EventId.make(`event:compacted:${id}`), + type: "provider-thread.updated", + threadId, + occurredAt: created, + payload: { + id, + appThreadId, + driver, + providerInstanceId, + providerSessionId: null, + ownerNodeId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + status: "idle", + firstRunOrdinal: null, + createdAt: created, + updatedAt: created, + }, + }); + yield* addProviderThread(providerThreadId, threadId); + for (const [name, ordinal, delivery] of [ + ["old-delivered", 5, "delivered"], + ["old-pending", 5, "pending"], + ["old-own-thread", 5, "delivered"], + ["recent-delivered", 12, "delivered"], + ["recent-pending", 12, "pending"], + ] as const) { + const id = NodeId.make(`node:compacted:task:${name}`); + const taskProviderThreadId = + name === "old-own-thread" + ? providerThreadId + : ProviderThreadId.make(`provider-thread:compacted:task:${name}`); + if (taskProviderThreadId !== providerThreadId) { + yield* addProviderThread( + taskProviderThreadId, + ThreadId.make(`thread:compacted:child:${name}`), + ); + } + const task = { + id, + threadId, + runId: runId(ordinal), + parentNodeId: rootNodeId(ordinal), + status: "completed", + startedAt: at(ordinal), + completedAt: at(ordinal), + } as const; + yield* store.apply({ + id: EventId.make(`event:compacted:task:${name}`), + type: "subagent.updated", + threadId, + runId: runId(ordinal), + nodeId: id, + driver, + providerInstanceId, + occurredAt: at(ordinal), + payload: { + ...task, + origin: "app_owned", + createdBy: "agent", + driver, + providerInstanceId, + providerThreadId: taskProviderThreadId, + childThreadId: null, + nativeTaskRef: null, + prompt: "Do the work", + title: name, + model: null, + completionDelivery: { state: delivery, observedByRunId: null }, + result: null, + updatedAt: at(ordinal), + }, + }); + yield* store.apply({ + id: EventId.make(`event:compacted:task-node:${name}`), + type: "node.updated", + threadId, + runId: runId(ordinal), + nodeId: id, + driver, + occurredAt: at(ordinal), + payload: { + ...task, + rootNodeId: rootNodeId(ordinal), + kind: "subagent", + countsForRun: false, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + }, + }); + } + const names = (ids: ReadonlyArray) => ids.map((id) => id.split(":").at(-1)).toSorted(); + const readWithTasks = (filter: ProjectionStore.ProjectionRecordFilter) => + store + .getThreadRecords( + threadId, + [ + "runs", + "attempts", + "nodes", + "providerThreads", + "providerTurns", + "messages", + "turnItems", + ], + { messageRoles: ["user"], omitSettledLeafNodes: true, ...filter }, + ) + .pipe( + Effect.map((records) => ({ + runs: names(records.runs.map((run) => run.id)), + turnItems: names(records.turnItems.map((row) => row.id)), + attempts: names(records.attempts.map((attempt) => attempt.id)), + rootNodes: names( + records.nodes.flatMap((node) => (node.kind === "root_turn" ? [node.id] : [])), + ), + taskNodes: names( + records.nodes.flatMap((node) => (node.kind === "subagent" ? [node.id] : [])), + ), + providerThreads: names(records.providerThreads.map((thread) => thread.id)), + providerTurns: names(records.providerTurns.map((turn) => turn.id)), + messages: names(records.messages.map((message) => message.id)), + })), + ); + const window = { + sinceLastCompaction: { keepRunIds: [runId(3)], keepMessageId: messageId(6) }, + }; + const windowed = yield* readWithTasks(window); + const unsettled = yield* readWithTasks({ omitSettledSubagentRecords: true }); + const both = yield* readWithTasks({ ...window, omitSettledSubagentRecords: true }); + // Each filter alone keeps records the other one drops. + assert.deepEqual(windowed.taskNodes, ["recent-delivered", "recent-pending"]); + assert.lengthOf(windowed.providerThreads, 5); + assert.deepEqual(unsettled.taskNodes, ["old-pending", "recent-pending"]); + assert.deepEqual(unsettled.providerThreads, ["compacted", "old-pending", "recent-pending"]); + assert.lengthOf(unsettled.rootNodes, all.length); + assert.lengthOf(unsettled.runs, 15); + // Together: every run, every turn item and the provider thread the runs use stay, and the + // windowed records are read for the recent and named runs, less each settled task's node. + const recentRuns = ["11", "12", "13", "14", "15", "2", "3", "4", "6"]; + assert.deepEqual(both, { + ...unsettled, + attempts: recentRuns, + rootNodes: recentRuns, + taskNodes: ["recent-pending"], + providerTurns: recentRuns, + messages: recentRuns, + }); + const keptUnsettled = new Map(Object.entries(unsettled)); + assert.deepEqual( + Object.fromEntries( + Object.entries(windowed).map(([key, kept]) => [ + key, + kept.filter((name) => keptUnsettled.get(key)?.includes(name)), + ]), + ), + both, + ); }).pipe(Effect.provide(layerTest)), );