diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 076960f4dc1d5..c186c40f9e73a 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?: ProjectionRecordFilter["sinceLastCompaction"], + ) => projectionStore .getThreadRecords( threadId, @@ -1557,6 +1560,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio omitSettledSubagentRecords: true, // Commands act on pending forks and merge-backs only; consumed ones stay unread. contextTransferStatuses: ["pending"], + ...(recent === undefined ? {} : { sinceLastCompaction: recent }), }, ) .pipe( @@ -1645,9 +1649,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.fnUntraced(function* ( threadId: ThreadId, events: Ref.Ref>, + recent?: ProjectionRecordFilter["sinceLastCompaction"], ) { 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; @@ -6307,7 +6312,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: Ref.Ref>, effects: Ref.Ref>, ) { - let projection = yield* getProjectionWithPendingEvents(command.threadId, events); + // 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 ( @@ -6344,6 +6370,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); return; } + const activeProviderInstanceId = projection.providerThreads.find( + (candidate) => candidate.id === projection.thread.activeProviderThreadId, + )?.providerInstanceId; + // 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 ( + 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); @@ -6547,15 +6585,27 @@ 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); + // 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" && onlyMaintenanceHistory)) + (command.createdBy === "user" && (yield* onlyMaintenanceHistory))) ) { 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 bcd6732c23782..8ab27060636fe 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -795,6 +795,441 @@ it.effect("idle compaction filters keep the latest run, live work and the newest }).pipe(Effect.provide(layerTest)), ); +// 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 created = yield* DateTime.now; + const at = (minutes: number) => DateTime.add(created, { minutes }); + const runId = (ordinal: number) => RunId.make(`run:compacted:${ordinal}`); + 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, + }, + }); + for (const [ordinal, status, endedAt] of [ + [2, "queued", null], + [3, "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, "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", + threadId, + runId: runId(ordinal), + nodeId, + driver, + providerInstanceId, + occurredAt: at(ordinal), + payload: { + id: runId(ordinal), + threadId, + ordinal, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: messageId(ordinal), + rootNodeId: nodeId, + activeAttemptId: attemptId(ordinal), + status, + requestedAt: at(ordinal), + startedAt: status === "queued" ? null : at(ordinal), + completedAt: ended, + checkpointId: null, + contextHandoffId: null, + }, + }); + 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({ + id: EventId.make(`event:compacted:message-${ordinal}`), + type: "message.updated", + threadId, + runId: runId(ordinal), + occurredAt: at(ordinal), + payload: { + ...authored, + id: messageId(ordinal), + threadId, + runId: runId(ordinal), + nodeId, + role: "user", + text, + attachments: [], + streaming: false, + createdAt: at(ordinal), + updatedAt: at(ordinal), + }, + }); + yield* store.apply({ + id: EventId.make(`event:compacted:prompt-${ordinal}`), + type: "turn-item.updated", + threadId, + runId: runId(ordinal), + nodeId, + driver, + occurredAt: at(ordinal), + payload: { + ...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.deepEqual(yield* read(), expected(all)); + + // 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])); + + // The latest compaction wins. + yield* compaction(11, "completed"); + const latest = expected([2, 4, 11, 12, 13, 14, 15]); + assert.deepEqual(yield* read(), latest); + + // 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); + + // 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]), + ); + + // 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)), +); + 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 2bf1e896b4973..6264b36dfbd7d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -378,6 +378,16 @@ export interface ProjectionRecordFilter { * the thread stays. */ readonly omitSettledSubagentRecords?: boolean; + /** + * 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; + /** 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, @@ -3435,6 +3445,49 @@ export const layer: Layer.Layer = const threadColumn = (ids: ReadonlyArray | undefined, column = "thread_id") => sql.literal(ids === undefined ? column : `+${column}`); + /** 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_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' + AND item.type NOT IN ${CONVERSATION_INDEX_EXCLUDED_TYPES} + AND run.status <> 'rolled_back' + ORDER BY item.ordinal DESC + LIMIT 1 + ) + 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 + `; + // 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* ( threadId: ThreadId, @@ -3452,6 +3505,19 @@ export const layer: Layer.Layer = if (!threadRow) { return yield* new ProjectionStoreThreadNotFoundError({ threadId }); } + const recentRunIds = + window === undefined && filter?.sinceLastCompaction !== undefined + ? 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; @@ -3729,6 +3795,7 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_run_attempts WHERE thread_id = ${threadId} + ${inRecentRuns("run_id")} ORDER BY run_id ASC, attempt_ordinal ASC ` : sql` @@ -3749,13 +3816,15 @@ export const layer: Layer.Layer = 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. + // 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")} ${ filter.omitSettledSubagentRecords === true ? // A subagent's node shares its id. Both indexes hold every column @@ -3781,7 +3850,8 @@ export const layer: Layer.Layer = 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 @@ -3912,6 +3982,13 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_provider_turns WHERE thread_id = ${threadId} + ${ + recentRunIds === null + ? sql`` + : sql`AND run_attempt_id IN ( + SELECT attempt_id FROM orchestration_v2_projection_run_attempts + 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)}))`} ${ @@ -3957,7 +4034,9 @@ export const layer: Layer.Layer = : window === undefined ? sql` SELECT payload_json FROM orchestration_v2_projection_messages - WHERE ${threadColumn(filter?.messageIds)} = ${threadId} + ${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``} diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 8d784a91f59cb..ad0c179d783fe 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1635,6 +1635,8 @@ const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( "events" | "started" | "startFirstTurn" >, siblingFinishesFirst: boolean, + userRunBeforeWake = false, + intentNamesTarget = false, ) { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; @@ -1736,6 +1738,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({ @@ -1769,19 +1789,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 origin run's records are older than the window. + 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"), @@ -1790,13 +1846,15 @@ 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", }); 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( @@ -1883,6 +1941,35 @@ it.effect("wakes the parent for a sibling result after a steer restarts the comp ), ); +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* 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( "offers the next completion wake when a steer restarts a wake a sibling finished behind", () =>