diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index e5c8d507..ef7f2bcc 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -1734,11 +1734,13 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { threadId: ThreadId.make("thread-stale-user-input"), activity: { id: EventId.make("activity-stale-user-input-requested"), + sequence: 1, tone: "info", kind: "user-input.requested", summary: "User input requested", payload: { requestId: "user-input-request-stale-1", + isBlocking: false, questions: [ { id: "sandbox_mode", @@ -1759,6 +1761,16 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }, }); + const pendingRows = yield* sql<{ + readonly pendingUserInputCount: number; + readonly blockingUserInputCount: number; + }>` + SELECT pending_user_input_count AS "pendingUserInputCount", + blocking_user_input_count AS "blockingUserInputCount" + FROM projection_threads WHERE thread_id = 'thread-stale-user-input' + `; + assert.deepEqual(pendingRows, [{ pendingUserInputCount: 1, blockingUserInputCount: 0 }]); + yield* appendAndProject({ type: "thread.activity-appended", eventId: EventId.make("evt-stale-user-input-4"), @@ -1773,6 +1785,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { threadId: ThreadId.make("thread-stale-user-input"), activity: { id: EventId.make("activity-stale-user-input-failed"), + sequence: 2, tone: "error", kind: "provider.user-input.respond.failed", summary: "Provider user input response failed", @@ -1782,19 +1795,21 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { "Provider adapter request failed (codex) for item/tool/requestUserInput: Unknown pending Codex user input request: user-input-request-stale-1", }, turnId: null, - createdAt: "2026-02-26T12:35:03.000Z", + createdAt: "2026-02-26T12:35:01.000Z", }, }, }); const threadRows = yield* sql<{ readonly pendingUserInputCount: number; + readonly blockingUserInputCount: number; }>` - SELECT pending_user_input_count AS "pendingUserInputCount" + SELECT pending_user_input_count AS "pendingUserInputCount", + blocking_user_input_count AS "blockingUserInputCount" FROM projection_threads WHERE thread_id = 'thread-stale-user-input' `; - assert.deepEqual(threadRows, [{ pendingUserInputCount: 0 }]); + assert.deepEqual(threadRows, [{ pendingUserInputCount: 0, blockingUserInputCount: 0 }]); }), ); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 340fac9d..617ef281 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -14,9 +14,8 @@ import * as Path from "effect/Path"; import * as Stream from "effect/Stream"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { - collectOpenPendingRequests, + countPendingUserInputs, isStalePendingRequestFailureDetail, - USER_INPUT_ACTIVITY_KINDS, } from "@threadlines/shared/pendingRequests"; import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../persistence/Errors.ts"; @@ -109,16 +108,6 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null { return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null; } -function derivePendingUserInputCountFromActivities( - activities: ReadonlyArray, -): number { - const ordered = [...activities].toSorted( - (left, right) => - compareTranscriptOrder(left, right) || left.activityId.localeCompare(right.activityId), - ); - return collectOpenPendingRequests(ordered, USER_INPUT_ACTIVITY_KINDS).length; -} - function activityMayAffectThreadShellSummary(activity: { readonly kind: string }): boolean { switch (activity.kind) { case "approval.requested": @@ -558,7 +547,13 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const pendingApprovalCount = pendingApprovals.filter( (approval) => approval.status === "pending", ).length; - const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities); + // Provider timestamps can arrive out of order; count in transcript order. + const userInputCounts = countPendingUserInputs( + [...activities].toSorted( + (left, right) => + compareTranscriptOrder(left, right) || left.activityId.localeCompare(right.activityId), + ), + ); const hasActionableProposedPlan = deriveHasActionableProposedPlan({ latestTurnId: existingRow.value.latestTurnId, proposedPlans, @@ -568,7 +563,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti ...existingRow.value, latestUserMessageAt, pendingApprovalCount, - pendingUserInputCount, + ...userInputCounts, hasActionableProposedPlan: hasActionableProposedPlan ? 1 : 0, }); }); @@ -603,6 +598,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, + blockingUserInputCount: 0, hasActionableProposedPlan: 0, deletedAt: null, }); @@ -1654,7 +1650,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti // rows. Other activity kinds that happen to carry a requestId // (e.g. user-input.requested / user-input.resolved) must not // pollute this projection — they have their own accounting via - // derivePendingUserInputCountFromActivities. + // countPendingUserInputs. if (event.payload.activity.kind !== "approval.requested") { return; } diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 234f3ad1..daa6562c 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -519,6 +519,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { latestUserMessageAt: "2026-02-24T00:00:04.000Z", hasPendingApprovals: true, hasPendingUserInput: false, + hasBlockingUserInput: false, hasActionableProposedPlan: false, cumulativeDiffStat: { additions: 2, deletions: 1 }, diffStatBaselineTurnCount: 0, @@ -1264,6 +1265,51 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { assert.equal(snapshot.threads[0]?.activities.length, MAX_THREAD_ACTIVITIES); assert.equal(snapshot.threads[0]?.subagents?.[0]?.id, "agent-durable"); + // Old open questions and approvals must survive a busy turn. Closed + // prompt history still leaves the recent activity window as usual. + const oldPrompts = [ + { + id: "open-question", + kind: "user-input.requested", + payload: { requestId: "question", isBlocking: false }, + }, + { id: "open-approval", kind: "approval.requested", payload: { requestId: "approval" } }, + { id: "closed-question", kind: "user-input.requested", payload: { requestId: "closed" } }, + { + id: "closed-question-resolved", + kind: "user-input.resolved", + payload: { requestId: "closed" }, + }, + ]; + for (const [index, prompt] of oldPrompts.entries()) { + yield* sql` + INSERT INTO projection_thread_activities ( + activity_id, thread_id, tone, kind, summary, payload_json, sequence, created_at + ) VALUES ( + ${prompt.id}, 'thread-activity-cap', 'info', ${prompt.kind}, 'Prompt', + ${JSON.stringify(prompt.payload)}, ${index}, '2026-03-01T00:00:00.000Z' + ) + `; + } + const retainedSnapshot = yield* snapshotQuery.getSnapshot(); + const retainedDetail = yield* snapshotQuery.getThreadDetailById( + ThreadId.make("thread-activity-cap"), + ); + assert.equal(retainedDetail._tag, "Some"); + if (retainedDetail._tag === "Some") { + const activities = retainedDetail.value.activities; + assert.equal(activities.length, MAX_THREAD_ACTIVITIES + 2); + assert.deepEqual( + activities.slice(0, 2).map((activity) => activity.id), + ["open-question", "open-approval"], + ); + assert.equal( + activities.some((activity) => activity.id === "closed-question"), + false, + ); + assert.deepEqual(retainedSnapshot.threads[0]?.activities, activities); + } + yield* sql`DELETE FROM projection_thread_activities`; yield* sql` INSERT INTO projection_thread_activities ( diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 3c4cf298..8a00caea 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -39,6 +39,7 @@ import * as Struct from "effect/Struct"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as SqlSchema from "effect/unstable/sql/SqlSchema"; import { MAX_THREAD_ACTIVITIES, MAX_THREAD_MESSAGES } from "@threadlines/shared/threadLimits"; +import { retainRecentActivitiesAndOpenRequests } from "@threadlines/shared/pendingRequests"; import { isPersistenceError, @@ -531,6 +532,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", + blocking_user_input_count AS "blockingUserInputCount", has_actionable_proposed_plan AS "hasActionableProposedPlan", deleted_at AS "deletedAt" FROM projection_threads @@ -568,6 +570,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", + blocking_user_input_count AS "blockingUserInputCount", has_actionable_proposed_plan AS "hasActionableProposedPlan", deleted_at AS "deletedAt" FROM projection_threads @@ -607,6 +610,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", + blocking_user_input_count AS "blockingUserInputCount", has_actionable_proposed_plan AS "hasActionableProposedPlan", deleted_at AS "deletedAt" FROM projection_threads @@ -729,6 +733,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { created_at AS "createdAt" FROM ranked_activities WHERE activity_rank <= ${MAX_THREAD_ACTIVITIES} + OR kind IN ( + 'approval.requested', 'approval.resolved', 'provider.approval.respond.failed', + 'user-input.requested', 'user-input.resolved', 'provider.user-input.respond.failed' + ) ORDER BY thread_id ASC, event_sequence ASC, @@ -1190,6 +1198,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", + blocking_user_input_count AS "blockingUserInputCount", has_actionable_proposed_plan AS "hasActionableProposedPlan", deleted_at AS "deletedAt" FROM projection_threads @@ -1260,6 +1269,15 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { created_at DESC, activity_id DESC LIMIT ${MAX_THREAD_ACTIVITIES} + ), retained_candidates AS ( + SELECT * FROM limited_activities + UNION + SELECT * FROM projection_thread_activities + WHERE thread_id = ${threadId} + AND kind IN ( + 'approval.requested', 'approval.resolved', 'provider.approval.respond.failed', + 'user-input.requested', 'user-input.resolved', 'provider.user-input.respond.failed' + ) ) SELECT activity_id AS "activityId", @@ -1272,7 +1290,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { payload_json AS "payload", sequence, created_at AS "createdAt" - FROM limited_activities + FROM retained_candidates ORDER BY event_sequence ASC, sequence ASC, @@ -1739,7 +1757,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { messages: messagesByThread.get(row.threadId) ?? [], proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], activities: dropStaleContextWindowActivities( - activitiesByThread.get(row.threadId) ?? [], + retainRecentActivitiesAndOpenRequests( + activitiesByThread.get(row.threadId) ?? [], + MAX_THREAD_ACTIVITIES, + ), ), subagents: subagentsByThread.get(row.threadId) ?? [], checkpoints: checkpointsByThread.get(row.threadId) ?? [], @@ -2130,6 +2151,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latestUserMessageAt: row.latestUserMessageAt, hasPendingApprovals: row.pendingApprovalCount > 0, hasPendingUserInput: row.pendingUserInputCount > 0, + hasBlockingUserInput: row.blockingUserInputCount > 0, hasActionableProposedPlan: row.hasActionableProposedPlan > 0, cumulativeDiffStat: mapThreadDiffStat(diffStatByThread.get(row.threadId)), diffStatBaselineTurnCount: row.diffStatBaselineTurnCount ?? 0, @@ -2280,6 +2302,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latestUserMessageAt: row.latestUserMessageAt, hasPendingApprovals: row.pendingApprovalCount > 0, hasPendingUserInput: row.pendingUserInputCount > 0, + hasBlockingUserInput: row.blockingUserInputCount > 0, hasActionableProposedPlan: row.hasActionableProposedPlan > 0, cumulativeDiffStat: mapThreadDiffStat(diffStatByThread.get(row.threadId)), diffStatBaselineTurnCount: row.diffStatBaselineTurnCount ?? 0, @@ -2556,6 +2579,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latestUserMessageAt: threadRow.value.latestUserMessageAt, hasPendingApprovals: threadRow.value.pendingApprovalCount > 0, hasPendingUserInput: threadRow.value.pendingUserInputCount > 0, + hasBlockingUserInput: threadRow.value.blockingUserInputCount > 0, hasActionableProposedPlan: threadRow.value.hasActionableProposedPlan > 0, cumulativeDiffStat: mapThreadDiffStat(Option.getOrUndefined(diffStatRow)), diffStatBaselineTurnCount: threadRow.value.diffStatBaselineTurnCount ?? 0, @@ -2667,7 +2691,12 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { deletedAt: null, messages: messageRows.map(mapThreadMessageRow), proposedPlans: proposedPlanRows.map(mapProposedPlanRow), - activities: dropStaleContextWindowActivities(activityRows.map(mapThreadActivityRow)), + activities: dropStaleContextWindowActivities( + retainRecentActivitiesAndOpenRequests( + activityRows.map(mapThreadActivityRow), + MAX_THREAD_ACTIVITIES, + ), + ), subagents: subagentRows.map(mapThreadSubagentRow), checkpoints: checkpointRows.map((row) => ({ turnId: row.turnId, diff --git a/apps/server/src/orchestration/Layers/ProviderActivityProjection.ts b/apps/server/src/orchestration/Layers/ProviderActivityProjection.ts index 8348f3cc..5f73868f 100644 --- a/apps/server/src/orchestration/Layers/ProviderActivityProjection.ts +++ b/apps/server/src/orchestration/Layers/ProviderActivityProjection.ts @@ -717,6 +717,7 @@ export function projectRuntimeEventToActivities( ...(event.payload.isBlocking !== undefined ? { isBlocking: event.payload.isBlocking } : {}), + ...(event.payload.responseMode ? { responseMode: event.payload.responseMode } : {}), }, }), ]; @@ -727,10 +728,11 @@ export function projectRuntimeEventToActivities( id: event.eventId, tone: "info", kind: "user-input.resolved", - summary: "User input submitted", + summary: event.payload.reason ? "Question closed" : "User input submitted", payload: { ...(event.requestId ? { requestId: event.requestId } : {}), answers: event.payload.answers, + ...(event.payload.reason ? { reason: event.payload.reason } : {}), }, }), ]; diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 8125997e..56c47942 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -2753,7 +2753,7 @@ describe("ProviderCommandReactor", () => { it("does not restart when the provider reports the same workspace through a path alias", async () => { const realDir = fs.mkdtempSync(path.join(os.tmpdir(), "threadlines-cwd-alias-")); const linkPath = `${realDir}-link`; - fs.symlinkSync(realDir, linkPath); + fs.symlinkSync(realDir, linkPath, process.platform === "win32" ? "junction" : "dir"); try { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; @@ -3789,6 +3789,239 @@ describe("ProviderCommandReactor", () => { }); }); + describe("async question replies", () => { + const now = "2026-01-01T00:00:00.000Z"; + const requestId = asApprovalRequestId("async-sign-in-question"); + const answerText = + "Are both providers signed in?\nBoth are signed in\n\nWhich page should I check?\nClone picker"; + + async function prepareQuestion(status: "running" | "ready" | "stopped" = "running") { + const harness = await createHarness(); + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-async-question-session"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status, + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "approval-required", + activeTurnId: status === "running" ? asTurnId("turn-1") : null, + lastError: null, + updatedAt: now, + }, + createdAt: now, + }), + ); + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.activity.append", + commandId: CommandId.make("cmd-async-question-requested"), + threadId: ThreadId.make("thread-1"), + activity: { + id: EventId.make("activity-async-question-requested"), + tone: "info", + kind: "user-input.requested", + summary: "User input requested", + payload: { + requestId, + isBlocking: false, + responseMode: "message", + questions: [ + { + id: "signed_in", + header: "Sign in", + question: "Are both providers signed in?", + options: [], + }, + { id: "page", header: "Page", question: "Which page should I check?", options: [] }, + ], + }, + turnId: asTurnId("turn-1"), + createdAt: now, + }, + createdAt: now, + }), + ); + await harness.drain(); + return harness; + } + + async function answerQuestion( + harness: Awaited>, + commandId = "cmd-answer-async-question", + answers: Record = { signed_in: "Both are signed in", page: "Clone picker" }, + ) { + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.user-input.respond", + commandId: CommandId.make(commandId), + threadId: ThreadId.make("thread-1"), + requestId, + answers, + createdAt: now, + }), + ); + await harness.drain(); + } + + it("steers a running turn, saves the answer in the transcript, and closes the question once", async () => { + const harness = await prepareQuestion(); + await answerQuestion(harness); + await answerQuestion(harness, "cmd-answer-async-question-again"); + + expect(harness.steerTurn).toHaveBeenCalledTimes(1); + expect(harness.steerTurn.mock.calls[0]?.[0]).toMatchObject({ + threadId: ThreadId.make("thread-1"), + expectedTurnId: asTurnId("turn-1"), + input: answerText, + }); + expect(harness.sendTurn).not.toHaveBeenCalled(); + expect(harness.respondToUserInput).not.toHaveBeenCalled(); + const thread = (await harness.readModel()).threads[0]; + expect(thread?.messages.filter((message) => message.role === "user")).toEqual([ + expect.objectContaining({ text: answerText, turnId: asTurnId("turn-1") }), + ]); + expect( + thread?.activities.filter((activity) => activity.kind === "user-input.resolved"), + ).toEqual([ + expect.objectContaining({ + payload: { + requestId, + answers: { signed_in: "Both are signed in", page: "Clone picker" }, + }, + }), + ]); + }); + + for (const status of ["ready", "stopped"] as const) { + it(`starts a normal turn when an open question is answered with the provider ${status}`, async () => { + const harness = await prepareQuestion(status); + await answerQuestion(harness); + await answerQuestion(harness, "cmd-answer-finished-question-again"); + + expect(harness.sendTurn).toHaveBeenCalledTimes(1); + expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ input: answerText }); + expect(harness.steerTurn).not.toHaveBeenCalled(); + expect(harness.respondToUserInput).not.toHaveBeenCalled(); + const thread = (await harness.readModel()).threads[0]; + expect(thread?.messages.filter((message) => message.role === "user")).toEqual([ + expect.objectContaining({ text: answerText }), + ]); + expect( + thread?.activities.filter((activity) => activity.kind === "user-input.resolved"), + ).toHaveLength(1); + }); + } + + it("starts a new turn if Codex finishes just before the answer reaches it", async () => { + const harness = await prepareQuestion(); + harness.steerTurn.mockImplementationOnce(() => + Effect.fail( + new ProviderAdapterRequestError({ + provider: "codex", + method: "turn/steer", + detail: "no active turn to steer", + }), + ), + ); + await answerQuestion(harness); + + expect(harness.steerTurn).toHaveBeenCalledTimes(1); + expect(harness.sendTurn).toHaveBeenCalledTimes(1); + expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ input: answerText }); + const thread = (await harness.readModel()).threads[0]; + expect(thread?.messages.filter((message) => message.role === "user")).toEqual([ + expect.objectContaining({ text: answerText }), + ]); + expect( + thread?.activities.filter((activity) => activity.kind === "user-input.resolved"), + ).toHaveLength(1); + expect( + thread?.activities.some( + (activity) => activity.kind === "provider.user-input.respond.failed", + ), + ).toBe(false); + }); + + it("keeps an incomplete answer pending so the user can finish and submit it", async () => { + const harness = await prepareQuestion(); + await answerQuestion(harness, "cmd-answer-incomplete", { signed_in: "Both are signed in" }); + + expect(harness.steerTurn).not.toHaveBeenCalled(); + expect(harness.sendTurn).not.toHaveBeenCalled(); + let thread = (await harness.readModel()).threads[0]; + expect(thread?.activities.some((activity) => activity.kind === "user-input.resolved")).toBe( + false, + ); + expect(thread?.activities).toContainEqual( + expect.objectContaining({ kind: "provider.user-input.respond.failed" }), + ); + expect(thread?.messages.filter((message) => message.role === "user")).toHaveLength(0); + + await answerQuestion(harness, "cmd-answer-complete"); + expect(harness.steerTurn).toHaveBeenCalledTimes(1); + thread = (await harness.readModel()).threads[0]; + expect( + thread?.activities.filter((activity) => activity.kind === "user-input.resolved"), + ).toHaveLength(1); + }); + + it("closes open and delayed questions after an explicit Stop", async () => { + const harness = await prepareQuestion(); + const question = (await harness.readModel()).threads[0]?.activities.find( + (activity) => activity.kind === "user-input.requested", + ); + expect(question).toBeDefined(); + if (!question) throw new Error("The async question was not projected"); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.session.stop", + commandId: CommandId.make("cmd-stop-open-async-question"), + threadId: ThreadId.make("thread-1"), + createdAt: now, + }), + ); + await harness.drain(); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.activity.append", + commandId: CommandId.make("cmd-late-async-question-after-stop"), + threadId: ThreadId.make("thread-1"), + activity: { + ...question, + id: EventId.make("activity-late-async-question"), + payload: { + ...(question.payload as Record), + requestId: "late-async-question", + }, + }, + createdAt: now, + }), + ); + await harness.drain(); + await answerQuestion(harness, "cmd-answer-question-after-stop"); + + const thread = (await harness.readModel()).threads[0]; + expect(thread?.session?.status).toBe("stopped"); + expect( + thread?.activities.filter((activity) => activity.kind === "user-input.resolved"), + ).toEqual([ + expect.objectContaining({ payload: expect.objectContaining({ requestId }) }), + expect.objectContaining({ + payload: expect.objectContaining({ requestId: "late-async-question" }), + }), + ]); + expect(harness.steerTurn).not.toHaveBeenCalled(); + expect(harness.sendTurn).not.toHaveBeenCalled(); + expect(thread?.messages.filter((message) => message.role === "user")).toHaveLength(0); + }); + }); + it("surfaces stale provider approval request failures without faking approval resolution", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 39952766..e291ca20 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -7,7 +7,7 @@ import { CheckoutMissingError, CommandId, EventId, - type MessageId, + MessageId, type ModelSelection, type OrchestrationEvent, ProviderDriverKind, @@ -33,6 +33,7 @@ import { isTemporaryWorktreeBranch, WORKTREE_BRANCH_PREFIX } from "@threadlines/ import { APPROVAL_ACTIVITY_KINDS, collectOpenPendingRequests, + extractPendingRequestId, PENDING_REQUEST_EXPIRED_REASON, PENDING_REQUEST_INTERRUPTED_REASON, USER_INPUT_ACTIVITY_KINDS, @@ -58,6 +59,7 @@ import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts import { ensureGeneralChatThreadScratchCwd } from "../generalChats.ts"; import { pauseActiveThreadGoalForStop } from "../threadGoalLifecycle.ts"; import { canReplaceThreadTitle } from "../threadTitle.ts"; +import { formatUserInputReply, readRequestedUserInput } from "../userInput.ts"; import { increment, orchestrationEventsProcessedTotal, @@ -356,6 +358,10 @@ const make = Effect.gen(function* () { ); const threadModelSelections = new Map(); + // Only explicit stops cancel durable message questions. Runtime reaping and + // reconnects leave them answerable. Keep shutdown notifications covered until + // a new session starts; no provider can send late notifications after restart. + const explicitlyStoppedThreads = new Set(); /** * Threads whose queued checkout switch is currently deferred because the @@ -656,6 +662,7 @@ const make = Effect.gen(function* () { }>; readonly detail: string; readonly reason?: string; + readonly includeMessageQuestions?: boolean; }) { const expirations = [ ...collectOpenPendingRequests(input.activities, APPROVAL_ACTIVITY_KINDS).map((open) => ({ @@ -663,11 +670,20 @@ const make = Effect.gen(function* () { kind: APPROVAL_ACTIVITY_KINDS.resolved, summary: "Approval request expired", })), - ...collectOpenPendingRequests(input.activities, USER_INPUT_ACTIVITY_KINDS).map((open) => ({ - open, - kind: USER_INPUT_ACTIVITY_KINDS.resolved, - summary: "User input request expired", - })), + ...collectOpenPendingRequests(input.activities, USER_INPUT_ACTIVITY_KINDS) + .filter((open) => { + const question = readRequestedUserInput(open.activity.payload); + return ( + input.includeMessageQuestions || + Option.isNone(question) || + question.value.responseMode !== "message" + ); + }) + .map((open) => ({ + open, + kind: USER_INPUT_ACTIVITY_KINDS.resolved, + summary: "User input request expired", + })), ]; if (expirations.length === 0) { return; @@ -710,6 +726,7 @@ const make = Effect.gen(function* () { activities: input.activities, detail: "The provider turn was interrupted before the request was answered.", reason: PENDING_REQUEST_INTERRUPTED_REASON, + includeMessageQuestions: true, }); }); @@ -2069,6 +2086,131 @@ const make = Effect.gen(function* () { if (!thread) { return; } + const openQuestion = collectOpenPendingRequests( + thread.activities, + USER_INPUT_ACTIVITY_KINDS, + ).find((open) => open.requestId === event.payload.requestId); + const requested = openQuestion + ? readRequestedUserInput(openQuestion.activity.payload) + : Option.none(); + if (!openQuestion) { + const previous = thread.activities.findLast( + (activity) => + activity.kind === USER_INPUT_ACTIVITY_KINDS.requested && + extractPendingRequestId(activity.payload) === event.payload.requestId, + ); + const previousQuestion = previous + ? readRequestedUserInput(previous.payload) + : Option.none(); + if (Option.isSome(previousQuestion) && previousQuestion.value.responseMode === "message") + return; + } + if (Option.isSome(requested) && requested.value.responseMode === "message") { + const input = formatUserInputReply(requested.value.questions, event.payload.answers); + const fail = (detail: string) => + appendProviderFailureActivity({ + threadId: thread.id, + kind: "provider.user-input.respond.failed", + summary: "Could not send answer", + detail, + turnId: openQuestion?.activity.turnId ?? null, + createdAt: event.payload.createdAt, + requestId: event.payload.requestId, + }); + if (!input) return yield* fail("Answer each question before submitting."); + + // A stable message id lets a replay finish projection without sending + // the same answer twice after it has already reached the transcript. + const messageId = MessageId.make(`question-answer:${event.payload.requestId}`); + yield* Effect.gen(function* () { + if (!thread.messages.some((message) => message.id === messageId)) { + const activeTurnId = thread.session?.activeTurnId; + let steered = false; + if (thread.session?.status === "running" && activeTurnId) { + steered = yield* providerService + .steerTurn({ + threadId: thread.id, + expectedTurnId: activeTurnId, + messageId, + input, + }) + .pipe( + Effect.as(true), + Effect.catchCause((cause) => + isNoActiveTurnSteerError(cause) + ? Effect.succeed(false) + : Effect.failCause(cause), + ), + ); + if (steered) { + yield* orchestrationEngine.dispatch({ + type: "thread.follow-up.accept", + commandId: CommandId.make(`question-answer-accepted:${event.payload.requestId}`), + threadId: thread.id, + turnId: activeTurnId, + message: { messageId, role: "user", text: input, attachments: [] }, + createdAt: event.payload.createdAt, + }); + } else { + // Codex may finish between displaying the question and receiving + // its answer. Only release the exact turn it rejected. + const latestThread = yield* resolveThread(thread.id); + const latestSession = latestThread?.session; + if ( + latestSession?.status === "interrupted" || + explicitlyStoppedThreads.has(thread.id) + ) { + return; + } + if (latestSession?.activeTurnId && latestSession.activeTurnId !== activeTurnId) { + return yield* fail("The active turn changed. Submit your answer again."); + } + if (latestSession?.activeTurnId === activeTurnId) { + yield* setThreadSession({ + threadId: thread.id, + session: { + ...latestSession, + status: "ready", + activeTurnId: null, + updatedAt: event.payload.createdAt, + }, + createdAt: event.payload.createdAt, + }); + } + } + } + if (!steered) { + // Persist the answer through the normal turn path. If starting + // the provider fails, the transcript keeps the reply for Retry. + yield* orchestrationEngine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`question-answer-start:${event.payload.requestId}`), + threadId: thread.id, + message: { messageId, role: "user", text: input, attachments: [] }, + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + createdAt: event.payload.createdAt, + }); + } + } + yield* orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId: CommandId.make(`question-answer-resolved:${event.payload.requestId}`), + threadId: thread.id, + activity: { + id: EventId.make(`question-answer-resolved:${event.payload.requestId}`), + kind: "user-input.resolved", + tone: "info", + summary: "User input submitted", + payload: { requestId: event.payload.requestId, answers: event.payload.answers }, + turnId: openQuestion?.activity.turnId ?? null, + createdAt: event.payload.createdAt, + }, + createdAt: event.payload.createdAt, + }); + }).pipe(Effect.catchCause((cause) => fail(formatFailureDetail(cause)))); + return; + } const hasSession = thread.session && thread.session.status !== "stopped"; if (!hasSession) { return yield* appendProviderFailureActivity({ @@ -2142,6 +2284,15 @@ const make = Effect.gen(function* () { yield* providerService.stopSession({ threadId: thread.id }); } + explicitlyStoppedThreads.add(thread.id); + const stoppedThread = yield* resolveThread(thread.id); + yield* expireOpenPendingRequests({ + threadId: thread.id, + activities: stoppedThread?.activities ?? thread.activities, + detail: "The session was stopped before the question was answered.", + includeMessageQuestions: true, + }); + yield* setThreadSession({ threadId: thread.id, session: { @@ -2240,17 +2391,19 @@ const make = Effect.gen(function* () { }); /** - * Pending approval / user-input prompts are answered through the live - * provider session; once that session stops (explicit stop, inactivity - * reap, startup reconcile after a server restart) the provider-side - * request is gone and the prompt can never be answered. Close each open - * prompt with an expiry activity so clients stop offering a Submit that - * is guaranteed to fail. + * RPC prompts need their live provider session and expire when it stops. + * Message questions survive automatic session recycling and reconnects. */ const processSessionSet = Effect.fn("processSessionSet")(function* ( event: Extract, ) { if (event.payload.session.status !== "stopped") { + if ( + event.payload.session.status === "starting" || + event.payload.session.status === "running" + ) { + explicitlyStoppedThreads.delete(event.payload.threadId); + } // Cheap payload precheck; the apply path re-validates against the // freshly projected thread before touching the session. const session = event.payload.session; @@ -2325,6 +2478,7 @@ const make = Effect.gen(function* () { threadId: event.payload.threadId, activities: thread.activities, detail: "The provider session stopped before the request was answered.", + includeMessageQuestions: explicitlyStoppedThreads.has(thread.id), }); } }, diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 680ae8ee..8ea815f7 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -14,6 +14,7 @@ import { } from "@threadlines/shared/threadLimits"; import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; +import { retainRecentActivitiesAndOpenRequests } from "@threadlines/shared/pendingRequests"; import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts"; import { @@ -929,15 +930,16 @@ export function projectEvent( const existingActivity = thread.activities.find( (entry) => entry.id === payload.activity.id, ); - const activities = [ - ...thread.activities.filter((entry) => entry.id !== payload.activity.id), - { - ...payload.activity, - eventSequence: existingActivity ? existingActivity.eventSequence : event.sequence, - }, - ] - .toSorted(compareThreadActivities) - .slice(-MAX_THREAD_ACTIVITIES); + const activities = retainRecentActivitiesAndOpenRequests( + [ + ...thread.activities.filter((entry) => entry.id !== payload.activity.id), + { + ...payload.activity, + eventSequence: existingActivity ? existingActivity.eventSequence : event.sequence, + }, + ].toSorted(compareThreadActivities), + MAX_THREAD_ACTIVITIES, + ); const subagents = projectSubagentActivity(thread.subagents ?? [], { ...payload.activity, eventSequence: event.sequence, diff --git a/apps/server/src/orchestration/userInput.ts b/apps/server/src/orchestration/userInput.ts new file mode 100644 index 00000000..87c24233 --- /dev/null +++ b/apps/server/src/orchestration/userInput.ts @@ -0,0 +1,28 @@ +import { UserInputRequestedPayload, type ProviderUserInputAnswers } from "@threadlines/contracts"; +import * as Schema from "effect/Schema"; + +const decodeRequestedInput = Schema.decodeUnknownOption(UserInputRequestedPayload); + +/** Read durable questions so message replies survive a provider process restart. */ +export const readRequestedUserInput = decodeRequestedInput; + +/** Include each question in the reply so later answers retain their context. */ +export function formatUserInputReply( + questions: UserInputRequestedPayload["questions"], + answers: ProviderUserInputAnswers, +): string | undefined { + const replies: string[] = []; + for (const question of questions) { + const answer = answers[question.id]; + const values: ReadonlyArray = + typeof answer === "string" ? [answer] : Array.isArray(answer) ? answer : []; + const text = values.flatMap((value) => + typeof value === "string" && value.trim() ? [value.trim()] : [], + ); + if (text.length === 0 || text.length !== values.length) { + return undefined; + } + replies.push(`${question.question}\n${text.join(", ")}`); + } + return replies.length > 0 ? replies.join("\n\n") : undefined; +} diff --git a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts index 18cc6c4f..39a57b67 100644 --- a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts @@ -100,6 +100,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, + blockingUserInputCount: 0, hasActionableProposedPlan: 0, deletedAt: null, }); diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index 6052be5a..d1949ee0 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -56,6 +56,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { latest_user_message_at, pending_approval_count, pending_user_input_count, + blocking_user_input_count, has_actionable_proposed_plan, deleted_at ) @@ -84,6 +85,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.latestUserMessageAt}, ${row.pendingApprovalCount}, ${row.pendingUserInputCount}, + ${row.blockingUserInputCount}, ${row.hasActionableProposedPlan}, ${row.deletedAt} ) @@ -112,6 +114,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { latest_user_message_at = excluded.latest_user_message_at, pending_approval_count = excluded.pending_approval_count, pending_user_input_count = excluded.pending_user_input_count, + blocking_user_input_count = excluded.blocking_user_input_count, has_actionable_proposed_plan = excluded.has_actionable_proposed_plan, deleted_at = excluded.deleted_at `, @@ -147,6 +150,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", + blocking_user_input_count AS "blockingUserInputCount", has_actionable_proposed_plan AS "hasActionableProposedPlan", deleted_at AS "deletedAt" FROM projection_threads @@ -184,6 +188,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", + blocking_user_input_count AS "blockingUserInputCount", has_actionable_proposed_plan AS "hasActionableProposedPlan", deleted_at AS "deletedAt" FROM projection_threads diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 90c9df20..258844f8 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -63,6 +63,7 @@ import Migration0047 from "./Migrations/047_ProjectionThreadSubagents.ts"; import Migration0048 from "./Migrations/048_BackfillThreadSubagents.ts"; import Migration0049 from "./Migrations/049_ProjectionThreadSubagentsBackgrounded.ts"; import Migration0050 from "./Migrations/050_ProjectionTranscriptEventSequence.ts"; +import Migration0051 from "./Migrations/051_ProjectionThreadsBlockingUserInput.ts"; /** * Migration loader with all migrations defined inline. @@ -125,6 +126,7 @@ export const migrationEntries = [ [48, "BackfillThreadSubagents", Migration0048], [49, "ProjectionThreadSubagentsBackgrounded", Migration0049], [50, "ProjectionTranscriptEventSequence", Migration0050], + [51, "ProjectionThreadsBlockingUserInput", Migration0051], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/051_ProjectionThreadsBlockingUserInput.test.ts b/apps/server/src/persistence/Migrations/051_ProjectionThreadsBlockingUserInput.test.ts new file mode 100644 index 00000000..ad83411c --- /dev/null +++ b/apps/server/src/persistence/Migrations/051_ProjectionThreadsBlockingUserInput.test.ts @@ -0,0 +1,61 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as NodeSqliteClient from "../NodeSqliteClient.ts"; + +it.layer(NodeSqliteClient.layerMemory())("051_ProjectionThreadsBlockingUserInput", (it) => { + it.effect("backfills mixed open questions without reviving resolved or stale requests", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 50 }); + yield* sql` + INSERT INTO projection_threads ( + thread_id, project_id, title, model_selection_json, runtime_mode, + interaction_mode, created_at, updated_at, pending_user_input_count + ) VALUES ( + 'thread-1', 'project-1', 'Questions', '{"instanceId":"codex","model":"gpt-5-codex"}', + 'full-access', 'default', '2026-09-05T00:00:00.000Z', '2026-09-05T00:00:00.000Z', 3 + ) + `; + const activities = [ + { kind: "user-input.requested", payload: { requestId: "legacy" } }, + { kind: "user-input.requested", payload: { requestId: "blocking", isBlocking: true } }, + { kind: "user-input.requested", payload: { requestId: "async", isBlocking: false } }, + { kind: "user-input.requested", payload: { requestId: "resolved" } }, + { kind: "user-input.requested", payload: { requestId: "stale" } }, + { kind: "user-input.resolved", payload: { requestId: "resolved" } }, + { + kind: "provider.user-input.respond.failed", + payload: { requestId: "stale", detail: "Unknown pending user-input request" }, + }, + { + kind: "provider.user-input.respond.failed", + payload: { requestId: "legacy", detail: "Temporary connection failure" }, + }, + ]; + for (const [index, activity] of activities.entries()) { + // Sequence is authoritative even when delayed events have older + // timestamps and their IDs sort before the original request. + const reverseIndex = activities.length - index; + const createdAt = new Date(Date.UTC(2026, 8, 5, 0, 0, reverseIndex)).toISOString(); + yield* sql` + INSERT INTO projection_thread_activities ( + activity_id, thread_id, tone, kind, summary, payload_json, sequence, created_at + ) VALUES ( + ${`activity-${reverseIndex}`}, 'thread-1', 'info', ${activity.kind}, 'Question', + ${JSON.stringify(activity.payload)}, ${index}, ${createdAt} + ) + `; + } + + yield* runMigrations({ toMigrationInclusive: 51 }); + const rows = yield* sql` + SELECT pending_user_input_count, blocking_user_input_count + FROM projection_threads WHERE thread_id = 'thread-1' + `; + assert.deepEqual(rows, [{ pending_user_input_count: 3, blocking_user_input_count: 2 }]); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/051_ProjectionThreadsBlockingUserInput.ts b/apps/server/src/persistence/Migrations/051_ProjectionThreadsBlockingUserInput.ts new file mode 100644 index 00000000..9a6267f9 --- /dev/null +++ b/apps/server/src/persistence/Migrations/051_ProjectionThreadsBlockingUserInput.ts @@ -0,0 +1,39 @@ +import { countPendingUserInputs } from "@threadlines/shared/pendingRequests"; +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` + ALTER TABLE projection_threads + ADD COLUMN blocking_user_input_count INTEGER NOT NULL DEFAULT 0 + `; + + // Only threads with open questions need a backfill. Replay the same request + // rules as the projection so old prompts and failed answers stay consistent. + const threads = yield* sql<{ threadId: string }>` + SELECT thread_id AS "threadId" FROM projection_threads + WHERE pending_user_input_count > 0 + `; + for (const { threadId } of threads) { + const activities = yield* sql<{ kind: string; payloadJson: string }>` + SELECT kind, payload_json AS "payloadJson" + FROM projection_thread_activities + WHERE thread_id = ${threadId} + AND json_valid(payload_json) + 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 counts = countPendingUserInputs( + activities.map((activity) => ({ + kind: activity.kind, + payload: JSON.parse(activity.payloadJson) as unknown, + })), + ); + yield* sql` + UPDATE projection_threads + SET blocking_user_input_count = ${counts.blockingUserInputCount} + WHERE thread_id = ${threadId} + `; + } +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index d55972f6..41e9d500 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -67,6 +67,7 @@ export const ProjectionThread = Schema.Struct({ latestUserMessageAt: Schema.NullOr(IsoDateTime), pendingApprovalCount: NonNegativeInt, pendingUserInputCount: NonNegativeInt, + blockingUserInputCount: NonNegativeInt, hasActionableProposedPlan: NonNegativeInt, deletedAt: Schema.NullOr(IsoDateTime), }); diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 908b4e17..54618da5 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -2444,6 +2444,94 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }), ); + it.effect("projects async question messages once with choices and free-text questions", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const eventsFiber = yield* Stream.runCollect(Stream.take(adapter.streamEvents, 2)).pipe( + Effect.forkChild, + ); + const item = { + id: "async-question-1", + type: "agentMessage", + delivery: "async", + phase: "final_answer", + text: "Which providers are signed in?\n- Both\n- Only one", + questions: [ + { title: "Which providers are signed in?", options: ["Both", "Only one"] }, + { title: "Anything else I should know?" }, + ], + }; + const base = { + kind: "notification" as const, + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), + createdAt: "2026-01-01T00:00:00.000Z", + }; + yield* runtime.emit({ + ...base, + id: asEventId("async-started"), + method: "item/started", + payload: { item, threadId: "thread-1", turnId: "turn-1", startedAtMs: 0 }, + }); + yield* runtime.emit({ + ...base, + id: asEventId("async-completed"), + method: "item/completed", + payload: { item, threadId: "thread-1", turnId: "turn-1", completedAtMs: 0 }, + }); + yield* runtime.emit({ + ...base, + id: asEventId("async-following-text"), + method: "item/agentMessage/delta", + payload: { + itemId: "work-continues", + threadId: "thread-1", + turnId: "turn-1", + delta: "I can keep checking the layout.", + }, + }); + const events = Array.from(yield* Fiber.join(eventsFiber)); + assert.equal(events[0]?.type, "user-input.requested"); + if (events[0]?.type === "user-input.requested") { + assert.equal(events[0].requestId, "codex-question:thread-1:async-question-1"); + assert.equal(events[0].payload.isBlocking, false); + assert.equal(events[0].payload.responseMode, "message"); + assert.deepEqual( + events[0].payload.questions.map((question) => + question.options.map((option) => option.label), + ), + [["Both", "Only one"], []], + ); + } + assert.equal(events[1]?.type, "content.delta"); + }), + ); + + it.effect("projects provider-cleared input as closed without an answer", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const eventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + yield* runtime.emit({ + id: asEventId("question-closed"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), + requestId: ApprovalRequestId.make("input-1"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "item/tool/requestUserInput/resolved", + payload: { reason: "turn-completed" }, + }); + const event = yield* Fiber.join(eventFiber); + assert.equal(event._tag, "Some"); + if (event._tag === "Some") { + assert.equal(event.value.type, "user-input.resolved"); + assert.deepEqual(event.value.payload, { answers: {}, reason: "turn-completed" }); + } + }), + ); + 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 71f94372..e3766e3f 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -1225,7 +1225,7 @@ function toUserInputQuestions(questions: ReadonlyArray { + const title = question.title.trim(); + if (!title) return []; + return [ + { + id: `question-${index + 1}`, + header: questionCount > 1 ? `Question ${index + 1}` : "Question", + question: title, + options: (question.options ?? []) + .filter((label) => label.trim().length > 0) + .map((label) => ({ label: label.trim(), description: label.trim() })), + multiSelect: false, + }, + ]; + }); + if (questions.length > 0) { + return [ + { + ...runtimeEventBase(event, canonicalThreadId), + eventId: EventId.make(`codex-question:${canonicalThreadId}:${item.id}`), + requestId: RuntimeRequestId.make(`codex-question:${canonicalThreadId}:${item.id}`), + type: "user-input.requested", + payload: { questions, isBlocking: false, responseMode: "message" }, + }, + ]; + } + } const itemType = toCanonicalItemType(item.type); if (itemType === "plan") { const detail = itemDetail(item); @@ -2279,6 +2317,17 @@ export function mapToRuntimeEvents( ]; } + if (event.method === "item/tool/requestUserInput/resolved") { + const reason = firstStringField(event.payload, ["reason"]) ?? "provider-resolved"; + return [ + { + ...runtimeEventBase(event, canonicalThreadId), + type: "user-input.resolved", + payload: { answers: {}, reason }, + }, + ]; + } + if (event.method === "item/tool/requestUserInput/answered") { const payload = readPayload(EffectCodexSchema.ToolRequestUserInputResponse, event.payload); if (!payload) { diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index afd43052..3a89a2da 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -1,9 +1,14 @@ import assert from "node:assert/strict"; +import { fileURLToPath } from "node:url"; import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Stream from "effect/Stream"; import * as Schema from "effect/Schema"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { describe, it } from "vite-plus/test"; -import { ThreadId, TurnId } from "@threadlines/contracts"; +import { ThreadId, TurnId, type ProviderEvent } from "@threadlines/contracts"; import * as CodexErrors from "effect-codex-app-server/errors"; import * as CodexRpc from "effect-codex-app-server/rpc"; @@ -20,6 +25,7 @@ import { isNativeThreadForkUnsupportedError, isRecoverableThreadResumeError, makeCodexStderrLineClassifier, + makeCodexSessionRuntime, openCodexThread, readCollabChildThreadMetadata, readCollabParentTurnId, @@ -32,6 +38,69 @@ import { } from "./CodexSessionRuntime.ts"; const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); +describe("Codex user input lifecycle", () => { + for (const [input, reason] of [ + ["resolve", "resolved"], + ["complete", "turn-completed"], + ] as const) { + it(`closes a pending question when Codex sends ${reason}`, async () => { + await Effect.runPromise( + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const runtime = yield* makeCodexSessionRuntime({ + threadId: ThreadId.make("local-question-thread"), + serverPort: 0, + binaryPath: process.execPath, + cwd: process.cwd(), + runtimeMode: "full-access", + }).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, { + ...spawner, + spawn: () => + spawner.spawn( + ChildProcess.make(process.execPath, [ + fileURLToPath( + new URL( + "../../../../../packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts", + import.meta.url, + ), + ), + ]), + ), + }), + ); + const requested = yield* Deferred.make(); + const resolved = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.runForEach((event) => { + if (event.method === "item/tool/requestUserInput") + return Deferred.succeed(requested, event); + if (event.method === "item/tool/requestUserInput/resolved") + return Deferred.succeed(resolved, event); + return Effect.void; + }), + Effect.forkScoped, + ); + yield* runtime.start(); + yield* runtime.sendTurn({ input: "Ask a question" }); + const request = yield* Deferred.await(requested); + assert.ok(request.requestId); + assert.equal((request.payload as { isBlocking: boolean }).isBlocking, true); + yield* runtime.steerTurn({ expectedTurnId: TurnId.make("turn-1"), input }); + const resolution = yield* Deferred.await(resolved); + assert.equal(resolution.requestId, request.requestId); + assert.equal(resolution.turnId, request.turnId); + assert.deepEqual(resolution.payload, { reason }); + const lateAnswer = yield* runtime + .respondToUserInput(request.requestId, { proceed: ["yes"] }) + .pipe(Effect.result); + assert.equal(lateAnswer._tag, "Failure"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + }); + } +}); + function makeThreadOpenResponse( threadId: string, ): CodexRpc.ClientRequestResponsesByMethod["thread/start"] { diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index c470d997..2149b1a2 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -389,6 +389,7 @@ interface ApprovalCorrelation { interface PendingUserInput { readonly requestId: ApprovalRequestId; + readonly jsonRpcId: string; readonly turnId: TurnId | undefined; readonly itemId: ProviderItemId | undefined; readonly answers: Deferred.Deferred; @@ -1581,7 +1582,7 @@ export const makeCodexSessionRuntime = ( ); const settlePendingUserInputs = (answers: ProviderUserInputAnswers) => - Ref.get(pendingUserInputsRef).pipe( + Ref.getAndSet(pendingUserInputsRef, new Map()).pipe( Effect.flatMap((pendingUserInputs) => Effect.forEach( Array.from(pendingUserInputs.values()), @@ -1592,6 +1593,31 @@ export const makeCodexSessionRuntime = ( ), ); + const expirePendingUserInputs = ( + matches: (pending: PendingUserInput) => boolean, + reason: "resolved" | "turn-completed", + ) => + Effect.gen(function* () { + const expired = yield* Ref.modify(pendingUserInputsRef, (current) => { + const next = new Map(current); + const expired = Array.from(current.values()).filter(matches); + for (const pending of expired) next.delete(pending.requestId); + return [expired, next]; + }); + for (const pending of expired) { + yield* Deferred.succeed(pending.answers, {}); + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "item/tool/requestUserInput/resolved", + requestId: pending.requestId, + ...(pending.turnId ? { turnId: pending.turnId } : {}), + ...(pending.itemId ? { itemId: pending.itemId } : {}), + payload: { reason }, + }); + } + }); + const handleRawNotification = (notification: CodexServerNotification) => Effect.gen(function* () { const route = readRouteFields(notification); @@ -1658,6 +1684,10 @@ export const makeCodexSessionRuntime = ( typeof notification.params.requestId === "string" ? notification.params.requestId : String(notification.params.requestId); + yield* expirePendingUserInputs( + (pending) => pending.jsonRpcId === rawRequestId, + "resolved", + ); const correlation = rawRequestId ? (yield* Ref.get(approvalCorrelationsRef)).get(rawRequestId) : undefined; @@ -1758,7 +1788,14 @@ export const makeCodexSessionRuntime = ( status: payload.turn.status === "failed" ? "error" : "ready", activeTurnId: undefined, ...(lastError ? { lastError } : {}), - }); + }).pipe( + Effect.andThen( + expirePendingUserInputs( + (pending) => pending.turnId === payload.turn.id, + "turn-completed", + ), + ), + ); }), ), ); @@ -1785,63 +1822,65 @@ export const makeCodexSessionRuntime = ( ), ); - yield* client.handleServerRequest("item/commandExecution/requestApproval", (payload) => - Effect.gen(function* () { - const requestId = ApprovalRequestId.make(yield* randomUUIDv4); - const turnId = TurnId.make(payload.turnId); - const itemId = ProviderItemId.make(payload.itemId); - const decision = yield* Deferred.make(); + yield* client.handleServerRequest( + "item/commandExecution/requestApproval", + (payload, metadata) => + Effect.gen(function* () { + const requestId = ApprovalRequestId.make(yield* randomUUIDv4); + const turnId = TurnId.make(payload.turnId); + const itemId = ProviderItemId.make(payload.itemId); + const decision = yield* Deferred.make(); - yield* Ref.update(pendingApprovalsRef, (current) => { - const next = new Map(current); - next.set(requestId, { - requestId, - jsonRpcId: payload.approvalId ?? payload.itemId, - requestKind: "command", - turnId, - itemId, - decision, + yield* Ref.update(pendingApprovalsRef, (current) => { + const next = new Map(current); + next.set(requestId, { + requestId, + jsonRpcId: String(metadata.id), + requestKind: "command", + turnId, + itemId, + decision, + }); + return next; }); - return next; - }); - yield* Ref.update(approvalCorrelationsRef, (current) => { - const next = new Map(current); - next.set(payload.approvalId ?? payload.itemId, { + yield* Ref.update(approvalCorrelationsRef, (current) => { + const next = new Map(current); + next.set(String(metadata.id), { + requestId, + requestKind: "command", + turnId, + itemId, + }); + return next; + }); + + yield* emitEvent({ + kind: "request", + threadId: options.threadId, + method: "item/commandExecution/requestApproval", requestId, requestKind: "command", - turnId, - itemId, + ...(turnId ? { turnId } : {}), + ...(itemId ? { itemId } : {}), + payload, }); - return next; - }); - yield* emitEvent({ - kind: "request", - threadId: options.threadId, - method: "item/commandExecution/requestApproval", - requestId, - requestKind: "command", - ...(turnId ? { turnId } : {}), - ...(itemId ? { itemId } : {}), - payload, - }); - - const resolved = yield* Deferred.await(decision).pipe( - Effect.ensuring( - Ref.update(pendingApprovalsRef, (current) => { - const next = new Map(current); - next.delete(requestId); - return next; - }), - ), - ); - return { - decision: resolved, - } satisfies EffectCodexSchema.CommandExecutionRequestApprovalResponse; - }), + const resolved = yield* Deferred.await(decision).pipe( + Effect.ensuring( + Ref.update(pendingApprovalsRef, (current) => { + const next = new Map(current); + next.delete(requestId); + return next; + }), + ), + ); + return { + decision: resolved, + } satisfies EffectCodexSchema.CommandExecutionRequestApprovalResponse; + }), ); - yield* client.handleServerRequest("item/fileChange/requestApproval", (payload) => + yield* client.handleServerRequest("item/fileChange/requestApproval", (payload, metadata) => Effect.gen(function* () { const requestId = ApprovalRequestId.make(yield* randomUUIDv4); const turnId = TurnId.make(payload.turnId); @@ -1852,7 +1891,7 @@ export const makeCodexSessionRuntime = ( const next = new Map(current); next.set(requestId, { requestId, - jsonRpcId: payload.itemId, + jsonRpcId: String(metadata.id), requestKind: "file-change", turnId, itemId, @@ -1862,7 +1901,7 @@ export const makeCodexSessionRuntime = ( }); yield* Ref.update(approvalCorrelationsRef, (current) => { const next = new Map(current); - next.set(payload.itemId, { + next.set(String(metadata.id), { requestId, requestKind: "file-change", turnId, @@ -1897,7 +1936,7 @@ export const makeCodexSessionRuntime = ( }), ); - yield* client.handleServerRequest("item/permissions/requestApproval", (payload) => + yield* client.handleServerRequest("item/permissions/requestApproval", (payload, metadata) => Effect.gen(function* () { const requestId = ApprovalRequestId.make(yield* randomUUIDv4); const turnId = TurnId.make(payload.turnId); @@ -1908,7 +1947,7 @@ export const makeCodexSessionRuntime = ( const next = new Map(current); next.set(requestId, { requestId, - jsonRpcId: payload.itemId, + jsonRpcId: String(metadata.id), requestKind: "permissions", turnId, itemId, @@ -1918,7 +1957,7 @@ export const makeCodexSessionRuntime = ( }); yield* Ref.update(approvalCorrelationsRef, (current) => { const next = new Map(current); - next.set(payload.itemId, { + next.set(String(metadata.id), { requestId, requestKind: "permissions", turnId, @@ -1951,7 +1990,7 @@ export const makeCodexSessionRuntime = ( }), ); - yield* client.handleServerRequest("item/tool/requestUserInput", (payload) => + yield* client.handleServerRequest("item/tool/requestUserInput", (payload, metadata) => Effect.gen(function* () { const requestId = ApprovalRequestId.make(yield* randomUUIDv4); const turnId = TurnId.make(payload.turnId); @@ -1962,6 +2001,7 @@ export const makeCodexSessionRuntime = ( const next = new Map(current); next.set(requestId, { requestId, + jsonRpcId: String(metadata.id), turnId, itemId, answers, @@ -1976,7 +2016,7 @@ export const makeCodexSessionRuntime = ( requestId, ...(turnId ? { turnId } : {}), ...(itemId ? { itemId } : {}), - payload, + payload: { ...payload, isBlocking: payload.isBlocking ?? true }, }); const resolvedAnswers = yield* Deferred.await(answers).pipe( @@ -2599,18 +2639,17 @@ export const makeCodexSessionRuntime = ( }), respondToUserInput: (requestId, answers) => Effect.gen(function* () { - const pending = (yield* Ref.get(pendingUserInputsRef)).get(requestId); + const codexAnswers = yield* toCodexUserInputAnswers(answers); + const pending = yield* Ref.modify(pendingUserInputsRef, (current) => { + const next = new Map(current); + next.delete(requestId); + return [current.get(requestId), next]; + }); if (!pending) { return yield* new CodexSessionRuntimePendingUserInputNotFoundError({ requestId, }); } - const codexAnswers = yield* toCodexUserInputAnswers(answers); - yield* Ref.update(pendingUserInputsRef, (current) => { - const next = new Map(current); - next.delete(requestId); - return next; - }); yield* Deferred.succeed(pending.answers, answers); yield* emitEvent({ kind: "notification", diff --git a/apps/web/src/components/ChatView.browser.tsx b/apps/web/src/components/ChatView.browser.tsx index 54422e33..83da9a01 100644 --- a/apps/web/src/components/ChatView.browser.tsx +++ b/apps/web/src/components/ChatView.browser.tsx @@ -1153,10 +1153,17 @@ function createSnapshotWithSecondaryProject(options?: { }; } -function createSnapshotWithPendingUserInput(): OrchestrationReadModel { +function createSnapshotWithPendingUserInput(options?: { + isBlocking?: boolean; + running?: boolean; + signIn?: boolean; +}): OrchestrationReadModel { const snapshot = createSnapshotForTargetUser({ targetMessageId: "msg-user-pending-input-target" as MessageId, - targetText: "question thread", + targetText: options?.signIn ? "Check that the VM is ready to use." : "question thread", + ...(options?.running + ? { sessionStatus: "running", sessionActiveTurnId: "turn-pending-input" as TurnId } + : {}), }); return { @@ -1165,6 +1172,25 @@ function createSnapshotWithPendingUserInput(): OrchestrationReadModel { thread.id === THREAD_ID ? Object.assign({}, thread, { interactionMode: "plan", + ...(options?.signIn + ? { + title: "Set up the VM", + messages: [ + createUserMessage({ + id: "msg-sign-in-user" as MessageId, + text: "Check that the VM is ready to use.", + offsetSeconds: 0, + }), + createAssistantMessage({ + id: "msg-sign-in-agent" as MessageId, + text: options.isBlocking + ? "The repository is ready. I need your sign-in result before I can check the provider connections." + : "The repository is ready. I will check the workspace files while you sign into the providers.", + offsetSeconds: 1, + }), + ], + } + : {}), activities: [ { id: EventId.make("activity-user-input-requested"), @@ -1173,38 +1199,62 @@ function createSnapshotWithPendingUserInput(): OrchestrationReadModel { summary: "User input requested", payload: { requestId: "req-browser-user-input", - questions: [ - { - id: "scope", - header: "Scope", - question: "What should this change cover?", - options: [ + isBlocking: options?.isBlocking ?? true, + questions: options?.signIn + ? [ { - label: "Tight", - description: "Touch only the footer layout logic.", + id: "sign_in", + header: "Provider sign-in", + question: + "Please sign into Codex and Claude using their buttons in the VM, then tell me when both are done.", + options: [ + { + label: "Both are signed in", + description: "Ready to verify the clone picker.", + }, + { + label: "Only one worked", + description: "One provider still needs help.", + }, + { + label: "I am still signing in", + description: "Keep the VM page where it is.", + }, + ], }, + ] + : [ { - label: "Broad", - description: "Also adjust the related composer controls.", + id: "scope", + header: "Scope", + question: "What should this change cover?", + options: [ + { + label: "Tight", + description: "Touch only the footer layout logic.", + }, + { + label: "Broad", + description: "Also adjust the related composer controls.", + }, + ], }, - ], - }, - { - id: "risk", - header: "Risk", - question: "How aggressive should the imaginary plan be?", - options: [ { - label: "Conservative", - description: "Favor reliability and low-risk changes.", - }, - { - label: "Balanced", - description: "Mix quick wins with one structural improvement.", + id: "risk", + header: "Risk", + question: "How aggressive should the imaginary plan be?", + options: [ + { + label: "Conservative", + description: "Favor reliability and low-risk changes.", + }, + { + label: "Balanced", + description: "Mix quick wins with one structural improvement.", + }, + ], }, ], - }, - ], }, turnId: null, sequence: 1, @@ -1788,6 +1838,13 @@ async function waitForButtonByText(text: string): Promise { return waitForElement(() => findButtonByText(text), `Unable to find "${text}" button.`); } +async function waitForButtonByAriaLabel(label: string): Promise { + return waitForElement( + () => document.querySelector(`button[aria-label="${label}"]`), + `Unable to find "${label}" button.`, + ); +} + /** The proposed-plan timeline card renders its own "Implement" button, so * composer-footer assertions must scope their lookup to the footer. */ async function waitForButtonByTextWithin( @@ -1873,6 +1930,37 @@ async function expectComposerActionsContained(): Promise { ); } +/** The question panel carries its own actions while the composer footer is + * hidden behind a blocking question, so containment is checked against the + * panel instead. */ +async function expectQuestionActionsContained(): Promise { + const panel = await waitForElement( + () => document.querySelector('[data-composer-questions-expanded="true"]'), + "Unable to find expanded question panel.", + ); + + await vi.waitFor( + () => { + const panelRect = panel.getBoundingClientRect(); + const actionButtons = Array.from(panel.querySelectorAll("button")).filter( + (button) => + /^(Previous|Next question|Submit answers?)$/.test(button.textContent?.trim() ?? ""), + ); + expect(actionButtons.length).toBeGreaterThanOrEqual(1); + + const buttonRects = actionButtons.map((button) => button.getBoundingClientRect()); + const firstTop = buttonRects[0]?.top ?? 0; + + for (const rect of buttonRects) { + expect(rect.right).toBeLessThanOrEqual(panelRect.right + 0.5); + expect(rect.bottom).toBeLessThanOrEqual(panelRect.bottom + 0.5); + expect(Math.abs(rect.top - firstTop)).toBeLessThanOrEqual(4.5); + } + }, + { timeout: 8_000, interval: 16 }, + ); +} + async function waitForInteractionModeButton( expectedLabel: "Build" | "Plan", ): Promise { @@ -5106,65 +5194,70 @@ describe("ChatView timeline estimator parity (full app)", () => { } }); - it("refreshes stale thread detail state after an accepted send has no projection ack", async () => { - const mounted = await mountChatView({ - viewport: DEFAULT_VIEWPORT, - snapshot: createSnapshotForTargetUser({ - targetMessageId: "msg-user-stale-send" as MessageId, - targetText: "stale send", - }), - resolveRpc: (body) => { - if (body._tag === ORCHESTRATION_WS_METHODS.dispatchCommand) { - return { - sequence: fixture.snapshot.snapshotSequence + 1, - }; - } - return undefined; - }, - }); + it.each([false, true])( + "refreshes stale thread detail after an accepted send with async question %s", + async (hasAsyncQuestion) => { + const mounted = await mountChatView({ + viewport: DEFAULT_VIEWPORT, + snapshot: hasAsyncQuestion + ? createSnapshotWithPendingUserInput({ isBlocking: false }) + : createSnapshotForTargetUser({ + targetMessageId: "msg-user-stale-send" as MessageId, + targetText: "stale send", + }), + resolveRpc: (body) => { + if (body._tag === ORCHESTRATION_WS_METHODS.dispatchCommand) { + return { + sequence: fixture.snapshot.snapshotSequence + 1, + }; + } + return undefined; + }, + }); - try { - await waitForThreadDetailSubscription(THREAD_ID); - const initialSubscriptionCount = wsRequests.filter( - (request) => - request._tag === ORCHESTRATION_WS_METHODS.subscribeThread && - request.threadId === THREAD_ID, - ).length; + try { + await waitForThreadDetailSubscription(THREAD_ID); + const initialSubscriptionCount = wsRequests.filter( + (request) => + request._tag === ORCHESTRATION_WS_METHODS.subscribeThread && + request.threadId === THREAD_ID, + ).length; - useComposerDraftStore.getState().setPrompt(THREAD_REF, "reconcile this send"); - const sendButton = await waitForSendButton(); - sendButton.click(); + useComposerDraftStore.getState().setPrompt(THREAD_REF, "reconcile this send"); + const sendButton = await waitForSendButton(); + sendButton.click(); - await vi.waitFor( - () => { - expect( - wsRequests.some( - (request) => - request._tag === ORCHESTRATION_WS_METHODS.dispatchCommand && - request.type === "thread.turn.start", - ), - ).toBe(true); - expect(document.querySelector('button[aria-label="Sending"]')).toBeTruthy(); - }, - { timeout: 8_000, interval: 16 }, - ); + await vi.waitFor( + () => { + expect( + wsRequests.some( + (request) => + request._tag === ORCHESTRATION_WS_METHODS.dispatchCommand && + request.type === "thread.turn.start", + ), + ).toBe(true); + expect(document.querySelector('button[aria-label="Sending"]')).toBeTruthy(); + }, + { timeout: 8_000, interval: 16 }, + ); - await vi.waitFor( - () => { - expect( - wsRequests.filter( - (request) => - request._tag === ORCHESTRATION_WS_METHODS.subscribeThread && - request.threadId === THREAD_ID, - ).length, - ).toBeGreaterThan(initialSubscriptionCount); - }, - { timeout: 8_000, interval: 16 }, - ); - } finally { - await mounted.cleanup(); - } - }); + await vi.waitFor( + () => { + expect( + wsRequests.filter( + (request) => + request._tag === ORCHESTRATION_WS_METHODS.subscribeThread && + request.threadId === THREAD_ID, + ).length, + ).toBeGreaterThan(initialSubscriptionCount); + }, + { timeout: 8_000, interval: 16 }, + ); + } finally { + await mounted.cleanup(); + } + }, + ); it("toggles plan mode with Shift+Tab only while the composer is focused", async () => { const mounted = await mountChatView({ @@ -9231,7 +9324,7 @@ describe("ChatView timeline estimator parity (full app)", () => { } }); - it("keeps pending-question footer actions inside the composer after a real resize", async () => { + it("keeps pending-question actions inside the question panel after a real resize", async () => { const mounted = await mountChatView({ viewport: WIDE_FOOTER_VIEWPORT, snapshot: createSnapshotWithPendingUserInput(), @@ -9240,12 +9333,14 @@ describe("ChatView timeline estimator parity (full app)", () => { try { const firstOption = await waitForButtonContainingText("Tight"); firstOption.click(); + (await waitForButtonByText("Next question")).click(); await waitForButtonByText("Previous"); await waitForButtonByText("Submit answers"); await mounted.setContainerSize(COMPACT_FOOTER_VIEWPORT); - await expectComposerActionsContained(); + await expectQuestionActionsContained(); + expect(document.querySelector('[data-chat-composer-footer="true"]')).toBeNull(); } finally { await mounted.cleanup(); } @@ -9439,7 +9534,7 @@ describe("ChatView timeline estimator parity (full app)", () => { } }); - it("submits pending user input after the final option selection resolves the draft answers", async () => { + it("submits pending user input only after explicitly reviewing the final answer", async () => { let resolveDispatch!: (value: { sequence: number }) => void; const pendingDispatch = new Promise<{ sequence: number }>((resolve) => { resolveDispatch = resolve; @@ -9458,9 +9553,18 @@ describe("ChatView timeline estimator parity (full app)", () => { try { const firstOption = await waitForButtonContainingText("Tight"); firstOption.click(); + (await waitForButtonByText("Next question")).click(); const finalOption = await waitForButtonContainingText("Conservative"); finalOption.click(); + expect( + wsRequests.some( + (request) => + request._tag === ORCHESTRATION_WS_METHODS.dispatchCommand && + request.type === "thread.user-input.respond", + ), + ).toBe(false); + (await waitForButtonByText("Submit answers")).click(); await vi.waitFor( () => { @@ -9505,6 +9609,210 @@ describe("ChatView timeline estimator parity (full app)", () => { } }); + it.each([ + { name: "async desktop", viewport: DEFAULT_VIEWPORT, isBlocking: false }, + { name: "async phone", viewport: PHONE_VIEWPORT, isBlocking: false }, + { name: "blocking desktop", viewport: DEFAULT_VIEWPORT, isBlocking: true }, + { name: "blocking phone", viewport: PHONE_VIEWPORT, isBlocking: true }, + ])( + "keeps answer typing separate from the Stop-to-Steer composer switch ($name)", + async ({ name, viewport, isBlocking }) => { + useComposerDraftStore + .getState() + .setPrompt(THREAD_REF, isBlocking ? "Keep checking the repository while I sign in." : ""); + const mounted = await mountChatView({ + viewport, + snapshot: createSnapshotWithPendingUserInput({ isBlocking, running: true, signIn: true }), + configureFixture: (nextFixture) => { + nextFixture.serverConfig = { + ...nextFixture.serverConfig, + environment: { + ...nextFixture.serverConfig.environment, + serverVersion: import.meta.env.APP_VERSION, + }, + }; + nextFixture.welcome = { + ...nextFixture.welcome, + environment: { + ...nextFixture.welcome.environment, + serverVersion: import.meta.env.APP_VERSION, + }, + }; + }, + }); + try { + document.documentElement.classList.add("dark"); + const stop = await waitForElement( + () => document.querySelector('button[aria-label="Stop generation"]'), + "Stop must remain available when only the answer field has text.", + ); + (await waitForButtonContainingText("Only one worked")).click(); + await page + .getByLabelText("Custom answer", { exact: true }) + .fill("Codex is ready. Claude still needs help.", { timeout: 5_000 }); + expect(useComposerDraftStore.getState().draftsByThreadKey[THREAD_KEY]?.prompt ?? "").toBe( + isBlocking ? "Keep checking the repository while I sign in." : "", + ); + // Plain DOM clicks: Playwright hit-tests the click point, and on phone + // the scroll-to-bottom pill can cover the panel header while the + // timeline is still settling to the end. + (await waitForButtonByAriaLabel("Collapse questions")).click(); + await expect.element(stop, { timeout: 5_000 }).toBeVisible(); + expect( + wsRequests.some( + (request) => + request._tag === ORCHESTRATION_WS_METHODS.dispatchCommand && + request.type === "thread.user-input.respond", + ), + ).toBe(false); + (await waitForButtonByAriaLabel("Expand questions")).click(); + await expect + .element(page.getByLabelText("Custom answer", { exact: true }), { timeout: 5_000 }) + .toHaveValue("Codex is ready. Claude still needs help."); + if (!isBlocking) { + const editor = page.getByTestId("composer-editor"); + await editor.fill("Keep checking the repository while I sign in.", { timeout: 5_000 }); + await waitForComposerText("Keep checking the repository while I sign in."); + await waitForButtonByAriaLabel("Steer active turn"); + expect(document.querySelector('button[aria-label="Stop generation"]')).toBeNull(); + // Clear through the draft store. Playwright's empty fill presses Delete + // over a select-all, which the editor does not reliably turn into an + // empty draft, and the point here is the empty-draft state, not the key. + useComposerDraftStore.getState().setPrompt(THREAD_REF, ""); + await waitForButtonByAriaLabel("Stop generation"); + expect(document.querySelector('button[aria-label="Steer active turn"]')).toBeNull(); + await editor.fill("Keep checking the repository while I sign in.", { timeout: 5_000 }); + await waitForComposerText("Keep checking the repository while I sign in."); + await waitForButtonByAriaLabel("Steer active turn"); + } + await expectQuestionActionsContained(); + if (isBlocking) { + // The blocked row keeps the saved draft readable and Stop inside it. + const blockedRow = await waitForElement( + () => + document.querySelector( + '[data-chat-composer-blocked-by-question="true"]', + ), + "Unable to find the blocked composer row.", + ); + expect(blockedRow.textContent).toContain("Keep checking the repository while I sign in."); + expect(blockedRow.contains(stop)).toBe(true); + expect(document.querySelector('[data-chat-composer-footer="true"]')).toBeNull(); + // The draft check above is the regression guard; the screenshot shows + // the common case of an empty composer. + useComposerDraftStore.getState().setPrompt(THREAD_REF, ""); + await vi.waitFor(() => { + expect(blockedRow.textContent).toContain("Answer the question above, or stop the turn"); + }); + } else { + await expectComposerActionsContained(); + } + await page.screenshot({ + path: `__screenshots__/question-${name.replaceAll(" ", "-")}.png`, + }); + if (isBlocking) { + expect(document.querySelector('[contenteditable="true"]')).toBeNull(); + stop.click(); + await vi.waitFor(() => { + expect( + wsRequests.find( + (request) => + request._tag === ORCHESTRATION_WS_METHODS.dispatchCommand && + request.type === "thread.turn.interrupt", + ), + ).toMatchObject({ turnId: "turn-pending-input" }); + }); + } else { + expect(document.body.textContent).toContain("The agent keeps working while you answer."); + await expect + .element(await waitForComposerEditor(), { timeout: 5_000 }) + .toHaveTextContent("Keep checking the repository while I sign in."); + (await waitForButtonByText("Submit answer")).click(); + await vi.waitFor(() => { + expect( + wsRequests.find( + (request) => + request._tag === ORCHESTRATION_WS_METHODS.dispatchCommand && + request.type === "thread.user-input.respond", + ), + ).toMatchObject({ answers: { sign_in: "Codex is ready. Claude still needs help." } }); + }); + await page.getByLabelText("Steer active turn").click({ timeout: 5_000 }); + await vi.waitFor(() => { + expect( + wsRequests.find( + (request) => + request._tag === ORCHESTRATION_WS_METHODS.dispatchCommand && + request.type === "thread.follow-up.submit", + ), + ).toMatchObject({ message: { text: "Keep checking the repository while I sign in." } }); + }); + } + } finally { + document.documentElement.classList.remove("dark"); + await mounted.cleanup(); + } + }, + ); + + it.each([ + { name: "desktop", viewport: DEFAULT_VIEWPORT }, + { name: "phone", viewport: PHONE_VIEWPORT }, + ])( + "keeps Stop reachable when an approval overlaps an async question ($name)", + async ({ viewport }) => { + const snapshot = createSnapshotWithPendingUserInput({ isBlocking: false, running: true }); + const mounted = await mountChatView({ + viewport, + snapshot: { + ...snapshot, + threads: snapshot.threads.map((thread) => + thread.id !== THREAD_ID + ? thread + : { + ...thread, + activities: [ + ...thread.activities, + { + id: EventId.make("activity-overlapping-approval"), + kind: "approval.requested", + tone: "info", + summary: "Approval requested", + payload: { + requestId: "req-overlapping-approval", + requestKind: "command", + detail: "git status", + }, + turnId: "turn-pending-input" as TurnId, + sequence: 2, + createdAt: isoAt(1_001), + }, + ], + }, + ), + }, + }); + try { + await expect.element(page.getByText("Command approval requested")).toBeVisible(); + const stop = page.getByLabelText("Stop generation"); + await expect.element(stop).toBeVisible(); + expect(document.querySelectorAll('button[aria-label="Stop generation"]')).toHaveLength(1); + await stop.click(); + await vi.waitFor(() => { + expect( + wsRequests.find( + (request) => + request._tag === ORCHESTRATION_WS_METHODS.dispatchCommand && + request.type === "thread.turn.interrupt", + ), + ).toMatchObject({ turnId: "turn-pending-input" }); + }); + } finally { + await mounted.cleanup(); + } + }, + ); + it("keeps plan follow-up footer actions fused and aligned after a real resize", async () => { const mounted = await mountChatView({ viewport: WIDE_FOOTER_VIEWPORT, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index ccd29cb0..3a868288 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -1967,9 +1967,7 @@ export default function ChatView(props: ChatViewProps) { : null, [activePendingDraftAnswers, activePendingUserInput], ); - const activePendingIsResponding = activePendingUserInput - ? respondingUserInputRequestIds.includes(activePendingUserInput.requestId) - : false; + const activeProposedPlan = useMemo(() => { if (!latestTurnSettled) { return null; @@ -2047,7 +2045,7 @@ export default function ChatView(props: ChatViewProps) { activeLatestTurn, phase, activePendingApproval: activePendingApproval?.requestId ?? null, - activePendingUserInput: activePendingUserInput?.requestId ?? null, + activePendingUserInput: pendingUserInputs.find(isBlockingUserInput)?.requestId ?? null, threadError: activeThread?.error, }); // Raised for the whole bootstrap request; dropped as soon as the thread @@ -4608,10 +4606,7 @@ export default function ChatView(props: ChatViewProps) { sendInFlightRef.current ) return; - if (activePendingProgress) { - onAdvanceActivePendingUserInput(); - return; - } + if (pendingUserInputs.some(isBlockingUserInput)) return; const sendCtx = composerRef.current?.getSendContext(); if (!sendCtx) return; const { @@ -5693,24 +5688,15 @@ export default function ChatView(props: ChatViewProps) { }, }; }); - promptRef.current = ""; - composerRef.current?.resetCursorState({ cursor: 0 }); }, [activePendingProgress?.activeQuestion, activePendingUserInput], ); const onChangeActivePendingUserInputCustomAnswer = useCallback( - ( - questionId: string, - value: string, - nextCursor: number, - expandedCursor: number, - _cursorAdjacentToMention: boolean, - ) => { + (questionId: string, value: string) => { if (!activePendingUserInput) { return; } - promptRef.current = value; setPendingUserInputAnswersByRequestId((existing) => ({ ...existing, [activePendingUserInput.requestId]: { @@ -5721,14 +5707,6 @@ export default function ChatView(props: ChatViewProps) { ), }, })); - const snapshot = composerRef.current?.readSnapshot(); - if ( - snapshot?.value !== value || - snapshot.cursor !== nextCursor || - snapshot.expandedCursor !== expandedCursor - ) { - composerRef.current?.focusAt(nextCursor); - } }, [activePendingUserInput], ); @@ -6856,9 +6834,6 @@ export default function ChatView(props: ChatViewProps) { activePendingApproval={activePendingApproval} pendingApprovals={pendingApprovals} pendingUserInputs={pendingUserInputs} - activePendingProgress={activePendingProgress} - activePendingResolvedAnswers={activePendingResolvedAnswers} - activePendingIsResponding={activePendingIsResponding} activePendingDraftAnswers={activePendingDraftAnswers} activePendingQuestionIndex={activePendingQuestionIndex} respondingRequestIds={respondingRequestIds} diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index f7643de2..f53955b2 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -491,6 +491,18 @@ describe("resolveThreadStatusPill", () => { }); }); + it("keeps working visible while a non-blocking question is open", () => { + expect( + resolveThreadStatusPill({ + thread: { + ...baseThread, + hasPendingUserInput: true, + hasBlockingUserInput: false, + }, + }), + ).toMatchObject({ label: "Working", pulse: true }); + }); + it("shows working from the orchestration running status even if the legacy status lags", () => { expect( resolveThreadStatusPill({ diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index c25b17d0..6c10e3bd 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -54,6 +54,7 @@ type ThreadStatusInput = Pick< | "hasActionableProposedPlan" | "hasPendingApprovals" | "hasPendingUserInput" + | "hasBlockingUserInput" | "interactionMode" | "latestTurn" | "session" @@ -302,7 +303,7 @@ export function resolveThreadStatusPill(input: { }; } - if (thread.hasPendingUserInput) { + if (thread.hasBlockingUserInput ?? thread.hasPendingUserInput) { return { label: "Awaiting Input", colorClass: "text-amber-600 dark:text-amber-300/90", diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index abcf9fbb..24a68db0 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -110,7 +110,7 @@ import { ComposerPendingApprovalActions } from "./ComposerPendingApprovalActions import { CompactComposerControlsMenu } from "./CompactComposerControlsMenu"; import { ComposerAttachmentMenu } from "./ComposerAttachmentMenu"; import { ComposerStashControl } from "./ComposerStashControl"; -import { ComposerPrimaryActions } from "./ComposerPrimaryActions"; +import { ComposerPrimaryActions, ComposerStopButton } from "./ComposerPrimaryActions"; import { ComposerPendingApprovalPanel } from "./ComposerPendingApprovalPanel"; import { ComposerPendingUserInputPanel } from "./ComposerPendingUserInputPanel"; import { ComposerGoalBar, type ComposerGoalSetInput } from "./ComposerGoalBar"; @@ -381,13 +381,6 @@ const ComposerFooterPrimaryActions = memo(function ComposerFooterPrimaryActions( providerAccountUsage: ProviderAccountUsagePresentation | null; contextWindowLabel: string | null; isPreparingWorktree: boolean; - pendingAction: { - questionIndex: number; - isLastQuestion: boolean; - canAdvance: boolean; - isResponding: boolean; - isComplete: boolean; - } | null; isRunning: boolean; showPlanFollowUpPrompt: boolean; promptHasText: boolean; @@ -399,7 +392,6 @@ const ComposerFooterPrimaryActions = memo(function ComposerFooterPrimaryActions( runtimeMode: RuntimeMode; runtimeModeOptions: ReadonlyArray; onRuntimeModeChange: (mode: RuntimeMode) => void; - onPreviousPendingQuestion: () => void; onInterrupt: () => void; onImplementPlanInNewThread: () => void; onResetAccountUsage?: (() => void) | undefined; @@ -427,7 +419,6 @@ const ComposerFooterPrimaryActions = memo(function ComposerFooterPrimaryActions( ) : null} @@ -537,15 +527,6 @@ export interface ChatComposerProps { activePendingApproval: PendingApproval | null; pendingApprovals: PendingApproval[]; pendingUserInputs: PendingUserInput[]; - activePendingProgress: { - questionIndex: number; - isLastQuestion: boolean; - canAdvance: boolean; - customAnswer: string; - activeQuestion: { id: string; multiSelect?: boolean | undefined } | null; - } | null; - activePendingResolvedAnswers: Record | null; - activePendingIsResponding: boolean; activePendingDraftAnswers: Record; activePendingQuestionIndex: number; respondingRequestIds: ApprovalRequestId[]; @@ -622,13 +603,7 @@ export interface ChatComposerProps { onSelectActivePendingUserInputOption: (questionId: string, optionLabel: string) => void; onAdvanceActivePendingUserInput: () => void; onPreviousActivePendingUserInputQuestion: () => void; - onChangeActivePendingUserInputCustomAnswer: ( - questionId: string, - value: string, - nextCursor: number, - expandedCursor: number, - cursorAdjacentToMention: boolean, - ) => void; + onChangeActivePendingUserInputCustomAnswer: (questionId: string, value: string) => void; onProviderModelSelect: (instanceId: ProviderInstanceId, model: string) => void; toggleInteractionMode: () => void; @@ -665,9 +640,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) activePendingApproval, pendingApprovals, pendingUserInputs, - activePendingProgress, - activePendingResolvedAnswers, - activePendingIsResponding, activePendingDraftAnswers, activePendingQuestionIndex, respondingRequestIds, @@ -1383,7 +1355,14 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) ); const isComposerApprovalState = activePendingApproval !== null; - const activePendingUserInput = pendingUserInputs[0] ?? null; + const hasBlockingQuestion = pendingUserInputs.some((input) => input.isBlocking !== false); + // A blocking question owns the composer: the editor and toolbar give way to a + // thin row that keeps the saved draft visible and Stop reachable. + const isComposerBlockedByQuestion = hasBlockingQuestion && !isComposerApprovalState; + const hasActiveTurn = + phase === "running" || + activeThread?.session?.activeTurnId != null || + activeThread?.latestTurn?.state === "running"; const hasComposerHeader = isComposerApprovalState || pendingUserInputs.length > 0 || @@ -1425,11 +1404,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) ? formatPromptSuggestionDisplayText(latestPromptSuggestion) : null; - const composerFooterHasWideActions = showPlanFollowUpPrompt || activePendingProgress !== null; + const composerFooterHasWideActions = showPlanFollowUpPrompt; const composerFooterActionLayoutKey = useMemo(() => { - if (activePendingProgress) { - return `pending:${activePendingProgress.questionIndex}:${activePendingProgress.isLastQuestion}:${activePendingIsResponding}`; - } if (phase === "running") { return `running:${composerSendState.hasSendableContent}:${isSendBusy}:${isConnecting}`; } @@ -1438,8 +1414,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) } return `idle:${composerSendState.hasSendableContent}:${isSendBusy}:${isConnecting}:${isPreparingWorktree}`; }, [ - activePendingIsResponding, - activePendingProgress, composerSendState.hasSendableContent, isConnecting, isPreparingWorktree, @@ -1513,31 +1487,16 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) modelOptions: composerModelOptions?.[selectedInstanceId], iconOnly: composerFooterTier !== "full", }); - const pendingPrimaryAction = useMemo( - () => - activePendingProgress - ? { - questionIndex: activePendingProgress.questionIndex, - isLastQuestion: activePendingProgress.isLastQuestion, - canAdvance: activePendingProgress.canAdvance, - isResponding: activePendingIsResponding, - isComplete: Boolean(activePendingResolvedAnswers), - } - : null, - [activePendingIsResponding, activePendingProgress, activePendingResolvedAnswers], - ); const collapsedComposerPrimaryActionDisabled = isSendBusy || isConnecting || !composerSendState.hasSendableContent; const collapsedComposerPrimaryActionLabel = phase === "running" ? "Steer active turn" : "Send message"; - const showMobilePendingAnswerActions = - isMobileViewport && !isComposerCollapsedMobile && pendingPrimaryAction !== null; // Shared gate for every "Add" action (upload + screenshot). The in-flight // capture only blocks the screenshot item, not uploading images, so it is // handled inside the menu rather than here. Models without image input // still accept document attachments, so modality no longer disables the // whole menu — image files are rejected per-file during ingest instead. - const attachmentsDisabled = isComposerApprovalState || pendingUserInputs.length > 0; + const attachmentsDisabled = isComposerApprovalState || hasBlockingQuestion; const attachmentsDisabledReason = attachmentsDisabled ? "Finish the pending prompt before adding attachments" : null; @@ -2120,51 +2079,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) composerMenuSearchKey, ]); - const lastSyncedPendingInputRef = useRef<{ - requestId: string | null; - questionId: string | null; - } | null>(null); - - useEffect(() => { - const nextCustomAnswer = activePendingProgress?.customAnswer; - if (typeof nextCustomAnswer !== "string") { - lastSyncedPendingInputRef.current = null; - return; - } - - const nextRequestId = activePendingUserInput?.requestId ?? null; - const nextQuestionId = activePendingProgress?.activeQuestion?.id ?? null; - const questionChanged = - lastSyncedPendingInputRef.current?.requestId !== nextRequestId || - lastSyncedPendingInputRef.current?.questionId !== nextQuestionId; - const textChangedExternally = promptRef.current !== nextCustomAnswer; - - lastSyncedPendingInputRef.current = { - requestId: nextRequestId, - questionId: nextQuestionId, - }; - - if (!questionChanged && !textChangedExternally) { - return; - } - - promptRef.current = nextCustomAnswer; - const nextCursor = collapseExpandedComposerCursor(nextCustomAnswer, nextCustomAnswer.length); - setComposerCursor(nextCursor); - setComposerTrigger( - detectComposerTrigger( - nextCustomAnswer, - expandCollapsedComposerCursor(nextCustomAnswer, nextCursor), - ), - ); - setComposerHighlightedItemId(null); - }, [ - activePendingProgress?.customAnswer, - activePendingProgress?.activeQuestion?.id, - activePendingUserInput?.requestId, - promptRef, - ]); - // ------------------------------------------------------------------ // Reset compositor state on thread/draft change // ------------------------------------------------------------------ @@ -2338,20 +2252,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) cursorAdjacentToMention: boolean, terminalContextIds: string[], ) => { - if (activePendingProgress?.activeQuestion && pendingUserInputs.length > 0) { - setComposerCursor(nextCursor); - setComposerTrigger( - cursorAdjacentToMention ? null : detectComposerTrigger(nextPrompt, expandedCursor), - ); - onChangeActivePendingUserInputCustomAnswer( - activePendingProgress.activeQuestion.id, - nextPrompt, - nextCursor, - expandedCursor, - cursorAdjacentToMention, - ); - return; - } const previousPrompt = promptRef.current; promptRef.current = nextPrompt; setPrompt(nextPrompt); @@ -2374,9 +2274,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) ); }, [ - activePendingProgress?.activeQuestion, - pendingUserInputs.length, - onChangeActivePendingUserInputCustomAnswer, promptRef, setPrompt, composerDraftTarget, @@ -2408,18 +2305,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const nextCursor = collapseExpandedComposerCursor(next.text, next.cursor); const nextExpandedCursor = expandCollapsedComposerCursor(next.text, nextCursor); promptRef.current = next.text; - const activePendingQuestion = activePendingProgress?.activeQuestion; - if (activePendingQuestion && activePendingUserInput) { - onChangeActivePendingUserInputCustomAnswer( - activePendingQuestion.id, - next.text, - nextCursor, - nextExpandedCursor, - false, - ); - } else { - setPrompt(next.text); - } + setPrompt(next.text); setComposerCursor(nextCursor); setComposerTrigger(detectComposerTrigger(next.text, nextExpandedCursor)); if (options?.focusEditorAfterReplace !== false) { @@ -2429,13 +2315,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) } return true; }, - [ - activePendingProgress?.activeQuestion, - activePendingUserInput, - onChangeActivePendingUserInputCustomAnswer, - promptRef, - setPrompt, - ], + [promptRef, setPrompt], ); const readComposerSnapshot = useCallback((): { @@ -2607,13 +2487,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const shouldBlurMobileComposerOnSubmit = useCallback(() => { if (!isMobileViewport) return false; if (isSendBusy || isConnecting || phase === "running") return false; - if (activePendingProgress) { - return activePendingProgress.isLastQuestion && Boolean(activePendingResolvedAnswers); - } return showPlanFollowUpPrompt || composerSendState.hasSendableContent; }, [ - activePendingProgress, - activePendingResolvedAnswers, composerSendState.hasSendableContent, isConnecting, isMobileViewport, @@ -2624,6 +2499,10 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const submitComposer = useCallback( (event?: { preventDefault: () => void }) => { + if (hasBlockingQuestion) { + event?.preventDefault(); + return; + } if (composerAttachments.length > 0 && !selectedModelSupportsImages) { event?.preventDefault(); toastManager.add({ @@ -2645,6 +2524,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) }, [ activeThread?.latestTurn, + hasBlockingQuestion, blurMobileComposerAfterSend, composerAttachments.length, composerSendState.hasSendableContent, @@ -2719,7 +2599,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) * confirm the attachment rather than the attempt. */ const addComposerFiles = (files: File[]): boolean => { if (!activeThreadId || files.length === 0) return false; - if (pendingUserInputs.length > 0) { + if (hasBlockingQuestion) { toastManager.add({ type: "error", title: "Attach files after answering plan questions.", @@ -3249,6 +3129,10 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) isTimelineScrolledAway={isTimelineScrolledAway} onToggleOption={onSelectActivePendingUserInputOption} onAdvance={onAdvanceActivePendingUserInput} + onPrevious={onPreviousActivePendingUserInputQuestion} + onCustomAnswerChange={onChangeActivePendingUserInputCustomAnswer} + isUnavailable={environmentUnavailable !== null} + isAgentRunning={hasActiveTurn} /> ) : showPlanFollowUpPrompt && activeProposedPlan ? ( @@ -3275,67 +3159,12 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) isResponding={respondingRequestIds.includes(activePendingApproval.requestId)} onRespondToApproval={onRespondToApproval} /> - - - ) : isComposerCollapsedMobile && pendingUserInputs.length > 0 ? ( -
- -
-
- - {activePendingProgress?.activeQuestion?.multiSelect ? ( - - ) : null} -
+ {hasActiveTurn ? ( + + ) : null}
) : null} @@ -3346,18 +3175,13 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) type="button" className={cn( "min-w-0 flex-1 truncate bg-transparent p-0 text-left text-[14px] focus:outline-none", - (activePendingProgress ? activePendingProgress.customAnswer : prompt.trim()) - ? "text-foreground" - : "text-muted-foreground/35", + prompt.trim() ? "text-foreground" : "text-muted-foreground/35", )} onPointerDown={(event) => event.preventDefault()} onClick={expandMobileComposer} aria-label="Expand composer" > - {activePendingProgress - ? activePendingProgress.customAnswer || - "Type your own answer, or leave this blank to use the selected option" - : prompt.trim() || "Ask anything..."} + {prompt.trim() || "Ask anything..."} - ) : ( - - )} - {nonPersistedComposerImageIdSet.has(attachment.id) && ( - - - + {attachment.previewUrl ? ( + + ) : ( + + + + + ))} - ))} - - )} + )} - {!isComposerCollapsedMobile && - !isComposerApprovalState && - pendingUserInputs.length === 0 && - composerTerminalContexts.length > 0 && ( - - )} + {!isComposerCollapsedMobile && + !isComposerApprovalState && + !hasBlockingQuestion && + composerTerminalContexts.length > 0 && ( + + )} - {!isComposerCollapsedMobile && - !isComposerApprovalState && - pendingUserInputs.length === 0 && - composerPickedElementContexts.length > 0 && ( - - )} + {!isComposerCollapsedMobile && + !isComposerApprovalState && + !hasBlockingQuestion && + composerPickedElementContexts.length > 0 && ( + + )} - {!isComposerCollapsedMobile && - !isComposerApprovalState && - pendingUserInputs.length === 0 && - composerDrawingContexts.length > 0 && ( - - )} + {!isComposerCollapsedMobile && + !isComposerApprovalState && + !hasBlockingQuestion && + composerDrawingContexts.length > 0 && ( + + )} - {!isComposerCollapsedMobile && - !isComposerApprovalState && - pendingUserInputs.length === 0 && - composerTranscriptHighlightContexts.length > 0 && ( - - )} + {!isComposerCollapsedMobile && + !isComposerApprovalState && + !hasBlockingQuestion && + composerTranscriptHighlightContexts.length > 0 && ( + + )} - {!isComposerCollapsedMobile && - !isComposerApprovalState && - pendingUserInputs.length === 0 && - composerFileSelectionContexts.length > 0 && ( - - )} + {!isComposerCollapsedMobile && + !isComposerApprovalState && + !hasBlockingQuestion && + composerFileSelectionContexts.length > 0 && ( + + )} -
- - {showMobilePendingAnswerActions ? ( -
- +
- ) : null} -
- - - {/* Bottom toolbar */} - {isComposerCollapsedMobile ? null : activePendingApproval ? ( -
- -
- ) : ( -
-
- { - setIsComposerModelPickerOpen(open); - }} - onInstanceModelChange={onProviderModelSelect} - /> - {activeModelFallback && activeFallbackModelDisplayName ? ( - - - } - > - - - {activeModelFallback.detail ?? - `Using ${activeModelFallback.activeModel} instead of ${activeModelFallback.requestedModel}.`} - - - ) : null} +
- {isComposerFooterCompact ? ( - + - ) : ( - <> - {providerTraitsPicker ? ( + {hasActiveTurn ? ( + + ) : null} +
+ ) : ( +
+
+ { + setIsComposerModelPickerOpen(open); + }} + onInstanceModelChange={onProviderModelSelect} + /> + {activeModelFallback && activeFallbackModelDisplayName ? ( + + + } + > + + + {activeModelFallback.detail ?? + `Using ${activeModelFallback.activeModel} instead of ${activeModelFallback.requestedModel}.`} + + + ) : null} + + {isComposerFooterCompact ? ( + + ) : ( <> - - {providerTraitsPicker} + {providerTraitsPicker ? ( + <> + + {providerTraitsPicker} + + ) : null} + - ) : null} - + + {/* Right side: stash + add attachments + send / stop button */} +
+ + + + + {voiceControl ? : null} + 0} + isSendBusy={isSendBusy} + isConnecting={isConnecting} + isEnvironmentUnavailable={environmentUnavailable !== null} + isPreparingWorktree={isPreparingWorktree} + hasSendableContent={ + !hasBlockingQuestion && composerSendState.hasSendableContent + } + preserveComposerFocusOnPointerDown={isMobileViewport} runtimeMode={runtimeMode} runtimeModeOptions={composerProviderControls.runtimeModeOptions} - onToggleInteractionMode={toggleInteractionMode} onRuntimeModeChange={handleRuntimeModeChange} + onInterrupt={handleInterruptPrimaryAction} + onImplementPlanInNewThread={handleImplementPlanInNewThreadPrimaryAction} + onResetAccountUsage={ + canResetSelectedProviderUsage + ? requestSelectedProviderUsageReset + : undefined + } + accountUsageResetInFlight={isConsumingRateLimitResetCredit} + onCompactContext={onCompactContext} + contextCompactDisabled={contextCompactDisabled} + contextCompactInFlight={contextCompactInFlight} + contextCompactDisabledReason={contextCompactDisabledReason} /> - - )} -
- - {/* Right side: stash + add attachments + send / stop button */} -
- - - - - {voiceControl ? : null} - 0} - isSendBusy={isSendBusy} - isConnecting={isConnecting} - isEnvironmentUnavailable={environmentUnavailable !== null} - isPreparingWorktree={isPreparingWorktree} - hasSendableContent={composerSendState.hasSendableContent} - preserveComposerFocusOnPointerDown={isMobileViewport} - runtimeMode={runtimeMode} - runtimeModeOptions={composerProviderControls.runtimeModeOptions} - onRuntimeModeChange={handleRuntimeModeChange} - onPreviousPendingQuestion={onPreviousActivePendingUserInputQuestion} - onInterrupt={handleInterruptPrimaryAction} - onImplementPlanInNewThread={handleImplementPlanInNewThreadPrimaryAction} - onResetAccountUsage={ - canResetSelectedProviderUsage ? requestSelectedProviderUsageReset : undefined - } - accountUsageResetInFlight={isConsumingRateLimitResetCredit} - onCompactContext={onCompactContext} - contextCompactDisabled={contextCompactDisabled} - contextCompactInFlight={contextCompactInFlight} - contextCompactDisabledReason={contextCompactDisabledReason} - /> -
-
+
+ + )} + )} diff --git a/apps/web/src/components/chat/ComposerPendingUserInputPanel.browser.tsx b/apps/web/src/components/chat/ComposerPendingUserInputPanel.browser.tsx index 48b8e0e7..bb3c71f0 100644 --- a/apps/web/src/components/chat/ComposerPendingUserInputPanel.browser.tsx +++ b/apps/web/src/components/chat/ComposerPendingUserInputPanel.browser.tsx @@ -38,20 +38,43 @@ describe("ComposerPendingUserInputPanel", () => { respondingRequestIds={[]} answers={{}} questionIndex={0} + isAgentRunning onToggleOption={vi.fn()} + onPrevious={vi.fn()} + onCustomAnswerChange={vi.fn()} onAdvance={vi.fn()} />, ); await expect.element(page.getByText("The agent keeps working while you answer.")).toBeVisible(); - // A blocking question (the default) says nothing extra: the turn is paused on it. + screen.rerender( + , + ); + await expect + .element(page.getByText("The agent finished. Your answer starts a follow-up.")) + .toBeVisible(); + + // Blocking questions carry no hint; the transcript already says the turn is waiting. screen.rerender( , ); @@ -67,7 +90,10 @@ describe("ComposerPendingUserInputPanel", () => { respondingRequestIds={[]} answers={{}} questionIndex={0} + isAgentRunning onToggleOption={onToggleOption} + onPrevious={vi.fn()} + onCustomAnswerChange={vi.fn()} onAdvance={vi.fn()} />, ); @@ -77,6 +103,7 @@ describe("ComposerPendingUserInputPanel", () => { .toBeVisible(); await page.getByLabelText("Collapse questions").click(); + await expect.element(page.getByLabelText("Expand questions")).toHaveFocus(); expect(screen.container.textContent).not.toContain("Split + pinned default"); await expect @@ -88,6 +115,7 @@ describe("ComposerPendingUserInputPanel", () => { expect(onToggleOption).not.toHaveBeenCalled(); await page.getByLabelText("Expand questions").click(); + await expect.element(page.getByLabelText("Collapse questions")).toHaveFocus(); await expect .element(page.getByRole("button", { name: /Split \+ pinned default/ })) .toBeVisible(); @@ -104,8 +132,11 @@ describe("ComposerPendingUserInputPanel", () => { respondingRequestIds: [], answers: {}, questionIndex: 0, + isAgentRunning: true, onToggleOption: vi.fn(), onAdvance: vi.fn(), + onPrevious: vi.fn(), + onCustomAnswerChange: vi.fn(), }; const screen = await render( , diff --git a/apps/web/src/components/chat/ComposerPendingUserInputPanel.tsx b/apps/web/src/components/chat/ComposerPendingUserInputPanel.tsx index e0b155ca..060212b7 100644 --- a/apps/web/src/components/chat/ComposerPendingUserInputPanel.tsx +++ b/apps/web/src/components/chat/ComposerPendingUserInputPanel.tsx @@ -1,12 +1,14 @@ import { type ApprovalRequestId } from "@threadlines/contracts"; -import { memo, useEffect, useEffectEvent, useRef, useState } from "react"; +import { memo, useEffect, useEffectEvent, useLayoutEffect, useRef, useState } from "react"; import { type PendingUserInput } from "../../session-logic"; import { derivePendingUserInputProgress, type PendingUserInputDraftAnswer, } from "../../pendingUserInput"; -import { CheckIcon, ChevronsDownUpIcon, ChevronsUpDownIcon } from "lucide-react"; +import { CheckIcon, ChevronsDownUpIcon, ChevronsUpDownIcon, PencilLineIcon } from "lucide-react"; import { cn } from "~/lib/utils"; +import { Button } from "../ui/button"; +import { formatPendingPrimaryActionLabel } from "./ComposerPrimaryActions"; interface PendingUserInputPanelProps { pendingUserInputs: PendingUserInput[]; @@ -17,6 +19,10 @@ interface PendingUserInputPanelProps { isTimelineScrolledAway?: boolean; onToggleOption: (questionId: string, optionLabel: string) => void; onAdvance: () => void; + onPrevious: () => void; + onCustomAnswerChange: (questionId: string, value: string) => void; + isUnavailable?: boolean; + isAgentRunning: boolean; } export const ComposerPendingUserInputPanel = memo(function ComposerPendingUserInputPanel({ @@ -27,6 +33,10 @@ export const ComposerPendingUserInputPanel = memo(function ComposerPendingUserIn isTimelineScrolledAway = false, onToggleOption, onAdvance, + onPrevious, + onCustomAnswerChange, + isUnavailable = false, + isAgentRunning, }: PendingUserInputPanelProps) { if (pendingUserInputs.length === 0) return null; const activePrompt = pendingUserInputs[0]; @@ -42,6 +52,10 @@ export const ComposerPendingUserInputPanel = memo(function ComposerPendingUserIn isTimelineScrolledAway={isTimelineScrolledAway} onToggleOption={onToggleOption} onAdvance={onAdvance} + onPrevious={onPrevious} + onCustomAnswerChange={onCustomAnswerChange} + isUnavailable={isUnavailable} + isAgentRunning={isAgentRunning} /> ); }); @@ -54,6 +68,10 @@ const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard( isTimelineScrolledAway, onToggleOption, onAdvance, + onPrevious, + onCustomAnswerChange, + isUnavailable, + isAgentRunning, }: { prompt: PendingUserInput; isResponding: boolean; @@ -62,6 +80,10 @@ const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard( isTimelineScrolledAway: boolean; onToggleOption: (questionId: string, optionLabel: string) => void; onAdvance: () => void; + onPrevious: () => void; + onCustomAnswerChange: (questionId: string, value: string) => void; + isUnavailable: boolean; + isAgentRunning: boolean; }) { const progress = derivePendingUserInputProgress(prompt.questions, answers, questionIndex); const activeQuestion = progress.activeQuestion; @@ -70,42 +92,27 @@ const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard( // toggle wins until the next scroll boundary change. State lives in this card // (keyed by requestId), so each prompt re-derives from the current scroll state. const [isCollapsed, setIsCollapsed] = useState(isTimelineScrolledAway); + const collapseButtonRef = useRef(null); + const restoreToggleFocusRef = useRef(false); + const toggleCollapsed = () => { + restoreToggleFocusRef.current = true; + setIsCollapsed((collapsed) => !collapsed); + }; + useLayoutEffect(() => { + if (restoreToggleFocusRef.current) { + collapseButtonRef.current?.focus({ preventScroll: true }); + restoreToggleFocusRef.current = false; + } + }, [isCollapsed]); useEffect(() => { setIsCollapsed(isTimelineScrolledAway); }, [isTimelineScrolledAway]); - const autoAdvanceTimerRef = useRef(null); - const onAdvanceRef = useRef(onAdvance); - - useEffect(() => { - onAdvanceRef.current = onAdvance; - }, [onAdvance]); - - // Clear auto-advance timer on unmount - useEffect(() => { - return () => { - if (autoAdvanceTimerRef.current !== null) { - window.clearTimeout(autoAdvanceTimerRef.current); - } - }; - }, []); - const handleOptionSelection = useEffectEvent((questionId: string, optionLabel: string) => { onToggleOption(questionId, optionLabel); - if (activeQuestion?.multiSelect) { - return; - } - if (autoAdvanceTimerRef.current !== null) { - window.clearTimeout(autoAdvanceTimerRef.current); - } - autoAdvanceTimerRef.current = window.setTimeout(() => { - autoAdvanceTimerRef.current = null; - onAdvanceRef.current(); - }, 200); }); // Keyboard shortcut: number keys 1-9 select corresponding options when focus is - // outside editable fields. Multi-select prompts toggle options in place; single- - // select prompts keep the existing auto-advance behavior. + // outside editable fields. Answers are only sent by the explicit submit action. useEffect(() => { if (!activeQuestion || isResponding || isCollapsed) return; const handler = (event: globalThis.KeyboardEvent) => { @@ -143,7 +150,8 @@ const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard( type="button" aria-expanded={false} aria-label="Expand questions" - onClick={() => setIsCollapsed(false)} + ref={collapseButtonRef} + onClick={toggleCollapsed} className="group flex w-full min-w-0 cursor-pointer items-center gap-2 px-4 py-2.5 text-left transition-colors hover:bg-muted/30 sm:px-5" > {prompt.questions.length > 1 ? ( @@ -151,26 +159,41 @@ const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard( {questionIndex + 1}/{prompt.questions.length} ) : null} - + {activeQuestion.header} {activeQuestion.question} - + ); } + const submitLabel = formatPendingPrimaryActionLabel({ + compact: false, + isLastQuestion: progress.isLastQuestion, + isResponding, + questionIndex, + }); + const submitDisabled = + isUnavailable || + isResponding || + (progress.isLastQuestion ? !progress.isComplete : !progress.canAdvance); + return ( // The data attribute lets ChatView measure the expanded height to derive the // scroll distance at which auto-collapse becomes safe (no layout feedback). -
+

{activeQuestion.question}

+ {/* Blocking questions need no hint: the transcript already says the turn + is waiting. Async ones do, since answering later is the new behavior. */} {prompt.isBlocking === false ? (

- The agent keeps working while you answer. + {isAgentRunning + ? "The agent keeps working while you answer." + : "The agent finished. Your answer starts a follow-up."}

) : null}
@@ -210,6 +237,7 @@ const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard( key={`${activeQuestion.id}:${option.label}`} type="button" disabled={isResponding} + aria-pressed={isSelected} onClick={() => handleOptionSelection(activeQuestion.id, option.label)} className={cn( "group flex w-full items-center gap-2.5 rounded-lg border px-2.5 py-1.5 text-left transition-all duration-150", @@ -225,7 +253,7 @@ const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard( "flex size-4.5 shrink-0 items-center justify-center rounded text-[10px] font-medium tabular-nums transition-colors duration-150", isSelected ? "bg-primary/20 text-primary-readable" - : "bg-muted/40 text-muted-foreground/50 group-hover:bg-muted/60 group-hover:text-muted-foreground/70", + : "bg-muted/40 text-muted-foreground/70 group-hover:bg-muted/60 group-hover:text-muted-foreground/70", )} > {shortcutKey} @@ -234,7 +262,7 @@ const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard(
{option.label} {option.description && option.description !== option.label ? ( - + {option.description} ) : null} @@ -245,6 +273,67 @@ const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard( ); })} + {/* The free-text answer is one more row in the list, styled like an + option, with the submit action at its end so the panel needs no + separate button row. Enter submits; Shift+Enter adds a line. */} +
+ + + +