Skip to content
Merged
96 changes: 92 additions & 4 deletions apps/server/src/orchestration-v2/ProjectionStore.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3179,6 +3179,48 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
);
assert.isNull(shell?.latestRunCompletedAt);

const requestId = RuntimeRequestId.make("request:projection-shell-question");
for (const [responseCapability, blocking] of [
[{ type: "message" }, false],
[{ type: "live", providerSessionId: ProviderSessionId.make("session:question") }, true],
] as const) {
yield* projectionStore.apply({
id: EventId.make(`event:projection-shell-question:${responseCapability.type}`),
type: "runtime-request.updated",
threadId,
nodeId: rootNodeId,
driver,
occurredAt: now,
payload: {
id: requestId,
nodeId: rootNodeId,
providerTurnId: null,
nativeRequestRef: null,
kind: "user_input",
status: "pending",
responseCapability,
createdAt: now,
resolvedAt: null,
},
});
const projection = yield* projectionStore.getThreadProjection(threadId);
const sqlShell = (yield* projectionStore.getShellSnapshot()).threads.find(
(row) => row.id === threadId,
)!;
for (const questionShell of [
sqlShell,
ProjectionStore.threadShellFromProjection(projection),
]) {
assert.equal(questionShell.activeRunId, runId);
assert.deepEqual(questionShell.pendingRuntimeRequest, {
id: requestId,
kind: "user_input",
createdAt: now,
blocking,
});
}
}

