authorgravatar for 76947376+Andrew-Forster@users.noreply.github.comAndrew Forster <76947376+Andrew-Forster@users.noreply.github.com> 2026-07-20 02:04:48-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-20 11:04:48+02:00
log501ce27b81826960dcbb9a93a6b3407efcf09366
tree9cd3c6b910c8052967dcd73def7e7ca0d640e53f
parent40c0ab088308fb010a3db488c545f10aa7179ed2
signature Signed by PGP key B5690EEEBB952194

[orchestration] Clear stale active turn when session becomes inactive (#3159)

Co-authored-by: Julius Marminge <julius0216@outlook.com>

2 files changed, 61 insertions(+), 6 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+47
...@@ -448,6 +448,51 @@ describe("ProviderRuntimeIngestion", () => {...@@ -448,6 +448,51 @@ describe("ProviderRuntimeIngestion", () => {
448 expect(thread.session?.lastError).toBeNull();448 expect(thread.session?.lastError).toBeNull();
449 });449 });
450450
451 it("clears active turn when provider session becomes ready", async () => {
452 const harness = await createHarness();
453 const now = "2026-01-01T00:00:00.000Z";
454
455 harness.emit({
456 type: "turn.started",
457 eventId: asEventId("evt-turn-started-session-ready"),
458 provider: ProviderDriverKind.make("codex"),
459 createdAt: now,
460 threadId: asThreadId("thread-1"),
461 turnId: asTurnId("turn-session-ready"),
462 });
463
464 await waitForThread(
465 harness.readModel,
466 (thread) =>
467 thread.session?.status === "running" &&
468 thread.session?.activeTurnId === "turn-session-ready",
469 10_000,
470 );
471
472 harness.emit({
473 type: "session.state.changed",
474 eventId: asEventId("evt-session-state-ready-with-active-turn"),
475 provider: ProviderDriverKind.make("codex"),
476 threadId: asThreadId("thread-1"),
477 createdAt: "2026-01-01T00:00:01.000Z",
478 payload: {
479 state: "ready",
480 },
481 });
482
483 const thread = await waitForThread(
484 harness.readModel,
485 (entry) =>
486 entry.session?.status === "ready" &&
487 entry.session?.activeTurnId === null &&
488 entry.session?.lastError === null,
489 10_000,
490 );
491 expect(thread.session?.status).toBe("ready");
492 expect(thread.session?.activeTurnId).toBeNull();
493 expect(thread.session?.lastError).toBeNull();
494 });
495
451 it("does not clear active turn when session/thread started arrives mid-turn", async () => {496 it("does not clear active turn when session/thread started arrives mid-turn", async () => {
452 const harness = await createHarness();497 const harness = await createHarness();
453 const now = "2026-01-01T00:00:00.000Z";498 const now = "2026-01-01T00:00:00.000Z";
...@@ -466,6 +511,7 @@ describe("ProviderRuntimeIngestion", () => {...@@ -466,6 +511,7 @@ describe("ProviderRuntimeIngestion", () => {
466 (thread) =>511 (thread) =>
467 thread.session?.status === "running" &&512 thread.session?.status === "running" &&
468 thread.session?.activeTurnId === "turn-midturn-lifecycle",513 thread.session?.activeTurnId === "turn-midturn-lifecycle",
514 10_000,
469 );515 );
470516
471 harness.emit({517 harness.emit({
...@@ -502,6 +548,7 @@ describe("ProviderRuntimeIngestion", () => {...@@ -502,6 +548,7 @@ describe("ProviderRuntimeIngestion", () => {
502 await waitForThread(548 await waitForThread(
503 harness.readModel,549 harness.readModel,
504 (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null,550 (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null,
551 10_000,
505 );552 );
506 });553 });
507554
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+14-6
...@@ -283,6 +283,12 @@ function orchestrationSessionStatusFromRuntimeState(...@@ -283,6 +283,12 @@ function orchestrationSessionStatusFromRuntimeState(
283 }283 }
284}284}
285285
286function sessionStatusAllowsActiveTurn(
287 status: ReturnType<typeof orchestrationSessionStatusFromRuntimeState>,
288): boolean {
289 return status === "starting" || status === "running";
290}
291
286function requestKindFromCanonicalRequestType(292function requestKindFromCanonicalRequestType(
287 requestType: string | undefined,293 requestType: string | undefined,
288): "command" | "file-read" | "file-change" | undefined {294): "command" | "file-read" | "file-change" | undefined {
...@@ -1362,12 +1368,6 @@ const make = Effect.gen(function* () {...@@ -1362,12 +1368,6 @@ const make = Effect.gen(function* () {
1362 event.type === "turn.started" ||1368 event.type === "turn.started" ||
1363 event.type === "turn.completed"1369 event.type === "turn.completed"
1364 ) {1370 ) {
1365 const nextActiveTurnId =
1366 event.type === "turn.started"
1367 ? (eventTurnId ?? null)
1368 : event.type === "turn.completed" || event.type === "session.exited"
1369 ? null
1370 : activeTurnId;
1371 const status = (() => {1371 const status = (() => {
1372 switch (event.type) {1372 switch (event.type) {
1373 case "session.state.changed":1373 case "session.state.changed":
...@@ -1387,6 +1387,14 @@ const make = Effect.gen(function* () {...@@ -1387,6 +1387,14 @@ const make = Effect.gen(function* () {
1387 return activeTurnId !== null ? "running" : "ready";1387 return activeTurnId !== null ? "running" : "ready";
1388 }1388 }
1389 })();1389 })();
1390 const nextActiveTurnId =
1391 event.type === "turn.started"
1392 ? (eventTurnId ?? null)
1393 : event.type === "turn.completed" || event.type === "session.exited"
1394 ? null
1395 : event.type === "session.state.changed" && !sessionStatusAllowsActiveTurn(status)
1396 ? null
1397 : activeTurnId;
1390 const lastError =1398 const lastError =
1391 event.type === "session.state.changed" && event.payload.state === "error"1399 event.type === "session.state.changed" && event.payload.state === "error"
1392 ? (event.payload.reason ?? thread.session?.lastError ?? "Provider session error")1400 ? (event.payload.reason ?? thread.session?.lastError ?? "Provider session error")