diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index e3b18d74a9a7..ac5c45f3233e 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -241,6 +241,306 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { ); }); +it.layer(makeProjectionPipelinePrefixedTestLayer("t3-abort-rebuild-test-"))( + "OrchestrationProjectionPipeline abort rebuild", + (it) => { + it.effect( + "rebuilds interrupted lifecycle and pending requests identically from persisted events", + () => + Effect.gen(function* () { + const eventStore = yield* OrchestrationEventStore; + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-abort-rebuild"); + const turnId = TurnId.make("turn-abort-rebuild"); + const createdAt = "2026-08-10T12:00:00.000Z"; + const startedAt = "2026-08-10T12:00:01.000Z"; + const approvalAt = "2026-08-10T12:00:02.000Z"; + const userInputAt = "2026-08-10T12:00:03.000Z"; + const abortedAt = "2026-08-10T12:00:04.000Z"; + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); + + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.make("evt-abort-rebuild-created"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: createdAt, + commandId: CommandId.make("cmd-abort-rebuild-created"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-abort-rebuild-created"), + metadata: {}, + payload: { + threadId, + projectId: ProjectId.make("project-abort-rebuild"), + title: "Abort rebuild", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "approval-required", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt, + updatedAt: createdAt, + }, + }); + yield* appendAndProject({ + type: "thread.session-set", + eventId: EventId.make("evt-abort-rebuild-running"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: startedAt, + commandId: CommandId.make("cmd-abort-rebuild-running"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-abort-rebuild-running"), + metadata: {}, + payload: { + threadId, + session: { + threadId, + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: turnId, + lastError: null, + updatedAt: startedAt, + }, + }, + }); + yield* appendAndProject({ + type: "thread.activity-appended", + eventId: EventId.make("evt-abort-rebuild-approval"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: approvalAt, + commandId: CommandId.make("cmd-abort-rebuild-approval"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-abort-rebuild-approval"), + metadata: {}, + payload: { + threadId, + activity: { + id: EventId.make("activity-abort-rebuild-approval"), + tone: "approval", + kind: "approval.requested", + summary: "Command approval requested", + payload: { + requestId: "approval-abort-rebuild", + requestKind: "command", + }, + turnId, + createdAt: approvalAt, + }, + }, + }); + yield* appendAndProject({ + type: "thread.activity-appended", + eventId: EventId.make("evt-abort-rebuild-user-input"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: userInputAt, + commandId: CommandId.make("cmd-abort-rebuild-user-input"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-abort-rebuild-user-input"), + metadata: {}, + payload: { + threadId, + activity: { + id: EventId.make("activity-abort-rebuild-user-input"), + tone: "info", + kind: "user-input.requested", + summary: "User input requested", + payload: { + requestId: "user-input-abort-rebuild", + questions: [ + { + id: "choice", + header: "Choice", + question: "Continue?", + options: [{ label: "Yes", description: "Continue" }], + }, + ], + }, + turnId, + createdAt: userInputAt, + }, + }, + }); + yield* appendAndProject({ + type: "thread.session-set", + eventId: EventId.make("evt-abort-rebuild-interrupted"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: abortedAt, + commandId: CommandId.make("cmd-abort-rebuild-interrupted"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-abort-rebuild-interrupted"), + metadata: {}, + payload: { + threadId, + session: { + threadId, + status: "interrupted", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: abortedAt, + }, + }, + }); + + const readLifecycle = () => + Effect.all({ + threads: sql<{ + readonly latestTurnId: string | null; + readonly pendingApprovalCount: number; + readonly pendingUserInputCount: number; + readonly updatedAt: string; + }>` + SELECT + latest_turn_id AS "latestTurnId", + pending_approval_count AS "pendingApprovalCount", + pending_user_input_count AS "pendingUserInputCount", + updated_at AS "updatedAt" + FROM projection_threads + WHERE thread_id = ${threadId} + `, + sessions: sql<{ + readonly status: string; + readonly activeTurnId: string | null; + readonly lastError: string | null; + readonly updatedAt: string; + }>` + SELECT + status, + active_turn_id AS "activeTurnId", + last_error AS "lastError", + updated_at AS "updatedAt" + FROM projection_thread_sessions + WHERE thread_id = ${threadId} + `, + turns: sql<{ + readonly turnId: string | null; + readonly state: string; + readonly requestedAt: string; + readonly startedAt: string | null; + readonly completedAt: string | null; + }>` + SELECT + turn_id AS "turnId", + state, + requested_at AS "requestedAt", + started_at AS "startedAt", + completed_at AS "completedAt" + FROM projection_turns + WHERE thread_id = ${threadId} + ORDER BY requested_at ASC, turn_id ASC + `, + activities: sql<{ + readonly activityId: string; + readonly kind: string; + readonly turnId: string | null; + }>` + SELECT + activity_id AS "activityId", + kind, + turn_id AS "turnId" + FROM projection_thread_activities + WHERE thread_id = ${threadId} + ORDER BY created_at ASC, activity_id ASC + `, + pendingApprovals: sql<{ + readonly requestId: string; + readonly status: string; + readonly turnId: string | null; + }>` + SELECT + request_id AS "requestId", + status, + turn_id AS "turnId" + FROM projection_pending_approvals + WHERE thread_id = ${threadId} + ORDER BY request_id ASC + `, + }); + + const live = yield* readLifecycle(); + assert.deepEqual(live.threads, [ + { + latestTurnId: turnId, + pendingApprovalCount: 1, + pendingUserInputCount: 1, + updatedAt: abortedAt, + }, + ]); + assert.deepEqual(live.sessions, [ + { + status: "interrupted", + activeTurnId: null, + lastError: null, + updatedAt: abortedAt, + }, + ]); + assert.deepEqual(live.turns, [ + { + turnId, + state: "interrupted", + requestedAt: startedAt, + startedAt, + completedAt: abortedAt, + }, + ]); + + yield* sql.withTransaction( + Effect.gen(function* () { + yield* sql` + DELETE FROM projection_pending_approvals + WHERE thread_id = ${threadId} + `; + yield* sql` + DELETE FROM projection_thread_activities + WHERE thread_id = ${threadId} + `; + yield* sql` + DELETE FROM projection_thread_sessions + WHERE thread_id = ${threadId} + `; + yield* sql` + DELETE FROM projection_turns + WHERE thread_id = ${threadId} + `; + yield* sql` + DELETE FROM projection_threads + WHERE thread_id = ${threadId} + `; + yield* sql` + DELETE FROM projection_state + WHERE projector IN ( + ${ORCHESTRATION_PROJECTOR_NAMES.threadActivities}, + ${ORCHESTRATION_PROJECTOR_NAMES.threadSessions}, + ${ORCHESTRATION_PROJECTOR_NAMES.threadTurns}, + ${ORCHESTRATION_PROJECTOR_NAMES.pendingApprovals}, + ${ORCHESTRATION_PROJECTOR_NAMES.threads} + ) + `; + }), + ); + + yield* projectionPipeline.bootstrap; + + const rebuilt = yield* readLifecycle(); + assert.deepEqual(rebuilt, live); + }), + ); + }, +); + it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-base-")))( "OrchestrationProjectionPipeline", (it) => { diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 258aa010e3e6..eb8951af7797 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -316,12 +316,31 @@ describe("ProviderRuntimeIngestion", () => { engine, dispatch, readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()), + readShell: () => Effect.runPromise(snapshotQuery.getShellSnapshot()), + readEvents: () => + Effect.runPromise( + Stream.runCollect(engine.readEvents(0)).pipe(Effect.map((events) => Array.from(events))), + ), emit: provider.emit, setProviderSession: provider.setSession, drain, }; } + async function drainAndRead(harness: Awaited>) { + await harness.drain(); + const threadId = asThreadId("thread-1"); + const [detailSnapshot, shellSnapshot] = await Promise.all([ + harness.readModel(), + harness.readShell(), + ]); + const detail = detailSnapshot.threads.find((entry) => entry.id === threadId); + const shell = shellSnapshot.threads.find((entry) => entry.id === threadId); + expect(detail).toBeDefined(); + expect(shell).toBeDefined(); + return { detail: detail!, shell: shell! }; + } + it("maps turn started/completed events into thread session updates", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; @@ -364,6 +383,565 @@ describe("ProviderRuntimeIngestion", () => { expect(thread.session?.lastError).toBe("turn failed"); }); + it.each(["codex", "opencode"] as const)( + "projects an aborted active turn as interrupted for %s", + async (provider) => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-aborted"); + const startedAt = "2026-01-01T00:00:01.000Z"; + const abortedAt = "2026-01-01T00:00:02.000Z"; + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-before-abort"), + provider: ProviderDriverKind.make(provider), + threadId, + createdAt: startedAt, + turnId, + }); + await harness.drain(); + + let thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId); + expect(thread?.session?.status).toBe("running"); + expect(thread?.session?.activeTurnId).toBe(turnId); + expect(thread?.latestTurn?.state).toBe("running"); + expect(thread?.latestTurn?.completedAt).toBeNull(); + + harness.emit({ + type: "turn.aborted", + eventId: asEventId("evt-turn-aborted"), + provider: ProviderDriverKind.make(provider), + threadId, + createdAt: abortedAt, + turnId, + payload: { reason: "Interrupted by user." }, + }); + await harness.drain(); + + thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId); + expect(thread?.session?.status).toBe("interrupted"); + expect(thread?.session?.activeTurnId).toBeNull(); + expect(thread?.session?.lastError).toBeNull(); + expect(thread?.latestTurn?.state).toBe("interrupted"); + expect(thread?.latestTurn?.completedAt).toBe(abortedAt); + const interruptedSessionEvents = (await harness.readEvents()).filter( + (event) => + event.type === "thread.session-set" && + event.payload.session.status === "interrupted" && + event.payload.session.activeTurnId === null, + ); + expect(interruptedSessionEvents).toHaveLength(1); + }, + ); + + effectIt.effect("converges native Codex and OpenCode interruption ordering", () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => createHarness()); + const codexThreadId = asThreadId("thread-1"); + const openCodeThreadId = asThreadId("thread-opencode-interruption"); + const codexTurnId = asTurnId("turn-codex-interruption"); + const openCodeTurnId = asTurnId("turn-opencode-interruption"); + const startedAt = "2026-01-01T00:00:01.000Z"; + const interruptedAt = "2026-01-01T00:00:02.000Z"; + const completedAt = "2026-01-01T00:00:03.000Z"; + + yield* harness.engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-create-opencode-interruption-thread"), + threadId: openCodeThreadId, + projectId: asProjectId("project-1"), + title: "OpenCode interruption", + modelSelection: { + instanceId: ProviderInstanceId.make("opencode"), + model: "openai/gpt-5", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: "2026-01-01T00:00:00.000Z", + }); + yield* harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-seed-opencode-interruption-session"), + threadId: openCodeThreadId, + session: { + threadId: openCodeThreadId, + status: "ready", + providerName: "opencode", + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: "2026-01-01T00:00:00.000Z", + }, + createdAt: "2026-01-01T00:00:00.000Z", + }); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-native-codex-interruption-started"), + provider: ProviderDriverKind.make("codex"), + threadId: codexThreadId, + turnId: codexTurnId, + createdAt: startedAt, + }); + yield* Effect.promise(() => harness.drain()); + yield* harness.engine.dispatch({ + type: "thread.turn.interrupt", + commandId: CommandId.make("cmd-native-codex-interruption"), + threadId: codexThreadId, + turnId: codexTurnId, + createdAt: interruptedAt, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-native-codex-interruption-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: codexThreadId, + turnId: codexTurnId, + createdAt: completedAt, + payload: { state: "interrupted" }, + }); + yield* Effect.promise(() => harness.drain()); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-native-opencode-interruption-started"), + provider: ProviderDriverKind.make("opencode"), + threadId: openCodeThreadId, + turnId: openCodeTurnId, + createdAt: startedAt, + }); + yield* Effect.promise(() => harness.drain()); + harness.emit({ + type: "turn.aborted", + eventId: asEventId("evt-native-opencode-interruption-aborted"), + provider: ProviderDriverKind.make("opencode"), + threadId: openCodeThreadId, + turnId: openCodeTurnId, + createdAt: interruptedAt, + payload: { reason: "Interrupted by user." }, + }); + yield* Effect.promise(() => harness.drain()); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-native-opencode-interruption-completed"), + provider: ProviderDriverKind.make("opencode"), + threadId: openCodeThreadId, + turnId: openCodeTurnId, + createdAt: completedAt, + payload: { state: "completed" }, + }); + yield* Effect.promise(() => harness.drain()); + + const snapshot = yield* Effect.promise(() => harness.readModel()); + const lifecycle = (threadId: ThreadId) => { + const thread = snapshot.threads.find((entry) => entry.id === threadId); + return { + sessionStatus: thread?.session?.status, + activeTurnId: thread?.session?.activeTurnId, + turnState: thread?.latestTurn?.state, + turnCompletedAt: thread?.latestTurn?.completedAt, + }; + }; + const expectedLifecycle = { + sessionStatus: "ready", + activeTurnId: null, + turnState: "interrupted", + turnCompletedAt: interruptedAt, + }; + expect(lifecycle(codexThreadId)).toEqual(expectedLifecycle); + expect(lifecycle(openCodeThreadId)).toEqual(expectedLifecycle); + }), + ); + + it("ignores unscoped, stale, duplicate, and post-terminal aborts", async () => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const firstTurnId = asTurnId("turn-abort-identity-a"); + const secondTurnId = asTurnId("turn-abort-identity-b"); + const abortedAt = "2026-01-01T00:00:04.000Z"; + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-abort-identity-start-a"), + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId: firstTurnId, + createdAt: "2026-01-01T00:00:01.000Z", + }); + await harness.drain(); + + harness.emit({ + type: "turn.aborted", + eventId: asEventId("evt-abort-identity-wrong-turn"), + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId: secondTurnId, + createdAt: "2026-01-01T00:00:02.000Z", + payload: { reason: "Stale turn" }, + }); + harness.emit({ + type: "turn.aborted", + eventId: asEventId("evt-abort-identity-unscoped"), + provider: ProviderDriverKind.make("opencode"), + threadId, + createdAt: "2026-01-01T00:00:03.000Z", + payload: { reason: "Missing turn identity" }, + }); + + let { detail } = await drainAndRead(harness); + expect(detail.session?.status).toBe("running"); + expect(detail.session?.activeTurnId).toBe(firstTurnId); + expect(detail.latestTurn?.state).toBe("running"); + + const abort = { + type: "turn.aborted" as const, + eventId: asEventId("evt-abort-identity-matching"), + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId: firstTurnId, + createdAt: abortedAt, + payload: { reason: "Interrupted by user." }, + }; + harness.emit(abort); + await harness.drain(); + harness.emit(abort); + + ({ detail } = await drainAndRead(harness)); + expect(detail.session?.status).toBe("interrupted"); + expect(detail.session?.activeTurnId).toBeNull(); + expect(detail.latestTurn?.state).toBe("interrupted"); + expect(detail.latestTurn?.completedAt).toBe(abortedAt); + expect( + (await harness.readEvents()).filter( + (event) => + event.type === "thread.session-set" && event.payload.session.status === "interrupted", + ), + ).toHaveLength(1); + + harness.emit({ + type: "session.state.changed", + eventId: asEventId("evt-abort-identity-ready"), + provider: ProviderDriverKind.make("opencode"), + threadId, + createdAt: "2026-01-01T00:00:05.000Z", + payload: { state: "ready" }, + }); + await harness.drain(); + harness.emit(abort); + + ({ detail } = await drainAndRead(harness)); + expect(detail.session?.status).toBe("ready"); + expect(detail.session?.activeTurnId).toBeNull(); + expect(detail.latestTurn?.state).toBe("interrupted"); + expect(detail.latestTurn?.completedAt).toBe(abortedAt); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-abort-identity-start-b"), + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId: secondTurnId, + createdAt: "2026-01-01T00:00:06.000Z", + }); + await harness.drain(); + harness.emit(abort); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-abort-identity-late-completion-a"), + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId: firstTurnId, + createdAt: "2026-01-01T00:00:07.000Z", + payload: { state: "completed" }, + }); + + ({ detail } = await drainAndRead(harness)); + expect(detail.session?.status).toBe("running"); + expect(detail.session?.activeTurnId).toBe(secondTurnId); + expect(detail.latestTurn?.turnId).toBe(secondTurnId); + expect(detail.latestTurn?.state).toBe("running"); + }); + + it.each([ + ["starting", "starting"], + ["running", "running"], + ["waiting", "running"], + ["ready", "ready"], + ["error", "error"], + ["stopped", "stopped"], + ] as const)( + "preserves an interrupted turn after a later session state of %s", + async (runtimeState, expectedStatus) => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId(`turn-abort-before-state-${runtimeState}`); + const abortedAt = "2026-01-01T00:00:02.000Z"; + + harness.emit({ + type: "turn.started", + eventId: asEventId(`evt-abort-before-state-${runtimeState}-start`), + provider: ProviderDriverKind.make("codex"), + threadId, + turnId, + createdAt: "2026-01-01T00:00:01.000Z", + }); + await harness.drain(); + const abort = { + type: "turn.aborted" as const, + eventId: asEventId(`evt-abort-before-state-${runtimeState}-abort`), + provider: ProviderDriverKind.make("codex"), + threadId, + turnId, + createdAt: abortedAt, + payload: { reason: "Interrupted by user." }, + }; + harness.emit(abort); + await harness.drain(); + harness.emit({ + type: "session.state.changed", + eventId: asEventId(`evt-abort-before-state-${runtimeState}-state`), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:00:03.000Z", + payload: + runtimeState === "error" + ? { state: runtimeState, reason: "Late provider error" } + : { state: runtimeState }, + }); + await harness.drain(); + harness.emit(abort); + + const { detail } = await drainAndRead(harness); + expect(detail.session?.status).toBe(expectedStatus); + expect(detail.session?.activeTurnId).toBeNull(); + expect(detail.session?.lastError).toBe( + runtimeState === "error" ? "Late provider error" : null, + ); + expect(detail.latestTurn?.turnId).toBe(turnId); + expect(detail.latestTurn?.state).toBe("interrupted"); + expect(detail.latestTurn?.completedAt).toBe(abortedAt); + }, + ); + + it.each([ + ["completed", "ready"], + ["interrupted", "ready"], + ["cancelled", "ready"], + ["failed", "error"], + ] as const)( + "preserves an interrupted turn after a later turn completion of %s", + async (completionState, expectedStatus) => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId(`turn-abort-before-completion-${completionState}`); + const abortedAt = "2026-01-01T00:00:02.000Z"; + + harness.emit({ + type: "turn.started", + eventId: asEventId(`evt-abort-before-completion-${completionState}-start`), + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId, + createdAt: "2026-01-01T00:00:01.000Z", + }); + await harness.drain(); + const abort = { + type: "turn.aborted" as const, + eventId: asEventId(`evt-abort-before-completion-${completionState}-abort`), + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId, + createdAt: abortedAt, + payload: { reason: "Interrupted by user." }, + }; + harness.emit(abort); + await harness.drain(); + harness.emit({ + type: "turn.completed", + eventId: asEventId(`evt-abort-before-completion-${completionState}-completed`), + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId, + createdAt: "2026-01-01T00:00:03.000Z", + payload: + completionState === "failed" + ? { state: completionState, errorMessage: "Late turn failure" } + : { state: completionState }, + }); + await harness.drain(); + harness.emit(abort); + + const { detail } = await drainAndRead(harness); + expect(detail.session?.status).toBe(expectedStatus); + expect(detail.session?.activeTurnId).toBeNull(); + expect(detail.session?.lastError).toBe( + completionState === "failed" ? "Late turn failure" : null, + ); + expect(detail.latestTurn?.turnId).toBe(turnId); + expect(detail.latestTurn?.state).toBe("interrupted"); + expect(detail.latestTurn?.completedAt).toBe(abortedAt); + }, + ); + + it("preserves an interrupted turn after a later session exit", async () => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-abort-before-exit"); + const abortedAt = "2026-01-01T00:00:02.000Z"; + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-abort-before-exit-start"), + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId, + createdAt: "2026-01-01T00:00:01.000Z", + }); + await harness.drain(); + const abort = { + type: "turn.aborted" as const, + eventId: asEventId("evt-abort-before-exit-abort"), + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId, + createdAt: abortedAt, + payload: { reason: "Interrupted by user." }, + }; + harness.emit(abort); + await harness.drain(); + harness.emit({ + type: "session.exited", + eventId: asEventId("evt-abort-before-exit-exited"), + provider: ProviderDriverKind.make("opencode"), + threadId, + createdAt: "2026-01-01T00:00:03.000Z", + payload: { reason: "Provider stopped", exitKind: "graceful" }, + }); + await harness.drain(); + harness.emit(abort); + + const { detail } = await drainAndRead(harness); + expect(detail.session?.status).toBe("stopped"); + expect(detail.session?.activeTurnId).toBeNull(); + expect(detail.latestTurn?.turnId).toBe(turnId); + expect(detail.latestTurn?.state).toBe("interrupted"); + expect(detail.latestTurn?.completedAt).toBe(abortedAt); + }); + + it("keeps approval and user-input requests pending when their waiting turn aborts", async () => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-abort-while-waiting"); + const approvalRequestId = ApprovalRequestId.make("approval-abort-waiting"); + const userInputRequestId = ApprovalRequestId.make("user-input-abort-waiting"); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-abort-waiting-start"), + provider: ProviderDriverKind.make("codex"), + threadId, + turnId, + createdAt: "2026-01-01T00:00:01.000Z", + }); + await harness.drain(); + harness.emit({ + type: "request.opened", + eventId: asEventId("evt-abort-waiting-approval-opened"), + provider: ProviderDriverKind.make("codex"), + threadId, + turnId, + requestId: approvalRequestId, + createdAt: "2026-01-01T00:00:02.000Z", + payload: { + requestType: "command_execution_approval", + detail: "pwd", + }, + }); + harness.emit({ + type: "user-input.requested", + eventId: asEventId("evt-abort-waiting-user-input-requested"), + provider: ProviderDriverKind.make("codex"), + threadId, + turnId, + requestId: userInputRequestId, + createdAt: "2026-01-01T00:00:03.000Z", + payload: { + questions: [ + { + id: "choice", + header: "Choice", + question: "Continue?", + options: [{ label: "Yes", description: "Continue" }], + }, + ], + }, + }); + harness.emit({ + type: "session.state.changed", + eventId: asEventId("evt-abort-waiting-session-state"), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:00:04.000Z", + payload: { state: "waiting", reason: "Awaiting user action" }, + }); + + let { detail, shell } = await drainAndRead(harness); + expect(detail.session?.status).toBe("running"); + expect(detail.session?.activeTurnId).toBe(turnId); + expect(shell.hasPendingApprovals).toBe(true); + expect(shell.hasPendingUserInput).toBe(true); + + harness.emit({ + type: "turn.aborted", + eventId: asEventId("evt-abort-waiting-aborted"), + provider: ProviderDriverKind.make("codex"), + threadId, + turnId, + createdAt: "2026-01-01T00:00:05.000Z", + payload: { reason: "Interrupted by user." }, + }); + + ({ detail, shell } = await drainAndRead(harness)); + expect(detail.session?.status).toBe("interrupted"); + expect(detail.session?.activeTurnId).toBeNull(); + expect(detail.latestTurn?.state).toBe("interrupted"); + expect(shell.hasPendingApprovals).toBe(true); + expect(shell.hasPendingUserInput).toBe(true); + + harness.emit({ + type: "request.resolved", + eventId: asEventId("evt-abort-waiting-approval-resolved"), + provider: ProviderDriverKind.make("codex"), + threadId, + turnId, + requestId: approvalRequestId, + createdAt: "2026-01-01T00:00:06.000Z", + payload: { + requestType: "command_execution_approval", + decision: "decline", + }, + }); + harness.emit({ + type: "user-input.resolved", + eventId: asEventId("evt-abort-waiting-user-input-resolved"), + provider: ProviderDriverKind.make("codex"), + threadId, + turnId, + requestId: userInputRequestId, + createdAt: "2026-01-01T00:00:07.000Z", + payload: { answers: { choice: "No" } }, + }); + + ({ detail, shell } = await drainAndRead(harness)); + expect(detail.session?.status).toBe("interrupted"); + expect(detail.session?.activeTurnId).toBeNull(); + expect(shell.hasPendingApprovals).toBe(false); + expect(shell.hasPendingUserInput).toBe(false); + }); + it("applies provider session.state.changed transitions directly", async () => { const harness = await createHarness(); const waitingAt = "2026-01-01T00:00:00.000Z"; @@ -2244,11 +2822,7 @@ describe("ProviderRuntimeIngestion", () => { expect(resumedMessage?.text).toBe(" second half"); expect(resumedMessage?.streaming).toBe(false); - const events = await Effect.runPromise( - Stream.runCollect(harness.engine.readEvents(0)).pipe( - Effect.map((chunk) => Array.from(chunk)), - ), - ); + const events = await harness.readEvents(); const assistantEvents = events.filter( (event): event is Extract<(typeof events)[number], { type: "thread.message-sent" }> => event.type === "thread.message-sent" && @@ -2600,11 +3174,7 @@ describe("ProviderRuntimeIngestion", () => { ), ); - const events = await Effect.runPromise( - Stream.runCollect(harness.engine.readEvents(0)).pipe( - Effect.map((chunk) => Array.from(chunk)), - ), - ); + const events = await harness.readEvents(); const completionEvents = events.filter((event) => { if (event.type !== "thread.message-sent") { return false; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 03253797242e..b22acb971fbe 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -1513,6 +1513,14 @@ const make = Effect.gen(function* () { : false; const shouldApplyThreadLifecycle = (() => { + if (event.type === "turn.aborted") { + // Aborts are only authoritative while their exact turn is active. + // This makes duplicate and delayed aborts lifecycle no-ops after + // another terminal event or a newer turn has taken ownership. + return ( + activeTurnId !== null && eventTurnId !== undefined && sameId(activeTurnId, eventTurnId) + ); + } if (!STRICT_PROVIDER_LIFECYCLE_GUARD) { return true; } @@ -1555,7 +1563,8 @@ const make = Effect.gen(function* () { event.type === "session.exited" || event.type === "thread.started" || event.type === "turn.started" || - event.type === "turn.completed" + event.type === "turn.completed" || + event.type === "turn.aborted" ) { const status = (() => { switch (event.type) { @@ -1571,6 +1580,8 @@ const make = Effect.gen(function* () { return normalizeRuntimeTurnState(event.payload.state) === "failed" ? "error" : "ready"; + case "turn.aborted": + return "interrupted"; case "session.started": case "thread.started": // Provider thread/session start notifications can arrive during an @@ -1581,7 +1592,9 @@ const make = Effect.gen(function* () { const nextActiveTurnId = event.type === "turn.started" ? (eventTurnId ?? null) - : event.type === "turn.completed" || event.type === "session.exited" + : event.type === "turn.completed" || + event.type === "turn.aborted" || + event.type === "session.exited" ? null : event.type === "session.state.changed" && !sessionStatusAllowsActiveTurn( @@ -1953,7 +1966,10 @@ const make = Effect.gen(function* () { } else if (!conflictsWithActiveTurn) { if (event.type === "turn.plan.updated") { threadPlanProgress.recordPlanProgress(thread.id, event.payload.plan); - } else if (event.type === "turn.completed" || event.type === "turn.aborted") { + } else if ( + event.type === "turn.completed" || + (event.type === "turn.aborted" && shouldApplyThreadLifecycle) + ) { threadPlanProgress.clearThreadPlanProgress(thread.id); } } diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 7b8fbec56660..7af3b9abc954 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -512,6 +512,74 @@ function startLifecycleRuntime() { } lifecycleLayer("CodexAdapterLive lifecycle", (it) => { + it.effect("maps interrupted turn/completed notifications to canonical completion", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-turn-interrupted"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "turn/completed", + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), + payload: { + threadId: "provider-thread-1", + turn: { + id: "turn-1", + items: [], + status: "interrupted", + }, + }, + } satisfies ProviderEvent); + + const firstEvent = yield* Fiber.join(firstEventFiber); + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + NodeAssert.equal(firstEvent.value.type, "turn.completed"); + if (firstEvent.value.type !== "turn.completed") { + return; + } + NodeAssert.equal(firstEvent.value.turnId, "turn-1"); + NodeAssert.equal(firstEvent.value.payload.state, "interrupted"); + NodeAssert.equal(firstEvent.value.payload.errorMessage, undefined); + }), + ); + + it.effect("maps raw turn/aborted provider events to canonical aborts", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-turn-aborted-compat"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "turn/aborted", + message: "Interrupted by user.", + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), + } satisfies ProviderEvent); + + const firstEvent = yield* Fiber.join(firstEventFiber); + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + NodeAssert.equal(firstEvent.value.type, "turn.aborted"); + if (firstEvent.value.type !== "turn.aborted") { + return; + } + NodeAssert.equal(firstEvent.value.turnId, "turn-1"); + NodeAssert.equal(firstEvent.value.payload.reason, "Interrupted by user."); + }), + ); + it.effect("maps completed agent message items to canonical item.completed events", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 1385ccbaabec..63d6dd7fe0bd 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -695,16 +695,27 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { it.effect("rolls back session state when sendTurn fails before OpenCode accepts the prompt", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-send-turn-failure"); yield* adapter.startSession({ provider: ProviderDriverKind.make("opencode"), - threadId: asThreadId("thread-send-turn-failure"), + threadId, runtimeMode: "full-access", }); + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + (event.type === "turn.started" || event.type === "turn.aborted"), + ), + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); runtimeMock.state.promptAsyncError = new Error("prompt failed"); const error = yield* adapter .sendTurn({ - threadId: asThreadId("thread-send-turn-failure"), + threadId, input: "Fix it", modelSelection: { instanceId: ProviderInstanceId.make("opencode"), @@ -712,8 +723,19 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }, }) .pipe(Effect.flip); + const events = Array.from(yield* Fiber.join(eventsFiber)); const sessions = yield* adapter.listSessions(); + NodeAssert.deepEqual( + events.map((event) => event.type), + ["turn.started", "turn.aborted"], + ); + NodeAssert.ok(events[0]?.turnId); + NodeAssert.equal(events[1]?.turnId, events[0]?.turnId); + NodeAssert.equal(events[1]?.type, "turn.aborted"); + if (events[1]?.type === "turn.aborted") { + NodeAssert.equal(events[1].payload.reason, "prompt failed"); + } NodeAssert.equal(error._tag, "ProviderAdapterRequestError"); if (error._tag !== "ProviderAdapterRequestError") { throw new Error("Unexpected error type"); @@ -730,6 +752,51 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("emits a canonical abort after OpenCode accepts an interrupt", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-interrupt"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + (event.type === "turn.started" || event.type === "turn.aborted"), + ), + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + + const turn = yield* adapter.sendTurn({ + threadId, + input: "Fix it", + modelSelection: { + instanceId: ProviderInstanceId.make("opencode"), + model: "openai/gpt-5", + }, + }); + yield* adapter.interruptTurn(threadId, turn.turnId); + + const events = Array.from(yield* Fiber.join(eventsFiber)); + NodeAssert.deepEqual( + events.map((event) => event.type), + ["turn.started", "turn.aborted"], + ); + NodeAssert.equal(events[0]?.turnId, turn.turnId); + NodeAssert.equal(events[1]?.turnId, turn.turnId); + NodeAssert.equal(events[1]?.type, "turn.aborted"); + if (events[1]?.type === "turn.aborted") { + NodeAssert.equal(events[1].payload.reason, "Interrupted by user."); + } + NodeAssert.deepEqual(runtimeMock.state.abortCalls, ["http://127.0.0.1:9999/session"]); + }), + ); + it.effect("steers a running turn instead of opening a new one on mid-turn sendTurn", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter;