diff --git a/src/components/conversations/conversation-detail-panel.tsx b/src/components/conversations/conversation-detail-panel.tsx index 09a308dde6..cdf395b30d 100644 --- a/src/components/conversations/conversation-detail-panel.tsx +++ b/src/components/conversations/conversation-detail-panel.tsx @@ -1944,6 +1944,9 @@ const ConversationTabView = memo(function ConversationTabView({ connectionId: conn.connectionId, connStatus, enabled: feedbackEnabled, + // Notes the transcript adopted as mid-turn user turns show as messages, + // not as strips above the composer. + steeredMessageIds: conn.steeredMessageIds, onResendAsPrompt: resendFeedbackAsPrompt, }) // Composer "insert into current turn" (native steering only). Rethrows — diff --git a/src/components/message/sub-agent-session-dialog.test.tsx b/src/components/message/sub-agent-session-dialog.test.tsx index a2a373221b..27aa47f1f6 100644 --- a/src/components/message/sub-agent-session-dialog.test.tsx +++ b/src/components/message/sub-agent-session-dialog.test.tsx @@ -219,6 +219,7 @@ function makeConnState(overrides: Partial): ConnectionState { parentConnectionId: "p1", isViewer: false, pendingUserMessage: null, + steeredMessageIds: [], configStale: false, configStaleKind: null, configStaleDismissed: false, diff --git a/src/contexts/acp-connections-context.test.tsx b/src/contexts/acp-connections-context.test.tsx index 656fc85b93..c0d893322d 100644 --- a/src/contexts/acp-connections-context.test.tsx +++ b/src/contexts/acp-connections-context.test.tsx @@ -4066,3 +4066,153 @@ describe("live surfaces that are not tabs", () => { } }) }) + +/** + * A message the user sends mid-turn over the native `_session/steering` + * channel is spliced into the live turn, so the transcript can render it as a + * user turn between the two halves of the reply. + * + * The discriminator is that the note is ALREADY `delivered` when it is + * submitted: `FeedbackItem::new_delivered` (src-tauri/src/acp/feedback.rs) has + * exactly one caller, the native push path, and it exists precisely because + * the adapter has already consumed the text by then. A `pending` note is the + * cooperative `check_user_feedback` pull channel, which the agent reads as a + * tool result and never as a user message. + */ +describe("AcpConnectionsProvider mid-turn steering messages", () => { + async function connectOwner(): Promise { + h.acpFindConnectionForConversation.mockResolvedValue(null) + await mountProvider() + await act(async () => { + await h.actions!.connect(TAB, "claude_code", "/tmp/x", "sess-1", 42) + }) + return latestAttachHandlers() + } + + function conn() { + return h.store!.getConnection(TAB)! + } + + function steeringBlocks() { + return (conn().liveMessage?.content ?? []).filter( + (b) => b.type === "steering" + ) + } + + function submitted( + seq: number, + id: string, + text: string, + status: "pending" | "delivered" + ): EventEnvelope { + return { + seq, + connection_id: "spawned-conn", + type: "feedback_submitted", + item: { id, text, created_at: "2026-06-07T00:00:00Z", status }, + } as unknown as EventEnvelope + } + + it("splices a delivered note into the running turn and records the adoption", async () => { + const handlers = await connectOwner() + emitAcpEvent(handlers, { + seq: 1, + connection_id: "spawned-conn", + type: "status_changed", + status: "prompting", + }) + emitAcpEvent(handlers, { + seq: 2, + connection_id: "spawned-conn", + type: "content_delta", + text: "half one", + } as unknown as EventEnvelope) + emitAcpEvent(handlers, submitted(3, "n1", "use the other API", "delivered")) + + expect(steeringBlocks()).toEqual([ + { type: "steering", id: "n1", text: "use the other API" }, + ]) + expect(conn().steeredMessageIds).toEqual(["n1"]) + }) + + it("ignores a pending note - the pull channel is not a user message", async () => { + const handlers = await connectOwner() + emitAcpEvent(handlers, { + seq: 1, + connection_id: "spawned-conn", + type: "status_changed", + status: "prompting", + }) + emitAcpEvent(handlers, submitted(2, "n1", "waiting note", "pending")) + + expect(steeringBlocks()).toEqual([]) + expect(conn().steeredMessageIds).toEqual([]) + }) + + it("is idempotent - the submit broadcast reaches the sender too", async () => { + const handlers = await connectOwner() + emitAcpEvent(handlers, { + seq: 1, + connection_id: "spawned-conn", + type: "status_changed", + status: "prompting", + }) + emitAcpEvent(handlers, submitted(2, "n1", "same note", "delivered")) + emitAcpEvent(handlers, submitted(3, "n1", "same note", "delivered")) + + expect(steeringBlocks()).toHaveLength(1) + expect(conn().steeredMessageIds).toEqual(["n1"]) + }) + + it("refuses a note that arrives with no turn running", async () => { + // The native submit is recorded ungated on the backend, so a note can land + // just after the turn settled. There is nothing to split then, and + // appending would graft it onto the finished turn. The note keeps its + // strip instead (it is absent from `steeredMessageIds`), and the agent + // recorded it either way, so a reload still shows it. + const handlers = await connectOwner() + emitAcpEvent(handlers, { + seq: 1, + connection_id: "spawned-conn", + type: "status_changed", + status: "prompting", + }) + emitAcpEvent(handlers, { + seq: 2, + connection_id: "spawned-conn", + type: "status_changed", + status: "connected", + }) + emitAcpEvent(handlers, submitted(3, "n1", "too late", "delivered")) + + expect(steeringBlocks()).toEqual([]) + expect(conn().steeredMessageIds).toEqual([]) + }) + + it("starts each turn with no adoptions carried over", async () => { + const handlers = await connectOwner() + emitAcpEvent(handlers, { + seq: 1, + connection_id: "spawned-conn", + type: "status_changed", + status: "prompting", + }) + emitAcpEvent(handlers, submitted(2, "n1", "first turn", "delivered")) + expect(conn().steeredMessageIds).toEqual(["n1"]) + + emitAcpEvent(handlers, { + seq: 3, + connection_id: "spawned-conn", + type: "status_changed", + status: "connected", + }) + emitAcpEvent(handlers, { + seq: 4, + connection_id: "spawned-conn", + type: "status_changed", + status: "prompting", + }) + expect(conn().steeredMessageIds).toEqual([]) + expect(steeringBlocks()).toEqual([]) + }) +}) diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index d6aa3fca78..f76b4ca25c 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -195,6 +195,17 @@ export type LiveContentBlock = | { type: "thinking"; text: string; parentToolUseId?: string } | { type: "plan"; entries: PlanEntryInfo[] } | { type: "tool_call"; info: ToolCallInfo } + /** + * A message the user sent WHILE this turn was running, injected into it via + * the native `_session/steering` channel. Not agent output: it marks the + * point in the stream where the user interrupted, so + * `buildStreamingTurnsFromLiveMessage` can close the assistant turn here, + * render the message as its own user turn, and start the reply to it as a + * new turn. Mirrors what the transcript projection already does with a + * mid-turn `user_message_chunk` (see `parsers/acp_native.rs`), so the live + * view and a reload agree. `id` is the feedback note id. + */ + | { type: "steering"; id: string; text: string } export interface LiveMessage { id: string @@ -225,6 +236,19 @@ export interface ConnectionState { * event or a snapshot's `pending_user_message`. A VIEWER mirrors this into * the runtime as a synthesized user turn; `null` outside an active turn. */ pendingUserMessage: PendingUserMessage | null + /** + * Feedback-note ids whose text this turn's `liveMessage` adopted as a + * `steering` block, i.e. the mid-turn messages now rendered as user turns in + * the transcript. The notes list above the composer reads this to drop their + * strips: one message shows in exactly one place. Reset with `liveMessage` + * at the start of every turn. + * + * The reducer is the single decider — a note it could NOT adopt (it arrived + * out of turn) is absent here, so its strip stays. Deriving this in the + * notes hook instead would race the reducer's own view of the status and + * could leave a message showing nowhere at all. + */ + steeredMessageIds: string[] pendingQuestion: PendingQuestion | null /** Awaiting-answer multiple-choice `ask_user_question` (the codeg-mcp blocking * tool). Set from a `question_request` event or a snapshot's @@ -591,6 +615,12 @@ type Action = contextKey: string entries: PlanEntryInfo[] } + | { + type: "STEERING_MESSAGE" + contextKey: string + id: string + text: string + } | { type: "CLAUDE_API_RETRY" contextKey: string @@ -1122,6 +1152,10 @@ function ensureLiveMessage(prev: LiveMessage | null): LiveMessage { } } +/** Shared empty `steeredMessageIds`, so a turn that steers nothing (almost all + * of them) keeps a stable reference through `connRenderEqual`. */ +const EMPTY_STEERED_MESSAGE_IDS: string[] = [] + /** Last time an out-of-turn drop was logged — module-level sampling clock. */ let lastOutOfTurnDropLogAt = 0 @@ -1320,6 +1354,7 @@ function connectionsReducer( liveMessage: null, pendingPermission: null, pendingUserMessage: null, + steeredMessageIds: EMPTY_STEERED_MESSAGE_IDS, pendingQuestion: null, pendingAskQuestion: null, pendingPlanApproval: null, @@ -1378,6 +1413,7 @@ function connectionsReducer( liveMessage: null, pendingPermission: null, pendingUserMessage: null, + steeredMessageIds: EMPTY_STEERED_MESSAGE_IDS, pendingQuestion: null, pendingAskQuestion: null, pendingPlanApproval: null, @@ -1572,6 +1608,8 @@ function connectionsReducer( updated.pendingQuestion = null updated.claudeApiRetry = null updated.error = null + // Steering adoptions belong to the turn whose stream they split. + updated.steeredMessageIds = EMPTY_STEERED_MESSAGE_IDS // Starting a prompt past an active AIR failure acknowledges it — // settle EVERYTHING (watermarks retained). A failure that is still // real re-arms via a higher revision on the same id. @@ -2347,6 +2385,34 @@ function connectionsReducer( return next } + case "STEERING_MESSAGE": { + const conn = state.get(action.contextKey) + if (!conn) return state + // Same out-of-turn guard as PLAN_UPDATE / TOOL_CALL / streaming deltas: + // there is no running turn to split, and appending would graft the + // message onto the PREVIOUS turn's completed liveMessage. The note keeps + // its strip in that case (it is absent from `steeredMessageIds`), and + // the agent recorded it either way, so a reload still shows it. + if (conn.status !== "prompting") return state + // Idempotent by note id: the submit broadcast reaches every attached + // client, and one client is also the sender. + if (conn.steeredMessageIds.includes(action.id)) return state + const prev = ensureLiveMessage(conn.liveMessage) + const next = new Map(state) + next.set(action.contextKey, { + ...conn, + liveMessage: { + ...prev, + content: [ + ...prev.content, + { type: "steering" as const, id: action.id, text: action.text }, + ], + }, + steeredMessageIds: [...conn.steeredMessageIds, action.id], + }) + return next + } + case "CLAUDE_API_RETRY": { const conn = state.get(action.contextKey) if (!conn) return state @@ -3529,6 +3595,28 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) scheduleToolCallUpdateFlush() break + case "feedback_submitted": { + // A note that is ALREADY `delivered` when it is submitted was pushed + // into the running turn over the native `_session/steering` channel + // (`FeedbackItem::new_delivered` is that path's only producer). The + // agent has the text as a user message, so the transcript shows it + // as one: it closes the assistant turn at this point in the stream + // and the reply to it starts a new turn. + // + // A `pending` note is the cooperative `check_user_feedback` pull + // channel — the agent has not read it, and when it does it arrives + // as a tool result, never a user message. Those stay in the notes + // list above the composer, which is where a reload leaves them too. + if (e.item.status !== "delivered") break + flushStreamingQueue() + dispatch({ + type: "STEERING_MESSAGE", + contextKey, + id: e.item.id, + text: e.item.text, + }) + break + } case "permission_resolved": // Backend signals a permission was answered (this window's local // respondPermission, a sibling window, a server-mode peer, or diff --git a/src/contexts/conversation-runtime-context.test.tsx b/src/contexts/conversation-runtime-context.test.tsx index fc491ccc7a..0cbf266937 100644 --- a/src/contexts/conversation-runtime-context.test.tsx +++ b/src/contexts/conversation-runtime-context.test.tsx @@ -2465,3 +2465,291 @@ describe("buildStreamingTurnsFromLiveMessage — codex search/list-files command ) }) }) + +/** + * A message sent WHILE the agent is replying (native `_session/steering`) + * reaches the agent as a user message, so the transcript shows it as one. + * + * Before this, the live view dropped it entirely: the strip above the composer + * was the only trace, and because no user turn landed between the two halves + * of the reply, the answer to the steered message continued inside the SAME + * assistant bubble - two separate replies rendered as one run-on paragraph. + * + * The persisted projection already did the right thing (see the + * `user_message_chunk` arm of `project_turns` in `parsers/acp_native.rs`, + * which flushes the assistant turn and pushes a user turn), so this is what + * makes the live view agree with a reload. + */ +describe("buildStreamingTurnsFromLiveMessage - mid-turn steering messages", () => { + function live(content: LiveContentBlock[]): LiveMessage { + return { id: "lm-steer", role: "assistant", content, startedAt: 0 } + } + + it("renders a delivered mid-turn message as its own user turn", () => { + const turns = buildStreamingTurnsFromLiveMessage( + 1, + live([ + { type: "text", text: "working on it" }, + { type: "steering", id: "note-1", text: "actually, use the other API" }, + ]) + ).turns + + const user = turns.filter((t) => t.role === "user") + expect(user).toHaveLength(1) + expect(user[0].blocks).toEqual([ + { type: "text", text: "actually, use the other API" }, + ]) + }) + + it("splits the reply at the boundary so two answers never share one bubble", () => { + const turns = buildStreamingTurnsFromLiveMessage( + 1, + live([ + { type: "text", text: "I will report both links once CI is green." }, + { type: "steering", id: "note-1", text: "not done" }, + { type: "text", text: "Not done - those are the two PRs..." }, + ]) + ).turns + + expect(turns.map((t) => t.role)).toEqual(["assistant", "user", "assistant"]) + // The two replies stay in separate turns; concatenating them into one + // block is exactly the run-on paragraph this fixes. + expect(turns[0].blocks).toEqual([ + { type: "text", text: "I will report both links once CI is green." }, + ]) + expect(turns[2].blocks).toEqual([ + { type: "text", text: "Not done - those are the two PRs..." }, + ]) + // Distinct ids, or the timeline dedup would collapse them back together. + expect(new Set(turns.map((t) => t.id)).size).toBe(3) + }) + + it("splits even when the round has no completed tool call before it", () => { + // The ordinary round split needs a settled tool call first; a user + // interrupting mid-sentence is a boundary regardless. + const turns = buildStreamingTurnsFromLiveMessage( + 1, + live([ + { type: "thinking", text: "hmm" }, + { type: "steering", id: "note-1", text: "stop" }, + { type: "text", text: "ok" }, + ]) + ).turns + expect(turns.map((t) => t.role)).toEqual(["assistant", "user", "assistant"]) + }) + + it("keeps prose either side of it in separate blocks", () => { + // `mainProseContinuations` fuses same-kind prose only across blocks that + // render nothing. A steering turn renders, so it must break the run. + const turns = buildStreamingTurnsFromLiveMessage( + 1, + live([ + { type: "text", text: "first" }, + { type: "steering", id: "note-1", text: "wait" }, + { type: "text", text: "second" }, + ]) + ).turns + expect(turns[0].blocks).toEqual([{ type: "text", text: "first" }]) + expect(turns[2].blocks).toEqual([{ type: "text", text: "second" }]) + }) + + it("carries several steering messages in the order they were sent", () => { + const turns = buildStreamingTurnsFromLiveMessage( + 1, + live([ + { type: "text", text: "a" }, + { type: "steering", id: "n1", text: "one" }, + { type: "text", text: "b" }, + { type: "steering", id: "n2", text: "two" }, + { type: "text", text: "c" }, + ]) + ).turns + expect(turns.map((t) => t.role)).toEqual([ + "assistant", + "user", + "assistant", + "user", + "assistant", + ]) + expect( + turns + .filter((t) => t.role === "user") + .map((t) => (t.blocks[0].type === "text" ? t.blocks[0].text : "")) + ).toEqual(["one", "two"]) + }) + + it("leaves a turn with no steering completely unchanged", () => { + const turns = buildStreamingTurnsFromLiveMessage( + 1, + live([ + { type: "thinking", text: "think" }, + { type: "text", text: "reply" }, + ]) + ).turns + expect(turns).toHaveLength(1) + expect(turns[0].role).toBe("assistant") + }) +}) + +/** + * The agent writes a steered message into its own transcript, so a detail + * fetch landing DURING the turn brings it back as an ordinary user turn under + * a parser-assigned id - which no id-keyed dedup can match to the live copy. + * Both would render. The live copy is kept because it sits between the two + * halves of the reply; the persisted one would land after the whole thing. + */ +describe("conversation timeline - a steered message survives a mid-turn reload once", () => { + const runtimeHolder: { + current: ReturnType | undefined + } = { current: undefined } + + function RuntimeCapture() { + const runtime = useConversationRuntime() + useEffect(() => { + runtimeHolder.current = runtime + }) + return null + } + + function turn( + id: string, + role: "user" | "assistant", + text: string + ): MessageTurn { + return { + id, + role, + blocks: [{ type: "text" as const, text }], + timestamp: "2026-05-28T00:00:00.000Z", + } + } + + function detailWith( + turns: MessageTurn[], + inFlightUserTurnId: string | null + ): DbConversationDetail { + return { + summary: { + id: 99, + folder_id: 1, + agent_type: "claude", + title: "c", + title_locked: false, + status: "in_progress", + kind: "regular", + model: null, + git_branch: null, + external_id: "ext-1", + message_count: turns.length, + child_count: 0, + created_at: "2026-05-28T00:00:00.000Z", + updated_at: "2026-05-28T00:00:00.000Z", + pinned_at: null, + }, + turns, + session_stats: null, + in_flight_user_turn_id: inFlightUserTurnId, + } as DbConversationDetail + } + + function userTexts( + items: ReturnType< + NonNullable["getTimelineTurns"] + > + ): string[] { + return items + .filter((t) => t.turn.role === "user") + .map((t) => + t.turn.blocks[0]?.type === "text" ? t.turn.blocks[0].text : "" + ) + } + + beforeEach(() => { + runtimeHolder.current = undefined + mockGetFolderConversation.mockReset() + mockGetFolderConversation.mockImplementation(() => new Promise(() => {})) + }) + + it("shows the steered message once, keeping the live copy's position", async () => { + renderProvider() + const api = () => runtimeHolder.current! + + // The turn is running and the reply has been split by a steering message. + act(() => { + api().setLiveMessage( + 99, + { + id: "lm-1", + role: "assistant", + content: [ + { type: "text", text: "half one" }, + { type: "steering", id: "note-1", text: "use the other API" }, + { type: "text", text: "half two" }, + ], + startedAt: 0, + }, + true + ) + }) + + // A mid-turn detail fetch lands, carrying the agent's own record of that + // same message under a parser id. + mockGetFolderConversation.mockResolvedValueOnce( + detailWith( + [ + turn("p-1", "user", "the original prompt"), + turn("p-2", "user", "use the other API"), + ], + "p-1" + ) + ) + await act(async () => { + api().refetchDetail(99, { preserveLive: true }) + }) + + const timeline = api().getTimelineTurns(99) + // Once, not twice - and the original prompt is untouched. + expect(userTexts(timeline)).toEqual([ + "the original prompt", + "use the other API", + ]) + const steered = timeline.filter( + (t) => + t.turn.role === "user" && + t.turn.blocks[0]?.type === "text" && + t.turn.blocks[0].text === "use the other API" + ) + // The surviving copy is the live one, between the halves of the reply. + expect(steered).toHaveLength(1) + expect(steered[0].phase).toBe("streaming") + }) + + it("never suppresses the in-flight prompt, even when a steer repeats it", async () => { + renderProvider() + const api = () => runtimeHolder.current! + act(() => { + api().setLiveMessage( + 99, + { + id: "lm-2", + role: "assistant", + content: [{ type: "steering", id: "note-1", text: "continue" }], + startedAt: 0, + }, + true + ) + }) + mockGetFolderConversation.mockResolvedValueOnce( + detailWith([turn("p-1", "user", "continue")], "p-1") + ) + await act(async () => { + api().refetchDetail(99, { preserveLive: true }) + }) + // Both survive: hiding a prompt is the one failure worse than showing a + // duplicate, so the in-flight turn is always exempt. + expect(userTexts(api().getTimelineTurns(99))).toEqual([ + "continue", + "continue", + ]) + }) +}) diff --git a/src/hooks/use-connection.ts b/src/hooks/use-connection.ts index 1e8b6f9447..cb45806152 100644 --- a/src/hooks/use-connection.ts +++ b/src/hooks/use-connection.ts @@ -34,6 +34,7 @@ const DEFAULT_PROMPT_CAPABILITIES: PromptCapabilitiesInfo = { /** Stable empty table so the no-failures common case never re-renders. */ const EMPTY_SESSION_FAILURES: SessionFailureRecord[] = [] +const EMPTY_STEERED_MESSAGE_IDS: string[] = [] export interface UseConnectionReturn { connectionId: string | null @@ -65,6 +66,10 @@ export interface UseConnectionReturn { availableCommands: AvailableCommandInfo[] | null pendingPermission: PendingPermission | null pendingUserMessage: PendingUserMessage | null + /** Feedback-note ids this turn's live message adopted as mid-turn user turns + * (native steering). The notes list drops their strips so one message shows + * in exactly one place. `[]` when nothing was steered. */ + steeredMessageIds: string[] pendingQuestion: PendingQuestion | null pendingAskQuestion: PendingQuestionState | null pendingPlanApproval: PendingPlanApprovalState | null @@ -225,6 +230,8 @@ export function useConnection(contextKey: string): UseConnectionReturn { const availableCommands = connection?.availableCommands ?? null const pendingPermission = connection?.pendingPermission ?? null const pendingUserMessage = connection?.pendingUserMessage ?? null + const steeredMessageIds = + connection?.steeredMessageIds ?? EMPTY_STEERED_MESSAGE_IDS const pendingQuestion = connection?.pendingQuestion ?? null const pendingAskQuestion = connection?.pendingAskQuestion ?? null const pendingPlanApproval = connection?.pendingPlanApproval ?? null @@ -330,6 +337,7 @@ export function useConnection(contextKey: string): UseConnectionReturn { availableCommands, pendingPermission, pendingUserMessage, + steeredMessageIds, pendingQuestion, pendingAskQuestion, pendingPlanApproval, @@ -370,6 +378,7 @@ export function useConnection(contextKey: string): UseConnectionReturn { availableCommands, pendingPermission, pendingUserMessage, + steeredMessageIds, pendingQuestion, pendingAskQuestion, pendingPlanApproval, diff --git a/src/hooks/use-session-feedback.test.ts b/src/hooks/use-session-feedback.test.ts index 3349393162..ace9594d79 100644 --- a/src/hooks/use-session-feedback.test.ts +++ b/src/hooks/use-session-feedback.test.ts @@ -465,3 +465,113 @@ describe("useSessionFeedback", () => { await waitFor(() => expect(result.current.canSubmit).toBe(true)) }) }) + +/** + * A note the transcript adopted as a mid-turn user turn is a MESSAGE now, so + * its strip above the composer goes away - otherwise the same text is on + * screen twice for the rest of the turn. + * + * The adoption decision belongs to the connection reducer (it is the only + * thing that knows whether there was a running turn to splice the message + * into), so the hook is told which ids were taken rather than guessing. A note + * that was NOT adopted keeps its strip, which is what makes "shows in exactly + * one place" true in both directions. + */ +describe("useSessionFeedback steered-note strips", () => { + // Widen the props type so a test can vary `steeredMessageIds`; + // `baseProps` alone would pin it to the three fields it declares. + const props: Parameters[0] = baseProps + + it("drops the strip for a note the transcript adopted", async () => { + const { result, rerender } = renderHook( + (props: Parameters[0]) => + useSessionFeedback(props), + { initialProps: props } + ) + act(() => { + capturedHandler?.({ + type: "feedback_submitted", + connection_id: "c1", + item: note("n1", "use the other API", "delivered"), + } as unknown as EventEnvelope) + }) + await waitFor(() => expect(result.current.notes).toHaveLength(1)) + + // The reducer spliced it into the live turn. + rerender({ ...baseProps, steeredMessageIds: ["n1"] }) + expect(result.current.notes).toHaveLength(0) + expect(result.current.showList).toBe(false) + }) + + it("keeps the strip for a note the transcript could not adopt", async () => { + const { result } = renderHook( + (props: Parameters[0]) => + useSessionFeedback(props), + { initialProps: { ...baseProps, steeredMessageIds: [] } } + ) + act(() => { + capturedHandler?.({ + type: "feedback_submitted", + connection_id: "c1", + item: note("n1", "landed after the turn ended", "delivered"), + } as unknown as EventEnvelope) + }) + // No adoption reported, so the note stays visible somewhere. + await waitFor(() => expect(result.current.notes).toHaveLength(1)) + expect(result.current.showList).toBe(true) + }) + + it("leaves pull-channel notes alone - they never become messages", async () => { + // A `check_user_feedback` note reaches the agent as a tool result, not as + // a user message, so it has no user turn on reload either. Strips are the + // right and only home for it, waiting or read. + const { result } = renderHook( + (props: Parameters[0]) => + useSessionFeedback(props), + { initialProps: { ...baseProps, steeredMessageIds: [] } } + ) + act(() => { + capturedHandler?.({ + type: "feedback_submitted", + connection_id: "c1", + item: note("n1", "waiting note"), + } as unknown as EventEnvelope) + }) + await waitFor(() => expect(result.current.notes).toHaveLength(1)) + act(() => { + capturedHandler?.({ + type: "feedback_consumed", + connection_id: "c1", + ids: ["n1"], + delivered_at: "2026-06-07T00:00:05Z", + } as unknown as EventEnvelope) + }) + // Read by the agent, still a strip. + expect(result.current.notes).toHaveLength(1) + expect(result.current.notes[0].status).toBe("delivered") + expect(result.current.showList).toBe(true) + }) + + it("only drops the ids it was given", async () => { + const { result, rerender } = renderHook( + (props: Parameters[0]) => + useSessionFeedback(props), + { initialProps: props } + ) + act(() => { + capturedHandler?.({ + type: "feedback_submitted", + connection_id: "c1", + item: note("n1", "one", "delivered"), + } as unknown as EventEnvelope) + capturedHandler?.({ + type: "feedback_submitted", + connection_id: "c1", + item: note("n2", "two", "delivered"), + } as unknown as EventEnvelope) + }) + await waitFor(() => expect(result.current.notes).toHaveLength(2)) + rerender({ ...baseProps, steeredMessageIds: ["n1"] }) + expect(result.current.notes.map((n) => n.id)).toEqual(["n2"]) + }) +}) diff --git a/src/hooks/use-session-feedback.ts b/src/hooks/use-session-feedback.ts index 04d70076d6..d99ad4aa09 100644 --- a/src/hooks/use-session-feedback.ts +++ b/src/hooks/use-session-feedback.ts @@ -46,6 +46,18 @@ export interface UseSessionFeedbackArgs { connStatus: ConnectionStatus | null /** Whether the live-feedback feature is enabled (global setting). */ enabled: boolean + /** + * Note ids the live transcript adopted as mid-turn user turns + * (`ConnectionState.steeredMessageIds`). Their strips are dropped: the note + * IS the message now, and showing both would print it twice. + * + * Taken from the connection rather than derived here on purpose. The + * transcript can only adopt a note while a turn is actually running, and a + * note submitted on the closing edge of one may miss that window; letting + * this hook guess would eventually guess the other way from the reducer and + * leave a message showing in neither place. + */ + steeredMessageIds?: readonly string[] /** Reroute a note as an ordinary prompt when the turn ended before it could be * submitted (turn-end race). */ onResendAsPrompt?: (text: string) => void @@ -85,6 +97,7 @@ export function useSessionFeedback({ connectionId, connStatus, enabled, + steeredMessageIds, onResendAsPrompt, }: UseSessionFeedbackArgs): UseSessionFeedback { const t = useTranslations("LiveFeedback") @@ -384,11 +397,20 @@ export function useSessionFeedback({ (toolAvailable || nativeSteering) && isPrompting const channel: "native" | "pull" = nativeSteering ? "native" : "pull" - const showList = notes.length > 0 && isPrompting + // Drop the notes the transcript is already rendering as user turns. Kept as + // a derivation rather than a filter on `setNotes` so a note stays recoverable + // as a strip if the transcript never took it. + const visibleNotes = useMemo(() => { + if (!steeredMessageIds || steeredMessageIds.length === 0) return notes + const adopted = new Set(steeredMessageIds) + const remaining = notes.filter((n) => !adopted.has(n.id)) + return remaining.length === notes.length ? notes : remaining + }, [notes, steeredMessageIds]) + const showList = visibleNotes.length > 0 && isPrompting return useMemo( () => ({ - notes, + notes: visibleNotes, featureEnabled: enabled, canSubmit, channel, @@ -401,7 +423,7 @@ export function useSessionFeedback({ steer, }), [ - notes, + visibleNotes, enabled, canSubmit, channel, diff --git a/src/stores/conversation-runtime-store.ts b/src/stores/conversation-runtime-store.ts index 31f02b749a..9385ae46f7 100644 --- a/src/stores/conversation-runtime-store.ts +++ b/src/stores/conversation-runtime-store.ts @@ -596,6 +596,13 @@ interface BuiltStreamingTurns { inProgressToolCallIds: Set } +/** One turn under construction inside a live message. Assistant groups are the + * reply's rounds; a `user` group is a message the user sent mid-turn. */ +interface StreamingGroup { + role: "assistant" | "user" + blocks: MessageTurn["blocks"] +} + // Cache joined chunk output keyed by chunks-array identity. The ACP reducer // creates a new chunks array only when streaming output actually changes, so // a WeakMap keyed on the array reference lets repeated renders reuse the @@ -1166,7 +1173,11 @@ export function buildStreamingTurnsFromLiveMessage( // pattern: each "round" (text/thinking + tool calls + tool results) is a // separate turn. A new turn starts when a text/thinking/plan block appears // after completed tool calls in the current group. - const groups: MessageTurn["blocks"][] = [[]] + // Each group becomes one turn. Assistant groups are the reply, split into + // rounds as before; a `user` group is a message the user sent mid-turn + // (native steering), which both ends the round before it and keeps the reply + // to it in a round of its own. + const groups: StreamingGroup[] = [{ role: "assistant", blocks: [] }] let currentGroupHasCompletedTool = false const inProgressToolCallIds = new Set() // Which main-thread prose blocks are a continuation of the previous one @@ -1189,17 +1200,32 @@ export function buildStreamingTurnsFromLiveMessage( continue } + // A mid-turn user message is a hard turn boundary in both directions: it + // closes whatever the agent had said so far and opens a fresh assistant + // group for the reply to it, so the two replies can never render as one + // run-on bubble. Unconditional — unlike a content block, it splits even + // when the current group has no completed tool call. + if (block.type === "steering") { + groups.push({ + role: "user", + blocks: [{ type: "text", text: block.text }], + }) + groups.push({ role: "assistant", blocks: [] }) + currentGroupHasCompletedTool = false + continue + } + const isContentBlock = block.type === "text" || block.type === "thinking" || block.type === "plan" if (isContentBlock && currentGroupHasCompletedTool) { - groups.push([]) + groups.push({ role: "assistant", blocks: [] }) currentGroupHasCompletedTool = false } - const currentBlocks = groups[groups.length - 1] + const currentBlocks = groups[groups.length - 1].blocks switch (block.type) { case "text": @@ -1431,14 +1457,14 @@ export function buildStreamingTurnsFromLiveMessage( const timestamp = new Date(liveMessage.startedAt).toISOString() const turns = groups - .filter((blocks) => blocks.length > 0) - .map((blocks, i) => ({ + .filter((group) => group.blocks.length > 0) + .map((group, i) => ({ id: i === 0 ? `live-${conversationId}-${liveMessage.id}` : `live-${conversationId}-${liveMessage.id}-${i}`, - role: "assistant" as const, - blocks, + role: group.role, + blocks: group.blocks, timestamp, })) @@ -3137,6 +3163,54 @@ function computeTimelinePrefix( return entry } +/** + * Hide the persisted copy of a message the user sent mid-turn, when the live + * stream is already showing it. + * + * The agent writes a steered message into its own transcript, so a detail + * fetch that lands DURING the turn brings it back as an ordinary user turn — + * under a parser id, which no id-keyed dedup can match to the live copy. Both + * would render. + * + * The live copy is the one to keep: it sits between the two halves of the + * reply, where the message was actually sent, while the persisted copy is + * appended after the in-flight prompt with the reply's first half suppressed + * around it (see `visiblePersistedTurns`), which would put the interruption + * before the text it interrupted. + * + * Matched on CONTENT, the same way `APPEND_VIEWER_USER_TURN` reconciles the + * two id namespaces of one prompt. Scoped tightly, because suppressing a user + * turn is the one failure that hides a message rather than duplicating it: + * only persisted-phase turns, never the in-flight prompt itself (a steer that + * repeats the prompt verbatim leaves the prompt alone), and only when the tail + * actually carries a steered turn — so a timeline with no steering does no + * work here at all. + */ +function suppressPersistedSteeredPrompts( + prefix: ConversationTimelineTurn[], + tail: ConversationTimelineTurn[], + session: ConversationRuntimeSession +): ConversationTimelineTurn[] { + let steeredKeys: Set | null = null + for (const item of tail) { + if (item.turn.role !== "user") continue + steeredKeys ??= new Set() + steeredKeys.add(userTurnContentKey(item.turn)) + } + if (!steeredKeys) return prefix + const inFlightPromptId = session.detail?.in_flight_user_turn_id ?? null + const filtered = prefix.filter( + (item) => + !( + item.phase === "persisted" && + item.turn.role === "user" && + item.turn.id !== inFlightPromptId && + steeredKeys.has(userTurnContentKey(item.turn)) + ) + ) + return filtered.length === prefix.length ? prefix : filtered +} + function computeTimeline( state: ConversationRuntimeState, conversationId: number @@ -3185,9 +3259,8 @@ function computeTimeline( } seenTailKeys?.add(key) } - deduped = collides - ? dedupeTimeline(prefix.concat(tail)) - : prefix.concat(tail) + const head = suppressPersistedSteeredPrompts(prefix, tail, session) + deduped = collides ? dedupeTimeline(head.concat(tail)) : head.concat(tail) } timelineCache.set(session, deduped)