diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index a225030b8b48..8f416f156f81 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -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", @@ -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", @@ -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", @@ -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], ); } }), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 9deee8f3850a..508365bd789a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -19,6 +19,7 @@ import type { OrchestrationV2CheckpointScope, OrchestrationV2ExecutionNode, OrchestrationV2PlanArtifact, + OrchestrationV2PendingRuntimeRequestSummary, PlanId, OrchestrationV2ConversationMessage, OrchestrationV2DomainEvent, @@ -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, }; } @@ -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( @@ -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, @@ -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; @@ -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, @@ -5791,14 +5789,21 @@ export const layer: Layer.Layer = 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 diff --git a/packages/client-runtime/src/state/entities.test.ts b/packages/client-runtime/src/state/entities.test.ts index 02d9fa11696f..63a1ac57f2bb 100644 --- a/packages/client-runtime/src/state/entities.test.ts +++ b/packages/client-runtime/src/state/entities.test.ts @@ -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"); @@ -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"); diff --git a/packages/client-runtime/src/state/models.ts b/packages/client-runtime/src/state/models.ts index 36535a273cc0..91fe95b8585b 100644 --- a/packages/client-runtime/src/state/models.ts +++ b/packages/client-runtime/src/state/models.ts @@ -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, @@ -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 ? {} @@ -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 ?? [], diff --git a/packages/client-runtime/src/state/threadSettled.ts b/packages/client-runtime/src/state/threadSettled.ts index c7172c330008..a48f22b28e33 100644 --- a/packages/client-runtime/src/state/threadSettled.ts +++ b/packages/client-runtime/src/state/threadSettled.ts @@ -1,7 +1,7 @@ // @effect-diagnostics globalDate:off -- UI snooze presets use local calendar boundaries and Intl labels. import * as DateTime from "effect/DateTime"; -import type { PendingSubagentRequest } from "./models.ts"; +import type { EnvironmentThreadShell, PendingSubagentRequest } from "./models.ts"; interface SettlementRunLike { readonly turnId?: unknown; @@ -71,6 +71,7 @@ export function hasQueuedTurnStart( } export interface ThreadSnoozeShell extends QueuedThreadShell { + readonly source?: Pick; readonly snoozedUntil?: string | null; readonly snoozedAt?: string | null; readonly hasPendingApprovals: boolean; @@ -80,6 +81,7 @@ export interface ThreadSnoozeShell extends QueuedThreadShell { export function threadRaisedHandWhileSnoozed(shell: ThreadSnoozeShell): boolean { if (shell.hasPendingApprovals || shell.hasPendingUserInput) return true; + if (shell.source?.pendingRuntimeRequest != null) return true; if (shell.pendingSubagentRequest != null) return true; const latestRun = shell.latestRun ?? shell.latestTurn ?? null; const failedAt = failureEdgeAt(shell); @@ -124,10 +126,13 @@ export function canSnooze( | "latestRun" | "session" | "runtime" + | "source" >, options: { readonly now: string }, ): boolean { if (shell.hasPendingApprovals || shell.hasPendingUserInput) return false; + // The server rejects snoozing even for nonblocking questions. + if (shell.source?.pendingRuntimeRequest != null) return false; if (shell.pendingSubagentRequest != null) return false; if (hasQueuedTurnStart(shell, options)) return false; return true; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 94a9eca331d1..2dc4296bb963 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1782,6 +1782,8 @@ export const OrchestrationV2PendingRuntimeRequestSummary = Schema.Struct({ id: RuntimeRequestId, kind: OrchestrationV2RuntimeRequest.fields.kind, createdAt: Schema.DateTimeUtc, + /** False for async questions the run keeps working past. Absent from older servers. */ + blocking: Schema.optional(Schema.Boolean), }); export type OrchestrationV2PendingRuntimeRequestSummary = typeof OrchestrationV2PendingRuntimeRequestSummary.Type; diff --git a/packages/shared/src/agentAwareness.test.ts b/packages/shared/src/agentAwareness.test.ts index c31312646c02..85b8d614150b 100644 --- a/packages/shared/src/agentAwareness.test.ts +++ b/packages/shared/src/agentAwareness.test.ts @@ -122,6 +122,28 @@ describe("projectThreadAwarenessV2", () => { ).toMatchObject({ phase: "waiting_for_input", headline: "Waiting for input" }); }); + it("reports an async question as input only after the working run completes", () => { + const input = { + environmentId: "env-1" as EnvironmentId, + project, + thread: v2Thread({ + pendingRuntimeRequest: { + id: RuntimeRequestId.make("request-async"), + kind: "user_input", + createdAt: updatedAt, + blocking: false, + }, + }), + }; + expect(projectThreadAwarenessV2(input)).toMatchObject({ + phase: "running", + headline: "Agent is working", + }); + expect( + projectThreadAwarenessV2({ ...input, thread: { ...input.thread, status: "completed" } }), + ).toMatchObject({ phase: "waiting_for_input", headline: "Waiting for input" }); + }); + it("does not present authentication refreshes as user approvals", () => { expect( projectThreadAwarenessV2({ diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index b8bd43766de5..b8f6a194a9d0 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -86,16 +86,22 @@ export function projectThreadAwarenessV2( function resolveThreadAwarenessPhaseV2( thread: ProjectThreadAwarenessV2Input["thread"], ): AgentAwarenessPhase | null { - if (thread.pendingRuntimeRequest?.kind === "user_input") { + const runStatus = thread.activityRunStatus ?? thread.status; + if ( + thread.pendingRuntimeRequest?.kind === "user_input" && + (thread.pendingRuntimeRequest.blocking !== false || + !(runStatus === "preparing" || runStatus === "starting" || runStatus === "running")) + ) { return "waiting_for_input"; } if ( thread.pendingRuntimeRequest !== null && + thread.pendingRuntimeRequest.kind !== "user_input" && thread.pendingRuntimeRequest.kind !== "auth_refresh" ) { return "waiting_for_approval"; } - switch (thread.activityRunStatus ?? thread.status) { + switch (runStatus) { case "preparing": case "starting": return "starting";