From 9217472cac5705c974f7eafa1cd68aadf6c693a3 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Wed, 11 Nov 2026 20:33:53 -0300 Subject: [PATCH 1/6] 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 | 43 +++- .../orchestration-v2/ProjectionStore.test.ts | 200 ++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 168 +++++++++++++-- .../ProviderTurnControlService.test.ts | 1 + .../SteeringCompletion.integration.test.ts | 133 ++++++++++++ apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/083_CommandWindowIndexes.ts | 31 +++ .../reconcileV2PreviewMigration.test.ts | 2 + 8 files changed, 559 insertions(+), 21 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 6e38afbf6430..847a1bcb078a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1458,7 +1458,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, @@ -1483,6 +1486,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, + ...(recent === undefined ? {} : { sinceLastCompaction: recent }), }, ) .pipe( @@ -1571,9 +1575,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; @@ -6166,7 +6171,19 @@ 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.dispatchMode.type === "steer_active" || + command.dispatchMode.type === "restart_active" + ? [command.dispatchMode.targetRunId] + : []), + ].flatMap((runId) => (runId === undefined ? [] : [runId])), + }); if (yield* isStaleQuestionWake(projection, command.messageId, command.notification)) { yield* emit( events, @@ -6179,6 +6196,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); @@ -6381,16 +6410,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 2ba22350330b..bfff55b4c85d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -762,6 +762,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 9e2721035064..cb55a58e1af7 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,6 +479,10 @@ 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: ( + threadId: ThreadId, + ) => 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: ( @@ -3296,6 +3307,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, @@ -3313,6 +3385,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; @@ -3334,11 +3410,13 @@ 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} ${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 @@ -3529,7 +3607,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)}))`} ${ @@ -3583,6 +3668,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` @@ -3599,7 +3685,9 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_nodes AS node - WHERE ${threadColumn(filter?.nodeIds, "node.thread_id")} = ${threadId} + ${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 @@ -3609,7 +3697,8 @@ export const layer: Layer.Layer = AND node.run_id IS NOT NULL AND NOT EXISTS ( SELECT 1 FROM orchestration_v2_projection_runs AS run - WHERE run.run_id = node.run_id + 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') ) )` @@ -3638,7 +3727,13 @@ 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)}))`} ${ @@ -3653,7 +3748,8 @@ export const layer: Layer.Layer = IN ('acknowledged', 'delivered', 'disposed') AND NOT EXISTS ( SELECT 1 FROM orchestration_v2_projection_runs AS run - WHERE run.run_id = subagent.run_id + 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') ) ))` @@ -3700,13 +3796,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 ` @@ -3727,7 +3827,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)}))`} ${ @@ -3773,7 +3881,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``} @@ -3889,7 +3999,11 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_context_transfers - WHERE (source_thread_id = ${threadId} OR target_thread_id = ${threadId}) + WHERE ${ + recentRuns === null + ? sql`(source_thread_id = ${threadId} OR target_thread_id = ${threadId})` + : 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 ` @@ -3936,7 +4050,13 @@ export const layer: Layer.Layer = decodeRows(decodeContextTransferPayload, threadId)(contextTransferRows), ]); const messageOrdinals = new Map(); - if (fields !== undefined && fields.includes("messages") && messages.length > 1) { + // The recent user turns, loaded above, already order their own messages. + if ( + recentRuns === null && + fields !== undefined && + fields.includes("messages") && + messages.length > 1 + ) { // Broad role reads keep the original query to avoid per-ID lookup overhead. const loadedMessageIds = filter?.messageIds === undefined && @@ -6372,15 +6492,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 @@ -7783,6 +7922,7 @@ export const layer: Layer.Layer = hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest), isNodeStored: readOnReader(isNodeStored), getMessageCount: readOnReader(getMessageCount), + hasOnlyMaintenanceUserMessages: readOnReader(hasOnlyMaintenanceUserMessages), getSubagentCount: readOnReader(getSubagentCount), listAppOwnedChildThreadIds: readOnReader(listAppOwnedChildThreadIds), getNextTurnItemOrdinal: readOnReader(getNextTurnItemOrdinal), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 25daa45a847f..101c64479e44 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -239,6 +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"), getSubagentCount: () => Effect.die("unused subagent count"), listAppOwnedChildThreadIds: () => Effect.die("unused listAppOwnedChildThreadIds"), getNextTurnItemOrdinal: () => Effect.die("unused ordinal read"), diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index e39c9bb07e93..067df95145aa 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1375,6 +1375,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)); + }), + ), +); + // regression: #26695 cancelling a Copilot prompt to steer it kills its background agents it.effect("queues a steer until the turn ends when the provider cannot steer it", () => Effect.scoped( diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index e5aa6c68dc75..abb06549ce3b 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -85,6 +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"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -170,6 +171,7 @@ const migrationEntries = [ [80, "ScheduledTaskThreadMode", Migration0080], [81, "ProjectionMessageSearchText", Migration0081], [82, "TurnItemConversationIndex", Migration0082], + [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 000000000000..c22f3682cc1b --- /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 88348b742b47..2ebef1789f09 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -62,6 +62,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], + [83, "CommandWindowIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -167,6 +168,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], + [83, "CommandWindowIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From 5bf39dac0cd816dc11f6d91298b1c9544dbbc1fc Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Wed, 11 Nov 2026 20:57:10 -0300 Subject: [PATCH 2/6] 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 | 35 +++++--- .../orchestration-v2/ProjectionStore.test.ts | 16 ++-- .../src/orchestration-v2/ProjectionStore.ts | 21 +++-- .../SteeringCompletion.integration.test.ts | 85 ++++++++++++++++++- 4 files changed, 128 insertions(+), 29 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 847a1bcb078a..06c3840e59e9 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6172,18 +6172,33 @@ 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, + ].flatMap((runId) => (runId === undefined ? [] : [runId])); let projection = yield* getProjectionWithPendingEvents(command.threadId, events, { - keepRunIds: [ - command.manualContinuationOfRunId, - command.restartContinuationOfRunId, - command.usageLimitContinuationOfRunId, - command.delegatedCompletion?.parentRunId, - ...(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 (yield* isStaleQuestionWake(projection, command.messageId, command.notification)) { yield* emit( events, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index bfff55b4c85d..14d43dc1c9b4 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -769,6 +769,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"], @@ -802,7 +805,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, @@ -821,7 +824,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, @@ -882,15 +885,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 cb55a58e1af7..be7c597328ba 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3320,18 +3320,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 067df95145aa..686d70c221ae 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1591,6 +1591,7 @@ const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( "events" | "started" | "startFirstTurn" >, siblingFinishesFirst: boolean, + userRunBeforeWake = false, ) { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; @@ -1692,6 +1693,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({ @@ -1725,19 +1744,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"), @@ -1752,7 +1807,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( @@ -1839,6 +1894,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 84292e141a5a3785fc0bc22413b332c2d0c9e2df Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Wed, 11 Nov 2026 20:59:03 -0300 Subject: [PATCH 3/6] 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 14d43dc1c9b4..720438f60149 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -779,6 +779,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"], @@ -788,6 +789,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}`), @@ -871,7 +874,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. @@ -884,13 +887,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 be7c597328ba..ba8b9b77fe43 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3351,6 +3351,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 @@ -6501,7 +6502,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 ( @@ -6519,7 +6520,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 caf6b880460f55a83ea79dde9a0a13e0f48cca16 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Wed, 11 Nov 2026 21:14:39 -0300 Subject: [PATCH 4/6] 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 06c3840e59e9..ff93e6d23d72 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6186,19 +6186,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 (yield* isStaleQuestionWake(projection, command.messageId, command.notification)) { yield* emit( events, @@ -6468,6 +6466,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 686d70c221ae..a87a0cd0c8dd 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1592,6 +1592,7 @@ const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( >, siblingFinishesFirst: boolean, userRunBeforeWake = false, + intentNamesTarget = false, ) { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; @@ -1801,7 +1802,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", }); @@ -1894,26 +1897,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 b13993569bb2a8af51929ff52c7cfae2490731ec Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Thu, 12 Nov 2026 00:19:56 -0300 Subject: [PATCH 5/6] 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 f2dfdcec11fa..95d58b320cca 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -762,6 +762,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; @@ -769,7 +815,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) => @@ -835,6 +881,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, @@ -885,19 +940,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 b5505f4c00e2..a7264da27b69 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3363,7 +3363,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( @@ -3374,6 +3374,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 e0a4190e46e1637da3d238661b8de040366517b9 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Thu, 12 Nov 2026 07:53:13 -0300 Subject: [PATCH 6/6] 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 d288e1c3cdcd..e8501640befa 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6206,6 +6206,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,