yield* projectionStore.apply({
id: EventId.make("event:projection-shell-interruptible:waiting"),
type: "run.updated",
Expand Down Expand Up @@ -4355,7 +4397,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
title: null,
startedAt: now,
completedAt: null,
updatedAt: now,
updatedAt: DateTime.add(now, { seconds: ordinal }),
type: "secret_request",
label: "Test credential",
reason: "Test pending input",
Expand All @@ -4366,6 +4408,45 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
yield* addSecret("secret:b", 3, "waiting");
yield* addSecret("secret:c", 4, "completed");

yield* sql`
UPDATE orchestration_v2_projection_runs
SET status = 'running', payload_json = json_set(payload_json, '$.status', 'running')
WHERE run_id = ${runId}
`;
const secretOnlyShell = (yield* projectionStore.getShellSnapshot()).threads.find(
(thread) => thread.id === threadId,
);
const secretOnlyProjection = yield* projectionStore.getThreadProjection(threadId);
for (const shell of [
secretOnlyShell,
ProjectionStore.threadShellFromProjection(secretOnlyProjection),
]) {
assert.equal(shell?.pendingRuntimeRequest?.id, "secret:b");
assert.isTrue(shell?.pendingRuntimeRequest?.blocking);
}

const requestId = RuntimeRequestId.make(`request:${suffix}:async`);
yield* projectionStore.apply({
id: EventId.make(`event:${suffix}:async`),
type: "runtime-request.updated",
threadId,
runId,
nodeId,
driver,
occurredAt: now,
payload: {
id: requestId,
nodeId,
providerTurnId: null,
nativeRequestRef: null,
kind: "user_input",
status: "pending",
responseCapability: { type: "message" },
createdAt: DateTime.subtract(now, { seconds: 1 }),
resolvedAt: null,
},
});

for (const status of [
"preparing",
"starting",
Expand All @@ -4382,9 +4463,16 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
const shell = yield* projectionStore.getShellSnapshot();
const thread = shell.threads.find((candidate) => candidate.id === threadId);
assert.isDefined(thread);
assert.equal(
thread?.pendingRuntimeRequest?.id ?? null,
status === "completed" || status === "rolled_back" ? null : "secret:b",
const projection = yield* projectionStore.getThreadProjection(threadId);
const memoryShell = ProjectionStore.threadShellFromProjection(projection);
const terminal = status === "completed" || status === "rolled_back";
assert.deepEqual(
[thread?.pendingRuntimeRequest?.id, memoryShell.pendingRuntimeRequest?.id],
terminal ? [requestId, requestId] : ["secret:b", "secret:b"],
);
assert.deepEqual(
[thread?.pendingRuntimeRequest?.blocking, memoryShell.pendingRuntimeRequest?.blocking],
terminal ? [false, false] : [true, true],
);
}
}),
Expand Down
99 changes: 52 additions & 47 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import type {
OrchestrationV2CheckpointScope,
OrchestrationV2ExecutionNode,
OrchestrationV2PlanArtifact,
OrchestrationV2PendingRuntimeRequestSummary,
PlanId,
OrchestrationV2ConversationMessage,
OrchestrationV2DomainEvent,
Expand Down Expand Up @@ -1427,20 +1428,27 @@ function buildVisibleTurnItems(input: {
]);
}

function pendingRuntimeRequestSummary(
request: OrchestrationV2ThreadProjection["runtimeRequests"][number] | null,
): OrchestrationV2PendingRuntimeRequestSummary | null {
if (request === null) return null;
return {
id: request.id,
kind: request.kind,
createdAt: request.createdAt,
blocking: request.responseCapability.type !== "message",
};
}

function secretRequestAsPendingInput(
item: OrchestrationV2TurnItem | null,
): OrchestrationV2ThreadProjection["runtimeRequests"][number] | null {
): OrchestrationV2PendingRuntimeRequestSummary | null {
if (item?.type !== "secret_request" || item.nodeId === null) return null;
return {
id: RuntimeRequestId.make(item.id),
nodeId: item.nodeId,
providerTurnId: item.providerTurnId,
nativeRequestRef: null,
kind: "user_input",
status: "pending",
responseCapability: { type: "message" },
createdAt: item.startedAt ?? item.updatedAt,
resolvedAt: null,
blocking: true,
};
}

Expand Down Expand Up @@ -1469,27 +1477,31 @@ export function threadShellFromProjection(
.filter(isActivityRunForShell)
.toSorted((left, right) => right.ordinal - left.ordinal)[0] ?? null;
const liveRunIds = new Set(projection.runs.filter(isActivityRunForShell).map((run) => run.id));
const pendingRuntimeRequest =
let pendingRuntimeRequest = pendingRuntimeRequestSummary(
projection.runtimeRequests
.filter((request) => request.status === "pending")
.toSorted(
(left, right) =>
DateTime.toEpochMillis(right.createdAt) - DateTime.toEpochMillis(left.createdAt),
)[0] ??
secretRequestAsPendingInput(
projection.turnItems
.filter(
(item) =>
item.type === "secret_request" &&
item.status === "waiting" &&
item.runId !== null &&
liveRunIds.has(item.runId),
)
.toSorted(
(left, right) =>
DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
)[0] ?? null,
);
)[0] ?? null,
);
if (pendingRuntimeRequest === null || pendingRuntimeRequest.blocking === false) {
pendingRuntimeRequest =
secretRequestAsPendingInput(
projection.turnItems
.filter(
(item) =>
item.type === "secret_request" &&
item.status === "waiting" &&
item.runId !== null &&
liveRunIds.has(item.runId),
)
.toSorted(
(left, right) =>
DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
)[0] ?? null,
) ?? pendingRuntimeRequest;
}
const userMessages = projection.messages
.filter((message) => message.role === "user")
.toSorted(
Expand Down Expand Up @@ -1546,14 +1558,7 @@ export function threadShellFromProjection(
latestRootProviderFailure(latestRun, projection.turnItems),
providerSession?.lastError ?? null,
),
pendingRuntimeRequest:
pendingRuntimeRequest === null
? null
: {
id: pendingRuntimeRequest.id,
kind: pendingRuntimeRequest.kind,
createdAt: pendingRuntimeRequest.createdAt,
},
pendingRuntimeRequest,
// Thread detail owns message bodies. Keeping them out of shell rows makes
// initial hydration and streaming updates independent of transcript size.
latestVisibleMessage: null,
Expand Down Expand Up @@ -1648,7 +1653,7 @@ type ShellThreadState = {
readonly lastError: string | null;
readonly lastErrorClass: OrchestrationV2ThreadShell["lastErrorClass"];
readonly usageLimitResetAt: OrchestrationV2ThreadShell["usageLimitResetAt"];
readonly pendingRuntimeRequest: OrchestrationV2ThreadProjection["runtimeRequests"][number] | null;
readonly pendingRuntimeRequest: OrchestrationV2PendingRuntimeRequestSummary | null;
readonly latestUserMessageAt: DateTime.Utc | null;
readonly latestUserAuthoredMessageAt: DateTime.Utc | null;
readonly hasActionableProposedPlan: boolean;
Expand Down Expand Up @@ -1801,14 +1806,7 @@ function shellFromState(input: {
lastError: input.state.lastError,
lastErrorClass: input.state.lastErrorClass,
usageLimitResetAt: input.state.usageLimitResetAt,
pendingRuntimeRequest:
input.state.pendingRuntimeRequest === null
? null
: {
id: input.state.pendingRuntimeRequest.id,
kind: input.state.pendingRuntimeRequest.kind,
createdAt: input.state.pendingRuntimeRequest.createdAt,
},
pendingRuntimeRequest: input.state.pendingRuntimeRequest,
latestVisibleMessage: null,
latestUserMessageAt: input.state.latestUserMessageAt,
latestUserAuthoredMessageAt: input.state.latestUserAuthoredMessageAt,
Expand Down Expand Up @@ -5791,14 +5789,21 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
pendingTurnItemsByThreadId,
} = input;
const thread = yield* decodeThreadPayload(row.payload_json);
const pendingRuntimeRequest =
let pendingRuntimeRequest =
row.pending_request_payload_json !== null
? yield* decodeRuntimeRequestPayload(row.pending_request_payload_json)
: row.pending_secret_request_payload_json !== null
? secretRequestAsPendingInput(
yield* decodeTurnItemPayload(row.pending_secret_request_payload_json),
)
: null;
? pendingRuntimeRequestSummary(
yield* decodeRuntimeRequestPayload(row.pending_request_payload_json),
)
: null;
if (
(pendingRuntimeRequest === null || pendingRuntimeRequest.blocking === false) &&
row.pending_secret_request_payload_json !== null
) {
pendingRuntimeRequest =
secretRequestAsPendingInput(
yield* decodeTurnItemPayload(row.pending_secret_request_payload_json),
) ?? pendingRuntimeRequest;
}
let terminalFailureItem =
row.terminal_failure_payload_json === null
? null
Expand Down
52 changes: 52 additions & 0 deletions packages/client-runtime/src/state/entities.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,9 @@ import {
} from "./models.ts";
import { v2Projection, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";
import { deriveLatestThreadRun, deriveThreadRuntime } from "./threadExecution.ts";
import { isThreadWorking } from "./threadInbox.ts";
import { derivePendingThreadRequests } from "./threadRequests.ts";
import { canSnooze, effectiveSnoozed } from "./threadSettled.ts";

const environmentId = EnvironmentId.make("environment-v2");

Expand Down Expand Up @@ -111,6 +113,56 @@ describe("V2 client presentation", () => {
});
});

it("keeps async questions in Working until the run completes", () => {
const runId = RunId.make("run-async-question");
const running = {
...v2ThreadShell,
snoozedAt: DateTime.makeUnsafe("2026-06-19T23:59:00.000Z"),
snoozedUntil: DateTime.makeUnsafe("2026-06-20T01:00:00.000Z"),
latestRunId: runId,
activeRunId: runId,
status: "running" as const,
pendingRuntimeRequest: {
id: RuntimeRequestId.make("request-async-question"),
kind: "user_input" as const,
createdAt: v2ThreadShell.updatedAt,
blocking: false,
},
};
const active = presentThreadShell(environmentId, running);
expect(active.hasPendingUserInput).toBe(false);
expect(isThreadWorking(active)).toBe(true);
expect(canSnooze(active, { now: "2026-06-20T00:05:00.000Z" })).toBe(false);

const completed = presentThreadShell(environmentId, {
...running,
activeRunId: null,
status: "completed",
});
expect(completed.hasPendingUserInput).toBe(true);
expect(isThreadWorking(completed)).toBe(false);
expect(effectiveSnoozed(active, { now: "2026-06-20T00:05:00.000Z" })).toBe(false);
});

it("keeps blocking and legacy questions out of Working", () => {
for (const capability of [{ blocking: true }, {}]) {
const shell = presentThreadShell(environmentId, {
...v2ThreadShell,
latestRunId: RunId.make("run-blocking-question"),
activeRunId: RunId.make("run-blocking-question"),
status: "running",
pendingRuntimeRequest: {
id: RuntimeRequestId.make("request-blocking-question"),
kind: "user_input",
createdAt: v2ThreadShell.updatedAt,
...capability,
},
});
expect(shell.hasPendingUserInput).toBe(true);
expect(isThreadWorking(shell)).toBe(false);
}
});

it("preserves shell run timestamps for sidebar activity clocks", () => {
const runId = RunId.make("run-working-clock");
const requestedAt = DateTime.makeUnsafe("2026-06-20T01:00:00.000Z");
Expand Down
12 changes: 10 additions & 2 deletions packages/client-runtime/src/state/models.ts
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,12 @@ export function presentThreadShell(
: nullableIso(thread.latestRunCompletedAt),
assistantMessageId: null,
} satisfies ThreadRunSummary);
const runtime = shellRuntime(thread);
// Waiting runs have ended their provider turn and are only draining effects.
const runtimeIsWorking =
runtime?.status === "preparing" ||
runtime?.status === "starting" ||
runtime?.status === "running";
return {
environmentId,
id: thread.id,
Expand All @@ -241,7 +247,7 @@ export function presentThreadShell(
forkedFrom: thread.forkedFrom,
activeProviderThreadId: thread.activeProviderThreadId,
latestRun,
runtime: shellRuntime(thread),
runtime,
latestUserMessageAt: nullableIso(thread.latestUserMessageAt),
...(thread.latestUserAuthoredMessageAt === undefined
? {}
Expand All @@ -250,7 +256,9 @@ export function presentThreadShell(
thread.pendingRuntimeRequest !== null &&
thread.pendingRuntimeRequest.kind !== "user_input" &&
thread.pendingRuntimeRequest.kind !== "auth_refresh",
hasPendingUserInput: thread.pendingRuntimeRequest?.kind === "user_input",
hasPendingUserInput:
thread.pendingRuntimeRequest?.kind === "user_input" &&
(thread.pendingRuntimeRequest.blocking !== false || !runtimeIsWorking),
pendingSubagentRequest,
hasActionableProposedPlan: thread.hasActionableProposedPlan,
pendingBackgroundTasks: thread.pendingBackgroundTasks ?? [],
Expand Down
Loading
Loading