Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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* () {
Expand Down
108 changes: 108 additions & 0 deletions apps/server/src/orchestration/decider.sessionLifecycle.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down
18 changes: 16 additions & 2 deletions apps/server/src/orchestration/decider.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand Down
Loading