diff --git a/apps/mobile/src/lib/threadActivity.test.ts b/apps/mobile/src/lib/threadActivity.test.ts index 7a4a0325a92c..0ffd0c03071f 100644 --- a/apps/mobile/src/lib/threadActivity.test.ts +++ b/apps/mobile/src/lib/threadActivity.test.ts @@ -88,6 +88,29 @@ const nativeQuestion = { } as const; describe("pending user input answers", () => { + it("accepts free-text answers to async questions without options", () => { + const question = { + id: "0", + header: "Question", + question: "What should it be named?", + options: [], + allowCustomAnswer: true, + multiSelect: false, + }; + const requested = makeActivity({ + id: EventId.make("async-question"), + kind: "user-input.requested", + summary: "User input requested", + createdAt: "2026-09-03T00:00:00.000Z", + payload: { requestId: "async-1", responseMode: "message", questions: [question] }, + }); + const questions = derivePendingUserInputs([requested])[0]?.questions; + expect(questions).toEqual([question]); + expect(buildPendingUserInputAnswers(questions!, { "0": { customAnswer: "Example" } })).toEqual({ + "0": "Example", + }); + }); + it("preserves native choice values and custom-answer rules from activities", () => { const requested = makeActivity({ id: EventId.make("native-question"), diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index 34389d46d302..796416d80200 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -245,7 +245,7 @@ function parseUserInputQuestions( }; }) .filter((option): option is UserInputQuestion["options"][number] => option !== null); - if (options.length === 0) { + if (options.length === 0 && question.allowCustomAnswer === false) { return null; } return { diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts index 55b6dd973494..cd4309057935 100644 --- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts @@ -75,6 +75,7 @@ describe("CheckpointDiffQuery.layer", () => { Layer.provideMerge(Layer.succeed(CheckpointStore.CheckpointStore, checkpointStore)), Layer.provideMerge( Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("CheckpointDiffQuery should not request the command read model"), getSnapshot: () => @@ -185,6 +186,7 @@ describe("CheckpointDiffQuery.layer", () => { Layer.provideMerge(Layer.succeed(CheckpointStore.CheckpointStore, checkpointStore)), Layer.provideMerge( Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("CheckpointDiffQuery should not request the command read model"), getSnapshot: () => @@ -270,6 +272,7 @@ describe("CheckpointDiffQuery.layer", () => { Layer.provideMerge(Layer.succeed(CheckpointStore.CheckpointStore, checkpointStore)), Layer.provideMerge( Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("CheckpointDiffQuery should not request the command read model"), getSnapshot: () => @@ -340,6 +343,7 @@ describe("CheckpointDiffQuery.layer", () => { Layer.provideMerge(Layer.succeed(CheckpointStore.CheckpointStore, checkpointStore)), Layer.provideMerge( Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("CheckpointDiffQuery should not request the command read model"), getSnapshot: () => @@ -395,6 +399,7 @@ describe("CheckpointDiffQuery.layer", () => { Layer.provideMerge(Layer.succeed(CheckpointStore.CheckpointStore, checkpointStore)), Layer.provideMerge( Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("CheckpointDiffQuery should not request the command read model"), getSnapshot: () => diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index a4c7942cfc18..d3d766084a28 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -1,4 +1,11 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeFSP from "node:fs/promises"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; + import { + ApprovalRequestId, + EventId, CheckpointRef, CommandId, DEFAULT_PROVIDER_INTERACTION_MODE, @@ -25,7 +32,10 @@ import { PersistenceSqlError } from "../../persistence/Errors.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; import * as OrchestrationCommandReceipts from "../../persistence/Services/OrchestrationCommandReceipts.ts"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; -import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; +import { + makeSqlitePersistenceLive, + SqlitePersistenceMemory, +} from "../../persistence/Layers/Sqlite.ts"; import { OrchestrationEventStore, type OrchestrationEventStoreShape, @@ -49,7 +59,10 @@ const asMessageId = (value: string): MessageId => MessageId.make(value); const asTurnId = (value: string): TurnId => TurnId.make(value); const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.make(value); -function makeOrchestrationLayer() { +function makeOrchestrationLayer(databasePath?: string) { + const persistence = databasePath + ? makeSqlitePersistenceLive(databasePath) + : SqlitePersistenceMemory; const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { prefix: "t3-orchestration-engine-test-", }); @@ -65,19 +78,21 @@ function makeOrchestrationLayer() { Layer.provide(OrchestrationEventStoreLive), Layer.provideMerge(OrchestrationCommandReceiptRepositoryLive), Layer.provide(RepositoryIdentityResolver.layer), - Layer.provide(SqlitePersistenceMemory), + Layer.provide(persistence), Layer.provideMerge(ServerConfigLayer), Layer.provideMerge(NodeServices.layer), ); } -async function createOrchestrationSystem() { - const runtime = ManagedRuntime.make(makeOrchestrationLayer()); +async function createOrchestrationSystem(databasePath?: string) { + const runtime = ManagedRuntime.make(makeOrchestrationLayer(databasePath)); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery)); return { engine, readModel: () => runtime.runPromise(snapshotQuery.getSnapshot()), + readThread: (threadId: ThreadId) => + runtime.runPromise(snapshotQuery.getThreadDetailById(threadId)), run: (effect: Effect.Effect) => runtime.runPromise(effect), dispose: () => runtime.dispose(), }; @@ -99,6 +114,196 @@ const hasMetricSnapshot = ( ); describe("OrchestrationEngine", () => { + it.each(["running", "stopped"] as const)( + "sends async answers with a %s session and rejects old duplicate replies", + async (status) => { + const directory = await NodeFSP.mkdtemp( + NodePath.join(NodeOS.tmpdir(), "t3-async-questions-"), + ); + const databasePath = NodePath.join(directory, "state.sqlite"); + let system = await createOrchestrationSystem(databasePath); + const threadId = ThreadId.make("async-thread"); + const projectId = ProjectId.make("async-project"); + const requestId = ApprovalRequestId.make("codex-async:question-1"); + try { + await system.run( + system.engine.dispatch({ + type: "project.create", + commandId: CommandId.make("async-project"), + projectId, + title: "Async questions", + workspaceRoot: "/tmp/async-questions", + createdAt: now(), + }), + ); + await system.run( + system.engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("async-thread"), + threadId, + projectId, + title: "Async questions", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: now(), + }), + ); + await system.run( + system.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("async-session"), + threadId, + createdAt: now(), + session: { + threadId, + status, + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: status === "running" ? TurnId.make("turn-1") : null, + lastError: null, + updatedAt: now(), + }, + }), + ); + await system.run( + system.engine.dispatch({ + type: "thread.activity.append", + commandId: CommandId.make("async-question"), + threadId, + createdAt: now(), + activity: { + id: EventId.make("async-question"), + kind: "user-input.requested", + summary: "User input requested", + tone: "info", + turnId: TurnId.make("turn-1"), + createdAt: now(), + payload: { + requestId, + responseMode: "message", + questions: [ + { + id: "0", + header: "Question", + question: "Which package manager?", + options: [{ label: "pnpm", description: "" }], + }, + { + id: "1", + header: "Question", + question: "What should it be named?", + options: [], + }, + ], + }, + }, + }), + ); + const appendWork = async (prefix: string, createdAt: string) => { + for (let index = 0; index < 501; index += 1) { + await system.run( + system.engine.dispatch({ + type: "thread.activity.append", + commandId: CommandId.make(`${prefix}-${index}`), + threadId, + createdAt, + activity: { + id: EventId.make(`${prefix}-${index}`), + kind: "tool.completed", + summary: "Work continued", + payload: {}, + tone: "info", + turnId: TurnId.make("turn-1"), + createdAt, + }, + }), + ); + } + }; + await appendWork("work", "2026-01-01T00:00:01.000Z"); + const before = await system.readModel(); + expect( + before.threads[0]?.activities.some((activity) => activity.id === "async-question"), + ).toBe(true); + if (status === "stopped") { + await system.dispose(); + system = await createOrchestrationSystem(databasePath); + } + const response = { + type: "thread.user-input.respond" as const, + commandId: CommandId.make("async-response"), + threadId, + requestId, + answers: { "0": "pnpm", "1": "Example" }, + createdAt: "2026-01-01T00:00:02.000Z", + }; + await expect( + system.run( + system.engine.dispatch({ + ...response, + commandId: CommandId.make("incomplete-answer"), + answers: { "0": "pnpm" }, + }), + ), + ).rejects.toThrow("Answer each question before sending."); + await system.run(system.engine.dispatch(response)); + const after = await system.readModel(); + const userMessages = after.threads[0]?.messages.filter( + (message) => message.role === "user", + ); + expect(userMessages).toHaveLength(1); + expect(userMessages?.[0]?.text).toBe( + "Which package manager?\npnpm\n\nWhat should it be named?\nExample", + ); + expect( + after.threads[0]?.activities.find((activity) => activity.kind === "user-input.resolved") + ?.payload, + ).toMatchObject({ requestId, responseMode: "message", answers: response.answers }); + const events = await system.run(Stream.runCollect(system.engine.readEvents(0))); + expect( + Array.from(events) + .filter((event) => event.commandId === response.commandId) + .map((event) => event.type), + ).toEqual([ + "thread.activity-appended", + "thread.message-sent", + "thread.turn-start-requested", + ]); + await expect( + system.run( + system.engine.dispatch({ + ...response, + commandId: CommandId.make("second-client-reply"), + }), + ), + ).rejects.toThrow("This question has already been answered."); + await appendWork("later-work", "2026-01-01T00:00:03.000Z"); + const afterEviction = Option.getOrThrow(await system.readThread(threadId)); + expect( + afterEviction.activities.some((activity) => activity.kind === "user-input.resolved"), + ).toBe(false); + if (status === "stopped") { + await system.dispose(); + system = await createOrchestrationSystem(databasePath); + } + await expect( + system.run( + system.engine.dispatch({ + ...response, + commandId: CommandId.make("reply-after-eviction"), + }), + ), + ).rejects.toThrow("This question has already been answered."); + } finally { + await system.dispose(); + await NodeFSP.rm(directory, { recursive: true, force: true }); + } + }, + ); + it("bootstraps command handling from persisted projections without reading the full snapshot", async () => { let nextSequence = 8; const eventStore: OrchestrationEventStoreShape = { @@ -183,6 +388,7 @@ describe("OrchestrationEngine", () => { const layer = OrchestrationEngineLive.pipe( Layer.provide( Layer.succeed(ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.succeed(commandReadModel), getSnapshot: () => Effect.sync(() => { diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 741e0fac7195..792dd0da3900 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -195,9 +195,18 @@ const makeOrchestrationEngine = Effect.gen(function* () { }); } + // Command snapshots omit activities at startup and cap them while running. + // Read this request's durable state before deciding how to send the answer. + const userInputActivity = + envelope.command.type === "thread.user-input.respond" + ? yield* projectionSnapshotQuery.getUserInputActivity(envelope.command) + : Option.none(); const eventBase = yield* decideOrchestrationCommand({ command: envelope.command, readModel: commandReadModel, + ...(Option.isSome(userInputActivity) + ? { userInputActivity: userInputActivity.value } + : {}), }).pipe( Effect.provideService(Crypto.Crypto, crypto), Effect.mapError((cause) => diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 24c7003204eb..0d064a1d6e9c 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -1,4 +1,5 @@ import { + ApprovalRequestId, ChatAttachment, CheckpointRef, IsoDateTime, @@ -1119,6 +1120,40 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const getUserInputActivityRow = SqlSchema.findOneOption({ + Request: Schema.Struct({ threadId: ThreadId, requestId: ApprovalRequestId }), + Result: ProjectionThreadActivityDbRowSchema, + execute: ({ threadId, requestId }) => 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 kind IN ('user-input.requested', 'user-input.resolved') + AND json_extract(payload_json, '$.requestId') = ${requestId} + ORDER BY sequence DESC, created_at DESC, activity_id DESC + LIMIT 1 + `, + }); + + const getUserInputActivity: ProjectionSnapshotQueryShape["getUserInputActivity"] = (input) => + getUserInputActivityRow(input).pipe( + Effect.map(Option.map(mapThreadActivityRow)), + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getUserInputActivity:query", + "ProjectionSnapshotQuery.getUserInputActivity:decodeRow", + ), + ), + ); + const listThreadActivityIdsByThread = SqlSchema.findAll({ Request: ThreadIdLookupInput, Result: ProjectionThreadActivityIdRowSchema, @@ -3151,6 +3186,7 @@ pending_approval_requests AS ( return { getCommandReadModel, + getUserInputActivity, getSnapshot, getShellSnapshot, getArchivedShellSnapshot, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 0d8c4f874909..b490d44726c0 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -2108,6 +2108,56 @@ describe("ProviderRuntimeIngestion", () => { expect(message?.streaming).toBe(false); }); + it("keeps streaming while an async question is pending", async () => { + const harness = await createHarness({ serverSettings: { enableLegacyTokenStreaming: true } }); + const base = { + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-async"), + }; + harness.emit({ ...base, type: "turn.started", eventId: asEventId("async-start") }); + harness.emit({ + ...base, + type: "content.delta", + eventId: asEventId("async-before"), + itemId: asItemId("message-1"), + payload: { streamKind: "assistant_text", delta: "Before. " }, + }); + harness.emit({ + ...base, + type: "user-input.requested", + eventId: asEventId("async-request"), + requestId: ApprovalRequestId.make("codex-async:question-1"), + payload: { + responseMode: "message", + questions: [ + { + id: "0", + header: "Question", + question: "Which name?", + options: [], + allowCustomAnswer: true, + }, + ], + }, + }); + harness.emit({ + ...base, + type: "content.delta", + eventId: asEventId("async-after"), + itemId: asItemId("message-1"), + payload: { streamKind: "assistant_text", delta: "After." }, + }); + await harness.drain(); + const thread = (await harness.readModel()).threads[0]; + expect(thread?.session?.status).toBe("running"); + expect(thread?.messages).toMatchObject([{ text: "Before. After.", streaming: true }]); + expect( + thread?.activities.find((activity) => activity.kind === "user-input.requested")?.payload, + ).toMatchObject({ responseMode: "message", requestId: "codex-async:question-1" }); + }); + it("does not create assistant segments for whitespace-only buffered text at approval boundaries", async () => { const harness = await createHarness(); const startedAt = "2026-03-28T06:28:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index d1a64544ccac..a503a5eebbcf 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -520,6 +520,7 @@ export function runtimeEventToActivities( payload: { ...(event.requestId ? { requestId: event.requestId } : {}), questions: event.payload.questions, + ...(event.payload.responseMode ? { responseMode: event.payload.responseMode } : {}), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -1739,7 +1740,8 @@ const make = Effect.gen(function* () { } const pauseForUserTurnId = - event.type === "request.opened" || event.type === "user-input.requested" + event.type === "request.opened" || + (event.type === "user-input.requested" && event.payload.responseMode !== "message") ? toTurnId(event.turnId) : undefined; if (pauseForUserTurnId) { diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index 5ad9ceef1d2a..0854492e255b 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -7,6 +7,7 @@ * @module ProjectionSnapshotQuery */ import type { + ApprovalRequestId, CheckpointRef, OrchestrationCheckpointSummary, OrchestrationProject, @@ -16,6 +17,7 @@ import type { OrchestrationSearchThreadsResult, OrchestrationShellSnapshot, OrchestrationThread, + OrchestrationThreadActivity, OrchestrationThreadDetailSnapshot, OrchestrationThreadDetailWindow, OrchestrationThreadShell, @@ -72,6 +74,12 @@ export interface ProjectionThreadDetailQuery { * ProjectionSnapshotQueryShape - Service API for read-model snapshots. */ export interface ProjectionSnapshotQueryShape { + /** Read the latest request or resolution without loading the thread history. */ + readonly getUserInputActivity: (input: { + readonly threadId: ThreadId; + readonly requestId: ApprovalRequestId; + }) => Effect.Effect, ProjectionRepositoryError>; + /** * Read the lightweight command snapshot used to bootstrap the in-memory * orchestration engine without hydrating message/activity/checkpoint bodies. diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 279413f669c1..b336053ac9e1 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1,13 +1,19 @@ import { EventId, + MessageId, + UserInputRequestedPayload, type OrchestrationCommand, type OrchestrationEvent, type OrchestrationReadModel, type OrchestrationThread, + type OrchestrationThreadActivity, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import * as Option from "effect/Option"; +import * as Predicate from "effect/Predicate"; import type * as PlatformError from "effect/PlatformError"; import { @@ -29,6 +35,7 @@ import { projectEvent } from "./projector.ts"; import { threadHasQueuedTurnStart } from "./ThreadSettlementPolicy.ts"; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); +const decodeUserInputRequestedPayload = Schema.decodeUnknownOption(UserInputRequestedPayload); /** * Blocked-on-you work derived from the thread's retained activities: an @@ -55,12 +62,8 @@ function isStaleRequestFailureDetail(payload: Record | null): b } // Scans the read model's activities, which the projector caps at the most -// recent 500. That bound is safe here: an OPEN approval/user-input request -// blocks its turn, so the thread cannot accumulate hundreds of later -// activities while one is outstanding — a request that has scrolled out of -// the window is one whose turn kept running, i.e. it was resolved or went -// stale. (The projection pipeline's pendingApprovalCount reads the same -// capped stream and stays consistent with this view.) +// recent 500 plus pending async questions. Async questions remain actionable +// while the agent works, so they must not expire with the activity window. function hasOpenBlockingRequest(thread: { readonly activities: ReadonlyArray<{ readonly kind: string; readonly payload: unknown }>; }): boolean { @@ -185,9 +188,11 @@ const decideCommandSequence = Effect.fn("decideCommandSequence")(function* ({ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand")(function* ({ command, readModel, + userInputActivity, }: { readonly command: OrchestrationCommand; readonly readModel: OrchestrationReadModel; + readonly userInputActivity?: OrchestrationThreadActivity; }): Effect.fn.Return< DecideOrchestrationCommandResult, OrchestrationCommandRejection | PlatformError.PlatformError, @@ -1061,11 +1066,76 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" } case "thread.user-input.respond": { - yield* requireThread({ + const thread = yield* requireThread({ readModel, command, threadId: command.threadId, }); + const request = userInputActivity; + if ( + request && + Predicate.isObject(request.payload) && + request.payload.responseMode === "message" + ) { + const payload = decodeUserInputRequestedPayload(request.payload); + if (request.kind !== "user-input.requested" || Option.isNone(payload)) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "This question has already been answered.", + }); + } + const replies: string[] = []; + for (const question of payload.value.questions) { + const answer = command.answers[question.id]; + if (typeof answer !== "string" || answer.trim().length === 0) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "Answer each question before sending.", + }); + } + replies.push(`${question.question}\n${answer.trim()}`); + } + // Commit the answer and its message together. The normal turn path + // steers a running agent or resumes an idle session. + return yield* decideCommandSequence({ + readModel, + commands: [ + { + type: "thread.activity.append", + commandId: command.commandId, + threadId: command.threadId, + createdAt: command.createdAt, + activity: { + id: EventId.make(`async-answer:${command.requestId}`), + kind: "user-input.resolved", + summary: "User input submitted", + tone: "info", + turnId: request.turnId, + createdAt: command.createdAt, + payload: { + requestId: command.requestId, + responseMode: "message", + answers: command.answers, + }, + }, + }, + { + type: "thread.turn.start", + commandId: command.commandId, + threadId: command.threadId, + createdAt: command.createdAt, + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + message: { + messageId: MessageId.make(`async-answer:${command.requestId}`), + role: "user", + text: replies.join("\n\n"), + attachments: [], + }, + }, + ], + }); + } return { ...(yield* withEventBase({ aggregateKind: "thread", diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index fbef8e40b50b..3cea194bbb44 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -7,6 +7,7 @@ import { } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; +import * as Predicate from "effect/Predicate"; import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts"; import { @@ -39,6 +40,28 @@ type ThreadPatch = Partial>; const MAX_THREAD_MESSAGES = 2_000; const MAX_THREAD_CHECKPOINTS = 500; +// Async questions can stay open while the agent produces more activity. +// Match the database snapshot's pending-question retention. +function retainThreadActivities(activities: OrchestrationThread["activities"]) { + const recentStart = activities.length - 500; + if (recentStart <= 0) return activities; + const pending = new Map(); + for (const activity of activities) { + if (!Predicate.isObject(activity.payload)) continue; + const requestId = activity.payload.requestId; + if (typeof requestId !== "string") continue; + if (activity.kind === "user-input.requested" && activity.payload.responseMode === "message") { + pending.set(requestId, activity); + } else if (activity.kind === "user-input.resolved") { + pending.delete(requestId); + } + } + const pendingActivities = new Set(pending.values()); + return activities.filter( + (activity, index) => index >= recentStart || pendingActivities.has(activity), + ); +} + function checkpointStatusToLatestTurnState(status: "ready" | "missing" | "error") { if (status === "error") return "error" as const; if (status === "missing") return "interrupted" as const; @@ -804,12 +827,12 @@ export function projectEvent( return nextBase; } - const activities = [ - ...thread.activities.filter((entry) => entry.id !== payload.activity.id), - payload.activity, - ] - .toSorted(compareThreadActivities) - .slice(-500); + const activities = retainThreadActivities( + [ + ...thread.activities.filter((entry) => entry.id !== payload.activity.id), + payload.activity, + ].toSorted(compareThreadActivities), + ); return { ...nextBase, diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts index b0679a94cc2b..8065d90917f0 100644 --- a/apps/server/src/project/ProjectSetupScriptRunner.test.ts +++ b/apps/server/src/project/ProjectSetupScriptRunner.test.ts @@ -26,6 +26,7 @@ const makeProject = (scripts: OrchestrationProject["scripts"]): OrchestrationPro const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) => Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index bcaad3cb5e57..fbfc48c53827 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -1648,6 +1648,81 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }), ); + it.effect("maps async agent questions without ending the turn", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const eventsFiber = yield* Stream.runCollect(Stream.take(adapter.streamEvents, 2)).pipe( + Effect.forkChild, + ); + yield* runtime.emit({ + id: asEventId("evt-async-question"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "item/completed", + payload: { + completedAtMs: 0, + threadId: "thread-1", + turnId: "turn-1", + item: { + type: "agentMessage", + id: "async-question-1", + text: "Which package manager?\n- pnpm\n- npm\n\nWhat should it be named?", + phase: "final_answer", + delivery: "async", + questions: [ + { title: "Which package manager?", options: ["pnpm", "npm"] }, + { title: "What should it be named?" }, + ], + }, + }, + }); + yield* runtime.emit({ + id: asEventId("evt-async-continued"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:01.000Z", + method: "item/agentMessage/delta", + payload: { + threadId: "thread-1", + turnId: "turn-1", + itemId: "message-2", + delta: "I will keep working.", + }, + }); + const events = Array.from(yield* Fiber.join(eventsFiber)); + NodeAssert.equal(events[0]?.type, "user-input.requested"); + NodeAssert.equal(events[0]?.requestId, "codex-async:thread-1:async-question-1"); + NodeAssert.deepEqual(events[0]?.payload, { + responseMode: "message", + questions: [ + { + id: "0", + header: "Question", + question: "Which package manager?", + options: [ + { label: "pnpm", description: "" }, + { label: "npm", description: "" }, + ], + allowCustomAnswer: true, + multiSelect: false, + }, + { + id: "1", + header: "Question", + question: "What should it be named?", + options: [], + allowCustomAnswer: true, + multiSelect: false, + }, + ], + }); + NodeAssert.equal(events[1]?.type, "content.delta"); + }), + ); + it.effect("unwraps Codex token usage payloads for context window events", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index fa8511ee09a1..d44e06c60e1d 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -8,6 +8,7 @@ * @module CodexAdapterLive */ import { + EventId, type CanonicalItemType, type CanonicalRequestType, type CodexSettings, @@ -1445,6 +1446,27 @@ function mapToRuntimeEvents( if (!item) { return []; } + if (item.type === "agentMessage" && item.delivery === "async" && item.questions?.length) { + return [ + { + ...runtimeEventBase(event, canonicalThreadId), + type: "user-input.requested", + requestId: RuntimeRequestId.make(`codex-async:${canonicalThreadId}:${item.id}`), + eventId: EventId.make(`codex-async:${canonicalThreadId}:${item.id}`), + payload: { + responseMode: "message", + questions: item.questions.map((question, index) => ({ + id: String(index), + header: "Question", + question: question.title, + options: (question.options ?? []).map((label) => ({ label, description: "" })), + allowCustomAnswer: true, + multiSelect: false, + })), + }, + }, + ]; + } const itemType = toCanonicalItemType(item.type); if (itemType === "plan") { const detail = itemDetail(itemType, item); diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 15e23c79010c..a796d1c4038a 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -204,6 +204,7 @@ describe("ProviderSessionReaper", () => { Layer.provideMerge(Layer.succeed(ProviderService, providerService)), Layer.provideMerge( Layer.succeed(ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 5ffa0cb9e594..aa3b9ad93b14 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -926,6 +926,7 @@ const buildAppUnderTest = (options?: { ), Layer.provide( Layer.mock(ProjectionSnapshotQuery.ProjectionSnapshotQuery)({ + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.succeed(makeDefaultOrchestrationReadModel()), getSnapshot: () => Effect.succeed(makeDefaultOrchestrationReadModel()), getShellSnapshot: () => diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index a7cb97ae9765..1de6a4247811 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -66,6 +66,7 @@ const makeProviderService = (liveThreadIds: ReadonlyArray = []) => const queryWithThreads = (threads: ReadonlyArray>) => ({ + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.succeed({ threads } as never), }) as unknown as ProjectionSnapshotQuery.ProjectionSnapshotQuery["Service"]; @@ -631,6 +632,7 @@ it.effect("does not fail startup when the live provider session inventory cannot let queried = false; return ServerRuntimeStartup.reconcileProviderSessions.pipe( Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.sync(() => { queried = true; diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index 52711857ee33..0a8afde249a4 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -118,6 +118,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa yield* ServerRuntimeStartup.launchStartupHeartbeat.pipe( Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), @@ -185,6 +186,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa autoBootstrapProjectFromCwd: true, } as never), Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), @@ -250,6 +252,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when autoBootstrapProjectFromCwd: true, } as never), Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), @@ -312,6 +315,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa autoBootstrapProjectFromCwd: true, } as never), Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index db259821509c..baa57e8f99fa 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -250,6 +250,26 @@ describe("derivePendingApprovals", () => { }); describe("derivePendingUserInputs", () => { + it("keeps free-text questions without suggested answers", () => { + const question = { + id: "0", + header: "Question", + question: "What should it be named?", + options: [], + allowCustomAnswer: true, + multiSelect: false, + }; + const activities = [ + makeActivity({ + id: "async-question", + kind: "user-input.requested", + summary: "User input requested", + payload: { requestId: "async-1", responseMode: "message", questions: [question] }, + }), + ]; + expect(derivePendingUserInputs(activities)[0]?.questions).toEqual([question]); + }); + it("preserves native choice values and the custom-answer restriction", () => { const question = { id: "interaction-result", diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 1397c34eb4e5..fcaa5e9e1755 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -532,7 +532,7 @@ function parseUserInputQuestions( }; }) .filter((option): option is UserInputQuestion["options"][number] => option !== null); - if (options.length === 0) { + if (options.length === 0 && question.allowCustomAnswer === false) { return null; } return { diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 4e1a7b774076..b4a5a19e6f5c 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -24,6 +24,27 @@ adapter in a child scope. Adapter implementations live beside them in [`ProviderAdapter.ts`][adapter]. Read the driver plus its adapter to see how a specific agent's transport, config, and event shapes are mapped. +## Codex async questions + +Codex 0.153 exposes `request_user_input_async` through `item/started` and `item/completed` +notifications. The item has `type: "agentMessage"`, `delivery: "async"`, and a `questions` array. +Each question has a `title` and an optional `options` array of strings. The tool returns `{"accepted":true}` +without waiting. This is separate from the `item/tool/requestUserInput` server request. +See the [Codex tool handler](https://github.com/openai/codex/blob/d979df154cf60e13eafb5453e75b6d84f21c67bf/codex-rs/core/src/tools/handlers/request_user_input_async.rs). + +The Codex adapter maps completed question items to `user-input.requested` with +`responseMode: "message"` and stable request and event IDs. Questions use the existing web, +desktop, and mobile panels. They stay pending while the turn runs and after it finishes. + +The engine reads the request's latest stored activity before deciding a reply. This works after +startup, when the command snapshot has no activities, and after a resolution leaves the recent +activity window. The query returns one activity, not the full thread history. + +For these requests, the decider saves the resolution and a user message in one transaction. +The standard turn path delivers the message, including session resume and active-turn input. +It does not send a JSON-RPC response to Codex. Other providers and blocking Codex questions +keep their existing response paths. + ## Registry and routing Two registries separate configuration from live processes: diff --git a/docs/user/providers-codex.md b/docs/user/providers-codex.md index 29b2bcffedf9..5556ef18f119 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -34,6 +34,15 @@ In an existing Codex thread, send `/feedback` or `/feedback` followed by a descr issue. T3 Code uploads the thread and Codex logs to OpenAI and shows a thread ID that you can copy and share with OpenAI employees. +## Answer questions while Codex works + +Codex can ask questions without stopping its work. Choose a suggested answer or enter your own +in the question panel. Questions without suggested answers accept text. + +Your answers are sent as a new message. They reach the current turn while Codex is working, or +start a new turn if it has finished. Unanswered questions stay available after you reconnect. +This works in the web, desktop, and mobile apps. Codex must support async questions. + ## Sub-agent models The web and desktop Agents panel shows each sub-agent's model and reasoning effort when Codex diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index f2ca5118fe07..adbb4b9cb6c4 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -491,7 +491,7 @@ export type RequestResolvedPayload = typeof RequestResolvedPayload.Type; const UserInputQuestionOption = Schema.Struct({ label: TrimmedNonEmptyStringSchema, - description: TrimmedNonEmptyStringSchema, + description: Schema.String, value: Schema.optional(Schema.String), }); export type UserInputQuestionOption = typeof UserInputQuestionOption.Type; @@ -508,8 +508,9 @@ export const UserInputQuestion = Schema.Struct({ }); export type UserInputQuestion = typeof UserInputQuestion.Type; -const UserInputRequestedPayload = Schema.Struct({ +export const UserInputRequestedPayload = Schema.Struct({ questions: Schema.Array(UserInputQuestion), + responseMode: Schema.optional(Schema.Literal("message")), }); export type UserInputRequestedPayload = typeof UserInputRequestedPayload.Type; diff --git a/packages/effect-codex-app-server/scripts/generate.ts b/packages/effect-codex-app-server/scripts/generate.ts index 3622cb8f5158..4b106d228564 100644 --- a/packages/effect-codex-app-server/scripts/generate.ts +++ b/packages/effect-codex-app-server/scripts/generate.ts @@ -330,6 +330,60 @@ function stripNullDefaults(value: Schema.Json): Schema.Json { ) as Schema.Json; } +// Codex 0.153 adds async questions to agent messages. Keep older protocol +// fields until the next full refresh, including every thread history namespace. +function addAsyncQuestionFields(value: Schema.Json): Schema.Json { + if (Array.isArray(value)) { + return value.map(addAsyncQuestionFields); + } + if (value === null || typeof value !== "object") { + return value; + } + const properties = "properties" in value ? value.properties : undefined; + const itemType = + properties && typeof properties === "object" && "type" in properties + ? properties.type + : undefined; + if ( + properties && + typeof properties === "object" && + itemType && + typeof itemType === "object" && + "enum" in itemType && + Array.isArray(itemType.enum) && + itemType.enum.includes("agentMessage") + ) { + return { + ...value, + properties: { + ...properties, + delivery: { anyOf: [{ type: "string", enum: ["async"] }, { type: "null" }] }, + questions: { + anyOf: [ + { + type: "array", + items: { + type: "object", + properties: { + title: { type: "string" }, + options: { + anyOf: [{ type: "array", items: { type: "string" } }, { type: "null" }], + }, + }, + required: ["title"], + }, + }, + { type: "null" }, + ], + }, + }, + }; + } + return Object.fromEntries( + Object.entries(value).map(([key, child]) => [key, addAsyncQuestionFields(child)]), + ); +} + function toPascalCaseMethod(method: string) { return method .split("/") @@ -648,7 +702,7 @@ const generateFiles = Effect.fn("generateFiles")(function* () { for (const [name, schema] of Object.entries(aggregateSchemas).toSorted(([left], [right]) => left.localeCompare(right), )) { - generator.addSchema(name, schema as never); + generator.addSchema(name, addAsyncQuestionFields(schema) as never); } const generatedEntries = new Map(); diff --git a/packages/effect-codex-app-server/src/_generated/schema.gen.ts b/packages/effect-codex-app-server/src/_generated/schema.gen.ts index a26c98f44e6c..c6f3b3519fef 100644 --- a/packages/effect-codex-app-server/src/_generated/schema.gen.ts +++ b/packages/effect-codex-app-server/src/_generated/schema.gen.ts @@ -20405,6 +20405,11 @@ export type ServerNotification__ThreadItem = readonly memoryCitation?: ServerNotification__MemoryCitation | null; readonly phase?: ServerNotification__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -20530,6 +20535,18 @@ export const ServerNotification__ThreadItem = Schema.Union( ), phase: Schema.optionalKey(Schema.Union([ServerNotification__MessagePhase, Schema.Null])), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -21499,6 +21516,11 @@ export type V2ItemCompletedNotification__ThreadItem = readonly memoryCitation?: V2ItemCompletedNotification__MemoryCitation | null; readonly phase?: V2ItemCompletedNotification__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -21628,6 +21650,18 @@ export const V2ItemCompletedNotification__ThreadItem = Schema.Union( Schema.Union([V2ItemCompletedNotification__MessagePhase, Schema.Null]), ), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -21965,6 +21999,11 @@ export type V2ItemStartedNotification__ThreadItem = readonly memoryCitation?: V2ItemStartedNotification__MemoryCitation | null; readonly phase?: V2ItemStartedNotification__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -22092,6 +22131,18 @@ export const V2ItemStartedNotification__ThreadItem = Schema.Union( Schema.Union([V2ItemStartedNotification__MessagePhase, Schema.Null]), ), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -22601,6 +22652,11 @@ export type V2ReviewStartResponse__ThreadItem = readonly memoryCitation?: V2ReviewStartResponse__MemoryCitation | null; readonly phase?: V2ReviewStartResponse__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -22726,6 +22782,18 @@ export const V2ReviewStartResponse__ThreadItem = Schema.Union( ), phase: Schema.optionalKey(Schema.Union([V2ReviewStartResponse__MessagePhase, Schema.Null])), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -23050,6 +23118,11 @@ export type V2ThreadForkResponse__ThreadItem = readonly memoryCitation?: V2ThreadForkResponse__MemoryCitation | null; readonly phase?: V2ThreadForkResponse__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -23175,6 +23248,18 @@ export const V2ThreadForkResponse__ThreadItem = Schema.Union( ), phase: Schema.optionalKey(Schema.Union([V2ThreadForkResponse__MessagePhase, Schema.Null])), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -23468,6 +23553,11 @@ export type V2ThreadListResponse__ThreadItem = readonly memoryCitation?: V2ThreadListResponse__MemoryCitation | null; readonly phase?: V2ThreadListResponse__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -23593,6 +23683,18 @@ export const V2ThreadListResponse__ThreadItem = Schema.Union( ), phase: Schema.optionalKey(Schema.Union([V2ThreadListResponse__MessagePhase, Schema.Null])), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -23886,6 +23988,11 @@ export type V2ThreadMetadataUpdateResponse__ThreadItem = readonly memoryCitation?: V2ThreadMetadataUpdateResponse__MemoryCitation | null; readonly phase?: V2ThreadMetadataUpdateResponse__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -24015,6 +24122,18 @@ export const V2ThreadMetadataUpdateResponse__ThreadItem = Schema.Union( Schema.Union([V2ThreadMetadataUpdateResponse__MessagePhase, Schema.Null]), ), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -24309,6 +24428,11 @@ export type V2ThreadReadResponse__ThreadItem = readonly memoryCitation?: V2ThreadReadResponse__MemoryCitation | null; readonly phase?: V2ThreadReadResponse__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -24434,6 +24558,18 @@ export const V2ThreadReadResponse__ThreadItem = Schema.Union( ), phase: Schema.optionalKey(Schema.Union([V2ThreadReadResponse__MessagePhase, Schema.Null])), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -24735,6 +24871,11 @@ export type V2ThreadResumeResponse__ThreadItem = readonly memoryCitation?: V2ThreadResumeResponse__MemoryCitation | null; readonly phase?: V2ThreadResumeResponse__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -24860,6 +25001,18 @@ export const V2ThreadResumeResponse__ThreadItem = Schema.Union( ), phase: Schema.optionalKey(Schema.Union([V2ThreadResumeResponse__MessagePhase, Schema.Null])), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -25153,6 +25306,11 @@ export type V2ThreadRollbackResponse__ThreadItem = readonly memoryCitation?: V2ThreadRollbackResponse__MemoryCitation | null; readonly phase?: V2ThreadRollbackResponse__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -25280,6 +25438,18 @@ export const V2ThreadRollbackResponse__ThreadItem = Schema.Union( Schema.Union([V2ThreadRollbackResponse__MessagePhase, Schema.Null]), ), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -25583,6 +25753,11 @@ export type V2ThreadStartedNotification__ThreadItem = readonly memoryCitation?: V2ThreadStartedNotification__MemoryCitation | null; readonly phase?: V2ThreadStartedNotification__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -25712,6 +25887,18 @@ export const V2ThreadStartedNotification__ThreadItem = Schema.Union( Schema.Union([V2ThreadStartedNotification__MessagePhase, Schema.Null]), ), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -26006,6 +26193,11 @@ export type V2ThreadStartResponse__ThreadItem = readonly memoryCitation?: V2ThreadStartResponse__MemoryCitation | null; readonly phase?: V2ThreadStartResponse__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -26131,6 +26323,18 @@ export const V2ThreadStartResponse__ThreadItem = Schema.Union( ), phase: Schema.optionalKey(Schema.Union([V2ThreadStartResponse__MessagePhase, Schema.Null])), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -26424,6 +26628,11 @@ export type V2ThreadUnarchiveResponse__ThreadItem = readonly memoryCitation?: V2ThreadUnarchiveResponse__MemoryCitation | null; readonly phase?: V2ThreadUnarchiveResponse__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -26551,6 +26760,18 @@ export const V2ThreadUnarchiveResponse__ThreadItem = Schema.Union( Schema.Union([V2ThreadUnarchiveResponse__MessagePhase, Schema.Null]), ), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -26845,6 +27066,11 @@ export type V2TurnCompletedNotification__ThreadItem = readonly memoryCitation?: V2TurnCompletedNotification__MemoryCitation | null; readonly phase?: V2TurnCompletedNotification__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -26974,6 +27200,18 @@ export const V2TurnCompletedNotification__ThreadItem = Schema.Union( Schema.Union([V2TurnCompletedNotification__MessagePhase, Schema.Null]), ), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -27268,6 +27506,11 @@ export type V2TurnStartedNotification__ThreadItem = readonly memoryCitation?: V2TurnStartedNotification__MemoryCitation | null; readonly phase?: V2TurnStartedNotification__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -27395,6 +27638,18 @@ export const V2TurnStartedNotification__ThreadItem = Schema.Union( Schema.Union([V2TurnStartedNotification__MessagePhase, Schema.Null]), ), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ @@ -27689,6 +27944,11 @@ export type V2TurnStartResponse__ThreadItem = readonly memoryCitation?: V2TurnStartResponse__MemoryCitation | null; readonly phase?: V2TurnStartResponse__MessagePhase | null; readonly text: string; + readonly delivery?: "async" | null; + readonly questions?: ReadonlyArray<{ + readonly title: string; + readonly options?: ReadonlyArray | null; + }> | null; readonly type: "agentMessage"; } | { readonly id: string; readonly text: string; readonly type: "plan" } @@ -27814,6 +28074,18 @@ export const V2TurnStartResponse__ThreadItem = Schema.Union( ), phase: Schema.optionalKey(Schema.Union([V2TurnStartResponse__MessagePhase, Schema.Null])), text: Schema.String, + delivery: Schema.optionalKey(Schema.Union([Schema.Literal("async"), Schema.Null])), + questions: Schema.optionalKey( + Schema.Union([ + Schema.Array( + Schema.Struct({ + title: Schema.String, + options: Schema.optionalKey(Schema.Union([Schema.Array(Schema.String), Schema.Null])), + }), + ), + Schema.Null, + ]), + ), type: Schema.Literal("agentMessage").annotate({ title: "AgentMessageThreadItemType" }), }).annotate({ title: "AgentMessageThreadItem" }), Schema.Struct({ diff --git a/packages/effect-codex-app-server/src/schema.test.ts b/packages/effect-codex-app-server/src/schema.test.ts index d7afec7db268..caa507f18862 100644 --- a/packages/effect-codex-app-server/src/schema.test.ts +++ b/packages/effect-codex-app-server/src/schema.test.ts @@ -5,6 +5,29 @@ import * as CodexSchema from "./schema.ts"; const isGetAccountResponse = Schema.is(CodexSchema.V2GetAccountResponse); +it("keeps async questions in live notifications and thread history", () => { + const item = { + type: "agentMessage", + id: "question-1", + text: "Which package?\n- pnpm\n- npm\n\nWhat should it be named?", + phase: "final_answer", + delivery: "async", + questions: [ + { title: "Which package manager?", options: ["pnpm", "npm"] }, + { title: "What should it be named?" }, + ], + } as const; + for (const schema of [ + CodexSchema.ServerNotification__ThreadItem, + CodexSchema.V2ItemStartedNotification__ThreadItem, + CodexSchema.V2ItemCompletedNotification__ThreadItem, + CodexSchema.V2ThreadReadResponse__ThreadItem, + CodexSchema.V2ThreadResumeResponse__ThreadItem, + ]) { + assert.deepEqual(Schema.decodeUnknownSync(schema)(item), item); + } +}); + it("accepts Codex 0.150 multi-agent values", () => { const schemas = [ CodexSchema.ServerNotification__SubAgentActivityKind,