diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index c1ba48108f..e9f231464e 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -502,6 +502,9 @@ describe("ProviderRuntimeIngestion", () => { thread.session?.status === "running" && thread.session?.activeTurnId === "turn-primary", ); + // A turn.completed from a different turn still transitions the session to + // "ready" — preventing the session from getting permanently stuck at + // "running" when turnIds mismatch (the runtime is done regardless). harness.emit({ type: "turn.completed", eventId: asEventId("evt-turn-completed-aux"), @@ -512,24 +515,6 @@ describe("ProviderRuntimeIngestion", () => { status: "completed", }); - await harness.drain(); - const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); - const midThread = midReadModel.threads.find( - (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), - ); - expect(midThread?.session?.status).toBe("running"); - expect(midThread?.session?.activeTurnId).toBe("turn-primary"); - - harness.emit({ - type: "turn.completed", - eventId: asEventId("evt-turn-completed-primary"), - provider: "codex", - createdAt: new Date().toISOString(), - threadId: asThreadId("thread-1"), - turnId: asTurnId("turn-primary"), - status: "completed", - }); - await waitForThread( harness.engine, (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null, @@ -556,6 +541,8 @@ describe("ProviderRuntimeIngestion", () => { thread.session?.activeTurnId === "turn-guarded-main", ); + // A turn.completed with a mismatched turnId still transitions the session + // to "ready" — preventing stuck sessions when turnIds don't align. harness.emit({ type: "turn.completed", eventId: asEventId("evt-turn-completed-guarded-other"), @@ -566,24 +553,6 @@ describe("ProviderRuntimeIngestion", () => { status: "completed", }); - await harness.drain(); - const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); - const midThread = midReadModel.threads.find( - (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), - ); - expect(midThread?.session?.status).toBe("running"); - expect(midThread?.session?.activeTurnId).toBe("turn-guarded-main"); - - harness.emit({ - type: "turn.completed", - eventId: asEventId("evt-turn-completed-guarded-main"), - provider: "codex", - createdAt: new Date().toISOString(), - threadId: asThreadId("thread-1"), - turnId: asTurnId("turn-guarded-main"), - status: "completed", - }); - await waitForThread( harness.engine, (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 3df47941af..ccb41fd330 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -883,19 +883,32 @@ const make = Effect.gen(function* () { case "turn.started": return !conflictsWithActiveTurn; case "turn.completed": - if (conflictsWithActiveTurn || missingTurnForActiveTurn) { - return false; - } - // Only the active turn may close the lifecycle state. - if (activeTurnId !== null && eventTurnId !== undefined) { - return sameId(activeTurnId, eventTurnId); - } - // If no active turn is tracked, accept completion scoped to this thread. + // Always apply turn.completed to prevent sessions stuck at "running". + // A completed turn must clear the running state regardless of turnId + // mismatch — the runtime is done and the UI must reflect that. return true; default: return true; } })(); + + if ( + event.type === "turn.completed" && + STRICT_PROVIDER_LIFECYCLE_GUARD && + (conflictsWithActiveTurn || missingTurnForActiveTurn) + ) { + yield* Effect.logWarning( + "turn.completed accepted despite turnId mismatch — clearing running state to prevent stuck session", + { + threadId: thread.id, + eventTurnId: eventTurnId ?? "", + activeTurnId: activeTurnId ?? "", + conflictsWithActiveTurn, + missingTurnForActiveTurn, + }, + ); + } + const acceptedTurnStartedSourcePlan = event.type === "turn.started" && shouldApplyThreadLifecycle ? yield* getSourceProposedPlanReferenceForAcceptedTurnStart(thread.id, eventTurnId)