Merge pull request #264 from pylon-code/fix/steer-requires-active-turn

fix(server): steer a turn only when one is actually running

Author: rynfar

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* () {

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");

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<Omit<OrchestrationEvent, "sequence">> = [];
-      if (targetThread.session?.status !== "running" || providerSettingsChanged) {
+      if (!hasSteerableTurn || providerSettingsChanged) {
         admissionPendingEvents.push({
           ...(yield* withEventBase({
             aggregateKind: "thread",