diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 0999000ed4f..343246901e5 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -1734,6 +1734,119 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect("keeps the active session turn until provider confirms an interrupt", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-interrupt-session"); + const turnId = TurnId.make("turn-interrupt-session"); + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); + + yield* appendAndProject({ + type: "thread.session-set", + eventId: EventId.make("evt-interrupt-session-1"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-02-26T13:00:00.000Z", + commandId: CommandId.make("cmd-interrupt-session-1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-interrupt-session-1"), + metadata: {}, + payload: { + threadId, + session: { + threadId, + status: "running", + providerName: "cursor", + runtimeMode: "full-access", + activeTurnId: turnId, + lastError: null, + updatedAt: "2026-02-26T13:00:00.000Z", + }, + }, + }); + + yield* appendAndProject({ + type: "thread.turn-interrupt-requested", + eventId: EventId.make("evt-interrupt-session-2"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-02-26T13:00:01.000Z", + commandId: CommandId.make("cmd-interrupt-session-2"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-interrupt-session-2"), + metadata: {}, + payload: { + threadId, + turnId, + createdAt: "2026-02-26T13:00:01.000Z", + }, + }); + + const sessions = yield* sql<{ + readonly status: string; + readonly activeTurnId: string | null; + }>` + SELECT status, active_turn_id AS "activeTurnId" + FROM projection_thread_sessions + WHERE thread_id = ${threadId} + `; + const turns = yield* sql<{ readonly state: string }>` + SELECT state + FROM projection_turns + WHERE thread_id = ${threadId} AND turn_id = ${turnId} + `; + + assert.deepEqual(sessions, [{ status: "running", activeTurnId: turnId }]); + assert.deepEqual(turns, [{ state: "interrupted" }]); + + yield* appendAndProject({ + type: "thread.session-set", + eventId: EventId.make("evt-interrupt-session-3"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-02-26T13:00:02.000Z", + commandId: CommandId.make("cmd-interrupt-session-3"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-interrupt-session-3"), + metadata: {}, + payload: { + threadId, + session: { + threadId, + status: "ready", + providerName: "cursor", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: "2026-02-26T13:00:02.000Z", + }, + }, + }); + + const confirmedSessions = yield* sql<{ + readonly status: string; + readonly activeTurnId: string | null; + }>` + SELECT status, active_turn_id AS "activeTurnId" + FROM projection_thread_sessions + WHERE thread_id = ${threadId} + `; + const confirmedTurns = yield* sql<{ readonly state: string }>` + SELECT state + FROM projection_turns + WHERE thread_id = ${threadId} AND turn_id = ${turnId} + `; + + assert.deepEqual(confirmedSessions, [{ status: "ready", activeTurnId: null }]); + assert.deepEqual(confirmedTurns, [{ state: "interrupted" }]); + }), + ); + it.effect("clears stale pending approvals from projected shell summaries", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index f12df850941..95acadd2d98 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -992,9 +992,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const applyThreadSessionsProjection: ProjectorDefinition["apply"] = Effect.fn( "applyThreadSessionsProjection", )(function* (event, _attachmentSideEffects) { - if (event.type !== "thread.session-set") { - return; - } + if (event.type !== "thread.session-set") return; yield* projectionThreadSessionRepository.upsert({ threadId: event.payload.threadId, status: event.payload.session.status, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 001ba388949..d451e315893 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -249,6 +249,10 @@ describe("ProviderRuntimeIngestion", () => { scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(ingestion.start().pipe(Scope.provide(scope))); const drain = () => Effect.runPromise(ingestion.drain); + const readEvents = () => + Effect.runPromise( + Stream.runCollect(engine.readEvents(0)).pipe(Effect.map((chunk) => Array.from(chunk))), + ); const createdAt = "2026-01-01T00:00:00.000Z"; await Effect.runPromise( @@ -315,6 +319,7 @@ describe("ProviderRuntimeIngestion", () => { emit: provider.emit, setProviderSession: provider.setSession, drain, + readEvents, }; } @@ -505,6 +510,81 @@ describe("ProviderRuntimeIngestion", () => { ); }); + it("does not restore an interrupted Cursor turn during session recovery", async () => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-interrupted-before-recovery"); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-before-cursor-recovery"), + provider: ProviderDriverKind.make("cursor"), + createdAt: "2026-01-01T00:00:01.000Z", + threadId, + turnId, + }); + + await waitForThread( + harness.readModel, + (thread) => thread.session?.status === "running" && thread.session.activeTurnId === turnId, + ); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.turn.interrupt", + commandId: CommandId.make("cmd-interrupt-before-cursor-recovery"), + threadId, + turnId, + createdAt: "2026-01-01T00:00:02.000Z", + }), + ); + + await waitForThread( + harness.readModel, + (thread) => + thread.session?.status === "running" && + thread.session.activeTurnId === turnId && + thread.latestTurn?.state === "interrupted", + ); + + // Cursor session/load emits this startup sequence while recovering the + // provider session. The interrupt request alone must not mark the session + // idle, but the provider's ready state is confirmation that the recovered + // session is no longer running the interrupted turn. + harness.emit({ + type: "session.started", + eventId: asEventId("evt-cursor-session-started-after-interrupt"), + provider: ProviderDriverKind.make("cursor"), + createdAt: "2026-01-01T00:00:03.000Z", + threadId, + }); + harness.emit({ + type: "session.state.changed", + eventId: asEventId("evt-cursor-session-ready-after-interrupt"), + provider: ProviderDriverKind.make("cursor"), + createdAt: "2026-01-01T00:00:04.000Z", + threadId, + payload: { + state: "ready", + reason: "Cursor ACP session ready", + }, + }); + harness.emit({ + type: "thread.started", + eventId: asEventId("evt-cursor-thread-started-after-interrupt"), + provider: ProviderDriverKind.make("cursor"), + createdAt: "2026-01-01T00:00:05.000Z", + threadId, + }); + + await harness.drain(); + const readModel = await harness.readModel(); + const thread = readModel.threads.find((entry) => entry.id === threadId); + expect(thread?.session?.status).toBe("ready"); + expect(thread?.session?.activeTurnId).toBeNull(); + expect(thread?.latestTurn?.state).not.toBe("running"); + }); + it("accepts claude turn lifecycle when seeded thread id is a synthetic placeholder", async () => { const harness = await createHarness(); const seededAt = "2026-01-01T00:00:00.000Z"; @@ -1938,11 +2018,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" && @@ -2294,11 +2370,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 3e5978f4846..9327afb5c95 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -1281,12 +1281,6 @@ const make = Effect.gen(function* () { event.type === "turn.started" || event.type === "turn.completed" ) { - const nextActiveTurnId = - event.type === "turn.started" - ? (eventTurnId ?? null) - : event.type === "turn.completed" || event.type === "session.exited" - ? null - : activeTurnId; const status = (() => { switch (event.type) { case "session.state.changed": @@ -1306,6 +1300,16 @@ const make = Effect.gen(function* () { return activeTurnId !== null ? "running" : "ready"; } })(); + const nextActiveTurnId = + event.type === "turn.started" + ? (eventTurnId ?? null) + : event.type === "turn.completed" || event.type === "session.exited" + ? null + : event.type === "session.state.changed" && + status !== "starting" && + status !== "running" + ? null + : activeTurnId; const lastError = event.type === "session.state.changed" && event.payload.state === "error" ? (event.payload.reason ?? thread.session?.lastError ?? "Provider session error")