diff --git a/src-tauri/src/acp/feedback.rs b/src-tauri/src/acp/feedback.rs index 8114fb9e57..2a688cba55 100644 --- a/src-tauri/src/acp/feedback.rs +++ b/src-tauri/src/acp/feedback.rs @@ -50,7 +50,9 @@ pub struct FeedbackItem { pub text: String, pub created_at: DateTime, pub status: FeedbackStatus, - /// When the agent read this note. `None` while `Pending`. + /// When the agent read this note. `None` while `Pending`. On the native + /// push path it is the submit instant (see [`Self::new_delivered`]) — a + /// lower bound on the read rather than the read itself. #[serde(default, skip_serializing_if = "Option::is_none")] pub delivered_at: Option>, } @@ -73,6 +75,12 @@ impl FeedbackItem { /// let `read_pending_feedback` hand the same text to a `check_user_feedback` /// call and double-deliver it (the Pending-only read is the mutual /// exclusion between the push and pull channels). + /// + /// `at` is the SUBMIT instant, taken before the injection is handed to the + /// adapter (`submit_feedback_native`), so `created_at` is provably earlier + /// than the agent's own transcript copy of the message — the ordering the + /// frontend needs to recognize that copy. `delivered_at` shares it, and is + /// therefore a lower bound on the read rather than the read itself. pub fn new_delivered(id: String, text: String, at: DateTime) -> Self { Self { id, diff --git a/src-tauri/src/acp/manager.rs b/src-tauri/src/acp/manager.rs index df23c66036..16fd94c587 100644 --- a/src-tauri/src/acp/manager.rs +++ b/src-tauri/src/acp/manager.rs @@ -2648,6 +2648,11 @@ impl ConnectionManager { /// note recorded right after `TurnComplete` is harmless — the notes /// list renders only while prompting, and the next turn's `UserMessage` /// clears `feedback`. + /// * `created_at` PRECEDES THE INJECTION. It is taken before the `Steer` + /// command is enqueued, so it is earlier than any transcript entry the + /// injection can cause. The frontend relies on that ordering to tell the + /// agent's own copy of the message from the same words sent in an + /// earlier round (`suppressPersistedSteeredPrompts`). async fn submit_feedback_native( conn_id: &str, state: Arc>, @@ -2664,6 +2669,18 @@ impl ConnectionManager { let conn_id_for_task = conn_id.to_string(); let handle = tokio::spawn(async move { let outcome: Result = async { + // Stamped BEFORE the command goes out, so the note's instant is + // causally earlier than anything the injection can cause. The + // adapter pushes the text to the agent before it answers, and + // the agent may write its own transcript copy of the message + // while this task is still awaiting that answer — a note + // stamped on the way back would then look NEWER than the copy + // it produced, and the frontend (which folds a persisted copy + // away only when it postdates the injection — see + // `suppressPersistedSteeredPrompts`) would show the message + // twice. Same clock, same host: `created_at` is when the note + // was created, which is also what the pull path records. + let created_at = chrono::Utc::now(); let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); cmd_tx .send(ConnectionCommand::Steer { @@ -2696,11 +2713,8 @@ impl ConnectionManager { state.write().await.native_steering_available = false; } } - let item = FeedbackItem::new_delivered( - uuid::Uuid::new_v4().to_string(), - text, - chrono::Utc::now(), - ); + let item = + FeedbackItem::new_delivered(uuid::Uuid::new_v4().to_string(), text, created_at); // Ungated on purpose — see the invariant on this fn's doc. emit_with_state( &state, @@ -7585,6 +7599,46 @@ mod tests { }) } + /// The note's instant must precede the injection reaching the agent. The + /// adapter hands the text to the agent BEFORE it answers `injected`, so the + /// agent can write its own transcript copy of the message while this call + /// is still awaiting that answer. A note stamped on the way back would + /// postdate the copy it caused, and the frontend — which folds a persisted + /// copy away only when it postdates the injection, so that the same words + /// sent in an earlier round are never hidden — would show the message both + /// as a transcript turn and as a live one. + #[tokio::test] + async fn native_submit_stamps_the_note_before_the_agent_can_see_it() { + let mgr = ConnectionManager::new(); + let mut rx = mgr + .insert_test_connection_live("c1", AgentType::ClaudeCode, None, EventEmitter::Noop) + .await; + mark_native_steering_ready(&mgr, "c1").await; + // Stand in for the adapter: note when the injection reached it (the + // earliest instant the agent could record the message), then dawdle + // before answering, as a real round-trip does. + let fake_loop = tokio::spawn(async move { + match rx.recv().await { + Some(ConnectionCommand::Steer { reply, .. }) => { + let seen_by_agent = chrono::Utc::now(); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + let _ = reply.send(Ok(SteerOutcome::Injected)); + seen_by_agent + } + _ => panic!("expected a Steer command"), + } + }); + + let item = mgr.submit_feedback("c1", "use the other API".into()).await.unwrap(); + let seen_by_agent = fake_loop.await.unwrap(); + assert!( + item.created_at <= seen_by_agent, + "created_at ({}) must precede the injection reaching the agent ({})", + item.created_at, + seen_by_agent + ); + } + #[tokio::test] async fn native_submit_injected_records_delivered_and_pull_stays_empty() { let mgr = ConnectionManager::new(); diff --git a/src/components/conversations/conversation-detail-panel.tsx b/src/components/conversations/conversation-detail-panel.tsx index e2687ddf3e..30b4f0d2f0 100644 --- a/src/components/conversations/conversation-detail-panel.tsx +++ b/src/components/conversations/conversation-detail-panel.tsx @@ -2013,6 +2013,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/message-list-view.test.tsx b/src/components/message/message-list-view.test.tsx index 6a5dbb10e6..8e9ad7d5a6 100644 --- a/src/components/message/message-list-view.test.tsx +++ b/src/components/message/message-list-view.test.tsx @@ -3,6 +3,8 @@ import { describe, expect, it } from "vitest" import { advanceReplyFold, extractDelegationSources, + isForkPointUnnamed, + markThreadTail, mergeConsecutiveAssistantTurns, singletonSourceTurns, type MergedAssistantRunCache, @@ -41,6 +43,7 @@ function assistantItem( isRoleTransition: false, previousUserIndex: null, isLastAssistantRun: false, + isThreadTail: false, sourceTurns: [], } } @@ -333,6 +336,7 @@ function makeItem( isRoleTransition: false, previousUserIndex: null, isLastAssistantRun: false, + isThreadTail: false, sourceTurns: singletonSourceTurns(turn(group.id)), } } @@ -607,3 +611,100 @@ describe("extractDelegationSources", () => { expect(extractDelegationSources([refused])).toEqual([]) }) }) + +/** + * The fork affordance sends a turn id to the backend, and the backend cannot + * resolve an id this client minted for its own live stream — it tail-forks + * instead of refusing. That is the right answer for the newest reply and a + * silent wrong one for any earlier reply, which a steered turn creates: it + * promotes as assistant / user message / assistant, so its first half sits + * settled and non-tail with a fork button while the parser's name is still a + * reparse away. + */ +describe("isForkPointUnnamed", () => { + function forkTurn(id: string, sourceTurnId?: string | null): MessageTurn { + return { + id, + role: "assistant", + blocks: [], + timestamp: "", + ...(sourceTurnId !== undefined ? { source_turn_id: sourceTurnId } : {}), + } + } + + it("withholds a live-named reply that is not the thread's last item", () => { + expect(isForkPointUnnamed(forkTurn("live-7-lm-1"), false)).toBe(true) + }) + + it("allows one at the end of the thread — there the tail IS the fork point", () => { + expect(isForkPointUnnamed(forkTurn("live-7-lm-1"), true)).toBe(false) + }) + + it("withholds the newest REPLY when a message follows it", () => { + // Steering at the very end of a turn promotes as assistant + user message + // with nothing after it: the reply is the newest one, and still not the + // tail. The backend's tail fork would land after the steered message, and + // a parse ending on a user turn never backfills a name to correct it — so + // "newest assistant run" is the wrong exception and `isThreadTail` is the + // right one. + expect(isForkPointUnnamed(forkTurn("live-7-lm"), false)).toBe(true) + }) + + it("allows it again once the reparse names it", () => { + expect(isForkPointUnnamed(forkTurn("live-7-lm-1", "turn-4"), false)).toBe( + false + ) + }) + + it("leaves parser-named history alone", () => { + // Every historical turn arrives under a parser id and no `source_turn_id`; + // treating that as unnamed would grey out the whole thread. + expect(isForkPointUnnamed(forkTurn("turn-4"), false)).toBe(false) + }) + + it("says nothing about a group with no turns", () => { + expect(isForkPointUnnamed(null, false)).toBe(false) + }) +}) + +describe("markThreadTail", () => { + const compaction: ThreadItem = { + key: "persisted-compact", + kind: "compaction", + meta: { contextCompaction: true }, + } + const tailFlags = (items: ThreadItem[]) => + items.map((it) => (it.kind === "turn" ? it.isThreadTail : null)) + + it("marks the last rendered turn", () => { + const items = [assistantItem("a"), assistantItem("b")] + markThreadTail(items) + expect(tailFlags(items)).toEqual([false, true]) + }) + + it("leaves a reply unmarked when a message follows it", () => { + // The shape a steer at the very end of a turn promotes to: the reply is + // still the newest one, and the tail is the message after it. + const items = [assistantItem("a"), makeUserItem("u", 1)] + markThreadTail(items) + expect(tailFlags(items)).toEqual([false, true]) + }) + + it("marks nothing when a compaction divider is last", () => { + const items = [assistantItem("a"), compaction] + markThreadTail(items) + expect(tailFlags(items)).toEqual([false, null]) + }) + + it("steps over a trailing turn that renders nothing", () => { + const items = [assistantItem("a"), assistantItem("empty", { parts: [] })] + markThreadTail(items) + expect(tailFlags(items)).toEqual([true, false]) + }) + + it("marks nothing in an empty thread", () => { + const items: ThreadItem[] = [] + markThreadTail(items) + expect(items).toEqual([]) + }) +}) diff --git a/src/components/message/message-list-view.tsx b/src/components/message/message-list-view.tsx index f9c8f7a363..4c5835d507 100644 --- a/src/components/message/message-list-view.tsx +++ b/src/components/message/message-list-view.tsx @@ -2,6 +2,7 @@ import { memo, useCallback, useEffect, useMemo, useRef, useState } from "react" import { + isLiveTurnId, selectTimelineTurns, useConversationRuntimeActions, useConversationRuntimeStore, @@ -174,6 +175,12 @@ export type ThreadRenderItem = * `armed` flag this is what makes a run "the current round" — see the * fold state below. */ isLastAssistantRun: boolean + /** Nothing follows this item in the thread — not a user message, not a + * compaction divider. Distinct from `isLastAssistantRun`, which is still + * true for a reply the user interrupted at its very end (that promotes + * as assistant then user message, so the newest REPLY is not the tail). + * Read by the fork gate, where the two differ by a wrong fork point. */ + isThreadTail: boolean /** Raw assistant sub-turn(s) that compose this reply — fed to the * per-reply artifacts card so it can list files changed this reply. */ sourceTurns: MessageTurn[] @@ -757,6 +764,52 @@ const UserMessageTaskButton = memo(function UserMessageTaskButton({ ) }) +/** + * Flag the thread's last rendered element, which is where the backend's tail + * fork would land — a user message or a compaction divider after the newest + * reply means that reply is NOT it. Blocks that render nothing are stepped + * over: they occupy an index without occupying the thread. + * + * Mutates in place, like the loop that resets these flags just before it (a + * cached merged item is reset there every render, so a stale `true` cannot + * survive). Exported for tests. + */ +export function markThreadTail(items: ThreadRenderItem[]): void { + for (let idx = items.length - 1; idx >= 0; idx--) { + const item = items[idx] + if (item.kind === "turn" && isEmptyTurnItem(item)) continue + if (item.kind === "turn") item.isThreadTail = true + break + } +} + +/** + * Whether forking at this reply would land somewhere other than where the user + * pointed — so the affordance greys out until it wouldn't. + * + * A turn this session streamed carries a `live-…` id until the post-turn + * reparse backfills the parser's name (`source_turn_id`). The backend cannot + * resolve such an id and deliberately degrades to a TAIL fork rather than + * refusing the click. That is exactly right at the END of the thread — the tail + * IS the fork point — and a silent lie anywhere before it. + * + * Anywhere before it is reachable: a reply the user steered mid-turn promotes + * as assistant / user message / assistant, so its first half is a settled + * group carrying a fork button while the backfill is a second and a half away. + * The exception is therefore the thread TAIL, not the newest assistant reply: + * steer at the very end of a turn and the promotion is assistant + user message + * with nothing after it, which leaves the newest reply one message short of the + * tail — and the parse ending on a user turn means `source_turn_id` never + * arrives to correct it (see `computeTurnMetadataPatches`). Exported for tests. + */ +export function isForkPointUnnamed( + forkPoint: MessageTurn | null, + isThreadTail: boolean +): boolean { + if (forkPoint === null || isThreadTail) return false + return forkPoint.source_turn_id == null && isLiveTurnId(forkPoint.id) +} + const HistoricalMessageGroup = memo(function HistoricalMessageGroup({ group, dimmed = false, @@ -770,6 +823,7 @@ const HistoricalMessageGroup = memo(function HistoricalMessageGroup({ foldEpoch = 0, onForkFromTurn, forkDisabled = false, + isThreadTail = false, }: { group: ResolvedMessageGroup dimmed?: boolean @@ -783,11 +837,21 @@ const HistoricalMessageGroup = memo(function HistoricalMessageGroup({ foldEpoch?: number onForkFromTurn?: (turnId: string) => void forkDisabled?: boolean + /** Whether nothing follows this group in the thread — the one position where + * a turn the backend cannot name still forks where the user pointed. */ + isThreadTail?: boolean }) { if (group.role === "system") { return } + // The fork point is the group's LAST turn: forking is "up to and including + // this reply", and a merged group ends where the reply does. + const forkPoint = sourceTurns?.length + ? sourceTurns[sourceTurns.length - 1] + : null + const forkPointUnnamed = isForkPointUnnamed(forkPoint, isThreadTail) + return (
@@ -835,22 +899,21 @@ const HistoricalMessageGroup = memo(function HistoricalMessageGroup({ isResponseComplete={isResponseComplete} copyText={extractTextFromParts(group.parts)} completedAt={group.completed_at} - forkDisabled={forkDisabled} + forkDisabled={forkDisabled || forkPointUnnamed} + forkDisabledReason={forkPointUnnamed ? "unnamed" : "busy"} onForkFromHere={ - // The group's LAST turn: forking is "up to and including this - // reply", and a merged group ends where the reply does. Gated on a - // settled turn — forking mid-stream would name a message the agent - // is still writing. + // Gated on a settled turn — forking mid-stream would name a message + // the agent is still writing. // // `source_turn_id` first: a turn produced in THIS session is named // `live-…`, which the backend cannot resolve against its own parse // — sending it forked at the tail and produced a copy of the parent. // The post-turn reparse backfills the parser's name; `id` is the - // right answer only for turns that came from the parser already. - onForkFromTurn && isResponseComplete && sourceTurns?.length + // right answer only for turns that came from the parser already, + // and for the newest reply, where the tail IS the fork point. + onForkFromTurn && isResponseComplete && forkPoint ? () => { - const turn = sourceTurns[sourceTurns.length - 1] - onForkFromTurn(turn.source_turn_id ?? turn.id) + onForkFromTurn(forkPoint.source_turn_id ?? forkPoint.id) } : undefined } @@ -1057,6 +1120,7 @@ export function MessageListView({ isRoleTransition: false, previousUserIndex: null, isLastAssistantRun: false, + isThreadTail: false, sourceTurns: singletonSourceTurns(allTurns[i]), } }) @@ -1080,6 +1144,7 @@ export function MessageListView({ item.isRoleTransition = false item.previousUserIndex = null item.isLastAssistantRun = false + item.isThreadTail = false // isRoleTransition: role differs from previous turn item if (idx > 0) { @@ -1108,6 +1173,7 @@ export function MessageListView({ lastAssistantItem.isLastAssistantRun = true lastAssistantRunning = !lastAssistantItem.isResponseComplete } + markThreadTail(items) const lastPhase = timelineTurns[timelineTurns.length - 1]?.phase ?? null if ( @@ -1207,6 +1273,7 @@ export function MessageListView({ foldEpoch={fold.epoch} onForkFromTurn={onForkFromTurn} forkDisabled={forkBusy} + isThreadTail={item.isThreadTail} />
) diff --git a/src/components/message/sub-agent-session-dialog.test.tsx b/src/components/message/sub-agent-session-dialog.test.tsx index aa580ae948..944702f0c4 100644 --- a/src/components/message/sub-agent-session-dialog.test.tsx +++ b/src/components/message/sub-agent-session-dialog.test.tsx @@ -220,6 +220,7 @@ function makeConnState(overrides: Partial): ConnectionState { parentConnectionId: "p1", isViewer: false, pendingUserMessage: null, + steeredMessageIds: [], configStale: false, configStaleKind: null, configStaleDismissed: false, diff --git a/src/components/message/turn-stats.test.tsx b/src/components/message/turn-stats.test.tsx index 2634a29ac8..d0d46eb0d6 100644 --- a/src/components/message/turn-stats.test.tsx +++ b/src/components/message/turn-stats.test.tsx @@ -119,4 +119,23 @@ describe("TurnStats fork-from-here gating", () => { enMessages.Folder.chat.messageList.forkBusy ) }) + + it("says so when the reply has no name to fork at yet", async () => { + // The other reason the button greys out: a reply this session streamed is + // named `live-…` until the post-turn reparse renames it, and the backend + // silently tail-forks such an id. Saying "a turn is running" there would + // be a lie about a session that is sitting idle. + renderStats( + + ) + await userEvent.hover(screen.getByLabelText(forkLabel)) + expect(await screen.findByRole("tooltip")).toHaveTextContent( + enMessages.Folder.chat.messageList.forkNotReady + ) + }) }) diff --git a/src/components/message/turn-stats.tsx b/src/components/message/turn-stats.tsx index ebd7287455..32707414af 100644 --- a/src/components/message/turn-stats.tsx +++ b/src/components/message/turn-stats.tsx @@ -37,10 +37,14 @@ interface TurnStatsProps { * session has no live connection, the agent has no `session/fork`, or this * surface doesn't own the conversation. */ onForkFromHere?: () => void - /** Forking is possible here but not right now (a turn is in flight). The - * button stays in place, greyed out, and says why on hover — it used to - * vanish for the length of every reply, which moved the whole icon row. */ + /** Forking is possible here but not right now. The button stays in place, + * greyed out, and says why on hover — it used to vanish for the length of + * every reply, which moved the whole icon row. */ forkDisabled?: boolean + /** Why it is greyed out: a turn is in flight (`busy`), or this reply has no + * name the backend can resolve yet (`unnamed` — the post-turn reparse fills + * it in a moment later). Only read while `forkDisabled`. */ + forkDisabledReason?: "busy" | "unnamed" } const iconButtonClass = @@ -57,6 +61,7 @@ export function TurnStats({ completedAt, onForkFromHere, forkDisabled = false, + forkDisabledReason = "busy", }: TurnStatsProps) { const locale = useLocale() const t = useTranslations("Folder.chat.messageList") @@ -205,7 +210,11 @@ export function TurnStats({ - {forkDisabled ? t("forkBusy") : t("forkFromHere")} + {forkDisabled + ? forkDisabledReason === "unnamed" + ? t("forkNotReady") + : t("forkBusy") + : t("forkFromHere")} )} diff --git a/src/contexts/acp-connections-context.test.tsx b/src/contexts/acp-connections-context.test.tsx index 2f10ad992d..f3823a4151 100644 --- a/src/contexts/acp-connections-context.test.tsx +++ b/src/contexts/acp-connections-context.test.tsx @@ -4184,3 +4184,215 @@ 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", () => { + /** The note's `created_at`: when the backend injected the text. Carried onto + * the block so the runtime store can tell the agent's own copy of THIS + * message from the same words sent in an earlier round. */ + const STEER_AT = "2026-06-07T00:00:00Z" + + 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: STEER_AT, 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", + createdAt: STEER_AT, + }, + ]) + 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([]) + }) + + it("gives the note its strip back when a snapshot replaces the live message", async () => { + // A mid-turn re-attach (WS reconnect in server mode) hydrates the backend's + // live message, which carries no steering block — the wire has no such kind + // — so the spliced message is gone from the transcript. Holding on to the + // adoption there would hide the strip for a message that is now rendered + // NOWHERE, the one failure worse than rendering it twice. + const handlers = await connectOwner() + emitAcpEvent(handlers, { + seq: 1, + connection_id: "spawned-conn", + type: "status_changed", + status: "prompting", + }) + emitAcpEvent(handlers, submitted(2, "n1", "use the other API", "delivered")) + expect(conn().steeredMessageIds).toEqual(["n1"]) + + h.denormalizeSnapshot.mockReturnValue({ + connectionId: "spawned-conn", + status: "prompting", + sessionId: null, + modes: null, + configOptions: null, + availableCommands: null, + usage: null, + liveMessage: { + id: "lm-server", + role: "assistant", + content: [{ type: "text", text: "half one" }], + startedAt: 0, + }, + pendingPermission: null, + pendingAskQuestion: null, + pendingUserMessage: null, + promptCapabilities: null, + selectorsReady: false, + supportsFork: false, + configStale: false, + configStaleKind: null, + backgroundOutstanding: 0, + activeDelegations: [], + lastError: null, + lastErrorDetails: null, + eventSeq: 9, + }) + hydrateSnapshot(handlers, { + event_seq: 9, + } as unknown as LiveSessionSnapshot) + + expect(steeringBlocks()).toEqual([]) + expect(conn().steeredMessageIds).toEqual([]) + }) +}) diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index cab2202a0e..f85709e809 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -203,6 +203,25 @@ 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. + * + * `createdAt` (ISO, the note's `created_at`) is taken before the backend + * hands the text to the agent (`submit_feedback_native`), on the machine the + * agent runs on — so it is directly comparable with, and earlier than, the + * timestamp the agent writes when it records this message in its own + * transcript. That ordering is what lets the runtime store tell the agent's + * copy of THIS message from the same words sent in an earlier round (see + * `suppressPersistedSteeredPrompts`), and it is the time the message shows. + */ + | { type: "steering"; id: string; text: string; createdAt: string } export interface LiveMessage { id: string @@ -233,6 +252,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 @@ -611,6 +643,14 @@ type Action = contextKey: string entries: PlanEntryInfo[] } + | { + type: "STEERING_MESSAGE" + contextKey: string + id: string + text: string + /** The note's `created_at` (ISO) — see the `steering` block. */ + createdAt: string + } | { type: "CLAUDE_API_RETRY" contextKey: string @@ -1142,6 +1182,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 @@ -1340,6 +1384,7 @@ function connectionsReducer( liveMessage: null, pendingPermission: null, pendingUserMessage: null, + steeredMessageIds: EMPTY_STEERED_MESSAGE_IDS, pendingQuestion: null, pendingAskQuestion: null, pendingPlanApproval: null, @@ -1399,6 +1444,7 @@ function connectionsReducer( liveMessage: null, pendingPermission: null, pendingUserMessage: null, + steeredMessageIds: EMPTY_STEERED_MESSAGE_IDS, pendingQuestion: null, pendingAskQuestion: null, pendingPlanApproval: null, @@ -1553,6 +1599,24 @@ function connectionsReducer( availableCommands: action.patch.availableCommands, usage: action.patch.usage, liveMessage: hydratedLiveMessage, + // The snapshot's live message REPLACES the local one, and the wire has + // no `steering` block (the backend never records one — see + // `snapshot-denormalize`), so every adopted mid-turn message is gone + // from the transcript with it. Keeping the adoption ids past that would + // hide the strips for messages that are no longer rendered anywhere, + // which is the one failure worse than showing them twice. Drop them: + // the notes list (hydrated from the same snapshot's `feedback`) shows + // those messages as strips again. + // + // Unconditional, including a null `liveMessage` — where the runtime + // mirror keeps the previous one (it never writes null) and the steered + // turn is still on screen for now. Holding the ids would be right for + // that frame and wrong from the next delta on, which rebuilds the live + // message without the block and would leave the message nowhere for + // the rest of the turn. The cost is the opposite way round: a message + // whose persisted copy the transcript is already showing gets a strip + // beside it until the turn ends. Turn-scoped, and visible. + steeredMessageIds: EMPTY_STEERED_MESSAGE_IDS, pendingPermission: hydratedPendingPermission, pendingAskQuestion: action.patch.pendingAskQuestion, pendingPlanApproval: action.patch.pendingPlanApproval, @@ -1625,6 +1689,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. @@ -2410,6 +2476,39 @@ 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, + createdAt: action.createdAt, + }, + ], + }, + steeredMessageIds: [...conn.steeredMessageIds, action.id], + }) + return next + } + case "CLAUDE_API_RETRY": { const conn = state.get(action.contextKey) if (!conn) return state @@ -3604,6 +3703,29 @@ 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, + createdAt: e.item.created_at, + }) + 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..ae0cefcae2 100644 --- a/src/contexts/conversation-runtime-context.test.tsx +++ b/src/contexts/conversation-runtime-context.test.tsx @@ -2465,3 +2465,495 @@ 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. + */ +/** When the backend injected the steered message — its note's `created_at`, + * stamped where the agent runs. Later than the `turn()` helper's default + * timestamp below, so an unrelated turn from earlier history is provably + * older than any steer in these tests. */ +const STEER_AT = "2026-05-28T00:05:00.000Z" +/** A turn the agent wrote after that injection — i.e. its own copy. */ +const AFTER_STEER = "2026-05-28T00:05:01.000Z" + +describe("buildStreamingTurnsFromLiveMessage - mid-turn steering messages", () => { + function live(content: LiveContentBlock[]): LiveMessage { + return { id: "lm-steer", role: "assistant", content, startedAt: 0 } + } + + it("stamps the message with the instant it was sent, not the turn's start", () => { + const turns = buildStreamingTurnsFromLiveMessage( + 1, + live([ + { type: "text", text: "working on it" }, + { type: "steering", id: "note-1", text: "stop", createdAt: STEER_AT }, + ]) + ).turns + const [reply, user] = turns + expect(user.timestamp).toBe(STEER_AT) + expect(reply.timestamp).toBe(new Date(0).toISOString()) + }) + + it("falls back to the turn's start when the stamp is unreadable", () => { + const turns = buildStreamingTurnsFromLiveMessage( + 1, + live([{ type: "steering", id: "note-1", text: "stop", createdAt: "" }]) + ).turns + expect(turns[0].timestamp).toBe(new Date(0).toISOString()) + }) + + 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", + createdAt: STEER_AT, + }, + ]) + ).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", + createdAt: STEER_AT, + }, + { 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", createdAt: STEER_AT }, + { 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", createdAt: STEER_AT }, + { 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", createdAt: STEER_AT }, + { type: "text", text: "b" }, + { type: "steering", id: "n2", text: "two", createdAt: STEER_AT }, + { 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, + timestamp = "2026-05-28T00:00:00.000Z" + ): MessageTurn { + return { + id, + role, + blocks: [{ type: "text" as const, text }], + timestamp, + } + } + + 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", + createdAt: STEER_AT, + }, + { 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 — written after the injection, which is + // what marks it as this message's copy. The backend cannot stamp an + // in-flight prompt here: it matches the pending prompt against the + // transcript TAIL, and the tail is now the steered message. + mockGetFolderConversation.mockResolvedValueOnce( + detailWith( + [ + turn("p-1", "user", "the original prompt"), + turn("p-2", "user", "use the other API", AFTER_STEER), + ], + null + ) + ) + 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 this round's prompt, even when a steer repeats it", async () => { + // And with NO in-flight stamp, which is the shape the backend produces + // once the steered message is on the transcript tail: the prompt is safe + // because the agent wrote it before the user steered, not because it was + // named. + renderProvider() + const api = () => runtimeHolder.current! + act(() => { + api().setLiveMessage( + 99, + { + id: "lm-2", + role: "assistant", + content: [ + { + type: "steering", + id: "note-1", + text: "continue", + createdAt: STEER_AT, + }, + ], + startedAt: 0, + }, + true + ) + }) + mockGetFolderConversation.mockResolvedValueOnce( + detailWith([turn("p-1", "user", "continue")], null) + ) + await act(async () => { + api().refetchDetail(99, { preserveLive: true }) + }) + // Both survive: hiding a prompt is the one failure worse than showing a + // duplicate. + expect(userTexts(api().getTimelineTurns(99))).toEqual([ + "continue", + "continue", + ]) + }) + + it("leaves an earlier round's identical prompt in history", async () => { + // Steered text is short and repeatable ("continue", "stop"), and content is + // the only thing linking the live copy to the persisted one. Matching it + // across the whole window would hide the SAME words the user sent three + // rounds ago for as long as this turn runs. Only a turn written after the + // injection can be a copy of it. + renderProvider() + const api = () => runtimeHolder.current! + act(() => { + api().setLiveMessage( + 99, + { + id: "lm-3", + role: "assistant", + content: [ + { type: "text", text: "half one" }, + { + type: "steering", + id: "note-1", + text: "continue", + createdAt: STEER_AT, + }, + { type: "text", text: "half two" }, + ], + startedAt: 0, + }, + true + ) + }) + mockGetFolderConversation.mockResolvedValueOnce( + detailWith( + [ + turn("p-1", "user", "continue"), // an earlier round, same words + turn("p-2", "assistant", "sure"), + turn("p-3", "user", "now do the thing"), // this turn's prompt + turn("p-4", "user", "continue", AFTER_STEER), // the agent's copy + ], + null + ) + ) + await act(async () => { + api().refetchDetail(99, { preserveLive: true }) + }) + // History intact; only the copy inside the running round is folded away. + expect(userTexts(api().getTimelineTurns(99))).toEqual([ + "continue", + "now do the thing", + "continue", + ]) + const steered = api() + .getTimelineTurns(99) + .filter( + (t) => + t.turn.role === "user" && + t.turn.blocks[0]?.type === "text" && + t.turn.blocks[0].text === "continue" + ) + expect(steered.map((t) => t.phase)).toEqual(["persisted", "streaming"]) + }) + + it("suppresses nothing when the message carries no readable instant", async () => { + // An unparseable stamp on either side leaves no way to tell this round's + // copy from an older message, so both copies render — a duplicate, never a + // disappearance. + renderProvider() + const api = () => runtimeHolder.current! + act(() => { + api().setLiveMessage( + 99, + { + id: "lm-4", + role: "assistant", + content: [ + { type: "steering", id: "note-1", text: "continue", createdAt: "" }, + ], + startedAt: Date.parse(STEER_AT), + }, + true + ) + }) + mockGetFolderConversation.mockResolvedValueOnce( + detailWith([turn("p-9", "user", "continue", AFTER_STEER)], null) + ) + await act(async () => { + api().refetchDetail(99, { preserveLive: true }) + }) + expect(userTexts(api().getTimelineTurns(99))).toEqual([ + "continue", + "continue", + ]) + }) + + it("leaves a promoted local turn from an earlier round alone", async () => { + // `localTurns` render as phase "persisted" but are NOT part of the detail's + // projection — a mid-turn refetch preserves them, and they are stamped from + // the client clock, so they are never compared against the injection + // instant. Only what the detail itself lists can be the agent's copy. + renderProvider() + const api = () => runtimeHolder.current! + const earlierReply: LiveMessage = { + id: "lm-earlier", + role: "assistant", + content: [{ type: "text", text: "done" }], + startedAt: 0, + } + act(() => { + api().appendOptimisticTurn(99, turn("o-1", "user", "continue"), "o-1") + }) + act(() => { + api().completeTurn(99, earlierReply) + }) + act(() => { + api().setLiveMessage( + 99, + { + id: "lm-5", + role: "assistant", + content: [ + { type: "text", text: "half one" }, + { + type: "steering", + id: "note-1", + text: "continue", + createdAt: STEER_AT, + }, + { type: "text", text: "half two" }, + ], + startedAt: 0, + }, + true + ) + }) + mockGetFolderConversation.mockResolvedValueOnce( + detailWith([turn("p-1", "user", "now do the thing")], "p-1") + ) + await act(async () => { + api().refetchDetail(99, { preserveLive: true }) + }) + expect(userTexts(api().getTimelineTurns(99))).toEqual([ + "now do the thing", + "continue", // the earlier round's promoted prompt + "continue", // this round's steer, live + ]) + }) +}) diff --git a/src/hooks/use-connection.ts b/src/hooks/use-connection.ts index 4dff03bebb..91e070478e 100644 --- a/src/hooks/use-connection.ts +++ b/src/hooks/use-connection.ts @@ -35,6 +35,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[] = [] // Stable empty reference: a new [] on every render would break the memo below // for every connection that has no async tasks — i.e. almost all of them. const EMPTY_ASYNC_TASKS: AsyncTaskRecord[] = [] @@ -69,6 +70,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 @@ -233,6 +238,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 @@ -339,6 +346,7 @@ export function useConnection(contextKey: string): UseConnectionReturn { availableCommands, pendingPermission, pendingUserMessage, + steeredMessageIds, pendingQuestion, pendingAskQuestion, pendingPlanApproval, @@ -380,6 +388,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/i18n/messages/ar.json b/src/i18n/messages/ar.json index 486b97c0d2..19958d8afb 100644 --- a/src/i18n/messages/ar.json +++ b/src/i18n/messages/ar.json @@ -3135,6 +3135,7 @@ "copyMessage": "نسخ", "forkFromHere": "تفريع من هنا", "forkBusy": "لا يمكن التفريع أثناء تنفيذ دور", + "forkNotReady": "لا يمكن التفريع من هذا الرد بعد، أعد المحاولة بعد قليل", "copied": "تم النسخ", "selectionActions": "إجراءات التحديد", "selectionCopy": "نسخ النص", diff --git a/src/i18n/messages/de.json b/src/i18n/messages/de.json index fcb9f2e74b..a5d7123cac 100644 --- a/src/i18n/messages/de.json +++ b/src/i18n/messages/de.json @@ -3135,6 +3135,7 @@ "copyMessage": "Kopieren", "forkFromHere": "Ab hier verzweigen", "forkBusy": "Verzweigen nicht möglich, während ein Zug läuft", + "forkNotReady": "Von dieser Antwort kann noch nicht verzweigt werden – bitte gleich erneut versuchen", "copied": "Kopiert", "selectionActions": "Auswahlaktionen", "selectionCopy": "Text kopieren", diff --git a/src/i18n/messages/en.json b/src/i18n/messages/en.json index 9b60e88e86..7e9d1cf926 100644 --- a/src/i18n/messages/en.json +++ b/src/i18n/messages/en.json @@ -3135,6 +3135,7 @@ "copyMessage": "Copy", "forkFromHere": "Fork from here", "forkBusy": "Can't fork while a turn is running", + "forkNotReady": "Can't fork from this reply just yet — try again in a moment", "copied": "Copied", "selectionActions": "Selection actions", "selectionCopy": "Copy Text", diff --git a/src/i18n/messages/es.json b/src/i18n/messages/es.json index d49ac8202d..3add30566c 100644 --- a/src/i18n/messages/es.json +++ b/src/i18n/messages/es.json @@ -3135,6 +3135,7 @@ "copyMessage": "Copiar", "forkFromHere": "Bifurcar desde aquí", "forkBusy": "No se puede bifurcar mientras hay un turno en curso", + "forkNotReady": "Todavía no se puede bifurcar desde esta respuesta; inténtalo en un momento", "copied": "Copiado", "selectionActions": "Acciones de selección", "selectionCopy": "Copiar texto", diff --git a/src/i18n/messages/fr.json b/src/i18n/messages/fr.json index 15ca47199c..08cb3977cc 100644 --- a/src/i18n/messages/fr.json +++ b/src/i18n/messages/fr.json @@ -3135,6 +3135,7 @@ "copyMessage": "Copier", "forkFromHere": "Bifurquer d'ici", "forkBusy": "Impossible de bifurquer pendant qu'un tour est en cours", + "forkNotReady": "Impossible de bifurquer depuis cette réponse pour le moment, réessayez dans un instant", "copied": "Copié", "selectionActions": "Actions de sélection", "selectionCopy": "Copier le texte", diff --git a/src/i18n/messages/ja.json b/src/i18n/messages/ja.json index e5edd06933..4f6a0800e6 100644 --- a/src/i18n/messages/ja.json +++ b/src/i18n/messages/ja.json @@ -3135,6 +3135,7 @@ "copyMessage": "コピー", "forkFromHere": "ここから分岐", "forkBusy": "ターンの実行中は分岐できません", + "forkNotReady": "この返信からはまだ分岐できません。少し待ってからお試しください", "copied": "コピー済み", "selectionActions": "選択範囲の操作", "selectionCopy": "テキストをコピー", diff --git a/src/i18n/messages/ko.json b/src/i18n/messages/ko.json index 93adb65cd5..0742dbba76 100644 --- a/src/i18n/messages/ko.json +++ b/src/i18n/messages/ko.json @@ -3135,6 +3135,7 @@ "copyMessage": "복사", "forkFromHere": "여기서 분기", "forkBusy": "턴이 실행 중일 때는 분기할 수 없습니다", + "forkNotReady": "아직 이 응답에서 분기할 수 없습니다. 잠시 후 다시 시도하세요", "copied": "복사됨", "selectionActions": "선택 영역 작업", "selectionCopy": "텍스트 복사", diff --git a/src/i18n/messages/pt.json b/src/i18n/messages/pt.json index aef785fb5a..66dfbfb5c9 100644 --- a/src/i18n/messages/pt.json +++ b/src/i18n/messages/pt.json @@ -3135,6 +3135,7 @@ "copyMessage": "Copiar", "forkFromHere": "Bifurcar a partir daqui", "forkBusy": "Não é possível bifurcar enquanto há um turno em andamento", + "forkNotReady": "Ainda não é possível bifurcar a partir desta resposta; tente novamente em instantes", "copied": "Copiado", "selectionActions": "Ações de seleção", "selectionCopy": "Copiar texto", diff --git a/src/i18n/messages/zh-CN.json b/src/i18n/messages/zh-CN.json index c223efe95a..e01a4638eb 100644 --- a/src/i18n/messages/zh-CN.json +++ b/src/i18n/messages/zh-CN.json @@ -3135,6 +3135,7 @@ "copyMessage": "复制", "forkFromHere": "从此处分叉", "forkBusy": "当前有回合正在进行，暂不可分叉", + "forkNotReady": "这条回复暂时还不能作为分叉点，请稍后再试", "copied": "已复制", "selectionActions": "选中内容操作", "selectionCopy": "复制文本", diff --git a/src/i18n/messages/zh-TW.json b/src/i18n/messages/zh-TW.json index 26dc4b2079..5b2e35e138 100644 --- a/src/i18n/messages/zh-TW.json +++ b/src/i18n/messages/zh-TW.json @@ -3135,6 +3135,7 @@ "copyMessage": "複製", "forkFromHere": "從此處分叉", "forkBusy": "目前有回合正在進行，暫不可分叉", + "forkNotReady": "這則回覆暫時還不能作為分叉點，請稍後再試", "copied": "已複製", "selectionActions": "選取內容操作", "selectionCopy": "複製文字", diff --git a/src/stores/conversation-runtime-store.ts b/src/stores/conversation-runtime-store.ts index f89b8b1bd6..e2c8f21cba 100644 --- a/src/stores/conversation-runtime-store.ts +++ b/src/stores/conversation-runtime-store.ts @@ -606,6 +606,21 @@ 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"] + /** + * Overrides the live message's start for this group. Only a `user` group sets + * it, to the instant the message was actually sent (the note's `created_at`) + * rather than the moment the reply it interrupted began. Display only — + * `suppressPersistedSteeredPrompts` reads that instant off the block itself, + * so an unreadable stamp falling back here can never widen its bound. + */ + timestamp?: string +} + // 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 @@ -981,6 +996,21 @@ function mainProseContinuations( return continues } +/** Prefix of every turn id minted from a live message below. */ +const LIVE_TURN_ID_PREFIX = "live-" + +/** + * True for a turn this client streamed itself, which therefore has no name in + * the agent's transcript yet. The backend cannot resolve such an id against its + * own parse — `fork_session` degrades an unresolvable fork point to a TAIL fork + * rather than refusing the click — so anything that sends a turn id to the + * backend must prefer the parser's name (`MessageTurn.source_turn_id`, filled + * in by the post-turn reparse) and treat this as "not namable yet". + */ +export function isLiveTurnId(id: string): boolean { + return id.startsWith(LIVE_TURN_ID_PREFIX) +} + export function buildStreamingTurnsFromLiveMessage( conversationId: number, liveMessage: LiveMessage, @@ -1176,7 +1206,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 @@ -1199,17 +1233,38 @@ 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 }], + // Display only; an unreadable stamp falls back to the turn's start. + // The persisted-copy match reads the stamp itself, not this, so it is + // never fooled by that fallback. + timestamp: Number.isFinite(Date.parse(block.createdAt)) + ? new Date(block.createdAt).toISOString() + : undefined, + }) + 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": @@ -1441,15 +1496,15 @@ 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, - timestamp, + ? `${LIVE_TURN_ID_PREFIX}${conversationId}-${liveMessage.id}` + : `${LIVE_TURN_ID_PREFIX}${conversationId}-${liveMessage.id}-${i}`, + role: group.role, + blocks: group.blocks, + timestamp: group.timestamp ?? timestamp, })) return { turns, inProgressToolCallIds } @@ -1714,6 +1769,17 @@ function userTurnContentKey(turn: MessageTurn): string { ) } +/** The same key for a mid-turn steered message, whose persisted copy is a user + * turn carrying exactly its text (see `suppressPersistedSteeredPrompts`). */ +function steeredContentKey(text: string): string { + return userTurnContentKey({ + id: "", + role: "user", + blocks: [{ type: "text", text }], + timestamp: "", + }) +} + /** * Rewrite the launching tool call's `[[codeg-background-task]]` marker in a turn * list so `AgentToolCallPart` flips from "running in background" to its @@ -3223,6 +3289,94 @@ 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 — but content ALONE cannot say which message it + * matched. Steered text is short and repeatable ("continue", "stop", "not + * done"), so a bare content match reaches back and hides the identical prompt + * the user sent three rounds ago, for as long as the turn runs. Suppressing a + * user turn is the one failure that hides a message rather than duplicating + * it, so the match is bounded by WHEN: + * + * - each `steering` block carries the note's `created_at`, taken on the + * agent's machine BEFORE the backend handed it the text (an invariant of + * `submit_feedback_native`); + * - the agent's copy is therefore written after it, so a persisted turn + * older than that instant is by construction a different message — + * including this round's own prompt, which the agent wrote before the user + * steered. + * + * Candidates are further limited to turns the DETAIL projected, so every + * timestamp compared comes from the agent's own clock; a promoted `localTurns` + * copy (client clock, and kept across a mid-turn refetch by `preserveLive`) is + * never a candidate. Anything unreadable — no parseable instant on either side + * — suppresses nothing, leaving the two copies to coexist: a visible duplicate, + * never a hidden message. + * + * Deliberately NOT anchored on `detail.in_flight_user_turn_id`: the backend + * stamps that by matching the pending prompt against the transcript TAIL (see + * `apply_in_flight_message_id`), and once the agent has written the steered + * message the tail is that message, not the prompt — so the stamp is gone in + * exactly the shape this function exists for. + */ +function suppressPersistedSteeredPrompts( + prefix: ConversationTimelineTurn[], + session: ConversationRuntimeSession +): ConversationTimelineTurn[] { + // Content key → the earliest instant a copy of it could have been written. + // Read from the blocks rather than from the built turns: a block with no + // readable stamp shows under the turn's start time, and treating THAT as the + // bound would put this round's own prompt in range. + let steeredAt: Map | null = null + let earliestSteerAt = Number.POSITIVE_INFINITY + for (const block of session.liveMessage?.content ?? []) { + if (block.type !== "steering") continue + const at = Date.parse(block.createdAt) + if (!Number.isFinite(at)) continue + const key = steeredContentKey(block.text) + steeredAt ??= new Map() + const known = steeredAt.get(key) + if (known === undefined || at < known) steeredAt.set(key, at) + if (at < earliestSteerAt) earliestSteerAt = at + } + if (!steeredAt) return prefix + const detailTurns = session.detail?.turns + if (!detailTurns) return prefix + // Ids are unique across the timeline's phases (a same-id copy in another + // phase is the same turn — see `dedupeTimeline`), so membership alone tells + // a detail-projected turn from a locally promoted one. + const detailUserIds = new Set() + for (const turn of detailTurns) { + if (turn.role === "user") detailUserIds.add(turn.id) + } + const filtered = prefix.filter((item) => { + if (item.phase !== "persisted" || item.turn.role !== "user") return true + if (!detailUserIds.has(item.turn.id)) return true + // Cheap gate first: everything written before the earliest steer is out, + // so history never reaches the content key (which serializes full text and + // full image data, and this runs on every streaming batch). + const at = Date.parse(item.turn.timestamp) + if (!Number.isFinite(at) || at < earliestSteerAt) return true + const steered = steeredAt.get(userTurnContentKey(item.turn)) + return steered === undefined || at < steered + }) + return filtered.length === prefix.length ? prefix : filtered +} + function computeTimeline( state: ConversationRuntimeState, conversationId: number @@ -3271,9 +3425,8 @@ function computeTimeline( } seenTailKeys?.add(key) } - deduped = collides - ? dedupeTimeline(prefix.concat(tail)) - : prefix.concat(tail) + const head = suppressPersistedSteeredPrompts(prefix, session) + deduped = collides ? dedupeTimeline(head.concat(tail)) : head.concat(tail) } timelineCache.set(session, deduped)