From 9c4d7a83e22e4a64803552d3920686f4beb07470 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Mar 2026 22:54:46 -0800 Subject: [PATCH] Handle session.state.changed in provider runtime ingestion - Map runtime session state transitions to orchestration session status - Persist error reasons from `session.state.changed` error events - Add coverage for waiting/error/stopped/ready transition handling --- .../Layers/ProviderRuntimeIngestion.test.ts | 92 +++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 31 ++++++- 2 files changed, 120 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 97001e29582a768e49f30bfa80e6737fe4c407a9..57a88bd6a32df119a4e31253af3af87f4aad9a6a 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -241,6 +241,98 @@ describe("ProviderRuntimeIngestion", () => { expect(thread.session?.lastError).toBe("turn failed"); }); + it("applies provider session.state.changed transitions directly", async () => { + const harness = await createHarness(); + const waitingAt = new Date().toISOString(); + + harness.emit({ + type: "session.state.changed", + eventId: asEventId("evt-session-state-waiting"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: waitingAt, + threadId: asProviderThreadId("provider-thread-1"), + payload: { + state: "waiting", + reason: "awaiting approval", + }, + }); + + let thread = await waitForThread( + harness.engine, + (entry) => entry.session?.status === "running" && entry.session?.activeTurnId === null, + ); + expect(thread.session?.status).toBe("running"); + expect(thread.session?.lastError).toBeNull(); + + harness.emit({ + type: "session.state.changed", + eventId: asEventId("evt-session-state-error"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("provider-thread-1"), + payload: { + state: "error", + reason: "provider crashed", + }, + }); + + thread = await waitForThread( + harness.engine, + (entry) => + entry.session?.status === "error" && + entry.session?.activeTurnId === null && + entry.session?.lastError === "provider crashed", + ); + expect(thread.session?.status).toBe("error"); + expect(thread.session?.lastError).toBe("provider crashed"); + + harness.emit({ + type: "session.state.changed", + eventId: asEventId("evt-session-state-stopped"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("provider-thread-1"), + payload: { + state: "stopped", + }, + }); + + thread = await waitForThread( + harness.engine, + (entry) => + entry.session?.status === "stopped" && + entry.session?.activeTurnId === null && + entry.session?.lastError === "provider crashed", + ); + expect(thread.session?.status).toBe("stopped"); + expect(thread.session?.lastError).toBe("provider crashed"); + + harness.emit({ + type: "session.state.changed", + eventId: asEventId("evt-session-state-ready"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("provider-thread-1"), + payload: { + state: "ready", + }, + }); + + thread = await waitForThread( + harness.engine, + (entry) => + entry.session?.status === "ready" && + entry.session?.activeTurnId === null && + entry.session?.lastError === null, + ); + expect(thread.session?.status).toBe("ready"); + expect(thread.session?.lastError).toBeNull(); + }); + it("does not clear active turn when session/thread started arrives mid-turn", async () => { const harness = await createHarness(); const now = new Date().toISOString(); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 25d291cf319739dd122cd5a89dbfbcf8ac76619a..46e68d439813310e06c29c97b7f1e8db534e2c90 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -143,6 +143,26 @@ function runtimeErrorMessageFromEvent(event: ProviderRuntimeEvent): string | und return payloadMessage ?? legacyMessage; } +function orchestrationSessionStatusFromRuntimeState( + state: ProviderRuntimeEvent extends { payload: { state: infer T } } ? T : never, +): "starting" | "running" | "ready" | "interrupted" | "stopped" | "error" { + switch (state) { + case "starting": + return "starting"; + case "running": + case "waiting": + return "running"; + case "ready": + return "ready"; + case "interrupted": + return "interrupted"; + case "stopped": + return "stopped"; + case "error": + return "error"; + } +} + function requestKindFromCanonicalRequestType( requestType: string | undefined, ): "command" | "file-change" | undefined { @@ -617,6 +637,7 @@ const make = Effect.gen(function* () { if ( event.type === "session.started" || + event.type === "session.state.changed" || event.type === "session.exited" || event.type === "thread.started" || event.type === "turn.started" || @@ -636,6 +657,8 @@ const make = Effect.gen(function* () { providerThreadIdFromEvent ?? scopedSessionProviderThreadId ?? null; const status = (() => { switch (event.type) { + case "session.state.changed": + return orchestrationSessionStatusFromRuntimeState(event.payload.state); case "turn.started": return "running"; case "session.exited": @@ -650,9 +673,11 @@ const make = Effect.gen(function* () { } })(); const lastError = - event.type === "turn.completed" && runtimeTurnState(event) === "failed" - ? (runtimeTurnErrorMessage(event) ?? thread.session?.lastError ?? "Turn failed") - : status === "ready" + event.type === "session.state.changed" && event.payload.state === "error" + ? (event.payload.reason ?? thread.session?.lastError ?? "Provider session error") + : event.type === "turn.completed" && runtimeTurnState(event) === "failed" + ? (runtimeTurnErrorMessage(event) ?? thread.session?.lastError ?? "Turn failed") + : status === "ready" ? null : (thread.session?.lastError ?? null); -- 2.54.0