From b397aecb5b2a26773a84b603c73c7e8231912074 Mon Sep 17 00:00:00 2001 From: Timothe F Date: Mon, 6 Jul 2026 20:46:23 +0000 Subject: [PATCH 1/3] fix: clear active turn after interruption --- .../Layers/ProjectionPipeline.test.ts | 72 +++++++++++++++++++ .../Layers/ProjectionPipeline.ts | 19 ++++- 2 files changed, 90 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 0999000ed4f..b1c5fa73b43 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -1734,6 +1734,78 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect("clears the active session turn when it is interrupted", () => + 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: "ready", activeTurnId: null }]); + assert.deepEqual(turns, [{ 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..23d1bb0a1cf 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -992,9 +992,26 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const applyThreadSessionsProjection: ProjectorDefinition["apply"] = Effect.fn( "applyThreadSessionsProjection", )(function* (event, _attachmentSideEffects) { - if (event.type !== "thread.session-set") { + if (event.type === "thread.turn-interrupt-requested" && event.payload.turnId !== undefined) { + const session = yield* projectionThreadSessionRepository.getByThreadId({ + threadId: event.payload.threadId, + }); + if ( + Option.isSome(session) && + session.value.activeTurnId === event.payload.turnId && + session.value.status === "running" + ) { + yield* projectionThreadSessionRepository.upsert({ + ...session.value, + status: "ready", + activeTurnId: null, + lastError: null, + updatedAt: event.payload.createdAt, + }); + } return; } + if (event.type !== "thread.session-set") return; yield* projectionThreadSessionRepository.upsert({ threadId: event.payload.threadId, status: event.payload.session.status, From db0bc967f24e8a0f9d80491f9c0a0673216d6f35 Mon Sep 17 00:00:00 2001 From: Timothe F Date: Mon, 6 Jul 2026 20:55:34 +0000 Subject: [PATCH 2/3] test: cover Cursor recovery after interruption --- .../Layers/ProviderRuntimeIngestion.test.ts | 91 +++++++++++++++++-- 1 file changed, 81 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 001ba388949..45f3a70da9b 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,80 @@ 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 === "ready" && + thread.session.activeTurnId === null && + thread.latestTurn?.state === "interrupted", + ); + + // Cursor session/load emits this startup sequence while recovering the + // provider session. It must use the cleared projected active turn rather + // than restoring the turn that was interrupted before recovery began. + 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 +2017,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 +2369,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; From d21bb7902c573a8db5137cf03d815f8b32d3ce51 Mon Sep 17 00:00:00 2001 From: Timothe F Date: Mon, 6 Jul 2026 21:20:51 +0000 Subject: [PATCH 3/3] fix: defer interrupt session clearing --- .../Layers/ProjectionPipeline.test.ts | 45 ++++++++++++++++++- .../Layers/ProjectionPipeline.ts | 19 -------- .../Layers/ProviderRuntimeIngestion.test.ts | 9 ++-- .../Layers/ProviderRuntimeIngestion.ts | 16 ++++--- 4 files changed, 58 insertions(+), 31 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index b1c5fa73b43..343246901e5 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -1734,7 +1734,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); - it.effect("clears the active session turn when it is interrupted", () => + it.effect("keeps the active session turn until provider confirms an interrupt", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; @@ -1801,8 +1801,49 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { WHERE thread_id = ${threadId} AND turn_id = ${turnId} `; - assert.deepEqual(sessions, [{ status: "ready", activeTurnId: null }]); + 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" }]); }), ); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 23d1bb0a1cf..95acadd2d98 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -992,25 +992,6 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const applyThreadSessionsProjection: ProjectorDefinition["apply"] = Effect.fn( "applyThreadSessionsProjection", )(function* (event, _attachmentSideEffects) { - if (event.type === "thread.turn-interrupt-requested" && event.payload.turnId !== undefined) { - const session = yield* projectionThreadSessionRepository.getByThreadId({ - threadId: event.payload.threadId, - }); - if ( - Option.isSome(session) && - session.value.activeTurnId === event.payload.turnId && - session.value.status === "running" - ) { - yield* projectionThreadSessionRepository.upsert({ - ...session.value, - status: "ready", - activeTurnId: null, - lastError: null, - updatedAt: event.payload.createdAt, - }); - } - return; - } if (event.type !== "thread.session-set") return; yield* projectionThreadSessionRepository.upsert({ threadId: event.payload.threadId, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 45f3a70da9b..d451e315893 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -542,14 +542,15 @@ describe("ProviderRuntimeIngestion", () => { await waitForThread( harness.readModel, (thread) => - thread.session?.status === "ready" && - thread.session.activeTurnId === null && + thread.session?.status === "running" && + thread.session.activeTurnId === turnId && thread.latestTurn?.state === "interrupted", ); // Cursor session/load emits this startup sequence while recovering the - // provider session. It must use the cleared projected active turn rather - // than restoring the turn that was interrupted before recovery began. + // 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"), 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")