diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 71b6f0ea994f..935922a609d2 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -1,6 +1,8 @@ import { CommandId, type OrchestrationV2Run, + OrchestrationV2ConversationMessageJson, + OrchestrationV2TurnItemJson, OrchestrationV2DomainEvent, OrchestrationV2StoredEvent, ProviderThreadId, @@ -33,6 +35,13 @@ import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProjectStore from "./ProjectStore.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; +const decodeMessagePayload = Schema.decodeUnknownEffect( + Schema.fromJsonString(OrchestrationV2ConversationMessageJson), +); +const decodeTurnItemPayload = Schema.decodeUnknownEffect( + Schema.fromJsonString(OrchestrationV2TurnItemJson), +); + /** * ERRORS */ @@ -78,6 +87,8 @@ export interface EventSinkV2Shape { }) => Effect.Effect, EventSinkV2Error>; readonly writeWithEffects: (input: { readonly guardPendingUserInputCancellations?: boolean; + /** Skip artifact cleanup when its node now belongs to a different run attempt. */ + readonly guardNodeRootId?: NodeId; readonly commandId?: CommandId; readonly events: ReadonlyArray; readonly effects: ReadonlyArray; @@ -90,6 +101,7 @@ export interface EventSinkV2Shape { readonly activeAttemptId: RunAttemptId; readonly expectedStatus: OrchestrationV2Run["status"]; readonly events: ReadonlyArray; + readonly effects?: ReadonlyArray; }) => Effect.Effect< { readonly committed: boolean; @@ -305,6 +317,139 @@ const baseLayer: Layer.Layer< }); }); + const guardNodeRootArtifacts = Effect.fnUntraced(function* ( + events: ReadonlyArray, + rootNodeId: NodeId, + ) { + const offeredNodes = new Map( + events.flatMap((event) => + event.type === "node.updated" && event.payload.rootNodeId === rootNodeId + ? [[event.payload.id, event.payload.providerTurnId] as const] + : [], + ), + ); + const accepted: Array = []; + for (const event of events) { + if ( + event.type !== "node.updated" && + event.type !== "message.updated" && + event.type !== "turn-item.updated" + ) { + accepted.push(event); + continue; + } + const nodeId = event.type === "node.updated" ? event.payload.id : event.payload.nodeId; + if (nodeId === null) continue; + if (event.type === "node.updated" && event.payload.rootNodeId !== rootNodeId) continue; + const nodes = yield* sql<{ + readonly root_node_id: string; + readonly status: string; + readonly provider_turn_id: string | null; + }>` + SELECT root_node_id, status, provider_turn_id FROM orchestration_v2_projection_nodes + WHERE thread_id = ${event.threadId} AND node_id = ${nodeId}`; + const node = nodes[0]; + if (node === undefined ? !offeredNodes.has(nodeId) : node.root_node_id !== rootNodeId) + continue; + // Assistant streaming may suppress the successor's running node update. + // Check item provenance too, even when the projected node still looks old. + // Root-linked reasoning shares a node with the user message. Its + // turn identity belongs to the item, and sibling items may have none. + const isRootTurnItem = + event.type === "turn-item.updated" && + nodeId === rootNodeId && + event.payload.providerTurnId !== null; + const expectedTurnId = isRootTurnItem + ? event.payload.providerTurnId + : (offeredNodes.get(nodeId) ?? node?.provider_turn_id); + const items = isRootTurnItem + ? [] + : yield* sql<{ readonly provider_turn_id: string | null }>` + SELECT provider_turn_id FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${event.threadId} AND node_id = ${nodeId}`; + if (items.some((item) => item.provider_turn_id !== expectedTurnId)) continue; + if (expectedTurnId == null && (node !== undefined || items.length > 0)) continue; + if (event.type === "node.updated") { + if ( + node !== undefined && + node.status !== "pending" && + node.status !== "running" && + node.status !== "waiting" + ) + continue; + } else if (event.type === "message.updated") { + const rows = yield* sql<{ + readonly node_id: string | null; + readonly streaming: number; + readonly payload_json: string; + }>` + SELECT node_id, streaming, payload_json FROM orchestration_v2_projection_messages + WHERE thread_id = ${event.threadId} AND message_id = ${event.payload.id}`; + const current = rows[0]; + if (current !== undefined && (current.node_id !== nodeId || current.streaming === 0)) + continue; + if (current !== undefined) { + if (node === undefined && items.length === 0) continue; + const message = yield* decodeMessagePayload(current.payload_json); + accepted.push({ + ...event, + payload: { + ...message, + text: + message.text.length > event.payload.text.length + ? message.text + : event.payload.text, + streaming: false, + updatedAt: event.payload.updatedAt, + }, + }); + continue; + } + } else { + const rows = yield* sql<{ + readonly node_id: string | null; + readonly status: string; + readonly provider_turn_id: string | null; + readonly payload_json: string; + }>` + SELECT node_id, status, provider_turn_id, payload_json FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${event.threadId} AND turn_item_id = ${event.payload.id}`; + const current = rows[0]; + if ( + current !== undefined && + (current.node_id !== nodeId || + current.provider_turn_id !== event.payload.providerTurnId || + (current.status !== "pending" && + current.status !== "running" && + current.status !== "waiting")) + ) + continue; + if (current !== undefined) { + const item = yield* decodeTurnItemPayload(current.payload_json); + const content = + "text" in item && + "text" in event.payload && + item.text.length < event.payload.text.length + ? event.payload + : item; + accepted.push({ + ...event, + payload: { + ...content, + ...("streaming" in content ? { streaming: false } : {}), + status: event.payload.status, + completedAt: event.payload.completedAt, + updatedAt: event.payload.updatedAt, + }, + }); + continue; + } + } + accepted.push(event); + } + return accepted; + }); + const normalizeEvents = (events: ReadonlyArray) => { const runOrdinals = new Map( events.flatMap((event) => @@ -369,10 +514,14 @@ const baseLayer: Layer.Layer< return yield* commitThenPublish( Effect.gen(function* () { + const guardedEvents = + input.guardNodeRootId === undefined + ? input.events + : yield* guardNodeRootArtifacts(input.events, input.guardNodeRootId); const normalized = yield* normalizeEvents( input.guardPendingUserInputCancellations === true - ? yield* guardUserInputCancellations(input.events) - : input.events, + ? yield* guardUserInputCancellations(guardedEvents) + : guardedEvents, ); const committed = yield* eventStore.append({ ...(input.commandId === undefined ? {} : { commandId: input.commandId }), @@ -437,9 +586,18 @@ const baseLayer: Layer.Layer< events: normalized, }); yield* applyStoredEvents(storedEvents); + yield* effectOutbox.enqueue(input.effects ?? []); return { committed: true as const, storedEvents }; }), - (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void), + (result) => + result.committed + ? Effect.gen(function* () { + if (input.effects !== undefined && input.effects.length > 0) { + yield* effectOutbox.notifyAvailable(input.effects.length); + } + yield* publishStoredEvents(result.storedEvents); + }) + : Effect.void, ); }, ); diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index dc7eacc59ba5..d547679de2ce 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -12,6 +12,8 @@ import { NodeId, type OrchestrationV2AppThread, type OrchestrationV2DomainEvent, + type OrchestrationV2ExecutionNode, + type OrchestrationV2ConversationMessage, type OrchestrationV2Run, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, @@ -20,6 +22,7 @@ import { ProviderInstanceId, ProviderSessionId, ProviderThreadId, + ProviderTurnId, RunAttemptId, RunId, ThreadId, @@ -1614,6 +1617,369 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { }), ); + it.effect.each([true, false])( + "enqueues finalization effects only when the current-run guard commits: %s", + (current) => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`thread:guarded-effects:${current}`); + const runId = RunId.make(`run:guarded-effects:${current}`); + const attemptId = RunAttemptId.make(`attempt:guarded-effects:${current}`); + const run: OrchestrationV2Run = { + id: runId, + threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:guarded-effects:${current}`), + rootNodeId: null, + activeAttemptId: attemptId, + status: "running", + queuePosition: null, + requestedAt: now, + startedAt: now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + yield* sink.write({ + events: [ + threadCreatedEvent({ + id: `event:guarded-effects:thread:${current}`, + thread: makeThread(threadId, now), + now, + }), + { + id: EventId.make(`event:guarded-effects:run:${current}`), + type: "run.created", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: run, + }, + ], + }); + const effect = { + id: `effect:guarded-effects:${current}`, + commandId: CommandId.make(`command:guarded-effects:${current}`), + threadId, + request: { + type: "checkpoint.capture" as const, + runId, + scopeId: CheckpointScopeId.make(`scope:guarded-effects:${current}`), + }, + }; + const result = yield* sink.writeIfRunCurrent({ + threadId, + runId, + activeAttemptId: current ? attemptId : RunAttemptId.make("attempt:superseded"), + expectedStatus: "running", + effects: [effect], + events: [ + { + id: EventId.make(`event:guarded-effects:finalized:${current}`), + type: "run.updated", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: { ...run, status: "waiting" }, + }, + ], + }); + assert.equal(result.committed, current); + assert.equal(Option.isSome(yield* outbox.get(effect.id)), current); + assert.equal( + (yield* projections.getThreadProjection(threadId)).runs[0]?.status, + current ? "waiting" : "running", + ); + if (current) yield* outbox.awaitAvailable; + else { + const wake = yield* outbox.awaitAvailable.pipe(Effect.forkChild); + yield* Effect.yieldNow; + assert.isUndefined(wake.pollUnsafe()); + } + }).pipe(Effect.provide(Layer.fresh(TestLayer))), + ); + + it.effect.each([ + "old-root", + "adopted-node", + "retargeted-records", + "buffered", + "already-closed", + "unprojected-successor", + "richer-text", + "root-reasoning", + "root-reasoning-retargeted", + "root-reasoning-successor-turn", + ] as const)("guards superseded artifact cleanup by current node ancestry: %s", (scenario) => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`thread:artifact-cleanup:${scenario}`); + const oldRoot = NodeId.make(`node:old-root:${scenario}`); + const newRoot = NodeId.make(`node:new-root:${scenario}`); + const rootLinked = scenario.startsWith("root-reasoning"); + const artifactNodeId = rootLinked ? oldRoot : NodeId.make(`node:artifact:${scenario}`); + const node: OrchestrationV2ExecutionNode = { + id: artifactNodeId, + threadId, + runId: null, + parentNodeId: rootLinked ? null : oldRoot, + rootNodeId: oldRoot, + kind: rootLinked ? "root_turn" : "assistant_message", + status: "running", + countsForRun: false, + providerThreadId: null, + providerTurnId: rootLinked ? null : ProviderTurnId.make(`turn:old:${scenario}`), + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }; + const message: OrchestrationV2ConversationMessage = { + id: MessageId.make(`message:artifact:${scenario}`), + threadId, + runId: null, + nodeId: artifactNodeId, + role: "assistant", + text: "Old partial text", + streaming: true, + attachments: [], + createdBy: "agent", + creationSource: "provider", + createdAt: now, + updatedAt: now, + }; + const item: OrchestrationV2TurnItem = { + id: TurnItemId.make(`item:artifact:${scenario}`), + threadId, + runId: null, + nodeId: artifactNodeId, + providerThreadId: null, + providerTurnId: ProviderTurnId.make(`turn:old:${scenario}`), + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + ...(rootLinked + ? { type: "reasoning" as const } + : { type: "assistant_message" as const, messageId: message.id }), + text: message.text, + streaming: true, + }; + const artifactEvents = ( + suffix: string, + currentNode: OrchestrationV2ExecutionNode, + currentMessage: OrchestrationV2ConversationMessage, + currentItem: OrchestrationV2TurnItem, + ): ReadonlyArray => { + const events: ReadonlyArray = [ + { + id: EventId.make(`event:artifact:${scenario}:node:${suffix}`), + type: "node.updated", + threadId, + occurredAt: now, + payload: currentNode, + }, + { + id: EventId.make(`event:artifact:${scenario}:message:${suffix}`), + type: "message.updated", + threadId, + occurredAt: now, + payload: currentMessage, + }, + { + id: EventId.make(`event:artifact:${scenario}:item:${suffix}`), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: currentItem, + }, + ]; + return events.filter( + (event) => + !rootLinked || + event.type === "turn-item.updated" || + (suffix !== "cleanup" && event.type === "node.updated"), + ); + }; + yield* sink.write({ + events: [ + threadCreatedEvent({ + id: `event:artifact:${scenario}:thread`, + thread: makeThread(threadId, now), + now, + }), + ...(scenario === "buffered" || scenario === "unprojected-successor" + ? [] + : artifactEvents("initial", node, message, item)), + ...(rootLinked + ? [ + { + id: EventId.make(`event:artifact:${scenario}:user`), + type: "turn-item.updated" as const, + threadId, + occurredAt: now, + payload: { + ...item, + id: TurnItemId.make(`item:user:${scenario}`), + type: "user_message" as const, + providerTurnId: null, + status: "completed" as const, + completedAt: now, + messageId: MessageId.make(`message:user:${scenario}`), + inputIntent: "turn_start" as const, + createdBy: "user" as const, + creationSource: "web" as const, + text: "Continue", + attachments: [], + }, + }, + ] + : []), + ], + }); + const retargeted = + scenario === "retargeted-records" || scenario === "root-reasoning-retargeted"; + const adopted = scenario === "adopted-node" || retargeted; + const newNodeId = retargeted ? NodeId.make(`node:successor-message:${scenario}`) : node.id; + if (adopted) + yield* sink.write({ + events: artifactEvents( + "successor", + { ...node, id: newNodeId, rootNodeId: newRoot, parentNodeId: newRoot }, + { ...message, nodeId: newNodeId, text: "Successor text" }, + { ...item, nodeId: newNodeId, text: "Successor text" }, + ), + }); + if (scenario === "unprojected-successor") + yield* sink.write({ + events: artifactEvents( + "unprojected", + node, + { ...message, text: "Successor text" }, + { + ...item, + text: "Successor text", + providerTurnId: ProviderTurnId.make(`turn:new:${scenario}`), + }, + ).filter((event) => event.type !== "node.updated"), + }); + if (scenario === "root-reasoning-successor-turn") + yield* sink.write({ + events: artifactEvents("successor-turn", node, message, { + ...item, + providerTurnId: ProviderTurnId.make(`turn:new:${scenario}`), + text: "Successor reasoning", + }).filter((event) => event.type === "turn-item.updated"), + }); + if (scenario === "richer-text") + yield* sink.write({ + events: artifactEvents( + "richer", + node, + { ...message, text: "Old partial text with more projected content" }, + { ...item, text: "Old partial text with more projected content" }, + ), + }); + if (scenario === "already-closed") + yield* sink.write({ + events: artifactEvents( + "completed", + { ...node, status: "completed", completedAt: now }, + { ...message, text: "Complete text", streaming: false }, + { + ...item, + text: "Complete text", + status: "completed", + streaming: false, + completedAt: now, + }, + ), + }); + const stored = yield* sink.writeWithEffects({ + effects: [], + guardNodeRootId: oldRoot, + events: artifactEvents( + "cleanup", + { ...node, status: "interrupted", completedAt: now }, + { ...message, streaming: false }, + { ...item, status: "interrupted", streaming: false, completedAt: now }, + ), + }); + const projection = yield* projections.getThreadProjection(threadId); + const actualMessage = projection.messages.find((candidate) => candidate.id === message.id); + const actualItem = projection.turnItems.find((candidate) => candidate.id === item.id); + if (rootLinked) { + const closed = scenario === "root-reasoning"; + assert.lengthOf(stored, closed ? 1 : 0); + assert.equal(actualItem?.status, closed ? "interrupted" : "running"); + assert.equal(actualItem?.type === "reasoning" && actualItem.streaming, !closed); + assert.equal(actualItem?.nodeId, retargeted ? newNodeId : oldRoot); + assert.equal( + actualItem?.providerTurnId, + ProviderTurnId.make( + `turn:${scenario === "root-reasoning-successor-turn" ? "new" : "old"}:${scenario}`, + ), + ); + const userItem = projection.turnItems.find( + (candidate) => candidate.id === `item:user:${scenario}`, + ); + assert.equal(userItem?.status, "completed"); + assert.isNull(userItem?.providerTurnId); + return; + } + if (adopted) { + assert.equal(actualMessage?.text, "Successor text"); + assert.isTrue(actualMessage?.streaming); + assert.equal(actualItem?.status, "running"); + assert.equal( + projection.nodes.find((candidate) => candidate.id === newNodeId)?.status, + "running", + ); + assert.lengthOf(stored, scenario === "retargeted-records" ? 1 : 0); + } else if (scenario === "unprojected-successor") { + assert.lengthOf(stored, 0); + assert.equal(actualMessage?.text, "Successor text"); + assert.isTrue(actualMessage?.streaming); + assert.equal(actualItem?.status, "running"); + assert.lengthOf(projection.nodes, 0); + } else if (scenario === "already-closed") { + assert.lengthOf(stored, 0); + assert.equal(actualMessage?.text, "Complete text"); + assert.equal(actualItem?.status, "completed"); + } else { + assert.lengthOf(stored, 3); + if (scenario === "richer-text") { + assert.equal(actualMessage?.text, "Old partial text with more projected content"); + assert.equal( + actualItem?.type === "assistant_message" && actualItem.text, + actualMessage?.text, + ); + } + assert.isFalse(actualMessage?.streaming); + assert.equal(actualItem?.status, "interrupted"); + assert.equal( + projection.nodes.find((candidate) => candidate.id === node.id)?.status, + "interrupted", + ); + } + }).pipe(Effect.provide(Layer.fresh(TestLayer))), + ); + it.effect("guards post-terminal provider-thread writes by attempt and run ordinal", () => Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 3d6dca6151ab..c0eba5868fd4 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -7,8 +7,10 @@ import { EventId, MessageId, NodeId, + PlanId, type OrchestrationV2AppThread, type OrchestrationV2CheckpointScope, + type OrchestrationV2ConversationMessage, type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderThread, @@ -3204,6 +3206,56 @@ it.effect("omits interrupt results and subagent cascade for a superseded attempt }), ); +it.effect("closes old attempt text after steering supersedes its run ownership", () => + Effect.gen(function* () { + const result = yield* captureRootRunTermination({ + key: "superseded-open-text", + shouldFinalizeRun: () => Effect.succeed(false), + events: (ids) => { + const node = { + ...makeRunOwnedSubagentNodeFixture({ ids, status: "running" }), + kind: "assistant_message" as const, + }; + const message: OrchestrationV2ConversationMessage = { + id: MessageId.make("message:superseded-open-text"), + threadId: ids.threadId, + runId: ids.runId, + nodeId: node.id, + role: "assistant", + text: "Partial text", + streaming: true, + attachments: [], + createdBy: "agent", + creationSource: "provider", + createdAt: DateTime.makeUnsafe(0), + updatedAt: DateTime.makeUnsafe(0), + }; + return Stream.fromIterable([ + { type: "node.updated", driver, node }, + { type: "message.updated", driver, message }, + { + type: "turn_item.updated", + driver, + turnItem: { + ...makeLinkedChildTurnItemFixture({ ids, driver, type: "assistant_message" }), + threadId: ids.threadId, + runId: ids.runId, + nodeId: node.id, + providerTurnId: ids.rootProviderTurnId, + }, + }, + rootTerminalEvent(ids, "interrupted"), + ]); + }, + }); + assert.lengthOf(result.messages, 1); + assert.isFalse(result.messages[0]?.streaming); + assert.equal(result.written[0]?.status, "interrupted"); + assert.isFalse(result.events.some((event) => event.type === "run.updated")); + assert.deepEqual(result.observed, []); + }), +); + it.effect("emits run_interrupt_result when superseded attempt still has a hard-stop request", () => Effect.gen(function* () { const { written, observed } = yield* captureRootRunTermination({ @@ -3292,6 +3344,46 @@ it.effect("records a finished run as failed when its ownership check cannot be r }), ); +it.effect( + "closes old reasoning after a failed ownership read without changing the successor run", + () => + Effect.gen(function* () { + const result = yield* captureRootRunTermination({ + key: "superseded-finalize-guard-read-failure", + runCurrentWriteCommitted: false, + shouldFinalizeRun: () => + Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId: ThreadId.make("thread:superseded-finalize-guard-read-failure"), + cause: "database unavailable", + }), + ), + events: (ids) => + Stream.fromIterable([ + { + type: "turn_item.updated", + driver, + turnItem: { + ...makeLinkedChildTurnItemFixture({ ids, driver, type: "reasoning" }), + id: ids.itemId, + threadId: ids.threadId, + runId: ids.runId, + nodeId: ids.rootNodeId, + providerTurnId: ids.rootProviderTurnId, + }, + }, + rootTerminalEvent(ids, "completed"), + ]), + }); + assert.deepEqual(result.observed, []); + assert.isFalse(result.events.some((event) => event.type === "run.updated")); + assert.isFalse(result.events.some((event) => event.type === "provider-thread.updated")); + const reasoning = result.written.find((item) => item.type === "reasoning"); + assert.equal(reasoning?.status, "failed"); + assert.isFalse(reasoning?.type === "reasoning" && reasoning.streaming); + }), +); + it.effect("does not refresh pull requests for auxiliary or stale provider terminals", () => Effect.gen(function* () { const { observed } = yield* captureRootRunTermination({ @@ -3307,9 +3399,10 @@ it.effect("does not refresh pull requests for auxiliary or stale provider termin { ...ids, rootProviderTurnId: ProviderTurnId.make("turn:previous") }, "interrupted", ), + rootTerminalEvent(ids, "completed"), ]), }); - assert.deepEqual(observed, []); + assert.deepEqual(observed, ["run:waiting", "pull-requests-refreshed"]); }), ); @@ -3379,6 +3472,214 @@ it.effect("keeps completed runs completed when pull request refresh fails", () = }), ); +it.effect("fails a clean provider stream drain without a root terminal", () => + Effect.gen(function* () { + const result = yield* captureRootRunTermination({ + key: "clean-stream-drain", + shouldFinalizeRun: () => Effect.succeed(true), + events: () => Stream.empty, + }); + assert.deepEqual(result.observed, ["run:failed", "pull-requests-refreshed"]); + assert.lengthOf(result.written, 1); + assert.equal(result.written[0]?.status, "failed"); + }), +); + +it.effect("does not correlate an early terminal by a reused thread and run ordinal", () => + Effect.gen(function* () { + const result = yield* captureRootRunTermination({ + key: "unseeded-early-terminal", + shouldFinalizeRun: () => Effect.succeed(true), + providerTurnIdKnown: false, + events: (ids) => Stream.make(rootTerminalEvent(ids, "completed")), + }); + assert.deepEqual(result.observed, ["run:failed", "pull-requests-refreshed"]); + }), +); + +it.effect("does not write another terminal batch when start fails after root completion", () => + Effect.gen(function* () { + const rootFinished = yield* Deferred.make(); + const result = yield* captureRootRunTermination({ + key: "completed-before-start-error", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => Stream.make(rootTerminalEvent(ids, "completed")), + refreshAfterTurn: Deferred.succeed(rootFinished, undefined), + startTurn: (input) => + Deferred.await(rootFinished).pipe( + Effect.andThen( + Effect.fail( + new ProviderAdapterTurnStartError({ + driver, + threadId: input.threadId, + providerThreadId: input.providerThread.id, + runId: input.runId, + cause: "late start failure", + }), + ), + ), + ), + }); + assert.deepEqual(result.observed, ["run:waiting", "pull-requests-refreshed"]); + assert.deepEqual(result.written, []); + }), +); + +it.effect( + "cancels a blocked observer when provider startup fails without writing another terminal batch", + () => + Effect.gen(function* () { + const refreshStarted = yield* Deferred.make(); + const refreshBlocked = yield* Deferred.make(); + const refreshExited = yield* Deferred.make(); + const result = yield* captureRootRunTermination({ + key: "blocked-observer-start-error", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => Stream.make(rootTerminalEvent(ids, "completed")), + refreshAfterTurn: Deferred.succeed(refreshStarted, undefined).pipe( + Effect.andThen(Deferred.await(refreshBlocked)), + Effect.ensuring(Deferred.succeed(refreshExited, undefined)), + ), + startTurn: (input) => + Deferred.await(refreshStarted).pipe( + Effect.andThen( + Effect.fail( + new ProviderAdapterTurnStartError({ + driver, + threadId: input.threadId, + providerThreadId: input.providerThread.id, + runId: input.runId, + cause: "late start failure", + }), + ), + ), + ), + }); + yield* Deferred.await(refreshExited); + assert.deepEqual(result.observed, ["run:waiting", "pull-requests-refreshed"]); + assert.deepEqual(result.written, []); + }), +); + +it.effect.each(["interrupted", "drained"] as const)( + "flushes buffered assistant text when the run is %s", + (status) => + Effect.gen(function* () { + const result = yield* captureRootRunTermination({ + key: `flush-buffered-text:${status}`, + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => { + const message: OrchestrationV2ConversationMessage = { + id: MessageId.make(`message:buffered:${status}`), + threadId: ids.threadId, + runId: ids.runId, + nodeId: ids.rootNodeId, + role: "assistant", + text: "Partial text without a markdown boundary", + streaming: true, + attachments: [], + createdBy: "agent", + creationSource: "provider", + createdAt: DateTime.makeUnsafe(0), + updatedAt: DateTime.makeUnsafe(0), + }; + return Stream.fromIterable([ + { type: "message.updated", driver, message }, + { + type: "turn_item.updated", + driver, + turnItem: { + ...makeLinkedChildTurnItemFixture({ ids, driver, type: "assistant_message" }), + threadId: ids.threadId, + runId: ids.runId, + text: message.text, + }, + }, + ...(status === "interrupted" ? [rootTerminalEvent(ids, "interrupted")] : []), + ]); + }, + }); + assert.lengthOf(result.messages, 1); + assert.isFalse(result.messages[0]?.streaming); + assert.equal(result.messages[0]?.text, "Partial text without a markdown boundary"); + const assistantItem = result.written.find((item) => item.type === "assistant_message"); + assert.isDefined(assistantItem); + assert.equal(assistantItem?.status, status === "drained" ? "failed" : status); + if (assistantItem?.type === "assistant_message") assert.isFalse(assistantItem.streaming); + }), +); + +it.effect.each(["completed", "interrupted"] as const)( + "closes plan execution records on %s while preserving unfinished plan steps", + (status) => + Effect.gen(function* () { + const result = yield* captureRootRunTermination({ + key: `open-plan:${status}`, + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => { + const planId = PlanId.make(`plan:open:${status}`); + const node = { + ...makeRunOwnedSubagentNodeFixture({ ids, status: "running" }), + kind: "todo_list" as const, + }; + const steps = [ + { id: "pending-step", text: "Finish the work", status: "pending" as const }, + ]; + return Stream.fromIterable([ + { type: "node.updated", driver, node }, + { + type: "plan.updated", + driver, + plan: { + id: planId, + threadId: ids.threadId, + runId: ids.runId, + nodeId: node.id, + kind: "todo_list", + status: "active", + steps, + }, + }, + { + type: "turn_item.updated", + driver, + turnItem: { + id: ids.childItemId, + threadId: ids.threadId, + runId: ids.runId, + nodeId: node.id, + providerThreadId: ids.providerThreadId, + providerTurnId: ids.rootProviderTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: "Plan", + startedAt: DateTime.makeUnsafe(0), + completedAt: null, + updatedAt: DateTime.makeUnsafe(0), + type: "todo_list", + planId, + steps, + }, + }, + rootTerminalEvent(ids, status), + ]); + }, + }); + const item = result.written.find((item) => item.type === "todo_list"); + assert.isDefined(item); + assert.equal(item?.status, status); + if (item?.type === "todo_list") assert.equal(item.steps[0]?.status, "pending"); + const node = result.events.find( + (event) => event.type === "node.updated" && event.payload.kind === "todo_list", + ); + assert.isDefined(node); + if (node?.type === "node.updated") assert.equal(node.payload.status, status); + assert.isFalse(result.events.some((event) => event.type === "plan.updated")); + }), +); + function captureRootRunTermination(input: { readonly key: string; readonly shouldFinalizeRun: () => Effect.Effect; @@ -3389,6 +3690,8 @@ function captureRootRunTermination(input: { ) => Stream.Stream; readonly startTurn?: ProviderAdapterV2SessionRuntime["startTurn"]; readonly refreshAfterTurn?: Effect.Effect; + readonly providerTurnIdKnown?: boolean; + readonly runCurrentWriteCommitted?: boolean; }) { return Effect.gen(function* () { const ids = backgroundScenarioIds(input.key); @@ -3400,13 +3703,16 @@ function captureRootRunTermination(input: { driver, status: "running", }); + const writtenEvents = yield* Ref.make>([]); const writtenItems = yield* Ref.make>([]); + const writtenMessages = yield* Ref.make>([]); const observed = yield* Ref.make>([]); const ingestionDone = yield* Deferred.make(); const captureTurnItem = (payload: OrchestrationV2TurnItem) => Ref.update(writtenItems, (current) => [...current, payload]); const captureFinalEvents = (events: ReadonlyArray) => Effect.gen(function* () { + yield* Ref.update(writtenEvents, (current) => [...current, ...events]); for (const event of events) { if (event.type === "turn-item.updated") { yield* captureTurnItem(event.payload); @@ -3414,6 +3720,9 @@ function captureRootRunTermination(input: { if (event.type === "run.updated") { yield* Ref.update(observed, (current) => [...current, `run:${event.payload.status}`]); } + if (event.type === "message.updated") { + yield* Ref.update(writtenMessages, (current) => [...current, event.payload]); + } } }); const testLayer = RunExecutionService.layer.pipe( @@ -3432,9 +3741,11 @@ function captureRootRunTermination(input: { }), writeWithEffects: (payload) => captureFinalEvents(payload.events).pipe(Effect.as([])), writeIfRunCurrent: (payload) => - captureFinalEvents(payload.events).pipe( - Effect.as({ committed: true, storedEvents: [] }), - ), + input.runCurrentWriteCommitted === false + ? Effect.succeed({ committed: false, storedEvents: [] }) + : captureFinalEvents(payload.events).pipe( + Effect.as({ committed: true, storedEvents: [] }), + ), }), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ @@ -3510,7 +3821,7 @@ function captureRootRunTermination(input: { } as OrchestrationV2ProviderThread, attempt: { id: ids.attemptId, - providerTurnId: ids.rootProviderTurnId, + providerTurnId: input.providerTurnIdKnown === false ? null : ids.rootProviderTurnId, } as OrchestrationV2RunAttempt, attemptId: ids.attemptId, providerTurnOrdinal: 1, @@ -3543,7 +3854,12 @@ function captureRootRunTermination(input: { }).pipe(Effect.provide(testLayer)); yield* Deferred.await(ingestionDone); - return { written: yield* Ref.get(writtenItems), observed: yield* Ref.get(observed) }; + return { + events: yield* Ref.get(writtenEvents), + written: yield* Ref.get(writtenItems), + messages: yield* Ref.get(writtenMessages), + observed: yield* Ref.get(observed), + }; }); } diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 6aff5da24f1e..eaa881923021 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -8,6 +8,7 @@ import { type NodeId, type OrchestrationV2AppThread, type OrchestrationV2CheckpointScope, + type OrchestrationV2ConversationMessage, type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderFailure, @@ -33,6 +34,7 @@ import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; @@ -562,9 +564,11 @@ export const layer: Layer.Layer< readonly shouldFinalizeRun?: () => Effect.Effect; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; + readonly openMessages?: ReadonlyArray; + readonly openTurnItems?: ReadonlyArray; + readonly openNodes?: ReadonlyArray; readonly terminal: ProviderTerminalEvent; readonly failureItemPersisted: boolean; - readonly refreshAfterTurn: Effect.Effect; readonly writeIfRunCurrent?: { readonly activeAttemptId: RunAttemptId; readonly expectedStatus: OrchestrationV2Run["status"]; @@ -577,9 +581,62 @@ export const layer: Layer.Layer< status: input.terminal.status, completedAt, }; + const allocateEventId = () => idAllocator.allocate.event({ threadId: input.run.threadId }); + const artifactEvents = [ + ...(yield* Effect.forEach(input.openNodes ?? [], (node) => + Effect.gen(function* () { + return { + id: yield* allocateEventId(), + type: "node.updated" as const, + threadId: input.run.threadId, + runId: input.run.id, + nodeId: node.id, + occurredAt: completedAt, + payload: { ...node, status: input.terminal.status, completedAt }, + }; + }), + )), + ...(yield* Effect.forEach(input.openMessages ?? [], (message) => + Effect.gen(function* () { + return { + id: yield* allocateEventId(), + type: "message.updated" as const, + threadId: input.run.threadId, + runId: input.run.id, + occurredAt: completedAt, + payload: { ...message, streaming: false, updatedAt: completedAt }, + }; + }), + )), + ...(yield* Effect.forEach(input.openTurnItems ?? [], (item) => + Effect.gen(function* () { + return { + id: yield* allocateEventId(), + type: "turn-item.updated" as const, + threadId: input.run.threadId, + runId: input.run.id, + occurredAt: completedAt, + payload: { + ...item, + ...("streaming" in item ? { streaming: false } : {}), + status: input.terminal.status, + completedAt, + updatedAt: completedAt, + }, + }; + }), + )), + ]; const shouldFinalizeRun = input.shouldFinalizeRun === undefined ? true : yield* input.shouldFinalizeRun(); if (!shouldFinalizeRun) { + if (artifactEvents.length > 0) { + yield* eventSink.writeWithEffects({ + events: artifactEvents, + effects: [], + guardNodeRootId: input.rootNode.id, + }); + } // Superseded attempt (steer / selection restart). Emit // run_interrupt_result only when hard Stop left an unpaired request // for this run; plain steers and already-paired stops emit nothing. @@ -610,12 +667,11 @@ export const layer: Layer.Layer< }, ], }); - yield* input.refreshAfterTurn; + return true; } } - return; + return false; } - const allocateEventId = () => idAllocator.allocate.event({ threadId: input.run.threadId }); const open = input.openRunOwnedSubagents ?? emptyOpenRunOwnedSubagentProjection(); const hasOpenSubagentProjection = open.subagents.size > 0 || @@ -683,6 +739,7 @@ export const layer: Layer.Layer< ] : [], events: [ + ...artifactEvents, // Terminalize open run-owned subagent rows before the root run // settles so projections never keep a forever-running subagent card. ...cascadedSubagentEvents, @@ -782,14 +839,22 @@ export const layer: Layer.Layer< activeAttemptId: input.writeIfRunCurrent.activeAttemptId, expectedStatus: input.writeIfRunCurrent.expectedStatus, events: finalization.events, + effects: finalization.effects, }); if (!result.committed) { - return; + if (artifactEvents.length > 0) { + yield* eventSink.writeWithEffects({ + events: artifactEvents, + effects: [], + guardNodeRootId: input.rootNode.id, + }); + } + return false; } } else { yield* eventSink.writeWithEffects(finalization); } - yield* input.refreshAfterTurn; + return true; }); return RunExecutionServiceV2.of({ @@ -866,7 +931,7 @@ export const layer: Layer.Layer< runId: input.run.id, cause, }); - yield* writeFinalRunEvents({ + const shouldRefresh = yield* writeFinalRunEvents({ run: input.run, rootNode: input.rootNode, checkpointScope: input.checkpointScope, @@ -883,12 +948,12 @@ export const layer: Layer.Layer< input.providerTurnOrdinal * 100 + 1, ), failureItemPersisted: false, - refreshAfterTurn, writeIfRunCurrent: { activeAttemptId: input.attemptId, expectedStatus: "running", }, }); + if (shouldRefresh) yield* refreshAfterTurn; return null; }), ), @@ -948,6 +1013,13 @@ export const layer: Layer.Layer< ); const rootTerminalSeen = yield* Ref.make(false); const rootRunFinalized = yield* Ref.make(false); + const finalizationPermit = yield* Semaphore.make(1); + const openMessages = new Map< + OrchestrationV2ConversationMessage["id"], + OrchestrationV2ConversationMessage + >(); + const openTurnItems = new Map(); + const openNodes = new Map(); const providerThreadOwnerLost = yield* Ref.make(false); const activeChildProviderTurns = yield* Ref.make>(new Set()); const activeChildSubagents = yield* Ref.make>(new Set()); @@ -955,31 +1027,44 @@ export const layer: Layer.Layer< ReadonlySet >(new Set(inheritedBackgroundTurnItemsById.keys())); const openRunOwnedSubagents = yield* Ref.make(emptyOpenRunOwnedSubagentProjection()); - const finalizeRootRun = (terminal: ProviderTerminalEvent) => + const finalizeRootRun = ( + terminal: ProviderTerminalEvent, + failureItemPersisted = terminal.status === "failed", + guardRunCurrent = false, + ) => Effect.gen(function* () { if (yield* Ref.get(rootRunFinalized)) { - return; + return false; } const providerThread = yield* Ref.get(latestProviderThread); const openSubagents = yield* Ref.get(openRunOwnedSubagents); - yield* writeFinalRunEvents({ + const shouldRefresh = yield* writeFinalRunEvents({ run: input.run, rootNode: input.rootNode, checkpointScope: input.checkpointScope, providerThread, attempt: input.attempt, - ...(input.shouldFinalizeRun === undefined - ? {} - : { shouldFinalizeRun: input.shouldFinalizeRun }), + ...(guardRunCurrent + ? { + writeIfRunCurrent: { + activeAttemptId: input.attempt.id, + expectedStatus: "running" as const, + }, + } + : input.shouldFinalizeRun === undefined + ? {} + : { shouldFinalizeRun: input.shouldFinalizeRun }), ...(input.hasUnpairedRunInterruptRequest === undefined ? {} : { hasUnpairedRunInterruptRequest: input.hasUnpairedRunInterruptRequest, }), openRunOwnedSubagents: openSubagents, + openMessages: [...openMessages.values()], + openTurnItems: [...openTurnItems.values()], + openNodes: [...openNodes.values()], terminal, - failureItemPersisted: terminal.status === "failed", - refreshAfterTurn, + failureItemPersisted, }).pipe( Effect.mapError( (cause) => new RunExecutionIngestError({ runId: input.run.id, cause }), @@ -989,7 +1074,15 @@ export const layer: Layer.Layer< yield* Ref.set(openRunOwnedSubagents, emptyOpenRunOwnedSubagentProjection()); } yield* Ref.set(rootRunFinalized, true); - }); + openMessages.clear(); + openTurnItems.clear(); + openNodes.clear(); + return shouldRefresh; + }).pipe( + finalizationPermit.withPermits(1), + Effect.uninterruptible, + Effect.flatMap((shouldRefresh) => (shouldRefresh ? refreshAfterTurn : Effect.void)), + ); const trackChildLifecycle = (event: ProviderAdapterV2Event, deliverable: boolean) => Effect.gen(function* () { const routing = yield* Ref.get(eventRouting); @@ -1175,6 +1268,34 @@ export const layer: Layer.Layer< ), Stream.tap((event) => Effect.gen(function* () { + if ( + event.type === "node.updated" && + event.node.runId === input.run.id && + (event.node.kind === "assistant_message" || + event.node.kind === "reasoning" || + event.node.kind === "plan" || + event.node.kind === "todo_list") + ) { + if (isOpenExecutionNodeStatus(event.node.status)) + openNodes.set(event.node.id, event.node); + else openNodes.delete(event.node.id); + } + if (event.type === "message.updated" && event.message.runId === input.run.id) { + if (event.message.streaming) openMessages.set(event.message.id, event.message); + else openMessages.delete(event.message.id); + } + if ( + event.type === "turn_item.updated" && + event.turnItem.runId === input.run.id && + (event.turnItem.type === "assistant_message" || + event.turnItem.type === "reasoning" || + event.turnItem.type === "proposed_plan" || + event.turnItem.type === "todo_list") + ) { + if (isSettledTurnItemStatus(event.turnItem.status)) + openTurnItems.delete(event.turnItem.id); + else openTurnItems.set(event.turnItem.id, event.turnItem); + } let storedEventCount = 0; const deliveredEvent = filterAssistantEvent( event, @@ -1261,67 +1382,37 @@ export const layer: Layer.Layer< Effect.gen(function* () { const terminal = yield* Ref.get(terminalEvent); if (terminal === null) { + yield* finalizeRootRun( + makeFailedTerminalEvent( + makeProviderFailure({ + message: "The provider event stream closed before the turn finished.", + class: "unknown", + }), + (yield* Ref.get(latestTurnItemOrdinal)) + 1, + ), + false, + ); return; } yield* finalizeRootRun(terminal); }), ), Effect.catchCause((cause) => - Ref.get(rootRunFinalized).pipe( - Effect.flatMap((finalized) => - Effect.logWarning("orchestration V2 provider event ingestion failed", { - runId: input.run.id, - cause, - }).pipe( - Effect.andThen( - finalized - ? Effect.void - : Ref.get(latestProviderThread).pipe( - Effect.flatMap((providerThread) => - Ref.get(latestTurnItemOrdinal).pipe( - Effect.flatMap((latestItemOrdinal) => - Ref.get(openRunOwnedSubagents).pipe( - Effect.flatMap((openSubagents) => - writeFinalRunEvents({ - run: input.run, - rootNode: input.rootNode, - checkpointScope: input.checkpointScope, - providerThread, - attempt: input.attempt, - // The failure may be the ownership - // read itself, so check in the write. - writeIfRunCurrent: { - activeAttemptId: input.attempt.id, - expectedStatus: "running", - }, - openRunOwnedSubagents: openSubagents, - terminal: makeFailedTerminalEvent( - makeProviderFailure({ - cause: Cause.squash(cause), - class: "unknown", - }), - latestItemOrdinal + 1, - ), - failureItemPersisted: false, - refreshAfterTurn, - }), - ), - ), - ), - ), - ), - ), - ), - Effect.mapError( - (writeCause) => - new RunExecutionIngestError({ - runId: input.run.id, - cause: { ingest: cause, write: writeCause }, - }), - ), + Effect.gen(function* () { + if (Cause.hasInterruptsOnly(cause)) return yield* Effect.failCause(cause); + yield* Effect.logWarning("orchestration V2 provider event ingestion failed", { + runId: input.run.id, + cause, + }); + yield* finalizeRootRun( + makeFailedTerminalEvent( + makeProviderFailure({ cause: Cause.squash(cause), class: "unknown" }), + (yield* Ref.get(latestTurnItemOrdinal)) + 1, ), - ), - ), + false, + true, + ); + }), ), Effect.ensuring(eventSubscription.close), Effect.forkDetach, @@ -1376,56 +1467,43 @@ export const layer: Layer.Layer< )) : input.session.startTurn(turnInput); yield* Effect.andThen(shouldStart, startTurn).pipe( + Effect.mapError( + (cause) => + new RunExecutionStartError({ + commandId: input.commandId, + runId: input.run.id, + cause, + }), + ), Effect.catchCause((cause) => - Effect.logError("orchestration V2 provider turn start failed", { - runId: input.run.id, - cause, - }).pipe( - Effect.andThen(Fiber.interrupt(providerEventFiber)), - Effect.andThen(Ref.get(latestProviderThread)), - Effect.flatMap((providerThread) => - Ref.get(latestTurnItemOrdinal).pipe( - Effect.flatMap((latestItemOrdinal) => - Ref.get(openRunOwnedSubagents).pipe( - Effect.flatMap((openSubagents) => - writeFinalRunEvents({ - run: input.run, - rootNode: input.rootNode, - checkpointScope: input.checkpointScope, - providerThread, - attempt: input.attempt, - // Checked in the write transaction, not by another - // read that can fail like the one before the start. - writeIfRunCurrent: { - activeAttemptId: input.attempt.id, - expectedStatus: "running", - }, - openRunOwnedSubagents: openSubagents, - terminal: makeFailedTerminalEvent( - makeProviderFailure({ - cause: Cause.squash(cause), - // A failed ownership read is not the provider's fault. - class: Exit.isFailure(shouldStart) ? "unknown" : "provider_error", - }), - latestItemOrdinal + 1, - ), - failureItemPersisted: false, - refreshAfterTurn, - }), - ), - ), - ), - ), - ), - Effect.mapError( - (writeCause) => - new RunExecutionStartError({ - commandId: input.commandId, - runId: input.run.id, - cause: { start: cause, write: writeCause }, + Effect.gen(function* () { + if (Cause.hasInterruptsOnly(cause)) return yield* Effect.failCause(cause); + yield* Effect.logError("orchestration V2 provider turn start failed", { + runId: input.run.id, + cause, + }); + yield* Fiber.interrupt(providerEventFiber); + yield* finalizeRootRun( + makeFailedTerminalEvent( + makeProviderFailure({ + cause: Cause.squash(cause), + class: Exit.isFailure(shouldStart) ? "unknown" : "provider_error", }), - ), - ), + (yield* Ref.get(latestTurnItemOrdinal)) + 1, + ), + false, + true, + ).pipe( + Effect.mapError( + (writeCause) => + new RunExecutionStartError({ + commandId: input.commandId, + runId: input.run.id, + cause: { start: cause, write: writeCause }, + }), + ), + ); + }), ), ); }), diff --git a/apps/server/src/orchestration-v2/RunFinalizationService.test.ts b/apps/server/src/orchestration-v2/RunFinalizationService.test.ts index a2e8f50901f4..7a77fd4d4e3f 100644 --- a/apps/server/src/orchestration-v2/RunFinalizationService.test.ts +++ b/apps/server/src/orchestration-v2/RunFinalizationService.test.ts @@ -7,6 +7,9 @@ import { } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Exit from "effect/Exit"; +import * as Cause from "effect/Cause"; +import * as Schema from "effect/Schema"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts"; @@ -15,6 +18,8 @@ import * as CheckpointCapture from "./CheckpointCaptureService.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as RunFinalization from "./RunFinalizationService.ts"; +const isRunFinalizationError = Schema.is(RunFinalization.RunFinalizationError); + it.effect("refreshes workspace after checkpoint capture without reading history", () => { const threadId = ThreadId.make("thread_finalize"); const runId = RunId.make("run_finalize"); @@ -129,3 +134,51 @@ it.effect.each( assert.deepEqual(refreshed, [...scenario.expected]); }).pipe(Effect.provide(layer)); }); + +it.effect.each(["missing-scope", "read-failure"] as const)( + "reports %s separately from workspace refresh failures", + (scenario) => { + const threadId = ThreadId.make("thread:finalization-error"); + const runId = RunId.make("run:finalization-error"); + const scopeId = CheckpointScopeId.make("scope:finalization-error"); + const refresh = vi.fn(() => Effect.void); + const layer = RunFinalization.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(CheckpointCapture.CheckpointCaptureServiceV2)({ execute: () => Effect.void }), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getCheckpointContext: () => + scenario === "missing-scope" + ? Effect.succeed({ runs: [], checkpointScopes: [], checkpoints: [] }) + : Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId, + cause: "read failed", + }), + ), + }), + Layer.succeed(RunFinalization.RunFinalizationObserver, { + refresh, + refreshAfterTurn: () => Effect.void, + }), + ), + ), + ); + return Effect.gen(function* () { + const service = yield* RunFinalization.RunFinalizationService; + const result = yield* Effect.exit(service.finalize({ threadId, runId, scopeId })); + assert.isTrue(Exit.isFailure(result)); + if (Exit.isFailure(result)) { + const error = Cause.squash(result.cause); + assert.isTrue(isRunFinalizationError(error)); + if (isRunFinalizationError(error)) { + assert.equal( + error.operation, + scenario === "missing-scope" ? "missing-checkpoint-scope" : "read-checkpoint-context", + ); + } + } + assert.equal(refresh.mock.calls.length, 0); + }).pipe(Effect.provide(layer)); + }, +); diff --git a/apps/server/src/orchestration-v2/RunFinalizationService.ts b/apps/server/src/orchestration-v2/RunFinalizationService.ts index d4a5dc342005..079a7982dee9 100644 --- a/apps/server/src/orchestration-v2/RunFinalizationService.ts +++ b/apps/server/src/orchestration-v2/RunFinalizationService.ts @@ -16,7 +16,12 @@ export class RunFinalizationError extends Schema.TaggedError new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }), + (cause) => + new RunFinalizationError({ ...input, operation: "read-checkpoint-context", cause }), ), ); const cwd = projection.checkpointScopes.find((scope) => scope.id === input.scopeId)?.cwd; - if (cwd !== undefined) { - yield* observer - .refresh({ cwd, threadId: input.threadId, runId: input.runId }) - .pipe( - Effect.mapError( - (cause) => - new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }), - ), - ); + if (cwd === undefined) { + return yield* new RunFinalizationError({ + ...input, + operation: "missing-checkpoint-scope", + cause: `Checkpoint scope ${input.scopeId} was not found after capture.`, + }); } + yield* observer + .refresh({ cwd, threadId: input.threadId, runId: input.runId }) + .pipe( + Effect.mapError( + (cause) => new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }), + ), + ); }); return RunFinalizationService.of({ finalize }); }); diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 6a8dded63caa..bf4988564d3d 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -417,7 +417,10 @@ const composerSelection = { options: [...runSelection.options, { id: "fastMode", value: false }], }; -const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* (name: string) { +const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* ( + name: string, + supportsRestart = false, +) { const cwd = yield* checkpointWorkspace(name); const events = yield* Queue.unbounded(); const started: ProviderAdapterV2TurnInput[] = []; @@ -427,7 +430,7 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* turns: { ...CodexProviderCapabilitiesV2.turns, supportsActiveSteering: true, - supportsSteeringByInterruptRestart: false, + supportsSteeringByInterruptRestart: supportsRestart, }, }; const adapter: ProviderAdapterV2Shape = { @@ -477,6 +480,9 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* startTurn: (turn) => Effect.gen(function* () { started.push(turn); + // The restart test releases this snapshot after stale predecessor + // frames, while production routing still has no current root ID. + if (supportsRestart && started.length > 1) return; yield* Queue.offer(events, { type: "provider_turn.updated", driver, @@ -501,7 +507,42 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* Effect.sync(() => { steered.push(turn.message.text); }), - interruptTurn: () => Effect.void, + interruptTurn: () => + Effect.gen(function* () { + if (!supportsRestart) return; + const turn = started.at(-1)!; + const completedAt = yield* DateTime.now; + const providerTurnId = ProviderTurnId.make(`provider-turn:${turn.attemptId}`); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: providerTurnId, + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: { + driver, + nativeId: `native:${turn.attemptId}`, + strength: "strong", + }, + ordinal: turn.providerTurnOrdinal, + status: "interrupted", + startedAt: now, + completedAt, + }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThread.id, + providerTurnId, + runOrdinal: turn.runOrdinal, + status: "interrupted", + failure: null, + threadDisposition: "reusable", + }); + }), respondToRuntimeRequest: () => Effect.void, readThreadSnapshot: () => Effect.die("unused"), rollbackThread: () => Effect.die("unused"), @@ -560,6 +601,175 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* return { events, started, steered, layer, startFirstTurn }; }); +it.effect("learns a restarted production attempt's root and rejects its predecessor terminal", () => + Effect.scoped( + Effect.gen(function* () { + const { events, started, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + "production-root-identity-restart", + true, + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + const firstProjection = yield* orchestrator.getThreadProjection(threadId); + assert.isNull(firstProjection.attempts[0]?.providerTurnId); + const oldTurn = firstProjection.providerTurns[0]!; + const replacementRunning = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.status === "running" && + event.payload.runAttemptId !== first.attemptId, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const admitted = yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("identity-restart"), + threadId, + messageId: MessageId.make("message:identity-restart"), + text: "restart", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "restart_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + const admission = admitted.storedEvents.find( + (stored) => stored.event.type === "run-attempt.created", + ); + assert.isDefined(admission); + if (admission.event.type !== "run-attempt.created") + return yield* Effect.die("missing attempt"); + assert.isNull(admission.event.payload.providerTurnId); + yield* worker.drain(); + assert.lengthOf(started, 2); + const second = started[1]!; + assert.equal(second.providerThread.id, first.providerThread.id); + assert.equal(second.runOrdinal, first.runOrdinal); + assert.notEqual(second.attemptId, first.attemptId); + assert.equal(second.attemptId, admission.event.payload.id); + const restartedAt = yield* DateTime.now; + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...oldTurn, status: "completed", completedAt: restartedAt }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: oldTurn.providerThreadId, + providerTurnId: oldTurn.id, + runOrdinal: second.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + ...oldTurn, + id: ProviderTurnId.make(`provider-turn:${second.attemptId}`), + runAttemptId: second.attemptId, + nodeId: second.rootNodeId, + ordinal: second.providerTurnOrdinal, + nativeTurnRef: { driver, nativeId: `native:${second.attemptId}`, strength: "strong" }, + status: "running", + startedAt: restartedAt, + completedAt: null, + }, + }); + yield* Fiber.join(replacementRunning); + const running = yield* orchestrator.getThreadProjection(threadId); + assert.equal(running.runs[0]?.status, "running"); + assert.equal(running.runs[0]?.activeAttemptId, second.attemptId); + const currentTurn = running.providerTurns.find( + (turn) => turn.runAttemptId === second.attemptId, + )!; + assert.isDefined(currentTurn); + + // Both frames are from the superseded attempt. A fresh current snapshot + // is the persisted receipt that this subscriber drained them in order. + const completedAt = yield* DateTime.now; + const receipt = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.id === currentTurn.id && + event.payload.completedAt !== null, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...oldTurn, status: "completed", completedAt }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: oldTurn.providerThreadId, + providerTurnId: oldTurn.id, + runOrdinal: second.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...currentTurn, status: "completed", completedAt }, + }); + yield* Fiber.join(receipt); + const afterStale = yield* orchestrator.getThreadProjection(threadId); + assert.equal(afterStale.runs[0]?.status, "running"); + assert.equal(afterStale.runs[0]?.activeAttemptId, second.attemptId); + assert.deepEqual( + afterStale.attempts.map((attempt) => attempt.status), + ["superseded", "running"], + ); + + const finalized = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === second.runId && + event.payload.status === "waiting", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: currentTurn.providerThreadId, + providerTurnId: currentTurn.id, + runOrdinal: second.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(finalized); + yield* worker.drain(); + const settled = yield* orchestrator.getThreadProjection(threadId); + assert.equal(settled.runs[0]?.status, "completed"); + assert.deepEqual( + settled.attempts.map((attempt) => attempt.status), + ["superseded", "completed"], + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + it.effect("steers a changed turn-scoped selection into a provider that cannot restart", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayFixtures.integration.test.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayFixtures.integration.test.ts index dd417ef162e0..1a03bd005030 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayFixtures.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayFixtures.integration.test.ts @@ -81,6 +81,35 @@ function isStreamingAssistantEvent(event: OrchestrationV2DomainEvent): boolean { } } +// These are production Orchestrator admissions, not pre-seeded routing inputs. +// Raw snapshot-before-terminal ordering is covered at the adapter boundary; +// here a successful finalization proves the initially unknown root was learned. +function assertRootIdentityLearnedFromSnapshot(events: ReadonlyArray) { + assert.isTrue(events.some((event) => event.type === "run-attempt.created")); + for (const [index, event] of events.entries()) { + if (event.type !== "run-attempt.created" && event.type !== "run-attempt.updated") continue; + const attempt = event.payload; + const admissionIndex = events.findIndex( + (candidate) => + candidate.type === "run-attempt.created" && candidate.payload.id === attempt.id, + ); + assert.isAtLeast(admissionIndex, 0, `${attempt.id} needs a production admission`); + if (index === admissionIndex) { + assert.isNull(attempt.providerTurnId, `${attempt.id} must enter with unknown root identity`); + } + if (attempt.status !== "completed" && attempt.status !== "interrupted") continue; + const snapshotIndex = events.findIndex( + (candidate) => + candidate.type === "provider-turn.updated" && + candidate.payload.runAttemptId === attempt.id && + candidate.payload.nodeId === attempt.rootNodeId && + candidate.payload.providerThreadId === attempt.providerThreadId, + ); + assert.isAbove(snapshotIndex, admissionIndex, `${attempt.id} needs an admitted root snapshot`); + assert.isBelow(snapshotIndex, index, `${attempt.id} must learn its root before finalization`); + } +} + const runFixtureProvider = Effect.fn("runOrchestratorReplayFixture")(function* < Transcript extends ProviderReplayTranscript, Error, @@ -128,6 +157,7 @@ const runFixtureProvider = Effect.fn("runOrchestratorReplayFixture")(function* < input.driver.runContinuationWorker === true ? { runContinuationWorker: true } : {}, ).pipe(provideDeterministicTestRuntime); input.driver.assertOutput(result, transcript); + assertRootIdentityLearnedFromSnapshot(result.domainEvents); assertProviderNativeSubagentRootTurns(result); const expectedAbsentWorkspacePaths = input.driver.expectedAbsentWorkspacePaths; if (expectedAbsentWorkspacePaths !== undefined) { diff --git a/docs/user/composer.md b/docs/user/composer.md index 8cb8d7805ac9..45aaae754f33 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -55,6 +55,10 @@ running the send button shows which action it will take. Long-press it to use th other action for a single message, or hold `Cmd` while sending from a hardware keyboard. The button only offers Steer when the running agent supports it. +When you stop or steer a turn, partial assistant text stays in the conversation. +Its streaming indicator ends with that turn; newer responses keep streaming until +their own turn finishes. + ## Queue messages offline on mobile Mobile keeps local copies of draft attachments, so you can preview them and queue