diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 28f229b22..16110da99 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -1663,6 +1663,87 @@ describe("ProviderCommandReactor", () => { }), ); + effectIt.effect("steers a running admitted turn without reopening admission", () => + Effect.gen(function* () { + // The everyday steer: a user message lands while an admitted turn is + // running. It must reach the provider tagged with the running turn's + // request id, and the session must stay as it was: no restart, no + // pending admission, same active turn. + const testClock = yield* TestClock.make(); + yield* testClock.setTime(PROVIDER_TURN_ADMISSION_TIMEOUT_MS + 1); + const requestId = CommandId.make("cmd-steer-admitted-boot"); + const messageId = asMessageId("message-steer-admitted-boot"); + const sessionIncarnationId = RuntimeSessionId.make("session-steer-admitted"); + const activeTurnId = asTurnId("turn-steer-admitted"); + const runningSession = { + provider: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex"), + status: "running" as const, + runtimeMode: "approval-required" as const, + threadId: ThreadId.make("thread-1"), + cwd: "/tmp/provider-project", + sessionIncarnationId, + activeTurnRequestId: requestId, + activeTurnId, + createdAt: isoAt(0), + updatedAt: isoAt(1), + }; + const harness = yield* Effect.promise(() => + createHarness({ + clock: testClock, + overdueTurnStartBeforeReactor: { + commandId: requestId, + messageId, + createdAt: isoAt(0), + sessionIncarnationId, + }, + inventoryEffect: () => Effect.succeed([runningSession]), + initialRuntimeSessions: [runningSession], + }), + ); + const before = yield* Effect.promise(() => harness.readModel()); + const runningThread = before.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + expect(runningThread?.session?.status).toBe("running"); + expect(runningThread?.session?.activeTurnRequestId).toBe(requestId); + + const steerRequestId = CommandId.make("cmd-steer-admitted"); + yield* harness.engine.dispatch({ + type: "thread.turn.start", + commandId: steerRequestId, + threadId: ThreadId.make("thread-1"), + message: { + messageId: asMessageId("user-message-steer-admitted"), + role: "user", + text: "actually, also run the tests", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt: isoAt(PROVIDER_TURN_ADMISSION_TIMEOUT_MS + 2), + }); + yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 1)); + + const sent = harness.sendTurn.mock.calls[0]?.[0] as ProviderSendTurnInput; + expect(sent.input).toBe("actually, also run the tests"); + expect(sent.admissionRequestId).toBe(requestId); + expect(harness.startSession.mock.calls.map((call) => JSON.stringify(call[1]))).toEqual([]); + expect(sent.sessionIncarnationId).toBe(sessionIncarnationId); + + const after = yield* Effect.promise(() => harness.readModel()); + const thread = after.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + expect(thread?.session?.status).toBe("running"); + expect(thread?.session?.activeTurnId).toBe(activeTurnId); + expect(thread?.session?.activeTurnRequestId).toBe(requestId); + expect(thread?.session?.pendingTurnRequestId).toBeUndefined(); + expect(thread?.messages.map((message) => message.id)).toContain( + asMessageId("user-message-steer-admitted"), + ); + expect( + thread?.activities.filter((activity) => activity.kind === "provider.turn.start.failed"), + ).toHaveLength(0); + }), + ); + for (const inventoryStatus of ["ready", "absent"] as const) { effectIt.effect(`fails an overdue ${inventoryStatus} per-instance inventory as absence`, () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration/decider.sessionLifecycle.test.ts b/apps/server/src/orchestration/decider.sessionLifecycle.test.ts index 60141d003..4e60e9c54 100644 --- a/apps/server/src/orchestration/decider.sessionLifecycle.test.ts +++ b/apps/server/src/orchestration/decider.sessionLifecycle.test.ts @@ -417,6 +417,114 @@ it.layer(NodeServices.layer)("session lifecycle CAS decider", (it) => { }), ); + it.effect("starts a new turn instead of steering when a running session has no active turn", () => + Effect.gen(function* () { + // Claude flips the session to "running" on its own system/status + // notifications between turns, so status alone cannot prove a turn + // exists. Steering nothing would hand the provider a turn the admission + // gate can never correlate, and the user's message would vanish. + const runningWithoutTurn = makeSession({ + status: "running", + pendingTurnRequestId: undefined, + pendingTurnMessageId: undefined, + pendingTurnRequestedAt: undefined, + pendingTurnDeadlineAt: undefined, + pendingTurnSessionId: undefined, + activeTurnRequestId: undefined, + activeTurnId: null, + }); + const commandId = CommandId.make("cmd-running-without-turn"); + const decided = yield* decideOrchestrationCommand({ + command: { + type: "thread.turn.start", + commandId, + threadId: THREAD_ID, + message: { + messageId: MessageId.make("message-running-without-turn"), + role: "user", + text: "nothing is running, start a turn", + attachments: [], + }, + modelSelection: { instanceId: INSTANCE_ID, model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + createdAt: NOW, + }, + readModel: makeReadModel(runningWithoutTurn), + }); + const events = Array.isArray(decided) ? decided : [decided]; + expect(events.map((event) => event.type)).toEqual([ + "thread.message-sent", + "thread.session-set", + "thread.turn-start-requested", + ]); + expect(events[1]).toMatchObject({ + type: "thread.session-set", + payload: { + session: { + status: "starting", + pendingTurnRequestId: commandId, + activeTurnId: null, + }, + }, + }); + expect(events[2]).toMatchObject({ + type: "thread.turn-start-requested", + payload: { admissionIntent: { kind: "start", expectedActiveTurnRequestId: null } }, + }); + }), + ); + + it.effect("starts a new turn instead of steering a provider-initiated turn", () => + Effect.gen(function* () { + // A turn the provider opened on its own (Claude continuing after a + // background task) is running but was never admitted, so it has no + // active request id. Steering it would tag the provider's next turn with + // a request id the admission gate cannot correlate. Start exactly. + const providerInitiated = makeSession({ + status: "running", + pendingTurnRequestId: undefined, + pendingTurnMessageId: undefined, + pendingTurnRequestedAt: undefined, + pendingTurnDeadlineAt: undefined, + pendingTurnSessionId: undefined, + activeTurnRequestId: undefined, + activeTurnId: TurnId.make("turn-provider-initiated"), + }); + const commandId = CommandId.make("cmd-provider-initiated-turn"); + const decided = yield* decideOrchestrationCommand({ + command: { + type: "thread.turn.start", + commandId, + threadId: THREAD_ID, + message: { + messageId: MessageId.make("message-provider-initiated-turn"), + role: "user", + text: "take over from the background continuation", + attachments: [], + }, + modelSelection: { instanceId: INSTANCE_ID, model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + createdAt: NOW, + }, + readModel: makeReadModel(providerInitiated), + }); + const events = Array.isArray(decided) ? decided : [decided]; + expect(events.map((event) => event.type)).toEqual([ + "thread.message-sent", + "thread.session-set", + "thread.turn-start-requested", + ]); + expect(events[1]).toMatchObject({ + payload: { session: { status: "starting", pendingTurnRequestId: commandId } }, + }); + expect(events[2]).toMatchObject({ + payload: { admissionIntent: { kind: "start" } }, + }); + }), + ); + it.effect("captures the exact stop target and projects stopped atomically", () => Effect.gen(function* () { const turnId = TurnId.make("turn-stop-target"); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index aeb383a36..52d7d70c6 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1033,12 +1033,26 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" updatedAt: command.createdAt, }, }; + // Steering needs an admitted turn to steer into. A session can sit in + // "running" with no active turn (Claude reports system/status between + // turns) or with a turn the provider opened on its own (Claude continuing + // after a background task). Steering there hands the provider a turn + // ingestion can never correlate to an admission, so the user's message + // silently disappears. Under an incarnation-tracked session every + // admitted turn carries its request id, so a running turn without one is + // provider-initiated and gets an exact start instead. Sessions without an + // incarnation predate admission tracking and keep plain steering. + const hasSteerableTurn = + targetThread.session?.status === "running" && + targetThread.session.activeTurnId !== null && + (targetThread.session.activeTurnRequestId !== undefined || + targetThread.session.sessionIncarnationId === undefined); const admissionIntent = { kind: targetThread.session?.providerInstanceId !== undefined && targetThread.session.providerInstanceId !== effectiveModelSelection.instanceId ? ("compatible-transition" as const) - : targetThread.session?.status === "running" && !providerSettingsChanged + : hasSteerableTurn && !providerSettingsChanged ? ("steer" as const) : ("start" as const), expectedProviderInstanceId: targetThread.session?.providerInstanceId ?? null, @@ -1071,7 +1085,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }, }; const admissionPendingEvents: Array> = []; - if (targetThread.session?.status !== "running" || providerSettingsChanged) { + if (!hasSteerableTurn || providerSettingsChanged) { admissionPendingEvents.push({ ...(yield* withEventBase({ aggregateKind: "thread",