authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-03-03 22:54:46-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-03-05 23:00:41-08:00
log9c4d7a83e22e4a64803552d3920686f4beb07470
treef0f50d7dafadb5627a67ce1b5c56a2b2de7f4d26
parent7425a3c4b3cb3bd664c0dd586bc0df3195966d39

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

2 files changed, 120 insertions(+), 3 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+92
...@@ -241,6 +241,98 @@ describe("ProviderRuntimeIngestion", () => {...@@ -241,6 +241,98 @@ describe("ProviderRuntimeIngestion", () => {
241 expect(thread.session?.lastError).toBe("turn failed");241 expect(thread.session?.lastError).toBe("turn failed");
242 });242 });
243243
244 it("applies provider session.state.changed transitions directly", async () => {
245 const harness = await createHarness();
246 const waitingAt = new Date().toISOString();
247
248 harness.emit({
249 type: "session.state.changed",
250 eventId: asEventId("evt-session-state-waiting"),
251 provider: "codex",
252 sessionId: asSessionId("sess-1"),
253 createdAt: waitingAt,
254 threadId: asProviderThreadId("provider-thread-1"),
255 payload: {
256 state: "waiting",
257 reason: "awaiting approval",
258 },
259 });
260
261 let thread = await waitForThread(
262 harness.engine,
263 (entry) => entry.session?.status === "running" && entry.session?.activeTurnId === null,
264 );
265 expect(thread.session?.status).toBe("running");
266 expect(thread.session?.lastError).toBeNull();
267
268 harness.emit({
269 type: "session.state.changed",
270 eventId: asEventId("evt-session-state-error"),
271 provider: "codex",
272 sessionId: asSessionId("sess-1"),
273 createdAt: new Date().toISOString(),
274 threadId: asProviderThreadId("provider-thread-1"),
275 payload: {
276 state: "error",
277 reason: "provider crashed",
278 },
279 });
280
281 thread = await waitForThread(
282 harness.engine,
283 (entry) =>
284 entry.session?.status === "error" &&
285 entry.session?.activeTurnId === null &&
286 entry.session?.lastError === "provider crashed",
287 );
288 expect(thread.session?.status).toBe("error");
289 expect(thread.session?.lastError).toBe("provider crashed");
290
291 harness.emit({
292 type: "session.state.changed",
293 eventId: asEventId("evt-session-state-stopped"),
294 provider: "codex",
295 sessionId: asSessionId("sess-1"),
296 createdAt: new Date().toISOString(),
297 threadId: asProviderThreadId("provider-thread-1"),
298 payload: {
299 state: "stopped",
300 },
301 });
302
303 thread = await waitForThread(
304 harness.engine,
305 (entry) =>
306 entry.session?.status === "stopped" &&
307 entry.session?.activeTurnId === null &&
308 entry.session?.lastError === "provider crashed",
309 );
310 expect(thread.session?.status).toBe("stopped");
311 expect(thread.session?.lastError).toBe("provider crashed");
312
313 harness.emit({
314 type: "session.state.changed",
315 eventId: asEventId("evt-session-state-ready"),
316 provider: "codex",
317 sessionId: asSessionId("sess-1"),
318 createdAt: new Date().toISOString(),
319 threadId: asProviderThreadId("provider-thread-1"),
320 payload: {
321 state: "ready",
322 },
323 });
324
325 thread = await waitForThread(
326 harness.engine,
327 (entry) =>
328 entry.session?.status === "ready" &&
329 entry.session?.activeTurnId === null &&
330 entry.session?.lastError === null,
331 );
332 expect(thread.session?.status).toBe("ready");
333 expect(thread.session?.lastError).toBeNull();
334 });
335
244 it("does not clear active turn when session/thread started arrives mid-turn", async () => {336 it("does not clear active turn when session/thread started arrives mid-turn", async () => {
245 const harness = await createHarness();337 const harness = await createHarness();
246 const now = new Date().toISOString();338 const now = new Date().toISOString();
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+28-3
...@@ -143,6 +143,26 @@ function runtimeErrorMessageFromEvent(event: ProviderRuntimeEvent): string | und...@@ -143,6 +143,26 @@ function runtimeErrorMessageFromEvent(event: ProviderRuntimeEvent): string | und
143 return payloadMessage ?? legacyMessage;143 return payloadMessage ?? legacyMessage;
144}144}
145145
146function orchestrationSessionStatusFromRuntimeState(
147 state: ProviderRuntimeEvent extends { payload: { state: infer T } } ? T : never,
148): "starting" | "running" | "ready" | "interrupted" | "stopped" | "error" {
149 switch (state) {
150 case "starting":
151 return "starting";
152 case "running":
153 case "waiting":
154 return "running";
155 case "ready":
156 return "ready";
157 case "interrupted":
158 return "interrupted";
159 case "stopped":
160 return "stopped";
161 case "error":
162 return "error";
163 }
164}
165
146function requestKindFromCanonicalRequestType(166function requestKindFromCanonicalRequestType(
147 requestType: string | undefined,167 requestType: string | undefined,
148): "command" | "file-change" | undefined {168): "command" | "file-change" | undefined {
...@@ -617,6 +637,7 @@ const make = Effect.gen(function* () {...@@ -617,6 +637,7 @@ const make = Effect.gen(function* () {
617637
618 if (638 if (
619 event.type === "session.started" ||639 event.type === "session.started" ||
640 event.type === "session.state.changed" ||
620 event.type === "session.exited" ||641 event.type === "session.exited" ||
621 event.type === "thread.started" ||642 event.type === "thread.started" ||
622 event.type === "turn.started" ||643 event.type === "turn.started" ||
...@@ -636,6 +657,8 @@ const make = Effect.gen(function* () {...@@ -636,6 +657,8 @@ const make = Effect.gen(function* () {
636 providerThreadIdFromEvent ?? scopedSessionProviderThreadId ?? null;657 providerThreadIdFromEvent ?? scopedSessionProviderThreadId ?? null;
637 const status = (() => {658 const status = (() => {
638 switch (event.type) {659 switch (event.type) {
660 case "session.state.changed":
661 return orchestrationSessionStatusFromRuntimeState(event.payload.state);
639 case "turn.started":662 case "turn.started":
640 return "running";663 return "running";
641 case "session.exited":664 case "session.exited":
...@@ -650,9 +673,11 @@ const make = Effect.gen(function* () {...@@ -650,9 +673,11 @@ const make = Effect.gen(function* () {
650 }673 }
651 })();674 })();
652 const lastError =675 const lastError =
653 event.type === "turn.completed" && runtimeTurnState(event) === "failed"676 event.type === "session.state.changed" && event.payload.state === "error"
654 ? (runtimeTurnErrorMessage(event) ?? thread.session?.lastError ?? "Turn failed")677 ? (event.payload.reason ?? thread.session?.lastError ?? "Provider session error")
655 : status === "ready"678 : event.type === "turn.completed" && runtimeTurnState(event) === "failed"
679 ? (runtimeTurnErrorMessage(event) ?? thread.session?.lastError ?? "Turn failed")
680 : status === "ready"
656 ? null681 ? null
657 : (thread.session?.lastError ?? null);682 : (thread.session?.lastError ?? null);
658683