diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index 7c232a24786b7..b09a1a6c34dfe 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -1,9 +1,11 @@ import { - ApprovalRequestId, - isToolLifecycleItemType, - ProviderApprovalOption, - ProviderRequestKind, -} from "@t3tools/contracts"; + derivePendingApprovals, + derivePendingUserInputs, + requestKindFromRequestType, + type PendingApproval, + type PendingUserInput, +} from "@t3tools/client-runtime/pending-requests"; +import { isToolLifecycleItemType } from "@t3tools/contracts"; import type { OrchestrationLatestTurn, OrchestrationThread, @@ -16,28 +18,11 @@ import { formatDuration } from "@t3tools/shared/orchestrationTiming"; import * as Arr from "effect/Array"; import * as Order from "effect/Order"; -import * as Schema from "effect/Schema"; import { buildBotStepMeters, type BotStepMeterData } from "../features/threads/botStepUsage"; -export interface PendingApproval { - readonly requestId: ApprovalRequestId; - readonly requestKind: ProviderRequestKind; - readonly createdAt: string; - readonly detail?: string; - readonly appName?: string; - readonly toolName?: string; - readonly options?: ReadonlyArray; -} - -const isProviderRequestKind = Schema.is(ProviderRequestKind); -const isProviderApprovalOption = Schema.is(ProviderApprovalOption); - -export interface PendingUserInput { - readonly requestId: ApprovalRequestId; - readonly createdAt: string; - readonly questions: ReadonlyArray; -} +export type { PendingApproval, PendingUserInput }; +export { derivePendingApprovals, derivePendingUserInputs }; export interface PendingUserInputDraftAnswer { readonly selectedOptionLabels?: ReadonlyArray; @@ -152,91 +137,6 @@ export type ThreadFeedLatestTurn = Pick< "turnId" | "state" | "startedAt" | "completedAt" >; -function requestKindFromRequestType(requestType: unknown): PendingApproval["requestKind"] | null { - switch (requestType) { - case "command_execution_approval": - case "exec_command_approval": - case "dynamic_tool_call": - return "command"; - case "file_read_approval": - return "file-read"; - case "file_change_approval": - case "apply_patch_approval": - return "file-change"; - case "mcp_elicitation_approval": - return "mcp-elicitation"; - default: - return null; - } -} - -function isStalePendingRequestFailureDetail(detail: string | undefined): boolean { - const normalized = detail?.toLowerCase(); - if (!normalized) { - return false; - } - return ( - normalized.includes("stale pending approval request") || - normalized.includes("stale pending user-input request") || - normalized.includes("unknown pending approval request") || - normalized.includes("unknown pending permission request") || - normalized.includes("unknown pending user-input request") - ); -} - -function parseApprovalRequestId(value: unknown): ApprovalRequestId | null { - return typeof value === "string" && value.length > 0 ? ApprovalRequestId.make(value) : null; -} - -function parseUserInputQuestions( - payload: Record | null, -): ReadonlyArray | null { - const questions = payload?.questions; - if (!Array.isArray(questions)) { - return null; - } - - const parsed = questions - .map((entry) => { - if (!entry || typeof entry !== "object") return null; - const question = entry as Record; - if ( - typeof question.id !== "string" || - typeof question.header !== "string" || - typeof question.question !== "string" || - !Array.isArray(question.options) - ) { - return null; - } - const options = question.options - .map((option) => { - if (!option || typeof option !== "object") return null; - const record = option as Record; - if (typeof record.label !== "string" || typeof record.description !== "string") { - return null; - } - return { - label: record.label, - description: record.description, - }; - }) - .filter((option): option is UserInputQuestion["options"][number] => option !== null); - if (options.length === 0) { - return null; - } - return { - id: question.id, - header: question.header, - question: question.question, - options, - multiSelect: question.multiSelect === true, - }; - }) - .filter((question): question is UserInputQuestion => question !== null); - - return parsed.length > 0 ? parsed : null; -} - function normalizeDraftAnswer(value: string | undefined): string | null { if (typeof value !== "string") { return null; @@ -1384,100 +1284,6 @@ export function sortThreadActivities( return Arr.sort(activities, activityOrder); } -export function derivePendingApprovals( - sortedActivities: ReadonlyArray, -): PendingApproval[] { - const openByRequestId = new Map(); - - for (const activity of sortedActivities) { - const payload = - activity.payload && typeof activity.payload === "object" - ? (activity.payload as Record) - : null; - const requestId = parseApprovalRequestId(payload?.requestId); - const requestKind = isProviderRequestKind(payload?.requestKind) - ? payload.requestKind - : requestKindFromRequestType(payload?.requestType); - const detail = typeof payload?.detail === "string" ? payload.detail : undefined; - const appName = typeof payload?.appName === "string" ? payload.appName : undefined; - const toolName = typeof payload?.toolName === "string" ? payload.toolName : undefined; - const options = Array.isArray(payload?.options) - ? payload.options.filter(isProviderApprovalOption) - : undefined; - - if (activity.kind === "approval.requested" && requestId) { - openByRequestId.set(requestId, { - requestId, - requestKind: requestKind ?? "command", - createdAt: activity.createdAt, - ...(detail ? { detail } : {}), - ...(appName ? { appName } : {}), - ...(toolName ? { toolName } : {}), - ...(options && options.length > 0 ? { options } : {}), - }); - continue; - } - - if (activity.kind === "approval.resolved" && requestId) { - openByRequestId.delete(requestId); - continue; - } - - if ( - activity.kind === "provider.approval.respond.failed" && - requestId && - isStalePendingRequestFailureDetail(detail) - ) { - openByRequestId.delete(requestId); - } - } - - return Arr.sortWith([...openByRequestId.values()], (s) => new Date(s.createdAt), Order.Date); -} - -export function derivePendingUserInputs( - sortedActivities: ReadonlyArray, -): PendingUserInput[] { - const openByRequestId = new Map(); - - for (const activity of sortedActivities) { - const payload = - activity.payload && typeof activity.payload === "object" - ? (activity.payload as Record) - : null; - const requestId = parseApprovalRequestId(payload?.requestId); - const detail = typeof payload?.detail === "string" ? payload.detail : undefined; - - if (activity.kind === "user-input.requested" && requestId) { - const questions = parseUserInputQuestions(payload); - if (!questions) { - continue; - } - openByRequestId.set(requestId, { - requestId, - createdAt: activity.createdAt, - questions, - }); - continue; - } - - if (activity.kind === "user-input.resolved" && requestId) { - openByRequestId.delete(requestId); - continue; - } - - if ( - activity.kind === "provider.user-input.respond.failed" && - requestId && - isStalePendingRequestFailureDetail(detail) - ) { - openByRequestId.delete(requestId); - } - } - - return Arr.sortWith(openByRequestId.values(), (s) => new Date(s.createdAt), Order.Date); -} - export function setPendingUserInputCustomAnswer( draft: PendingUserInputDraftAnswer | undefined, customAnswer: string, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index ef6ae6e2ec119..d3cb22a66b56a 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -39,6 +39,7 @@ import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as ManagedRuntime from "effect/ManagedRuntime"; +import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; @@ -466,6 +467,12 @@ describe("ProviderRuntimeIngestion", () => { }; }, readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()), + readThreadShell: () => + runtime!.runPromise( + snapshotQuery + .getThreadShellById(asThreadId("thread-1")) + .pipe(Effect.map(Option.getOrThrow)), + ), emit: provider.emit, setProviderSession: provider.setSession, drain, @@ -4679,6 +4686,203 @@ describe("ProviderRuntimeIngestion", () => { }); }); + function userInputEvent( + turnId: string, + requestId: string, + responseMode?: "message", + ): ProviderRuntimeEvent { + return { + type: "user-input.requested", + eventId: asEventId(`requested:${requestId}`), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + turnId: asTurnId(turnId), + requestId: RuntimeRequestId.make(requestId), + createdAt: "2026-01-01T00:00:01.000Z", + payload: { + ...(responseMode ? { responseMode } : {}), + questions: ["first", "second"].map((id) => ({ + id, + header: id, + question: `Choose ${id}`, + options: [{ label: "yes", description: "Continue" }], + multiSelect: false, + })), + }, + }; + } + + it.each(["completed", "interrupted", "failed"] as const)( + "resolves native questions when their turn is %s", + async (state) => { + const harness = await createHarness(); + const request = userInputEvent("question-turn", "question-request"); + harness.emit({ + type: "turn.started", + eventId: asEventId("question-started"), + provider: request.provider, + threadId: request.threadId, + turnId: request.turnId, + createdAt: request.createdAt, + }); + harness.emit(request); + await harness.drain(); + expect((await harness.readThreadShell()).hasPendingUserInput).toBe(true); + harness.emit({ + type: "turn.completed", + eventId: asEventId("question-completed"), + provider: request.provider, + threadId: request.threadId, + turnId: request.turnId, + createdAt: "2026-01-01T00:00:03.000Z", + payload: { state }, + }); + await harness.drain(); + const thread = (await harness.readModel()).threads[0]!; + expect(thread.session?.activeTurnId).toBeNull(); + expect((await harness.readThreadShell()).hasPendingUserInput).toBe(false); + expect( + thread.activities.filter((activity) => activity.kind === "user-input.resolved"), + ).toMatchObject([{ turnId: request.turnId, payload: { requestId: request.requestId } }]); + }, + ); + + it("resolves native questions when their turn is aborted", async () => { + const harness = await createHarness(); + const request = userInputEvent("abort-turn", "abort-question"); + harness.emit({ + type: "turn.started", + eventId: asEventId("abort-started"), + provider: request.provider, + threadId: request.threadId, + turnId: request.turnId, + createdAt: request.createdAt, + }); + harness.emit(request); + await harness.drain(); + harness.emit({ + type: "turn.aborted", + eventId: asEventId("abort-completed"), + provider: request.provider, + threadId: request.threadId, + turnId: request.turnId, + createdAt: "2026-01-01T00:00:03.000Z", + payload: { reason: "Interrupted by user." }, + }); + await harness.drain(); + expect((await harness.readThreadShell()).hasPendingUserInput).toBe(false); + expect( + (await harness.readModel()).threads[0]!.activities.filter( + (activity) => activity.kind === "user-input.resolved", + ), + ).toMatchObject([{ payload: { requestId: request.requestId } }]); + }); + + it("resolves a terminal native question after an ordinary response failure", async () => { + const harness = await createHarness(); + const request = userInputEvent("failed-response-turn", "failed-response-question"); + harness.emit({ + type: "turn.started", + eventId: asEventId("failed-response-started"), + provider: request.provider, + threadId: request.threadId, + turnId: request.turnId, + createdAt: request.createdAt, + }); + harness.emit(request); + await harness.drain(); + await harness.dispatch({ + type: "thread.activity.append", + commandId: CommandId.make("cmd-failed-user-input-response"), + threadId: request.threadId, + activity: { + id: asEventId("failed-user-input-response"), + createdAt: "2026-01-01T00:00:02.000Z", + tone: "error", + kind: "provider.user-input.respond.failed", + summary: "User input response failed", + payload: { + requestId: request.requestId, + detail: "Provider connection failed while sending the response", + }, + turnId: asTurnId("failed-response-turn"), + }, + createdAt: "2026-01-01T00:00:02.000Z", + }); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("failed-response-completed"), + provider: request.provider, + threadId: request.threadId, + turnId: request.turnId, + createdAt: "2026-01-01T00:00:03.000Z", + payload: { state: "failed" }, + }); + await harness.drain(); + + expect((await harness.readThreadShell()).hasPendingUserInput).toBe(false); + expect( + (await harness.readModel()).threads[0]!.activities.filter( + (activity) => activity.kind === "user-input.resolved", + ), + ).toMatchObject([{ payload: { requestId: request.requestId } }]); + }); + + it("preserves answered questions and leaves newer and message-mode questions pending", async () => { + const harness = await createHarness(); + const answered = userInputEvent("old-turn", "answered-question"); + const unresolved = userInputEvent("old-turn", "old-question"); + const newer = userInputEvent("new-turn", "new-question"); + const asynchronous = userInputEvent("old-turn", "async-question", "message"); + harness.emit(answered); + harness.emit(unresolved); + harness.emit(newer); + harness.emit(asynchronous); + harness.emit({ + type: "user-input.resolved", + eventId: asEventId("normal-answer"), + provider: answered.provider, + threadId: answered.threadId, + turnId: answered.turnId, + requestId: answered.requestId, + createdAt: "2026-01-01T00:00:02.000Z", + payload: { answers: { first: "yes", second: "yes" } }, + }); + harness.emit({ + type: "turn.started", + eventId: asEventId("new-turn-started"), + provider: newer.provider, + threadId: newer.threadId, + turnId: newer.turnId, + createdAt: "2026-01-01T00:00:03.000Z", + }); + await harness.drain(); + expect((await harness.readThreadShell()).hasPendingUserInput).toBe(true); + harness.emit({ + type: "turn.completed", + eventId: asEventId("old-turn-completed"), + provider: answered.provider, + threadId: answered.threadId, + turnId: answered.turnId, + createdAt: "2026-01-01T00:00:04.000Z", + payload: { state: "interrupted" }, + }); + await harness.drain(); + const thread = (await harness.readModel()).threads[0]!; + expect(thread.session?.activeTurnId).toBe(newer.turnId); + expect((await harness.readThreadShell()).hasPendingUserInput).toBe(true); + expect( + thread.activities.filter((activity) => activity.kind === "user-input.resolved"), + ).toMatchObject([ + { + id: "normal-answer", + payload: { requestId: answered.requestId, answers: { first: "yes", second: "yes" } }, + }, + { turnId: unresolved.turnId, payload: { requestId: unresolved.requestId } }, + ]); + }); + it("continues processing runtime events after a single event handler failure", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index f00cb8327bd26..22088a341194f 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -31,6 +31,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Exit from "effect/Exit"; +import * as Predicate from "effect/Predicate"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; @@ -532,6 +533,10 @@ export function runtimeEventToActivities( payload: { ...(event.requestId ? { requestId: event.requestId } : {}), questions: event.payload.questions, + ...("responseMode" in event.payload && + (event.payload as { responseMode?: unknown }).responseMode === "message" + ? { responseMode: "message" as const } + : {}), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -1000,6 +1005,64 @@ const make = Effect.gen(function* () { Effect.map((uuid) => CommandId.make(`provider:${event.eventId}:${tag}:${uuid}`)), ); + const resolveNativeUserInputForTerminalTurn = (input: { + readonly event: ProviderRuntimeEvent; + readonly threadId: ThreadId; + readonly turnId: TurnId; + readonly now: string; + }) => + Effect.gen(function* () { + const activities = yield* projectionThreadActivities.listUserInputLifecycleByThreadAndTurn({ + threadId: input.threadId, + turnId: input.turnId, + }); + const pendingRequestIds = new Set(); + for (const activity of activities) { + const payload = Predicate.isObject(activity.payload) ? activity.payload : undefined; + if (!payload || typeof payload.requestId !== "string") continue; + const requestId = payload.requestId; + + if (activity.kind === "user-input.requested" && payload.responseMode !== "message") { + pendingRequestIds.add(requestId); + continue; + } + if (activity.kind === "user-input.resolved") { + pendingRequestIds.delete(requestId); + continue; + } + if (activity.kind !== "provider.user-input.respond.failed") continue; + const detail = typeof payload.detail === "string" ? payload.detail.toLowerCase() : ""; + if ( + detail.includes("stale pending user-input request") || + detail.includes("unknown pending user-input request") || + detail.includes("unknown pending user input request") || + detail.includes("unknown pending codex user input request") + ) { + pendingRequestIds.delete(requestId); + } + } + + // Native callbacks cannot accept answers after their turn terminates. + // Message-mode questions can still receive a later chat message. + for (const requestId of pendingRequestIds) { + yield* orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId: yield* providerCommandId(input.event, "terminal-user-input-resolved"), + threadId: input.threadId, + activity: { + id: EventId.make(`${input.event.eventId}:user-input-resolved:${requestId}`), + createdAt: input.now, + tone: "info", + kind: "user-input.resolved", + summary: "User input dismissed", + payload: { requestId }, + turnId: input.turnId, + }, + createdAt: input.now, + }); + } + }); + const turnMessageIdsByTurnKey = yield* Cache.make>({ capacity: TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY, timeToLive: TURN_MESSAGE_IDS_BY_TURN_TTL, @@ -2075,6 +2138,18 @@ const make = Effect.gen(function* () { }); } + if (event.type === "turn.completed" || event.type === "turn.aborted") { + const turnId = toTurnId(event.turnId); + if (turnId) { + yield* resolveNativeUserInputForTerminalTurn({ + event, + threadId: thread.id, + turnId, + now, + }); + } + } + if (event.type === "turn.completed") { const turnId = toTurnId(event.turnId); if (turnId) { diff --git a/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts b/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts index fc93e5f3c8d43..9e1f1cfba8180 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts @@ -13,6 +13,7 @@ import { DeleteProjectionThreadActivitiesInput, GetLatestProjectionThreadTaskActivityInput, ListProjectionThreadActivitiesInput, + ListProjectionThreadUserInputLifecycleInput, ProjectionThreadActivity, ProjectionThreadActivityRepository, type ProjectionThreadActivityRepositoryShape, @@ -155,6 +156,33 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { `, }); + const listUserInputLifecycleActivityRowsForTurn = SqlSchema.findAll({ + Request: ListProjectionThreadUserInputLifecycleInput, + Result: ProjectionThreadActivityDbRowSchema, + execute: ({ threadId, turnId }) => + sql` + SELECT + activity_id AS "activityId", + thread_id AS "threadId", + turn_id AS "turnId", + tone, + kind, + summary, + payload_json AS "payload", + sequence, + created_at AS "createdAt" + FROM projection_thread_activities + WHERE thread_id = ${threadId} + AND turn_id = ${turnId} + AND kind IN ( + 'user-input.requested', + 'user-input.resolved', + 'provider.user-input.respond.failed' + ) + ORDER BY sequence ASC, created_at ASC, activity_id ASC + `, + }); + const getLatestProjectionThreadTaskActivityRow = SqlSchema.findOneOption({ Request: GetLatestProjectionThreadTaskActivityInput, Result: ProjectionThreadActivityDbRowSchema, @@ -231,6 +259,18 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { Effect.map((rows) => rows.map(toProjectionThreadActivity)), ); + const listUserInputLifecycleByThreadAndTurn: ProjectionThreadActivityRepositoryShape["listUserInputLifecycleByThreadAndTurn"] = + (input) => + listUserInputLifecycleActivityRowsForTurn(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionThreadActivityRepository.listUserInputLifecycleByThreadAndTurn:query", + "ProjectionThreadActivityRepository.listUserInputLifecycleByThreadAndTurn:decodeRows", + ), + ), + Effect.map((rows) => rows.map(toProjectionThreadActivity)), + ); + const getLatestTaskActivity: ProjectionThreadActivityRepositoryShape["getLatestTaskActivity"] = ( input, ) => @@ -255,6 +295,7 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { upsert, listByThreadId, listUserInputLifecycleByThreadId, + listUserInputLifecycleByThreadAndTurn, getLatestTaskActivity, deleteByThreadId, } satisfies ProjectionThreadActivityRepositoryShape; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 3c3c32d63360f..60d66ef793ca4 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -77,6 +77,7 @@ import Migration0061 from "./Migrations/061_ExternalChannels.ts"; import Migration0062 from "./Migrations/062_Routines.ts"; import Migration0063 from "./Migrations/063_RoutineTargetThreadRepair.ts"; import Migration0064 from "./Migrations/064_ExternalChannelProviders.ts"; +import Migration0065 from "./Migrations/065_ProjectionThreadUserInputLifecycleIndex.ts"; /** * Migration loader with all migrations defined inline. @@ -153,6 +154,7 @@ export const migrationEntries = [ [62, "Routines", Migration0062], [63, "RoutineTargetThreadRepair", Migration0063], [64, "ExternalChannelProviders", Migration0064], + [65, "ProjectionThreadUserInputLifecycleIndex", Migration0065], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/065_ProjectionThreadUserInputLifecycleIndex.ts b/apps/server/src/persistence/Migrations/065_ProjectionThreadUserInputLifecycleIndex.ts new file mode 100644 index 0000000000000..5b5162f9f6cd4 --- /dev/null +++ b/apps/server/src/persistence/Migrations/065_ProjectionThreadUserInputLifecycleIndex.ts @@ -0,0 +1,11 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projection_thread_activities_user_input_lifecycle + ON projection_thread_activities(thread_id, turn_id, kind, sequence, created_at, activity_id) + `; +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreadActivities.ts b/apps/server/src/persistence/Services/ProjectionThreadActivities.ts index 802ca1a96efe5..0b33c1586753b 100644 --- a/apps/server/src/persistence/Services/ProjectionThreadActivities.ts +++ b/apps/server/src/persistence/Services/ProjectionThreadActivities.ts @@ -41,6 +41,13 @@ export const ListProjectionThreadActivitiesInput = Schema.Struct({ }); export type ListProjectionThreadActivitiesInput = typeof ListProjectionThreadActivitiesInput.Type; +export const ListProjectionThreadUserInputLifecycleInput = Schema.Struct({ + threadId: ThreadId, + turnId: TurnId, +}); +export type ListProjectionThreadUserInputLifecycleInput = + typeof ListProjectionThreadUserInputLifecycleInput.Type; + export const GetLatestProjectionThreadTaskActivityInput = Schema.Struct({ threadId: ThreadId, taskId: Schema.String, @@ -86,6 +93,11 @@ export interface ProjectionThreadActivityRepositoryShape { input: ListProjectionThreadActivitiesInput, ) => Effect.Effect, ProjectionRepositoryError>; + /** List user-input lifecycle rows for one terminal turn. */ + readonly listUserInputLifecycleByThreadAndTurn: ( + input: ListProjectionThreadUserInputLifecycleInput, + ) => Effect.Effect, ProjectionRepositoryError>; + /** * Read the latest task-start or task-progress activity with a usable title. */ diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 3059994e437e7..921bcad2aecdc 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -1,19 +1,22 @@ import * as Option from "effect/Option"; import * as Arr from "effect/Array"; import * as Schema from "effect/Schema"; +import { + derivePendingApprovals, + derivePendingUserInputs, + requestKindFromRequestType, + type PendingApproval, + type PendingUserInput, +} from "@t3tools/client-runtime/pending-requests"; import { isBackgroundTaskActivity } from "@t3tools/client-runtime/state/subagentRuntime"; import { AkeruPluginSearchResult, - ApprovalRequestId, isToolLifecycleItemType, type OrchestrationLatestTurn, type OrchestrationThreadActivity, type OrchestrationProposedPlanId, ProviderDriverKind, - ProviderApprovalOption, - ProviderRequestKind, type ToolLifecycleItemType, - type UserInputQuestion, type ThreadId, type TurnId, } from "@t3tools/contracts"; @@ -125,25 +128,8 @@ const derivedWorkLogEntryByActivity = new WeakMap< >(); const isPluginSearchResult = Schema.is(AkeruPluginSearchResult); -export interface PendingApproval { - requestId: ApprovalRequestId; - requestKind: ProviderRequestKind; - createdAt: string; - detail?: string; - appName?: string; - options?: ReadonlyArray; - toolName?: string; - args?: unknown; -} - -const isProviderRequestKind = Schema.is(ProviderRequestKind); -const isProviderApprovalOption = Schema.is(ProviderApprovalOption); - -export interface PendingUserInput { - requestId: ApprovalRequestId; - createdAt: string; - questions: ReadonlyArray; -} +export type { PendingApproval, PendingUserInput }; +export { derivePendingApprovals, derivePendingUserInputs }; export interface ActivePlanState { createdAt: string; @@ -398,211 +384,6 @@ export function deriveActiveWorkStartedAt( return sendStartedAt; } -function requestKindFromRequestType(requestType: unknown): PendingApproval["requestKind"] | null { - switch (requestType) { - case "command_execution_approval": - case "exec_command_approval": - case "dynamic_tool_call": - return "command"; - case "file_read_approval": - return "file-read"; - case "file_change_approval": - case "apply_patch_approval": - return "file-change"; - case "mcp_elicitation_approval": - return "mcp-elicitation"; - default: - return null; - } -} - -function isStalePendingRequestFailureDetail(detail: string | undefined): boolean { - const normalized = detail?.toLowerCase(); - if (!normalized) { - return false; - } - return ( - normalized.includes("stale pending approval request") || - normalized.includes("stale pending user-input request") || - normalized.includes("unknown pending approval request") || - normalized.includes("unknown pending permission request") || - normalized.includes("unknown pending user-input request") || - normalized.includes("unknown pending user input request") || - normalized.includes("unknown pending codex user input request") - ); -} - -export function derivePendingApprovals( - activities: ReadonlyArray, -): PendingApproval[] { - const openByRequestId = new Map(); - const toolArgsByCallId = new Map(); - const ordered = [...activities].toSorted(compareActivitiesByOrder); - - for (const activity of ordered) { - const payload = - activity.payload && typeof activity.payload === "object" - ? (activity.payload as Record) - : null; - const toolCallId = - payload && typeof payload.toolCallId === "string" ? payload.toolCallId : null; - const data = - payload?.data && typeof payload.data === "object" - ? (payload.data as Record) - : null; - if (activity.kind === "tool.started" && toolCallId && data) { - const toolArgs = - data.args ?? (typeof data.command === "string" ? { command: data.command } : undefined); - if (toolArgs !== undefined) toolArgsByCallId.set(toolCallId, toolArgs); - } - const requestId = - payload && typeof payload.requestId === "string" - ? ApprovalRequestId.make(payload.requestId) - : null; - const requestKind = - payload && isProviderRequestKind(payload.requestKind) - ? payload.requestKind - : payload - ? requestKindFromRequestType(payload.requestType) - : null; - const detail = payload && typeof payload.detail === "string" ? payload.detail : undefined; - const appName = payload && typeof payload.appName === "string" ? payload.appName : undefined; - const toolName = payload && typeof payload.toolName === "string" ? payload.toolName : undefined; - const args = payload?.args ?? (requestId ? toolArgsByCallId.get(requestId) : undefined); - const options = Array.isArray(payload?.options) - ? payload.options.filter(isProviderApprovalOption) - : undefined; - - if (activity.kind === "approval.requested" && requestId) { - openByRequestId.set(requestId, { - requestId, - requestKind: requestKind ?? "command", - createdAt: activity.createdAt, - ...(detail ? { detail } : {}), - ...(appName ? { appName } : {}), - ...(toolName ? { toolName } : {}), - ...(args !== undefined ? { args } : {}), - ...(options && options.length > 0 ? { options } : {}), - }); - continue; - } - - if (activity.kind === "approval.resolved" && requestId) { - openByRequestId.delete(requestId); - continue; - } - - if ( - activity.kind === "provider.approval.respond.failed" && - requestId && - isStalePendingRequestFailureDetail(detail) - ) { - openByRequestId.delete(requestId); - continue; - } - } - - return [...openByRequestId.values()].toSorted((left, right) => - left.createdAt.localeCompare(right.createdAt), - ); -} - -function parseUserInputQuestions( - payload: Record | null, -): ReadonlyArray | null { - const questions = payload?.questions; - if (!Array.isArray(questions)) { - return null; - } - const parsed = questions - .map((entry) => { - if (!entry || typeof entry !== "object") return null; - const question = entry as Record; - if ( - typeof question.id !== "string" || - typeof question.header !== "string" || - typeof question.question !== "string" || - !Array.isArray(question.options) - ) { - return null; - } - const options = question.options - .map((option) => { - if (!option || typeof option !== "object") return null; - const optionRecord = option as Record; - if ( - typeof optionRecord.label !== "string" || - typeof optionRecord.description !== "string" - ) { - return null; - } - return { - label: optionRecord.label, - description: optionRecord.description, - }; - }) - .filter((option): option is UserInputQuestion["options"][number] => option !== null); - return { - id: question.id, - header: question.header, - question: question.question, - options, - multiSelect: question.multiSelect === true, - }; - }) - .filter((question): question is UserInputQuestion => question !== null); - return parsed.length > 0 ? parsed : null; -} - -export function derivePendingUserInputs( - activities: ReadonlyArray, -): PendingUserInput[] { - const openByRequestId = new Map(); - const ordered = [...activities].toSorted(compareActivitiesByOrder); - - for (const activity of ordered) { - const payload = - activity.payload && typeof activity.payload === "object" - ? (activity.payload as Record) - : null; - const requestId = - payload && typeof payload.requestId === "string" - ? ApprovalRequestId.make(payload.requestId) - : null; - const detail = payload && typeof payload.detail === "string" ? payload.detail : undefined; - - if (activity.kind === "user-input.requested" && requestId) { - const questions = parseUserInputQuestions(payload); - if (!questions) { - continue; - } - openByRequestId.set(requestId, { - requestId, - createdAt: activity.createdAt, - questions, - }); - continue; - } - - if (activity.kind === "user-input.resolved" && requestId) { - openByRequestId.delete(requestId); - continue; - } - - if ( - activity.kind === "provider.user-input.respond.failed" && - requestId && - isStalePendingRequestFailureDetail(detail) - ) { - openByRequestId.delete(requestId); - } - } - - return [...openByRequestId.values()].toSorted((left, right) => - left.createdAt.localeCompare(right.createdAt), - ); -} - function planStateFromActivity(activity: OrchestrationThreadActivity): ActivePlanState | null { const payload = activity.payload && typeof activity.payload === "object" diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index 47a074c73d4b2..895d8f2455092 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -19,6 +19,10 @@ "types": "./src/replyPlayback/index.ts", "default": "./src/replyPlayback/index.ts" }, + "./pending-requests": { + "types": "./src/pendingRequests.ts", + "default": "./src/pendingRequests.ts" + }, "./markdown-images": { "types": "./src/markdownImages.ts", "default": "./src/markdownImages.ts" diff --git a/packages/client-runtime/src/pendingRequests.test.ts b/packages/client-runtime/src/pendingRequests.test.ts new file mode 100644 index 0000000000000..768ca776d059a --- /dev/null +++ b/packages/client-runtime/src/pendingRequests.test.ts @@ -0,0 +1,260 @@ +import { EventId, TurnId, type OrchestrationThreadActivity } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { + derivePendingApprovals, + derivePendingRequests, + derivePendingUserInputs, +} from "./pendingRequests.ts"; + +let nextActivityId = 0; + +function makeActivity(overrides: { + id?: string; + createdAt?: string; + kind?: string; + summary?: string; + tone?: OrchestrationThreadActivity["tone"]; + payload?: Record; + turnId?: string; + sequence?: number; +}): OrchestrationThreadActivity { + return { + id: EventId.make(overrides.id ?? `activity-${nextActivityId++}`), + createdAt: overrides.createdAt ?? "2026-02-23T00:00:00.000Z", + kind: overrides.kind ?? "tool.started", + summary: overrides.summary ?? "Tool call", + tone: overrides.tone ?? "tool", + payload: overrides.payload ?? {}, + turnId: overrides.turnId ? TurnId.make(overrides.turnId) : null, + ...(overrides.sequence !== undefined ? { sequence: overrides.sequence } : {}), + }; +} + +describe("pending approvals", () => { + it("carries command arguments from the matching tool activity", () => { + const args = { command: 'printf "hi\\n"', cwd: null }; + const approvals = derivePendingApprovals([ + makeActivity({ + kind: "tool.started", + payload: { toolCallId: "shell-1", data: { args } }, + }), + makeActivity({ + kind: "approval.requested", + payload: { + requestId: "shell-1", + requestKind: "command", + toolName: "Shell", + }, + }), + ]); + + expect(approvals).toEqual([ + { + requestId: "shell-1", + requestKind: "command", + createdAt: "2026-02-23T00:00:00.000Z", + toolName: "Shell", + args, + }, + ]); + }); + + it("carries command arguments when approval arrives before its tool activity", () => { + const args = { command: "git status", cwd: "/workspace" }; + const approvals = derivePendingApprovals([ + makeActivity({ + kind: "approval.requested", + payload: { + requestId: "shell-reversed", + requestKind: "command", + toolName: "Shell", + }, + }), + makeActivity({ + kind: "tool.started", + payload: { toolCallId: "shell-reversed", data: { args } }, + }), + ]); + + expect(approvals[0]?.args).toEqual(args); + }); + + it("maps dynamic_tool_call approvals to the command kind", () => { + expect( + derivePendingApprovals([ + makeActivity({ + kind: "approval.requested", + payload: { + requestId: "req-dynamic-tool", + requestType: "dynamic_tool_call", + detail: "Search the web", + }, + }), + ]), + ).toEqual([ + { + requestId: "req-dynamic-tool", + requestKind: "command", + createdAt: "2026-02-23T00:00:00.000Z", + detail: "Search the web", + }, + ]); + }); + + it.each(["tool_user_input", "auth_tokens_refresh"])( + "does not turn %s into an approval", + (requestType) => { + expect( + derivePendingApprovals([ + makeActivity({ + kind: "approval.requested", + payload: { requestId: "not-an-approval", requestType }, + }), + ]), + ).toEqual([]); + }, + ); + + it("keeps a resolved approval closed when a later request activity arrives", () => { + const activities = [ + makeActivity({ + createdAt: "2026-02-23T00:00:02.000Z", + kind: "approval.resolved", + payload: { requestId: "req-closed" }, + }), + makeActivity({ + createdAt: "2026-02-23T00:00:01.000Z", + kind: "approval.requested", + payload: { requestId: "req-closed", requestKind: "command" }, + }), + ]; + expect(derivePendingRequests(activities).approvals).toEqual([]); + }); + + it("clears stale Codex approval callbacks", () => { + expect( + derivePendingApprovals([ + makeActivity({ + kind: "approval.requested", + payload: { requestId: "req-stale-1", requestKind: "command" }, + }), + makeActivity({ + kind: "provider.approval.respond.failed", + payload: { + requestId: "req-stale-1", + detail: "Unknown pending Codex approval request: req-stale-1", + }, + }), + ]), + ).toEqual([]); + }); +}); + +describe("pending questions", () => { + it("keeps free-text questions that do not have options", () => { + expect( + derivePendingUserInputs([ + makeActivity({ + kind: "user-input.requested", + payload: { + requestId: "req-free-text", + questions: [ + { + id: "answer", + header: "Question", + question: "What should the bot do next?", + options: [], + }, + ], + }, + }), + ]), + ).toMatchObject([ + { + requestId: "req-free-text", + questions: [{ id: "answer", question: "What should the bot do next?", options: [] }], + }, + ]); + }); + + it("keeps a resolved question closed when a later request activity arrives", () => { + expect( + derivePendingUserInputs([ + makeActivity({ + createdAt: "2026-02-23T00:00:02.000Z", + kind: "user-input.resolved", + payload: { requestId: "req-closed" }, + }), + makeActivity({ + createdAt: "2026-02-23T00:00:01.000Z", + kind: "user-input.requested", + payload: { + requestId: "req-closed", + questions: [ + { + id: "q", + header: "Q", + question: "Choose", + options: [{ label: "yes", description: "Yes" }], + }, + ], + }, + }), + ]), + ).toEqual([]); + }); + + it("clears stale pending user-input prompts including Codex callback errors", () => { + expect( + derivePendingUserInputs([ + makeActivity({ + kind: "user-input.requested", + payload: { + requestId: "req-user-input-stale-1", + questions: [ + { + id: "sandbox_mode", + header: "Sandbox", + question: "Which mode should be used?", + options: [{ label: "workspace-write", description: "Allow workspace writes only" }], + }, + ], + }, + }), + makeActivity({ + kind: "provider.user-input.respond.failed", + payload: { + requestId: "req-user-input-stale-1", + detail: + "Provider adapter request failed (codex) for item/tool/requestUserInput: Unknown pending Codex user input request: req-user-input-stale-1", + }, + }), + ]), + ).toEqual([]); + }); + + it("preserves native question ids and option labels without trimming", () => { + const question = { + id: " Which path?\n", + header: " Path ", + question: " Which path?\n", + options: [{ label: " Keep spaces ", description: "Keep" }], + multiSelect: false, + }; + expect( + derivePendingUserInputs([ + makeActivity({ + kind: "user-input.requested", + payload: { + requestId: "native-question", + questions: [ + question, + { id: "bad", header: "bad", question: "bad", options: [{ label: 42 }] }, + ], + }, + }), + ])[0]?.questions, + ).toEqual([question]); + }); +}); diff --git a/packages/client-runtime/src/pendingRequests.ts b/packages/client-runtime/src/pendingRequests.ts new file mode 100644 index 0000000000000..645dde594fc93 --- /dev/null +++ b/packages/client-runtime/src/pendingRequests.ts @@ -0,0 +1,218 @@ +import { + ApprovalRequestId, + type OrchestrationThreadActivity, + ProviderApprovalOption, + ProviderRequestKind, + UserInputQuestion, +} from "@t3tools/contracts"; +import * as Predicate from "effect/Predicate"; +import * as Schema from "effect/Schema"; + +export interface PendingApproval { + readonly requestId: ApprovalRequestId; + readonly requestKind: ProviderRequestKind; + readonly createdAt: string; + readonly detail?: string; + readonly appName?: string; + readonly options?: ReadonlyArray; + readonly toolName?: string; + readonly args?: unknown; +} + +export interface PendingUserInput { + readonly requestId: ApprovalRequestId; + readonly createdAt: string; + readonly questions: ReadonlyArray; +} + +const isRequestId = Schema.is(ApprovalRequestId); +const isProviderRequestKind = Schema.is(ProviderRequestKind); +const isProviderApprovalOption = Schema.is(ProviderApprovalOption); + +/** Older activities use native request types instead of a request kind. */ +export function requestKindFromRequestType(requestType: unknown): ProviderRequestKind | null { + switch (requestType) { + case "command_execution_approval": + case "exec_command_approval": + case "dynamic_tool_call": + return "command"; + case "file_read_approval": + return "file-read"; + case "file_change_approval": + case "apply_patch_approval": + return "file-change"; + case "mcp_elicitation_approval": + return "mcp-elicitation"; + default: + return null; + } +} + +function parseQuestions(value: unknown): UserInputQuestion[] { + if (!Array.isArray(value)) return []; + const parsed: UserInputQuestion[] = []; + for (const question of value) { + if (!Predicate.isObject(question) || !Array.isArray(question.options)) continue; + if ( + typeof question.id !== "string" || + typeof question.header !== "string" || + typeof question.question !== "string" + ) { + continue; + } + const options = question.options.flatMap((option) => { + if (!Predicate.isObject(option)) return []; + if (typeof option.label !== "string" || typeof option.description !== "string") return []; + return [{ label: option.label, description: option.description }]; + }); + if (question.options.length > 0 && options.length === 0) continue; + parsed.push({ + id: question.id, + header: question.header, + question: question.question, + options, + multiSelect: question.multiSelect === true, + }); + } + return parsed; +} + +const requestActivityKinds = new Set([ + "approval.requested", + "approval.resolved", + "provider.approval.respond.failed", + "user-input.requested", + "user-input.resolved", + "provider.user-input.respond.failed", + "tool.started", +]); + +const staleRequestFailureDetails = { + "provider.approval.respond.failed": [ + "stale pending approval request", + "unknown pending approval request", + "unknown pending permission request", + "unknown pending codex approval request", + ], + "provider.user-input.respond.failed": [ + "stale pending user-input request", + "unknown pending user-input request", + "unknown pending user input request", + "unknown pending codex user input request", + ], +} as const; + +function isStaleRequestFailure( + kind: keyof typeof staleRequestFailureDetails, + payload: Record, +): boolean { + const detail = typeof payload.detail === "string" ? payload.detail.toLowerCase() : ""; + return staleRequestFailureDetails[kind].some((fragment) => detail.includes(fragment)); +} + +/** Reduces request state once for web, desktop, and mobile. Layout stays with each client. */ +export function derivePendingRequests(activities: ReadonlyArray) { + const approvals = new Map(); + const userInputs = new Map(); + const closedApprovals = new Set(); + const closedUserInputs = new Set(); + const toolArgsByCallId = new Map(); + + // Tool and approval activities can arrive in either order. Index command + // details first so an earlier approval still receives its matching args. + for (const activity of activities) { + if (activity.kind !== "tool.started") continue; + const payload = Predicate.isObject(activity.payload) ? activity.payload : undefined; + if (!payload) continue; + const toolCallId = typeof payload.toolCallId === "string" ? payload.toolCallId : null; + const data = Predicate.isObject(payload.data) ? payload.data : null; + if (!toolCallId || !data) continue; + const toolArgs = + data.args ?? (typeof data.command === "string" ? { command: data.command } : undefined); + if (toolArgs !== undefined) toolArgsByCallId.set(toolCallId, toolArgs); + } + + // Request IDs are unique. A terminal event stays final even when provider + // sequences and server-generated activities arrive in a different order. + for (const activity of activities) { + if (!requestActivityKinds.has(activity.kind)) continue; + const payload = Predicate.isObject(activity.payload) ? activity.payload : undefined; + if (!payload) continue; + + if (activity.kind === "tool.started") continue; + + if (!isRequestId(payload.requestId)) continue; + const requestId = payload.requestId; + + if (activity.kind === "approval.requested") { + if ( + closedApprovals.has(requestId) || + payload.requestType === "tool_user_input" || + payload.requestType === "auth_tokens_refresh" + ) { + continue; + } + const requestKind = isProviderRequestKind(payload.requestKind) + ? payload.requestKind + : requestKindFromRequestType(payload.requestType); + const options = Array.isArray(payload.options) + ? payload.options.filter(isProviderApprovalOption) + : []; + const args = payload.args ?? toolArgsByCallId.get(requestId); + approvals.set(requestId, { + requestId, + requestKind: requestKind ?? "command", + createdAt: activity.createdAt, + ...(typeof payload.detail === "string" && payload.detail ? { detail: payload.detail } : {}), + ...(typeof payload.appName === "string" && payload.appName + ? { appName: payload.appName } + : {}), + ...(typeof payload.toolName === "string" && payload.toolName + ? { toolName: payload.toolName } + : {}), + ...(args !== undefined ? { args } : {}), + ...(options.length > 0 ? { options } : {}), + }); + } else if (activity.kind === "user-input.requested") { + if (closedUserInputs.has(requestId)) continue; + const questions = parseQuestions(payload.questions); + if (questions.length === 0) continue; + userInputs.set(requestId, { requestId, createdAt: activity.createdAt, questions }); + } else if ( + activity.kind === "approval.resolved" || + (activity.kind === "provider.approval.respond.failed" && + isStaleRequestFailure(activity.kind, payload)) + ) { + closedApprovals.add(requestId); + approvals.delete(requestId); + } else if ( + activity.kind === "user-input.resolved" || + (activity.kind === "provider.user-input.respond.failed" && + isStaleRequestFailure(activity.kind, payload)) + ) { + closedUserInputs.add(requestId); + userInputs.delete(requestId); + } + } + + const byCreatedAt = ( + left: { readonly createdAt: string }, + right: { readonly createdAt: string }, + ) => left.createdAt.localeCompare(right.createdAt); + return { + approvals: [...approvals.values()].sort(byCreatedAt), + userInputs: [...userInputs.values()].sort(byCreatedAt), + }; +} + +export function derivePendingApprovals( + activities: ReadonlyArray, +): PendingApproval[] { + return derivePendingRequests(activities).approvals; +} + +export function derivePendingUserInputs( + activities: ReadonlyArray, +): PendingUserInput[] { + return derivePendingRequests(activities).userInputs; +}