diff --git a/packages/ai/test/pi-native-client.test.ts b/packages/ai/test/pi-native-client.test.ts index f9adecbca37..b08712acb9f 100644 --- a/packages/ai/test/pi-native-client.test.ts +++ b/packages/ai/test/pi-native-client.test.ts @@ -48,12 +48,39 @@ function stalledBody(bytes: Uint8Array[] = []): ReadableStream { } function delayedBody(chunks: Array<{ atMs: number; bytes: Uint8Array }>): ReadableStream { + let closed = false; + const timers: Timer[] = []; + const clearTimers = () => { + closed = true; + for (const timer of timers) clearTimeout(timer); + timers.length = 0; + }; return new ReadableStream({ start(controller) { + const enqueue = (bytes: Uint8Array) => { + if (!closed) controller.enqueue(bytes); + }; for (const chunk of chunks) { - setTimeout(() => controller.enqueue(chunk.bytes), chunk.atMs); + if (chunk.atMs <= 0) { + enqueue(chunk.bytes); + } else { + timers.push(setTimeout(() => enqueue(chunk.bytes), chunk.atMs)); + } } - setTimeout(() => controller.close(), Math.max(...chunks.map(chunk => chunk.atMs)) + 1); + timers.push( + setTimeout( + () => { + if (!closed) { + clearTimers(); + controller.close(); + } + }, + Math.max(...chunks.map(chunk => chunk.atMs)) + 1, + ), + ); + }, + cancel() { + clearTimers(); }, }); } diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 459ec14a87e..1dd91d879e2 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,9 @@ ## [Unreleased] +### Fixed + +- Fixed rapid queued steer/follow-up image submissions racing into split or dropped pending entries by serializing queue mutations; added opt-in `coalescing` queue mode to merge rapid consecutive queued user entries while preserving attachments, hidden magic-keyword companions, restore behavior, delivery, and `[Image #N]` marker numbering. ## [16.3.11] - 2026-07-06 ### Changed diff --git a/packages/coding-agent/src/config/keybindings.ts b/packages/coding-agent/src/config/keybindings.ts index 385fa225812..99408fa3f51 100644 --- a/packages/coding-agent/src/config/keybindings.ts +++ b/packages/coding-agent/src/config/keybindings.ts @@ -33,6 +33,7 @@ interface AppKeybindings { "app.message.followUp": true; "app.retry": true; "app.message.dequeue": true; + "app.message.expandQueue": true; "app.clipboard.pasteImage": true; "app.clipboard.pasteTextRaw": true; "app.clipboard.copyLine": true; @@ -142,6 +143,10 @@ export const KEYBINDINGS = { defaultKeys: "alt+up", description: "Dequeue message", }, + "app.message.expandQueue": { + defaultKeys: "alt+o", + description: "Expand/collapse queued message preview", + }, "app.clipboard.pasteImage": { defaultKeys: getDefaultPasteImageKeys(), description: "Paste image or text from clipboard", @@ -247,6 +252,7 @@ const KEYBINDING_NAME_MIGRATIONS = { followUp: "app.message.followUp", retry: "app.retry", dequeue: "app.message.dequeue", + expandQueue: "app.message.expandQueue", pasteImage: "app.clipboard.pasteImage", pasteTextRaw: "app.clipboard.pasteTextRaw", copyLine: "app.clipboard.copyLine", diff --git a/packages/coding-agent/src/config/settings-schema.ts b/packages/coding-agent/src/config/settings-schema.ts index 00af4ed698d..946fc0be5f4 100644 --- a/packages/coding-agent/src/config/settings-schema.ts +++ b/packages/coding-agent/src/config/settings-schema.ts @@ -1407,25 +1407,25 @@ export const SETTINGS_SCHEMA = { // Conversation flow steeringMode: { type: "enum", - values: ["all", "one-at-a-time"] as const, + values: ["all", "one-at-a-time", "coalescing"] as const, default: "one-at-a-time", ui: { tab: "interaction", group: "Input", label: "Steering Mode", - description: "How to process queued messages while agent is working", + description: "How to process queued steering messages while the agent is working", }, }, followUpMode: { type: "enum", - values: ["all", "one-at-a-time"] as const, + values: ["all", "one-at-a-time", "coalescing"] as const, default: "one-at-a-time", ui: { tab: "interaction", group: "Input", label: "Follow-Up Mode", - description: "How to drain follow-up messages after a turn completes", + description: "How to drain queued follow-up messages after a turn completes", }, }, @@ -1440,6 +1440,23 @@ export const SETTINGS_SCHEMA = { description: "When steering messages interrupt tool execution", }, }, + pendingQueueCollapseLines: { + type: "number", + default: 5, + ui: { + tab: "interaction", + group: "Input", + label: "Queued Message Preview Lines", + description: + "How many leading lines of each queued steer/follow-up message the pending bar shows before collapsing the rest to `(+N)`. Alt+O expands every entry to its full text and toggles back to this collapsed preview.", + options: [ + { value: "1", label: "1 line" }, + { value: "3", label: "3 lines" }, + { value: "5", label: "5 lines" }, + { value: "10", label: "10 lines" }, + ], + }, + }, "loop.mode": { type: "enum", diff --git a/packages/coding-agent/src/modes/components/custom-editor.ts b/packages/coding-agent/src/modes/components/custom-editor.ts index bf144ff1160..00fc0f69289 100644 --- a/packages/coding-agent/src/modes/components/custom-editor.ts +++ b/packages/coding-agent/src/modes/components/custom-editor.ts @@ -25,6 +25,7 @@ type ConfigurableEditorAction = Extract< | "app.editor.external" | "app.history.search" | "app.message.dequeue" + | "app.message.expandQueue" | "app.retry" | "app.clipboard.pasteImage" | "app.clipboard.pasteTextRaw" @@ -47,6 +48,7 @@ const DEFAULT_ACTION_KEYS: Record = { "app.editor.external": ["ctrl+g"], "app.history.search": ["ctrl+r"], "app.message.dequeue": ["alt+up"], + "app.message.expandQueue": ["alt+o"], "app.retry": ["alt+r"], "app.clipboard.pasteImage": ["ctrl+v"], "app.clipboard.pasteTextRaw": ["ctrl+shift+v", "alt+shift+v"], @@ -414,12 +416,16 @@ export class CustomEditor extends Editor { onPasteTextRaw?: () => void; /** Called when the configured dequeue shortcut is pressed. */ onDequeue?: () => void; + /** Called when the configured expand-queue shortcut is pressed (toggle queued-message preview). */ + onExpandQueue?: () => void; /** Called when the configured retry shortcut is pressed. */ onRetry?: () => void; /** Called when Caps Lock is pressed. */ onCapsLock?: () => void; /** Called when left-arrow is pressed while the editor is empty (cursor necessarily at start). */ onLeftAtStart?: () => void; + /** Called when Up is pressed on an empty editor. Return true to consume it before history navigation. */ + onUpWhenEmpty?: () => boolean; /** Fired when a sustained space-bar hold is recognized — the push-to-talk STT start. The * optimistically-typed spaces have already been deleted by the time this runs. */ @@ -687,6 +693,10 @@ export class CustomEditor extends Editor { return; } + if (canonical === "up" && this.onUpWhenEmpty && this.getText().trim() === "" && this.onUpWhenEmpty()) { + return; + } + // Space-hold push-to-talk: a sustained space bar starts/stops STT instead of typing spaces. if (this.#handleSpaceHold(data, canonical)) return; @@ -801,6 +811,12 @@ export class CustomEditor extends Editor { return; } + // Intercept configured expand-queue shortcut (toggle queued-message preview) + if (this.#matchesAction(canonical, "app.message.expandQueue") && this.onExpandQueue) { + this.onExpandQueue(); + return; + } + // Intercept configured copy-prompt shortcut if (this.#matchesAction(canonical, "app.clipboard.copyPrompt") && this.onCopyPrompt) { this.onCopyPrompt(); diff --git a/packages/coding-agent/src/modes/components/queue-mode-selector.ts b/packages/coding-agent/src/modes/components/queue-mode-selector.ts index cb7d43cdd46..c51cade528c 100644 --- a/packages/coding-agent/src/modes/components/queue-mode-selector.ts +++ b/packages/coding-agent/src/modes/components/queue-mode-selector.ts @@ -1,5 +1,6 @@ import { Container, type SelectItem, SelectList, type SgrMouseEvent } from "@oh-my-pi/pi-tui"; import { getSelectListTheme } from "../../modes/theme/theme"; +import type { QueueMode } from "../../session/agent-session"; import { DynamicBorder } from "./dynamic-border"; import { routeSelectListMouseWithTopBorder } from "./select-list-mouse-routing"; @@ -9,18 +10,19 @@ import { routeSelectListMouseWithTopBorder } from "./select-list-mouse-routing"; export class QueueModeSelectorComponent extends Container { #selectList: SelectList; - constructor( - currentMode: "all" | "one-at-a-time", - onSelect: (mode: "all" | "one-at-a-time") => void, - onCancel: () => void, - ) { + constructor(currentMode: QueueMode, onSelect: (mode: QueueMode) => void, onCancel: () => void) { super(); const queueModes: SelectItem[] = [ { value: "one-at-a-time", label: "one-at-a-time", - description: "Process queued messages one by one (recommended)", + description: "Process queued messages one by one (default)", + }, + { + value: "coalescing", + label: "coalescing", + description: "Merge rapid consecutive queued messages into one pending entry", }, { value: "all", label: "all", description: "Process all queued messages at once" }, ]; @@ -38,7 +40,7 @@ export class QueueModeSelectorComponent extends Container { } this.#selectList.onSelect = item => { - onSelect(item.value as "all" | "one-at-a-time"); + onSelect(item.value as QueueMode); }; this.#selectList.onCancel = () => { diff --git a/packages/coding-agent/src/modes/components/queued-message-box.ts b/packages/coding-agent/src/modes/components/queued-message-box.ts new file mode 100644 index 00000000000..db243606424 --- /dev/null +++ b/packages/coding-agent/src/modes/components/queued-message-box.ts @@ -0,0 +1,198 @@ +import type { Component } from "@oh-my-pi/pi-tui"; +import { truncateToWidth, visibleWidth } from "@oh-my-pi/pi-tui"; +import { theme } from "../theme/theme"; +import { fit } from "./overlay-box"; + +/** + * A single queued steer / follow-up message rendered inside a bordered box. + * + * Model (matches the approved steering-preview design): + * - Body rows carry a tree-style gutter that distinguishes a hard newline + * (`├─` / `└─` for the first visual row of a logical line) from a soft + * wrap (`│ ` for continuation rows produced by word-wrapping a long line). + * - Each logical line is word-wrapped to the content width (Bun.wrapAnsi), + * so a long line flows onto continuation rows instead of being truncated + * mid-word. + * - When collapsed, the box keeps at most `collapseLines` visual rows and + * reports the remainder in the bottom border as `+N rows · M chars` + * (English), centered like the title sits in the top rule. + * - The horizontal rules use a light dashed glyph (`╌`) for the steering-box + * look; the vertical borders are dashed too (`╎`). This is local to queued + * overlays keep their own solid borders via `overlay-box`. + * - Callers may suppress the top rule to stack a custom header above the + * boxed body while preserving column-for-column alignment. + */ +const DASH = "╌"; +const VDASH = "╎"; + +/** Word-wrap a (plain, sanitized) line to `width` columns; returns visual rows. */ +function wrapLine(line: string, width: number): string[] { + if (width <= 0 || line.length === 0) return [""]; + // Bun.wrapAnsi is ANSI-aware and word-wraps at width, preserving interior + // whitespace runs; it joins visual rows with `\n`. + return Bun.wrapAnsi(line, width, { wordWrap: true, hard: true }).split("\n"); +} + +export interface QueuedMessageBoxOptions { + /** Max visual rows when not expanded (mirrors `pendingQueueCollapseLines`). */ + collapseLines: number; + /** When true, show every visual row and no truncation hint. */ + expanded: boolean; + /** When false, omit the top rule so a caller can supply a custom header. */ + showTopBorder?: boolean; + /** Optional footer text inset into the bottom rule (e.g. the queue hint). */ + footerText?: string; +} + +export class QueuedMessageBox implements Component { + #title: string; + #lines: readonly string[]; + #collapseLines: number; + #expanded: boolean; + #showTopBorder: boolean; + #footerText: string | undefined; + #cachedWidth = -1; + #cachedLines: string[] | undefined; + + /** + * @param title Box title inset into the top rule (e.g. `Steer`, `Follow-up`); + * empty for the streaming-steer box whose top rule is animated. + * @param lines The logical message lines (already sanitized by the caller); + * each is word-wrapped at render time, so pass every line — this + * component performs the collapse, not the caller. + */ + constructor(title: string, lines: readonly string[], opts: QueuedMessageBoxOptions) { + this.#title = title; + this.#lines = lines; + this.#collapseLines = opts.collapseLines; + this.#expanded = opts.expanded; + this.#showTopBorder = opts.showTopBorder ?? true; + this.#footerText = opts.footerText; + } + + invalidate(): void { + this.#cachedWidth = -1; + this.#cachedLines = undefined; + } + + render(width: number): readonly string[] { + if (this.#cachedLines && this.#cachedWidth === width) return this.#cachedLines; + // Below this the frame has no room; fall back to plain indented rows so a + // very narrow terminal still shows something without breaking layout. + if (width < 8) { + const flat = this.#title + ? [theme.fg("dim", `${this.#title}:`), ...this.#lines.map(l => theme.fg("dim", ` ${l}`))] + : this.#lines.map(l => theme.fg("dim", ` ${l}`)); + this.#cachedWidth = width; + this.#cachedLines = flat; + return flat; + } + // Body row layout: `│ │` → content = width - 7. + const contentW = Math.max(0, width - 7); + const lastLogical = this.#lines.length - 1; + const allRows: Array<{ li: number; vi: number; text: string }> = []; + const lineRowCount: number[] = []; + this.#lines.forEach((line, li) => { + const visual = wrapLine(line, contentW); + lineRowCount[li] = visual.length; + visual.forEach((v, vi) => { + allRows.push({ li, vi, text: v }); + }); + }); + let shown = allRows; + let hiddenRows = 0; + let hiddenChars = 0; + if (!this.#expanded && allRows.length > this.#collapseLines) { + shown = allRows.slice(0, this.#collapseLines); + hiddenRows = allRows.length - this.#collapseLines; + hiddenChars = allRows.slice(this.#collapseLines).reduce((a, r) => a + visibleWidth(r.text), 0); + } + // `└─` marks a clean single-row end (last logical line, not wrapped, nothing + // truncated). A wrapped last line continues onto `│` rows, so its first row + // is `├─`; a truncated queue likewise ends on `├─`. + const isTruncated = hiddenRows > 0; + const gp = (g: string) => theme.fg("muted", g); + const out: string[] = this.#showTopBorder ? [this.#topBorder(width)] : []; + for (const r of shown) { + const gutter = + r.vi === 0 ? (r.li === lastLogical && !isTruncated && lineRowCount[r.li] === 1 ? "└─ " : "├─ ") : "│ "; + out.push(this.#bodyRow(gutter, r.text, width, gp)); + } + if (allRows.length === 0) out.push(this.#bodyRow("│ ", "", width, gp)); + out.push(this.#bottomBorder(width, hiddenRows, hiddenChars, this.#footerText)); + + this.#cachedWidth = width; + this.#cachedLines = out; + return out; + } + + #topBorder(width: number): string { + const box = theme.boxRound; + const inner = Math.max(0, width - 2); + if (!this.#title) return theme.fg("border", box.topLeft + DASH.repeat(inner) + box.topRight); + const shown = truncateToWidth(` ${this.#title} `, Math.max(0, inner - 2)); + const fill = Math.max(0, inner - 1 - visibleWidth(shown)); + return ( + theme.fg("border", box.topLeft + DASH) + + theme.bold(theme.fg("accent", shown)) + + theme.fg("border", DASH.repeat(fill) + box.topRight) + ); + } + + #bodyRow(gutter: string, text: string, width: number, gp: (s: string) => string): string { + const contentW = Math.max(0, width - 7); + return `${theme.fg("border", VDASH)} ${gp(gutter)}${fit(text, contentW)} ${theme.fg("border", VDASH)}`; + } + + #bottomBorder(width: number, hiddenRows: number, hiddenChars: number, footerText: string | undefined): string { + const box = theme.boxRound; + const inner = Math.max(0, width - 2); + const parts: string[] = []; + if (footerText) parts.push(footerText); + if (hiddenRows > 0) parts.push(`+${hiddenRows} rows · ${hiddenChars} chars`); + const seg = parts.length ? ` ${parts.join(" · ")} ` : ""; + if (!seg) return theme.fg("border", box.bottomLeft + DASH.repeat(inner) + box.bottomRight); + const left = 1; // left-aligned, one cell in from the corner + const maxSeg = Math.max(0, inner - left); + const segShown = truncateToWidth(seg, maxSeg); + const segW = visibleWidth(segShown); + const right = Math.max(0, inner - left - segW); + return ( + theme.fg("border", box.bottomLeft + DASH.repeat(left)) + + theme.fg("muted", segShown) + + theme.fg("border", DASH.repeat(right) + box.bottomRight) + ); + } +} + +/** + * A dashed footer rule (no side borders) carrying queue-level hint text — used + * when the last visual pending entry is lightweight (no box to fold the hint + * into), so the affordance still reads as a border-style footer, not a thrown + * plain line. + */ +export class QueueFooter implements Component { + #text: string; + #cachedWidth = -1; + #cached: string[] | undefined; + constructor(text: string) { + this.#text = text; + } + invalidate(): void { + this.#cachedWidth = -1; + this.#cached = undefined; + } + render(width: number): readonly string[] { + if (this.#cached && this.#cachedWidth === width) return this.#cached; + const inner = Math.max(0, width); + const left = 1; + const seg = truncateToWidth(` ${this.#text} `, Math.max(0, inner - left)); + const segW = visibleWidth(seg); + const right = Math.max(0, inner - left - segW); + const line = + theme.fg("border", DASH.repeat(left)) + theme.fg("muted", seg) + theme.fg("border", DASH.repeat(right)); + this.#cached = [line]; + this.#cachedWidth = width; + return this.#cached; + } +} diff --git a/packages/coding-agent/src/modes/controllers/input-controller.ts b/packages/coding-agent/src/modes/controllers/input-controller.ts index ed16eb36cb5..564ba999a65 100644 --- a/packages/coding-agent/src/modes/controllers/input-controller.ts +++ b/packages/coding-agent/src/modes/controllers/input-controller.ts @@ -372,6 +372,11 @@ export class InputController { if (this.ctx.cancelPendingSubmission()) { return; } + const queued = this.ctx.session.getQueuedMessages(); + if (queued.steering.length > 0 || queued.followUp.length > 0) { + void this.ctx.session.abort({ reason: USER_INTERRUPT_LABEL }); + return; + } this.restoreQueuedMessagesToEditor({ abort: true }); } else if (this.ctx.session.isBashRunning) { this.ctx.session.abortBash(); @@ -469,6 +474,15 @@ export class InputController { this.ctx.editor.onExpandTools = () => this.toggleToolOutputExpansion(); this.ctx.editor.setActionKeys("app.message.dequeue", this.ctx.keybindings.getKeys("app.message.dequeue")); this.ctx.editor.onDequeue = () => this.handleDequeue(); + this.ctx.editor.setActionKeys("app.message.expandQueue", this.ctx.keybindings.getKeys("app.message.expandQueue")); + this.ctx.editor.onExpandQueue = () => this.togglePendingQueueExpansion(); + this.ctx.editor.onUpWhenEmpty = () => { + const queued = this.ctx.viewSession.getQueuedMessages(); + const hasQueued = + queued.steering.length > 0 || queued.followUp.length > 0 || this.ctx.compactionQueuedMessages.length > 0; + if (!hasQueued) return false; + return this.restoreQueuedMessagesToEditor() > 0; + }; this.ctx.editor.setActionKeys("app.retry", this.ctx.keybindings.getKeys("app.retry")); this.ctx.editor.onRetry = () => void this.handleRetry(); this.ctx.editor.clearCustomKeyHandlers(); @@ -1237,10 +1251,10 @@ export class InputController { } restoreQueuedMessagesToEditor(options?: { abort?: boolean; currentText?: string }): number { - this.ctx.locallySubmittedUserSignatures.clear(); + const targetSession = this.ctx.viewSession; // On Esc (abort) drop non-user internal steers so the post-abort drain can't // auto-resume; plain Alt+Up dequeue preserves them for the continuing stream. - const { steering, followUp } = this.ctx.session.clearQueue({ forInterrupt: options?.abort }); + const { steering, followUp } = targetSession.clearQueue({ forInterrupt: options?.abort }); // Messages typed while compacting live in `compactionQueuedMessages`, not the // agent queue `clearQueue()` drains — but the pending bar shows the same // "Alt+Up to edit" hint for them (ui-helpers `updatePendingMessagesDisplay`). @@ -1258,10 +1272,13 @@ export class InputController { if (allQueued.length === 0) { this.ctx.updatePendingMessagesDisplay(); if (options?.abort) { - void this.ctx.session.abort({ reason: USER_INTERRUPT_LABEL }); + void targetSession.abort({ reason: USER_INTERRUPT_LABEL }); } return 0; } + for (const entry of allQueued) { + this.ctx.locallySubmittedUserSignatures.delete(`${entry.text}\u0000${entry.images?.length ?? 0}`); + } // Image markers are positional: `[Image #N]` ↔ `pendingImages[N-1]`. Each // queued message numbered its markers against its own local image list // (1..K). Because we prepend the queued text but append the queued images @@ -1297,7 +1314,7 @@ export class InputController { } this.ctx.updatePendingMessagesDisplay(); if (options?.abort) { - void this.ctx.session.abort({ reason: USER_INTERRUPT_LABEL }); + void targetSession.abort({ reason: USER_INTERRUPT_LABEL }); } return allQueued.length; } @@ -1780,6 +1797,13 @@ export class InputController { this.ctx.showStatus(`Thinking blocks: ${this.ctx.hideThinkingBlock ? "hidden" : "visible"}`); } + /** Toggle expanded/collapsed queued-message preview. */ + togglePendingQueueExpansion(): void { + this.ctx.pendingQueueExpanded = !this.ctx.pendingQueueExpanded; + this.ctx.updatePendingMessagesDisplay(); + this.ctx.ui.requestRender(); + } + #getEditorTerminalPath(): string | null { if (process.platform === "win32") { return null; diff --git a/packages/coding-agent/src/modes/controllers/selector-controller.ts b/packages/coding-agent/src/modes/controllers/selector-controller.ts index 92b8a1ac7a2..619ea182c22 100644 --- a/packages/coding-agent/src/modes/controllers/selector-controller.ts +++ b/packages/coding-agent/src/modes/controllers/selector-controller.ts @@ -387,10 +387,10 @@ export class SelectorController { this.ctx.statusLine.setAutoCompactEnabled(value as boolean); break; case "steeringMode": - this.ctx.session.setSteeringMode(value as "all" | "one-at-a-time"); + this.ctx.session.setSteeringMode(value as "all" | "one-at-a-time" | "coalescing"); break; case "followUpMode": - this.ctx.session.setFollowUpMode(value as "all" | "one-at-a-time"); + this.ctx.session.setFollowUpMode(value as "all" | "one-at-a-time" | "coalescing"); break; case "interruptMode": this.ctx.session.setInterruptMode(value as "immediate" | "wait"); diff --git a/packages/coding-agent/src/modes/interactive-mode.ts b/packages/coding-agent/src/modes/interactive-mode.ts index 8e064d8eed3..4c0e45bc01a 100644 --- a/packages/coding-agent/src/modes/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive-mode.ts @@ -466,6 +466,7 @@ export class InteractiveMode implements InteractiveModeContext { return this.hideThinkingBlock || (thinkingOff && !this.hasDisplayableThinkingContent); } proseOnlyThinking = true; + pendingQueueExpanded = false; compactionQueuedMessages: CompactionQueuedMessage[] = []; pendingTools = new Map(); pendingBashComponents: BashExecutionComponent[] = []; @@ -588,6 +589,7 @@ export class InteractiveMode implements InteractiveModeContext { this.streamingMessage = undefined; this.lastAssistantUsage = undefined; this.pendingTools.clear(); + this.pendingQueueExpanded = false; } readonly #uiHelpers: UiHelpers; #sttController: STTController | undefined; @@ -630,6 +632,22 @@ export class InteractiveMode implements InteractiveModeContext { this.lspServers = lspServers; this.mcpManager = mcpManager; this.#eventBus = eventBus; + this.session.onLocalQueueCoalesced = ( + perSendText, + mergedText, + replacedText, + perSendImageCount, + mergedImageCount, + replacedImageCount, + ) => { + const droppedPerSend = this.locallySubmittedUserSignatures.delete(`${perSendText}\u0000${perSendImageCount}`); + const droppedReplaced = this.locallySubmittedUserSignatures.delete( + `${replacedText}\u0000${replacedImageCount}`, + ); + if (droppedPerSend || droppedReplaced) { + this.locallySubmittedUserSignatures.add(`${mergedText}\u0000${mergedImageCount}`); + } + }; if (eventBus) { this.#eventBusUnsubscribers.push( eventBus.on(LSP_STARTUP_EVENT_CHANNEL, data => { @@ -3372,6 +3390,9 @@ export class InteractiveMode implements InteractiveModeContext { // Clear the process-global consent handler so it doesn't outlive this // InteractiveMode instance (e.g. test harnesses, headless re-init). setAutoQaConsentHandler(null, null); + if (this.session.onLocalQueueCoalesced) { + this.session.onLocalQueueCoalesced = undefined; + } if (this.isInitialized) { this.ui.stop(); this.isInitialized = false; diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index 872064507b1..d55a91cfec7 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -588,14 +588,14 @@ export class RpcClient { /** * Set steering mode. */ - async setSteeringMode(mode: "all" | "one-at-a-time"): Promise { + async setSteeringMode(mode: "all" | "one-at-a-time" | "coalescing"): Promise { await this.#send({ type: "set_steering_mode", mode }); } /** * Set follow-up mode. */ - async setFollowUpMode(mode: "all" | "one-at-a-time"): Promise { + async setFollowUpMode(mode: "all" | "one-at-a-time" | "coalescing"): Promise { await this.#send({ type: "set_follow_up_mode", mode }); } diff --git a/packages/coding-agent/src/modes/rpc/rpc-types.ts b/packages/coding-agent/src/modes/rpc/rpc-types.ts index a877079cd53..50556031708 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-types.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-types.ts @@ -53,8 +53,8 @@ export type RpcCommand = | { id?: string; type: "cycle_thinking_level" } // Queue modes - | { id?: string; type: "set_steering_mode"; mode: "all" | "one-at-a-time" } - | { id?: string; type: "set_follow_up_mode"; mode: "all" | "one-at-a-time" } + | { id?: string; type: "set_steering_mode"; mode: "all" | "one-at-a-time" | "coalescing" } + | { id?: string; type: "set_follow_up_mode"; mode: "all" | "one-at-a-time" | "coalescing" } | { id?: string; type: "set_interrupt_mode"; mode: "immediate" | "wait" } // Compaction @@ -95,8 +95,8 @@ export interface RpcSessionState { thinkingLevel: ThinkingLevel | undefined; isStreaming: boolean; isCompacting: boolean; - steeringMode: "all" | "one-at-a-time"; - followUpMode: "all" | "one-at-a-time"; + steeringMode: "all" | "one-at-a-time" | "coalescing"; + followUpMode: "all" | "one-at-a-time" | "coalescing"; interruptMode: "immediate" | "wait"; sessionFile?: string; sessionId: string; diff --git a/packages/coding-agent/src/modes/types.ts b/packages/coding-agent/src/modes/types.ts index 6eb2c0f4a03..fe491121cef 100644 --- a/packages/coding-agent/src/modes/types.ts +++ b/packages/coding-agent/src/modes/types.ts @@ -151,6 +151,7 @@ export interface InteractiveModeContext { isBashMode: boolean; toolOutputExpanded: boolean; todoExpanded: boolean; + pendingQueueExpanded: boolean; planModeEnabled: boolean; goalModeEnabled: boolean; goalModePaused: boolean; diff --git a/packages/coding-agent/src/modes/utils/ui-helpers.ts b/packages/coding-agent/src/modes/utils/ui-helpers.ts index d5f2508244a..61519848c34 100644 --- a/packages/coding-agent/src/modes/utils/ui-helpers.ts +++ b/packages/coding-agent/src/modes/utils/ui-helpers.ts @@ -1,7 +1,8 @@ import type { AgentMessage } from "@oh-my-pi/pi-agent-core"; import type { AssistantMessage, ImageContent, Message, Usage } from "@oh-my-pi/pi-ai"; import { getStreamingPartialJson } from "@oh-my-pi/pi-ai/utils/block-symbols"; -import { type Component, Spacer, Text, TruncatedText } from "@oh-my-pi/pi-tui"; +import { type Component, Spacer, Text } from "@oh-my-pi/pi-tui"; +import { sanitizeText } from "@oh-my-pi/pi-utils"; import type { AdvisorMessageDetails } from "../../advisor"; import { COLLAB_PROMPT_MESSAGE_TYPE, type CollabPromptDetails } from "../../collab/protocol"; import { settings } from "../../config/settings"; @@ -24,6 +25,7 @@ import { type LateDiagnosticsFile, LateDiagnosticsMessageComponent, } from "../../modes/components/late-diagnostics-message"; +import { QueuedMessageBox } from "../../modes/components/queued-message-box"; import { ReadToolGroupComponent, readArgsHaveTarget, @@ -34,6 +36,7 @@ import { ToolExecutionComponent } from "../../modes/components/tool-execution"; import { TranscriptBlock } from "../../modes/components/transcript-container"; import { createUsageRowBlock } from "../../modes/components/usage-row"; import { UserMessageComponent } from "../../modes/components/user-message"; + import { decodeStreamedToolArgs, streamingStringKeysForTool } from "../../modes/controllers/tool-args-reveal"; import { materializeImageReferenceLinksSync } from "../../modes/image-references"; import { theme } from "../../modes/theme/theme"; @@ -58,6 +61,17 @@ import { resolveAssistantErrorPresentation, } from "./transcript-render-helpers"; +function queuedMessageVisualRowCount(message: string, boxWidth: number): number { + const contentWidth = Math.max(0, boxWidth - 7); + return message + .split("\n") + .map(line => sanitizeText(line.replace(/\t/g, " "))) + .reduce((rows, line) => { + if (contentWidth <= 0 || line.length === 0) return rows + 1; + return rows + Bun.wrapAnsi(line, contentWidth, { wordWrap: true, hard: true }).split("\n").length; + }, 0); +} + type TextBlock = { type: "text"; text: string }; interface RenderInitialMessagesOptions { preserveExistingChat?: boolean; @@ -671,15 +685,27 @@ export class UiHelpers { } const allMessages = [...steeringMessages, ...followUpMessages]; - if (allMessages.length > 0) { - this.ctx.pendingMessagesContainer.addChild(new Spacer(1)); - for (const entry of allMessages) { - const queuedText = theme.fg("dim", `${entry.label}: ${entry.message}`); - this.ctx.pendingMessagesContainer.addChild(new TruncatedText(queuedText, 1, 0)); - } - const dequeueKey = this.ctx.keybindings.getDisplayString("app.message.dequeue") || "Alt+Up"; - const hintText = theme.fg("dim", `${theme.tree.hook} ${dequeueKey} to edit`); - this.ctx.pendingMessagesContainer.addChild(new TruncatedText(hintText, 1, 0)); + if (allMessages.length === 0) return; + + this.ctx.pendingMessagesContainer.addChild(new Spacer(1)); + const expanded = this.ctx.pendingQueueExpanded; + const collapseLines = Math.max(1, this.ctx.settings?.get("pendingQueueCollapseLines") ?? 5); + const queueBoxWidth = Math.max(1, this.ctx.ui.terminal?.columns ?? 80); + const canExpandQueue = allMessages.some( + entry => queuedMessageVisualRowCount(entry.message, queueBoxWidth) > collapseLines, + ); + const dequeueKey = this.ctx.keybindings.getDisplayString("app.message.dequeue") || "Alt+Up"; + const expandKey = this.ctx.keybindings.getDisplayString("app.message.expandQueue") || "Alt+O"; + const expandHint = canExpandQueue ? `, ${expandKey} to ${expanded ? "collapse" : "expand"}` : ""; + const hint = `${dequeueKey} (or Up) to edit${expandHint}`; + + for (let idx = 0; idx < allMessages.length; idx++) { + const entry = allMessages[idx]; + const safeAll = entry.message.split("\n").map(line => sanitizeText(line.replace(/\t/g, " "))); + const footerText = idx === allMessages.length - 1 ? hint : undefined; + this.ctx.pendingMessagesContainer.addChild( + new QueuedMessageBox(entry.label, safeAll, { collapseLines, expanded, footerText }), + ); } } diff --git a/packages/coding-agent/src/sdk.ts b/packages/coding-agent/src/sdk.ts index afbbfa42deb..082862eed24 100644 --- a/packages/coding-agent/src/sdk.ts +++ b/packages/coding-agent/src/sdk.ts @@ -2698,8 +2698,8 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} deadline: options.deadline, transformContext, transformProviderContext, - steeringMode: settings.get("steeringMode") ?? "one-at-a-time", - followUpMode: settings.get("followUpMode") ?? "one-at-a-time", + steeringMode: settings.get("steeringMode") === "all" ? "all" : "one-at-a-time", + followUpMode: settings.get("followUpMode") === "all" ? "all" : "one-at-a-time", interruptMode: settings.get("interruptMode") ?? "immediate", thinkingBudgets: settings.getGroup("thinkingBudgets"), temperature: settings.get("temperature") >= 0 ? settings.get("temperature") : undefined, diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index c11d0bf3671..5587fe8d363 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -849,6 +849,8 @@ export interface PromptOptions { attribution?: MessageAttribution; /** Skip pre-send compaction checks for this prompt (internal use for maintenance flows). */ skipCompactionCheck?: boolean; + /** Notified when a prompt is queued behind an active stream. */ + onQueued?: QueuedUserMessageListener; } /** Options for AgentSession.followUp() */ @@ -1357,6 +1359,21 @@ function extractPermissionLocations( /** Entry returned by {@link AgentSession.clearQueue} / {@link AgentSession.popLastQueuedMessage}. */ export type RestoredQueuedMessage = { text: string; images?: ImageContent[] }; +type UserQueueMessage = Extract; +type QueuedUserMessageListener = (text: string, imageCount: number, replacedText?: string) => void; +export type QueueMode = "all" | "one-at-a-time" | "coalescing"; + +function coreQueueMode(mode: QueueMode): "all" | "one-at-a-time" { + return mode === "all" ? "all" : "one-at-a-time"; +} +export type LocalQueueCoalescedListener = ( + perSendText: string, + mergedText: string, + replacedText: string, + perSendImageCount: number, + mergedImageCount: number, + replacedImageCount: number, +) => void; function queuedTextContent(message: AgentMessage): string | undefined { if (!("content" in message)) return undefined; @@ -1391,7 +1408,7 @@ function isAdvisorCard(message: AgentMessage): message is CustomMessage { * steer/follow-up queues but must never be dumped into the editor on Esc/Alt+Up. */ function isUserQueuedMessage(message: AgentMessage): boolean { - if (message.role === "user") return true; + if (message.role === "user") return message.attribution !== "agent"; return message.role === "custom" && message.attribution === "user" && message.display !== false; } @@ -1407,6 +1424,16 @@ const MAGIC_KEYWORD_NOTICE_TYPES: ReadonlySet = new Set([ * attachments sent to a text-only model (see `#buildImageDescriptionNotice`). */ const IMAGE_ATTACHMENT_DESCRIPTION_TYPE = "image-attachment-description"; +const IMAGE_MARKER_REGEX = /\[Image #([1-9]\d*)((?:,[^\]\n]*)?)\]/g; + +function shiftQueuedImageMarkers(text: string, offset: number): string { + if (offset === 0) return text; + return text.replace( + IMAGE_MARKER_REGEX, + (_match, idx: string, tail: string) => `[Image #${Number(idx) + offset}${tail}]`, + ); +} + /** * A hidden, user-attributed companion of a queued user prompt: the magic-keyword * notices (`ultrathink`/`orchestrate`/`workflow`) enqueued alongside the user @@ -1437,6 +1464,21 @@ function toRestoredQueuedMessage(message: AgentMessage): RestoredQueuedMessage { return { text: queueChipText(message), images: queuedImageContent(message) }; } +function queuedUserDraftText(message: AgentMessage): string | undefined { + if (message.role !== "user" || message.attribution === "agent") return undefined; + if (!("content" in message)) return undefined; + if (typeof message.content === "string") return message.content; + const text = message.content.find((part): part is TextContent => part.type === "text")?.text ?? ""; + if (text) return text; + return queuedImageContent(message) ? "[Image]" : ""; +} + +function withQueuedUserContent(message: UserQueueMessage, text: string, images?: ImageContent[]): AgentMessage { + const content: (TextContent | ImageContent)[] = [{ type: "text", text }]; + if (images?.length) content.push(...images); + return { ...message, content, timestamp: Date.now() }; +} + function mergeLlmCompactionPreserveData( hookPreserveData: Record | undefined, resultPreserveData: Record | undefined, @@ -1564,6 +1606,7 @@ export class AgentSession { #pendingNextTurnMessages: CustomMessage[] = []; #scheduledHiddenNextTurnGeneration: number | undefined = undefined; #queuedMessageDrainScheduled = false; + #queuedUserMessageTail: Promise = Promise.resolve(); /** Latched true when the user deliberately interrupts (USER_INTERRUPT_LABEL); * suppresses advisor concern/blocker auto-resume until the user next resumes. * Advisor advice is still recorded into the transcript, just not auto-run. */ @@ -1574,6 +1617,7 @@ export class AgentSession { #goalModeState: GoalModeState | undefined; #goalRuntime: GoalRuntime; #advisorEnabled = false; + onLocalQueueCoalesced: LocalQueueCoalescedListener | undefined; #advisorTools?: AgentTool[]; #advisorWatchdogPrompt?: string; #advisorSharedInstructions?: string; @@ -6858,13 +6902,13 @@ export class AgentSession { } /** Current steering mode */ - get steeringMode(): "all" | "one-at-a-time" { - return this.agent.getSteeringMode(); + get steeringMode(): QueueMode { + return this.settings.get("steeringMode"); } /** Current follow-up mode */ - get followUpMode(): "all" | "one-at-a-time" { - return this.agent.getFollowUpMode(); + get followUpMode(): QueueMode { + return this.settings.get("followUpMode"); } /** Current interrupt mode */ @@ -7422,16 +7466,14 @@ export class AgentSession { if (!options?.streamingBehavior) { throw new AgentBusyError(); } - // Steer/follow-up the keyword notices BEFORE the queued user message so the - // model reads the steering notice ahead of the prompt it modifies. - for (const notice of keywordNotices) { - await this.sendCustomMessage(notice, { deliverAs: options.streamingBehavior }); - } - if (options.streamingBehavior === "followUp") { - await this.#queueUserMessage(expandedText, options?.images, "followUp"); - } else { - await this.#queueUserMessage(expandedText, options?.images, "steer"); - } + const queueMode = options.streamingBehavior === "followUp" ? "followUp" : "steer"; + await this.#queueUserMessageWithCompanions( + expandedText, + options?.images, + queueMode, + keywordNotices, + options.onQueued, + ); return true; } @@ -7882,13 +7924,13 @@ export class AgentSession { /** * Queue a steering message to interrupt the agent mid-run. */ - async steer(text: string, images?: ImageContent[]): Promise { + async steer(text: string, images?: ImageContent[], onQueued?: QueuedUserMessageListener): Promise { if (text.startsWith("/")) { this.#throwIfExtensionCommand(text); } const expandedText = expandPromptTemplate(text, [...this.#promptTemplates]); - await this.#queueUserMessage(expandedText, images, "steer"); + await this.#queueUserMessage(expandedText, images, "steer", onQueued); } /** @@ -7931,10 +7973,108 @@ export class AgentSession { this.#scheduleIdleQueueDrain(); } + async #withQueuedUserMessageLock(fn: () => Promise): Promise { + const previous = this.#queuedUserMessageTail; + const gate = Promise.withResolvers(); + this.#queuedUserMessageTail = previous.catch(() => undefined).then(() => gate.promise); + await previous.catch(() => undefined); + try { + return await fn(); + } finally { + gate.resolve(); + } + } + + async #queueUserMessageWithCompanions( + text: string, + images: ImageContent[] | undefined, + mode: "steer" | "followUp", + companionMessages: readonly CustomMessage[], + onQueued: QueuedUserMessageListener | undefined, + ): Promise { + await this.#withQueuedUserMessageLock(async () => { + for (const companion of companionMessages) { + await this.#queueCustomMessage(companion, mode); + } + await this.#queueUserMessageLocked(text, images, mode, onQueued); + }); + } + + #tryCoalesceQueuedUserMessage( + text: string, + images: ImageContent[] | undefined, + mode: "steer" | "followUp", + companionMessages: readonly CustomMessage[], + onQueued: QueuedUserMessageListener | undefined, + ): boolean { + const steering = [...this.agent.peekSteeringQueue()]; + const followUp = [...this.agent.peekFollowUpQueue()]; + const queue = mode === "steer" ? steering : followUp; + if (queue.length === 0) return false; + const perSendText = text || (images?.length ? "[Image]" : ""); + if (!perSendText) return false; + + let suffixCompanionStart = queue.length; + while (suffixCompanionStart > 0 && isHiddenUserCompanion(queue[suffixCompanionStart - 1])) { + suffixCompanionStart--; + } + const userIndex = suffixCompanionStart - 1; + const tail = queue[userIndex]; + if (tail?.role !== "user") return false; + const replacedText = queuedUserDraftText(tail); + if (replacedText === undefined) return false; + + let prefixCompanionStart = userIndex; + while (prefixCompanionStart > 0 && isHiddenUserCompanion(queue[prefixCompanionStart - 1])) { + prefixCompanionStart--; + } + const tailImages = queuedImageContent(tail); + const shiftedPerSendText = shiftQueuedImageMarkers(perSendText, tailImages?.length ?? 0); + const mergedImages = [...(tailImages ?? []), ...(images ?? [])]; + const mergedText = `${replacedText}\n${shiftedPerSendText}`; + const replacement = withQueuedUserContent(tail, mergedText, mergedImages.length > 0 ? mergedImages : undefined); + const nextQueue = [ + ...queue.slice(0, prefixCompanionStart), + ...queue.slice(prefixCompanionStart, userIndex), + ...queue.slice(userIndex + 1), + ...companionMessages, + replacement, + ]; + this.agent.replaceQueues( + mode === "steer" ? nextQueue : [...steering], + mode === "followUp" ? nextQueue : [...followUp], + ); + const perSendImageCount = images?.length ?? 0; + const replacedImageCount = tailImages?.length ?? 0; + const mergedImageCount = mergedImages.length; + onQueued?.(mergedText, mergedImageCount, replacedText); + if (!onQueued) { + this.onLocalQueueCoalesced?.( + perSendText, + mergedText, + replacedText, + perSendImageCount, + mergedImageCount, + replacedImageCount, + ); + } + return true; + } + async #queueUserMessage( text: string, images: ImageContent[] | undefined, mode: "steer" | "followUp", + onQueued?: QueuedUserMessageListener, + ): Promise { + await this.#withQueuedUserMessageLock(() => this.#queueUserMessageLocked(text, images, mode, onQueued)); + } + + async #queueUserMessageLocked( + text: string, + images: ImageContent[] | undefined, + mode: "steer" | "followUp", + onQueued?: QueuedUserMessageListener, ): Promise { // A queued user message (RPC/SDK/collab steer or follow-up, or a typed message // while streaming) is a deliberate resume; re-enable advisor auto-resume that @@ -7945,11 +8085,20 @@ export class AgentSession { if (normalizedImages?.length) { content.push(...normalizedImages); } - // Text-only model + image attachment: describe via a vision model and enqueue the - // description as a hidden companion immediately before the user message. const imageDescriptionNotice = normalizedImages?.length ? await this.#buildImageDescriptionNotice(normalizedImages) : undefined; + const companionMessages = imageDescriptionNotice ? [imageDescriptionNotice] : []; + const queueMode = mode === "followUp" ? this.settings.get("followUpMode") : this.settings.get("steeringMode"); + if ( + queueMode === "coalescing" && + this.#tryCoalesceQueuedUserMessage(text, normalizedImages, mode, companionMessages, onQueued) + ) { + this.#scheduleIdleQueueDrain(); + return; + } + // Text-only model + image attachment: describe via a vision model and enqueue the + // description as a hidden companion immediately before the user message. if (mode === "followUp") { if (imageDescriptionNotice) this.agent.followUp(imageDescriptionNotice); this.agent.followUp({ @@ -7968,6 +8117,7 @@ export class AgentSession { timestamp: Date.now(), }); } + onQueued?.(text, normalizedImages?.length ?? 0); this.#scheduleIdleQueueDrain(); } @@ -7980,17 +8130,28 @@ export class AgentSession { return; } this.#queuedMessageDrainScheduled = true; - this.#scheduleAgentContinue({ - shouldContinue: () => { - this.#queuedMessageDrainScheduled = false; - return this.#canAutoContinueForFollowUp() && this.agent.hasQueuedMessages(); - }, - onSkip: () => { - this.#queuedMessageDrainScheduled = false; - }, - onError: () => { + this.#schedulePostPromptTask(async signal => { + let observedTail: Promise; + do { + observedTail = this.#queuedUserMessageTail; + await observedTail.catch(() => undefined); + } while (observedTail !== this.#queuedUserMessageTail); + if (signal.aborted) { this.#queuedMessageDrainScheduled = false; - }, + return; + } + this.#scheduleAgentContinue({ + shouldContinue: () => { + this.#queuedMessageDrainScheduled = false; + return this.#canAutoContinueForFollowUp() && this.agent.hasQueuedMessages(); + }, + onSkip: () => { + this.#queuedMessageDrainScheduled = false; + }, + onError: () => { + this.#queuedMessageDrainScheduled = false; + }, + }); }); } @@ -9291,8 +9452,8 @@ export class AgentSession { * Set steering mode. * Saves to settings. */ - setSteeringMode(mode: "all" | "one-at-a-time"): void { - this.agent.setSteeringMode(mode); + setSteeringMode(mode: QueueMode): void { + this.agent.setSteeringMode(coreQueueMode(mode)); this.settings.set("steeringMode", mode); } @@ -9300,8 +9461,8 @@ export class AgentSession { * Set follow-up mode. * Saves to settings. */ - setFollowUpMode(mode: "all" | "one-at-a-time"): void { - this.agent.setFollowUpMode(mode); + setFollowUpMode(mode: QueueMode): void { + this.agent.setFollowUpMode(coreQueueMode(mode)); this.settings.set("followUpMode", mode); } diff --git a/packages/coding-agent/test/agent-session-queue-merge.test.ts b/packages/coding-agent/test/agent-session-queue-merge.test.ts new file mode 100644 index 00000000000..241ae9ef48b --- /dev/null +++ b/packages/coding-agent/test/agent-session-queue-merge.test.ts @@ -0,0 +1,829 @@ +/** + * Contract: consecutive user submissions on the same channel coalesce into a + * single queued entry (newline-joined) instead of piling up as separate messages + * — so rapid text/image sends read as one logical message (one pending chip, one + * delivery, one editor-restore block). + * + * Non-user queued entries (skill invocations / advisor cards) still break the + * run and keep their own identity. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "bun:test"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { Agent } from "@oh-my-pi/pi-agent-core"; +import { createMockModel, type MockResponse } from "@oh-my-pi/pi-ai/providers/mock"; +import { getBundledModel } from "@oh-my-pi/pi-catalog/models"; +import { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry"; +import { Settings } from "@oh-my-pi/pi-coding-agent/config/settings"; +import { AgentSession } from "@oh-my-pi/pi-coding-agent/session/agent-session"; +import { AuthStorage } from "@oh-my-pi/pi-coding-agent/session/auth-storage"; +import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manager"; +import * as imageLoading from "@oh-my-pi/pi-coding-agent/utils/image-loading"; +import * as imageVisionFallback from "@oh-my-pi/pi-coding-agent/utils/image-vision-fallback"; +import { Snowflake } from "@oh-my-pi/pi-utils"; + +describe("AgentSession queue coalescing", () => { + let tempDir: string; + let session: AgentSession; + const authStorages: AuthStorage[] = []; + + beforeEach(() => { + tempDir = path.join(os.tmpdir(), `pi-queue-merge-${Snowflake.next()}`); + fs.mkdirSync(tempDir, { recursive: true }); + }); + + afterEach(async () => { + vi.restoreAllMocks(); + await session?.dispose(); + for (const authStorage of authStorages.splice(0)) { + authStorage.close(); + } + fs.rmSync(tempDir, { recursive: true, force: true }); + }); + + async function createSession( + responses: MockResponse[], + modelRef: { api: Parameters[0]; id: string } = { + api: "anthropic", + id: "claude-sonnet-4-5", + }, + options?: { + steeringMode?: "all" | "one-at-a-time" | "coalescing"; + followUpMode?: "all" | "one-at-a-time" | "coalescing"; + }, + ): Promise { + const model = getBundledModel(modelRef.api, modelRef.id)!; + const mock = createMockModel({ responses }); + const agent = new Agent({ + getApiKey: () => "test-key", + initialState: { model, systemPrompt: ["Test"], tools: [] }, + streamFn: mock.stream, + }); + const sessionManager = SessionManager.inMemory(); + const settings = Settings.isolated({ + "compaction.enabled": false, + "images.describeForTextModels": true, + steeringMode: options?.steeringMode ?? "coalescing", + followUpMode: options?.followUpMode ?? "coalescing", + }); + const authStorage = await AuthStorage.create(path.join(tempDir, `auth-${Snowflake.next()}.db`)); + authStorages.push(authStorage); + authStorage.setRuntimeApiKey(modelRef.api, "test-key"); + const modelRegistry = new ModelRegistry(authStorage, path.join(tempDir, "models.yml")); + session = new AgentSession({ agent, sessionManager, settings, modelRegistry }); + return session; + } + + /** + * Run `fn` while the session is genuinely mid-prompt (isStreaming === true), so + * queued messages accumulate without the idle auto-drain delivering them. The + * queues are cleared afterwards so the outer prompt settles with one response. + */ + async function duringStream(target: AgentSession, fn: () => Promise): Promise { + let done = false; + let result: T | undefined; + target.agent.setOnBeforeYield(async () => { + if (done) return; + done = true; + result = await fn(); + target.agent.clearAllQueues(); + }); + await target.prompt("hello"); + return result as T; + } + + const steeringShapes = (target: AgentSession): string[] => + target.agent.peekSteeringQueue().map(m => (m.role === "custom" ? m.customType : m.role)); + + it("merges consecutive plain steers into one queued entry", async () => { + const target = await createSession([{ content: ["ok"] }]); + const steering = await duringStream(target, async () => { + await target.steer("Line1"); + await target.steer("Line2"); + await target.steer("Line3"); + return target.getQueuedMessages().steering.slice(); + }); + expect(steering).toEqual(["Line1\nLine2\nLine3"]); + }); + + it("merges consecutive plain follow-ups into one queued entry", async () => { + const target = await createSession([{ content: ["ok"] }]); + const followUp = await duringStream(target, async () => { + await target.followUp("first"); + await target.followUp("second"); + return target.getQueuedMessages().followUp.slice(); + }); + expect(followUp).toEqual(["first\nsecond"]); + }); + + it("keeps steer and follow-up channels separate", async () => { + const target = await createSession([{ content: ["ok"] }]); + const queued = await duringStream(target, async () => { + await target.steer("steer one"); + await target.followUp("follow one"); + await target.steer("steer two"); + await target.followUp("follow two"); + return target.getQueuedMessages(); + }); + expect(queued.steering).toEqual(["steer one\nsteer two"]); + expect(queued.followUp).toEqual(["follow one\nfollow two"]); + }); + + it("merges a plain steer into an image-bearing tail", async () => { + const target = await createSession([{ content: ["ok"] }]); + const shapes = await duringStream(target, async () => { + target.agent.steer({ + role: "user", + content: [ + { type: "text", text: "look at [Image #1]" }, + { type: "image", data: "QUJD", mimeType: "image/png" }, + ], + steering: true, + attribution: "user", + timestamp: Date.now(), + }); + await target.steer("plain follow-up text"); + return { + count: target.agent.peekSteeringQueue().length, + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + expect(shapes.count).toBe(1); + expect(shapes.chips).toEqual(["look at [Image #1]\nplain follow-up text"]); + expect(shapes.restored).toEqual([ + { + text: "look at [Image #1]\nplain follow-up text", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }, + ]); + }); + + it("merges a plain steer into an image-bearing tail that has a hidden text-model image notice", async () => { + const target = await createSession([{ content: ["ok"] }]); + const shapes = await duringStream(target, async () => { + target.agent.steer({ + role: "custom", + customType: "image-attachment-description", + content: [{ type: "text", text: 'described' }], + display: false, + attribution: "user", + timestamp: Date.now(), + }); + target.agent.steer({ + role: "user", + content: [ + { type: "text", text: "look at [Image #1]" }, + { type: "image", data: "QUJD", mimeType: "image/png" }, + ], + steering: true, + attribution: "user", + timestamp: Date.now(), + }); + await target.steer("plain follow-up text"); + return { + shapes: steeringShapes(target), + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + expect(shapes.shapes).toEqual(["image-attachment-description", "user"]); + expect(shapes.chips).toEqual(["look at [Image #1]\nplain follow-up text"]); + expect(shapes.restored).toEqual([ + { + text: "look at [Image #1]\nplain follow-up text", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }, + ]); + }); + + it("merges an image-only steer into a plain text tail", async () => { + const target = await createSession([{ content: ["ok"] }]); + const queued = await duringStream(target, async () => { + await target.steer("first text"); + await target.steer("", [{ type: "image", data: "QUJD", mimeType: "image/png" }]); + return { + count: target.agent.peekSteeringQueue().length, + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + expect(queued.count).toBe(1); + expect(queued.chips).toEqual(["first text\n[Image]"]); + expect(queued.restored).toEqual([ + { + text: "first text\n[Image]", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }, + ]); + }); + + it("does not merge a plain steer into a non-user (skill-like) tail", async () => { + const target = await createSession([{ content: ["ok"] }]); + const shapes = await duringStream(target, async () => { + target.agent.steer({ + role: "custom", + customType: "skill-prompt", + content: "/skill:review", + display: true, + attribution: "user", + timestamp: Date.now(), + }); + await target.steer("plain follow-up text"); + return steeringShapes(target); + }); + expect(shapes).toEqual(["skill-prompt", "user"]); + }); + + it("does not merge or restore an agent-attributed user steer", async () => { + const target = await createSession([{ content: ["ok"] }]); + const result = await duringStream(target, async () => { + target.agent.steer({ + role: "user", + content: "parent instruction", + attribution: "agent", + steering: true, + timestamp: Date.now(), + }); + await target.steer("plain user steer"); + const chips = target.getQueuedMessages().steering.slice(); + const cleared = target.clearQueue(); + return { + shapes: steeringShapes(target), + chips, + cleared, + remaining: target.agent.peekSteeringQueue().length, + }; + }); + + expect(result.shapes).toEqual(["user"]); + expect(result.chips).toEqual(["plain user steer"]); + expect(result.cleared.steering).toEqual([{ text: "plain user steer", images: undefined }]); + expect(result.remaining).toBe(1); + }); + + it("merges consecutive image steers into one entry and renumbers image markers", async () => { + const target = await createSession([{ content: ["ok"] }]); + const queued = await duringStream(target, async () => { + await target.steer("[Image #1, 638x450]", [{ type: "image", data: "QUJD", mimeType: "image/png" }]); + await target.steer("[Image #1, 638x450]", [{ type: "image", data: "REVG", mimeType: "image/png" }]); + await target.steer("[Image #1, 638x450]", [{ type: "image", data: "R0hJ", mimeType: "image/png" }]); + return { + count: target.agent.peekSteeringQueue().length, + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + expect(queued.count).toBe(1); + expect(queued.chips).toEqual(["[Image #1, 638x450]\n[Image #2, 638x450]\n[Image #3, 638x450]"]); + expect(queued.restored).toEqual([ + { + text: "[Image #1, 638x450]\n[Image #2, 638x450]\n[Image #3, 638x450]", + images: [ + { type: "image", data: "QUJD", mimeType: "image/png" }, + { type: "image", data: "REVG", mimeType: "image/png" }, + { type: "image", data: "R0hJ", mimeType: "image/png" }, + ], + }, + ]); + }); + + it("keeps hidden image-description companions when coalescing image steers for text-only models", async () => { + vi.spyOn(imageVisionFallback, "describeAttachedImagesForTextModel") + .mockResolvedValueOnce([{ type: "text", text: 'first description' }]) + .mockResolvedValueOnce([ + { type: "text", text: 'second description' }, + ]); + const target = await createSession( + [{ content: ["ok"] }], + { api: "aimlapi", id: "alibaba/qwen3-coder-480b-a35b-instruct" }, + { steeringMode: "coalescing" }, + ); + const result = await duringStream(target, async () => { + await target.steer("[Image #1]", [{ type: "image", data: "QUJD", mimeType: "image/png" }]); + await target.steer("[Image #1]", [{ type: "image", data: "REVG", mimeType: "image/png" }]); + return { + shapes: steeringShapes(target), + companions: target.agent + .peekSteeringQueue() + .flatMap(message => + message.role === "custom" && message.customType === "image-attachment-description" + ? [message.content] + : [], + ), + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + expect(result.shapes).toEqual(["image-attachment-description", "image-attachment-description", "user"]); + expect(result.companions).toEqual([ + [{ type: "text", text: 'first description' }], + [{ type: "text", text: 'second description' }], + ]); + expect(result.chips).toEqual(["[Image #1]\n[Image #2]"]); + expect(result.restored).toEqual([ + { + text: "[Image #1]\n[Image #2]", + images: [ + { type: "image", data: "QUJD", mimeType: "image/png" }, + { type: "image", data: "REVG", mimeType: "image/png" }, + ], + }, + ]); + }); + + it("merges an image steer after a magic-keyword prompt while keeping the hidden companion", async () => { + const target = await createSession([{ content: ["ok"] }]); + const result = await duringStream(target, async () => { + await target.prompt("ultrathink go", { streamingBehavior: "steer" }); + await target.steer("[Image #1]", [{ type: "image", data: "QUJD", mimeType: "image/png" }]); + return { + shapes: steeringShapes(target), + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + expect(result.shapes).toEqual(["ultrathink-notice", "user"]); + expect(result.chips).toEqual(["ultrathink go\n[Image #1]"]); + expect(result.restored).toEqual([ + { + text: "ultrathink go\n[Image #1]", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }, + ]); + }); + + it("merges a magic-keyword prompt after an image steer into one visible queued entry", async () => { + const target = await createSession([{ content: ["ok"] }]); + const result = await duringStream(target, async () => { + await target.steer("[Image #1]", [{ type: "image", data: "QUJD", mimeType: "image/png" }]); + await target.prompt("ultrathink go", { streamingBehavior: "steer" }); + return { + shapes: steeringShapes(target), + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + expect(result.shapes).toEqual(["ultrathink-notice", "user"]); + expect(result.chips).toEqual(["[Image #1]\nultrathink go"]); + expect(result.restored).toEqual([ + { + text: "[Image #1]\nultrathink go", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }, + ]); + }); + + it("coalesces concurrently spammed image steers before delivery", async () => { + const target = await createSession([{ content: ["ok"] }]); + const result = await duringStream(target, async () => { + const first = target.prompt("[Image #1, 216x200]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }); + const second = target.prompt("[Image #1, 216x200]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "REVG", mimeType: "image/png" }], + }); + await Promise.all([first, second]); + return { + count: target.agent.peekSteeringQueue().filter(message => message.role === "user").length, + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + + expect(result.count).toBe(1); + expect(result.chips).toEqual(["[Image #1, 216x200]\n[Image #2, 216x200]"]); + expect(result.restored).toEqual([ + { + text: "[Image #1, 216x200]\n[Image #2, 216x200]", + images: [ + { type: "image", data: "QUJD", mimeType: "image/png" }, + { type: "image", data: "REVG", mimeType: "image/png" }, + ], + }, + ]); + }); + + it("keeps magic-keyword companions attached during concurrent image steer spam", async () => { + const target = await createSession([{ content: ["ok"] }]); + const result = await duringStream(target, async () => { + const first = target.prompt("ultrathink [Image #1, 216x200]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }); + const second = target.prompt("[Image #1, 216x200]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "REVG", mimeType: "image/png" }], + }); + await Promise.all([first, second]); + return { + shapes: steeringShapes(target), + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + + expect(result.shapes).toEqual(["ultrathink-notice", "user"]); + expect(result.chips).toEqual(["ultrathink [Image #1, 216x200]\n[Image #2, 216x200]"]); + expect(result.restored).toEqual([ + { + text: "ultrathink [Image #1, 216x200]\n[Image #2, 216x200]", + images: [ + { type: "image", data: "QUJD", mimeType: "image/png" }, + { type: "image", data: "REVG", mimeType: "image/png" }, + ], + }, + ]); + }); + + it("coalesces concurrently spammed image follow-ups on the follow-up queue", async () => { + const target = await createSession([{ content: ["ok"] }]); + const result = await duringStream(target, async () => { + const first = target.prompt("[Image #1, 216x200]", { + streamingBehavior: "followUp", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }); + const second = target.prompt("[Image #1, 216x200]", { + streamingBehavior: "followUp", + images: [{ type: "image", data: "REVG", mimeType: "image/png" }], + }); + await Promise.all([first, second]); + return { + steering: target.getQueuedMessages().steering.slice(), + followUpCount: target.agent.peekFollowUpQueue().filter(message => message.role === "user").length, + followUp: target.getQueuedMessages().followUp.slice(), + restored: target.clearQueue().followUp, + }; + }); + + expect(result.steering).toEqual([]); + expect(result.followUpCount).toBe(1); + expect(result.followUp).toEqual(["[Image #1, 216x200]\n[Image #2, 216x200]"]); + expect(result.restored).toEqual([ + { + text: "[Image #1, 216x200]\n[Image #2, 216x200]", + images: [ + { type: "image", data: "QUJD", mimeType: "image/png" }, + { type: "image", data: "REVG", mimeType: "image/png" }, + ], + }, + ]); + }); + + it("coalesces concurrent text then image steer spam into one delivered message", async () => { + const target = await createSession([{ content: ["ok"] }]); + const result = await duringStream(target, async () => { + const first = target.prompt("existing steer text", { streamingBehavior: "steer" }); + const second = target.prompt("[Image #1, 216x200]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }); + await Promise.all([first, second]); + return { + count: target.agent.peekSteeringQueue().filter(message => message.role === "user").length, + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + + expect(result.count).toBe(1); + expect(result.chips).toEqual(["existing steer text\n[Image #1, 216x200]"]); + expect(result.restored).toEqual([ + { + text: "existing steer text\n[Image #1, 216x200]", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }, + ]); + }); + + it("coalesces concurrent image then text steer spam into one delivered message", async () => { + const target = await createSession([{ content: ["ok"] }]); + const result = await duringStream(target, async () => { + const first = target.prompt("[Image #1, 216x200]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }); + const second = target.prompt("text after image", { streamingBehavior: "steer" }); + await Promise.all([first, second]); + return { + count: target.agent.peekSteeringQueue().filter(message => message.role === "user").length, + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + + expect(result.count).toBe(1); + expect(result.chips).toEqual(["[Image #1, 216x200]\ntext after image"]); + expect(result.restored).toEqual([ + { + text: "[Image #1, 216x200]\ntext after image", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }, + ]); + }); + + it("coalesces three concurrent image steers in FIFO order", async () => { + const target = await createSession([{ content: ["ok"] }]); + const result = await duringStream(target, async () => { + const first = target.prompt("first [Image #1]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }); + const second = target.prompt("second [Image #1]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "REVG", mimeType: "image/png" }], + }); + const third = target.prompt("third [Image #1]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "R0hJ", mimeType: "image/png" }], + }); + await Promise.all([first, second, third]); + return { + count: target.agent.peekSteeringQueue().filter(message => message.role === "user").length, + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + + expect(result.count).toBe(1); + expect(result.chips).toEqual(["first [Image #1]\nsecond [Image #2]\nthird [Image #3]"]); + expect(result.restored).toEqual([ + { + text: "first [Image #1]\nsecond [Image #2]\nthird [Image #3]", + images: [ + { type: "image", data: "QUJD", mimeType: "image/png" }, + { type: "image", data: "REVG", mimeType: "image/png" }, + { type: "image", data: "R0hJ", mimeType: "image/png" }, + ], + }, + ]); + }); + + it("keeps concurrent steer and follow-up image submissions in separate queues", async () => { + const target = await createSession([{ content: ["ok"] }]); + const result = await duringStream(target, async () => { + const steer = target.prompt("steer [Image #1]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }); + const followUp = target.prompt("follow-up [Image #1]", { + streamingBehavior: "followUp", + images: [{ type: "image", data: "REVG", mimeType: "image/png" }], + }); + await Promise.all([steer, followUp]); + const queued = target.getQueuedMessages(); + const restored = target.clearQueue(); + return { queued, restored }; + }); + + expect(result.queued.steering).toEqual(["steer [Image #1]"]); + expect(result.queued.followUp).toEqual(["follow-up [Image #1]"]); + expect(result.restored.steering).toEqual([ + { text: "steer [Image #1]", images: [{ type: "image", data: "QUJD", mimeType: "image/png" }] }, + ]); + expect(result.restored.followUp).toEqual([ + { text: "follow-up [Image #1]", images: [{ type: "image", data: "REVG", mimeType: "image/png" }] }, + ]); + }); + + it("serializes concurrent image submissions even when coalescing is disabled", async () => { + const target = await createSession( + [{ content: ["ok"] }], + { api: "anthropic", id: "claude-sonnet-4-5" }, + { steeringMode: "one-at-a-time" }, + ); + const result = await duringStream(target, async () => { + const first = target.prompt("[Image #1, 216x200]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }); + const second = target.prompt("[Image #1, 216x200]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "REVG", mimeType: "image/png" }], + }); + await Promise.all([first, second]); + return { + count: target.agent.peekSteeringQueue().filter(message => message.role === "user").length, + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + + expect(result.count).toBe(2); + expect(result.chips).toEqual(["[Image #1, 216x200]", "[Image #1, 216x200]"]); + expect(result.restored).toEqual([ + { text: "[Image #1, 216x200]", images: [{ type: "image", data: "QUJD", mimeType: "image/png" }] }, + { text: "[Image #1, 216x200]", images: [{ type: "image", data: "REVG", mimeType: "image/png" }] }, + ]); + }); + + it("releases the queue lock when image normalization fails", async () => { + const target = await createSession([{ content: ["ok"] }]); + + const result = await duringStream(target, async () => { + const realNormalize = imageLoading.normalizeModelContextImages; + vi.spyOn(imageLoading, "normalizeModelContextImages").mockImplementation(async (images, options) => { + if (images?.[0]?.data === "QUJD") throw new Error("normalize failed"); + return realNormalize(images, options); + }); + const first = target.prompt("bad [Image #1]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "QUJD", mimeType: "image/png" }], + }); + const second = target.prompt("good [Image #1]", { + streamingBehavior: "steer", + images: [{ type: "image", data: "REVG", mimeType: "image/png" }], + }); + const settled = await Promise.allSettled([first, second]); + return { + settled, + chips: target.getQueuedMessages().steering.slice(), + restored: target.clearQueue().steering, + }; + }); + + expect(result.settled[0]?.status).toBe("rejected"); + expect(result.settled[1]?.status).toBe("fulfilled"); + expect(result.chips).toEqual(["good [Image #1]"]); + expect(result.restored).toEqual([ + { text: "good [Image #1]", images: [{ type: "image", data: "REVG", mimeType: "image/png" }] }, + ]); + }); + + it("reports the coalesced text and replaced tail to prompt's onQueued for signature tracking", async () => { + const target = await createSession([{ content: ["ok"] }]); + const calls: Array<[string, number, string | undefined]> = []; + const onQueued = (text: string, imageCount: number, replacedText?: string) => + calls.push([text, imageCount, replacedText]); + await duringStream(target, async () => { + await target.prompt("L1", { streamingBehavior: "steer", onQueued }); + await target.prompt("L2", { streamingBehavior: "steer", onQueued }); + await target.prompt("L3", { streamingBehavior: "steer", onQueued }); + return null; + }); + // Each send reports the FINAL queued text and the prior tail it replaced, so the + // caller can drop the replaced signature and record the merged one (no stale sigs). + expect(calls).toEqual([ + ["L1", 0, undefined], + ["L1\nL2", 0, "L1"], + ["L1\nL2\nL3", 0, "L1\nL2"], + ]); + }); + + it("reports replacedText through session.steer's onQueued on a coalescing merge", async () => { + const target = await createSession([{ content: ["ok"] }]); + const calls: Array<[string, number, string | undefined]> = []; + const onQueued = (text: string, imageCount: number, replacedText?: string) => + calls.push([text, imageCount, replacedText]); + await duringStream(target, async () => { + await target.steer("a", undefined, onQueued); + await target.steer("b", undefined, onQueued); + return null; + }); + // session.steer (the compaction-delivery path) reports the merge the same way, + // so #deliverQueuedMessage keeps the local-submit signature set exact. + expect(calls).toEqual([ + ["a", 0, undefined], + ["a\nb", 0, "a"], + ]); + }); + + it("routes callback-less coalescing through onLocalQueueCoalesced and clears stale local signatures", async () => { + const target = await createSession([{ content: ["ok"] }]); + const sigs = new Set(["a\u00000", "b\u00000"]); // UI recorded each per-send signature + // Mirror the interactive-mode wiring: drop per-send + replaced (both), record merged if either was local. + target.onLocalQueueCoalesced = (perSend, merged, replaced, perSendCount, mergedCount, replacedCount) => { + const droppedPerSend = sigs.delete(`${perSend}\u0000${perSendCount}`); + const droppedReplaced = sigs.delete(`${replaced}\u0000${replacedCount}`); + if (droppedPerSend || droppedReplaced) sigs.add(`${merged}\u0000${mergedCount}`); + }; + await duringStream(target, async () => { + await target.steer("a"); // no onQueued: pushes (no merge, no fire) + await target.steer("b"); // no onQueued: coalesces -> fires onLocalQueueCoalesced("b","a\nb","a",0,0,0) + return null; + }); + expect([...sigs]).toEqual(["a\nb\u00000"]); + }); + + it("does not mark a callback-less merge local when neither side was locally submitted (RPC/collab)", async () => { + const target = await createSession([{ content: ["ok"] }]); + const sigs = new Set(); // nothing was locally submitted + target.onLocalQueueCoalesced = (perSend, merged, replaced, perSendCount, mergedCount, replacedCount) => { + const droppedPerSend = sigs.delete(`${perSend}\u0000${perSendCount}`); + const droppedReplaced = sigs.delete(`${replaced}\u0000${replacedCount}`); + if (droppedPerSend || droppedReplaced) sigs.add(`${merged}\u0000${mergedCount}`); + }; + await duringStream(target, async () => { + await target.steer("a"); + await target.steer("b"); + return null; + }); + expect([...sigs]).toEqual([]); + }); +}); + +describe("AgentSession steering delivery contract", () => { + let tempDir: string; + let session: AgentSession; + const authStorages: AuthStorage[] = []; + + beforeEach(() => { + tempDir = path.join(os.tmpdir(), `pi-steer-deliver-${Snowflake.next()}`); + fs.mkdirSync(tempDir, { recursive: true }); + }); + + afterEach(async () => { + await session?.dispose(); + for (const authStorage of authStorages.splice(0)) { + authStorage.close(); + } + fs.rmSync(tempDir, { recursive: true, force: true }); + }); + + async function createSession(responses: MockResponse[]): Promise { + const model = getBundledModel("anthropic", "claude-sonnet-4-5")!; + const mock = createMockModel({ responses }); + const agent = new Agent({ + getApiKey: () => "test-key", + initialState: { model, systemPrompt: ["Test"], tools: [] }, + streamFn: mock.stream, + }); + const sessionManager = SessionManager.inMemory(); + const settings = Settings.isolated({ "compaction.enabled": false }); + const authStorage = await AuthStorage.create(path.join(tempDir, `auth-${Snowflake.next()}.db`)); + authStorages.push(authStorage); + authStorage.setRuntimeApiKey("anthropic", "test-key"); + const modelRegistry = new ModelRegistry(authStorage, path.join(tempDir, "models.yml")); + session = new AgentSession({ agent, sessionManager, settings, modelRegistry }); + return session; + } + + /** + * Pause mid-prompt so a steer can be queued, then RESUME the run (no clearAllQueues) + * so the agent dequeues the steer for the next turn and settles only once it has + * delivered + received the second response. Mirrors the live flow: user types a + * steer while streaming; once the turn ends, the steer is injected and delivered. + */ + async function steerMidStreamThenResume(target: AgentSession, steerText: string): Promise { + let queued = false; + target.agent.setOnBeforeYield(async () => { + if (queued) return; + queued = true; + await target.steer(steerText); + }); + await target.prompt("start"); + } + + it("clears a delivered steer from the queue in one-at-a-time mode (no ghost in pending display)", async () => { + // Regression: after a steer is delivered (dequeue + provider turn completes), the + // pending-display source must be empty — otherwise the steer reappears in the UI + // as a stale "Steer" chip even though the agent has already consumed it. + const target = await createSession([{ content: ["ok-1"] }, { content: ["ok-2"] }]); + await steerMidStreamThenResume(target, "queued-steer"); + + // Both responses consumed: prompt("start") → "ok-1", then auto-continue with the + // delivered steer → "ok-2". The queue must be empty. + expect(target.agent.peekSteeringQueue()).toEqual([]); + expect(target.getQueuedMessages().steering).toEqual([]); + + // Both turns reached the model; the steer text was delivered into state.messages. + const deliveredTexts = target.agent.state.messages + .filter(m => m.role === "user") + .map(m => + typeof m.content === "string" ? m.content : m.content.map(b => (b.type === "text" ? b.text : "")).join(""), + ); + expect(deliveredTexts).toContain("queued-steer"); + }); + it("delivers coalesced image steers without leaving ghosts", async () => { + // Q1 has an image and Q2 is text-only; they should deliver as one queued + // user turn and leave no stale pending-display source behind. + const target = await createSession([{ content: ["ok-1"] }, { content: ["ok-2"] }, { content: ["ok-3"] }]); + target.setSteeringMode("coalescing"); + let queued = false; + target.agent.setOnBeforeYield(async () => { + if (queued) return; + queued = true; + await Promise.all([ + target.steer("Q1 [Image #1]", [{ type: "image", data: "QUJD", mimeType: "image/png" }]), + target.steer("Q2"), + ]); + }); + await target.prompt("start"); + + const deliveredTexts = target.agent.state.messages + .filter(m => m.role === "user") + .map(m => + typeof m.content === "string" ? m.content : m.content.map(b => (b.type === "text" ? b.text : "")).join(""), + ); + expect(deliveredTexts).toEqual(["start", "Q1 [Image #1]\nQ2"]); + const deliveredImageTurn = target.agent.state.messages + .filter(m => m.role === "user") + .find(m => Array.isArray(m.content) && m.content.some(part => part.type === "image")); + expect(deliveredImageTurn?.content).toContainEqual({ type: "image", data: "QUJD", mimeType: "image/png" }); + expect(target.agent.peekSteeringQueue()).toEqual([]); + expect(target.getQueuedMessages().steering).toEqual([]); + }); +}); diff --git a/packages/coding-agent/test/input-controller-compaction-image.test.ts b/packages/coding-agent/test/input-controller-compaction-image.test.ts index a8618b72126..17484368eac 100644 --- a/packages/coding-agent/test/input-controller-compaction-image.test.ts +++ b/packages/coding-agent/test/input-controller-compaction-image.test.ts @@ -61,6 +61,7 @@ function makeCtx(initialQueue: CompactionQueuedMessage[] = []) { const ctx = { session, + viewSession: session, compactionQueuedMessages: [...initialQueue], pendingMessagesContainer: { clear: () => {}, addChild: () => {}, removeChild: () => {} }, editor: { @@ -214,6 +215,7 @@ describe("restoreQueuedMessagesToEditor image marker alignment", () => { }; const ctx = { session, + viewSession: session, editor, compactionQueuedMessages: [], locallySubmittedUserSignatures: new Set(), diff --git a/packages/coding-agent/test/input-controller-escape.test.ts b/packages/coding-agent/test/input-controller-escape.test.ts index e7ae3af47b6..e262c339f56 100644 --- a/packages/coding-agent/test/input-controller-escape.test.ts +++ b/packages/coding-agent/test/input-controller-escape.test.ts @@ -28,6 +28,7 @@ type FakeEditor = { onToggleThinking?: () => void; onExternalEditor?: () => void; onDequeue?: () => void; + onUpWhenEmpty?: () => boolean; onChange?: (text: string) => void; setText(text: string): void; getText(): string; @@ -179,6 +180,9 @@ function createContext(): { abortCompaction: vi.fn(), abortHandoff, abortRetry: vi.fn(), + abort, + clearQueue, + getQueuedMessages, } as unknown as InteractiveModeContext["viewSession"], sessionManager: { getSessionName: () => "existing session", @@ -821,3 +825,65 @@ describe("InputController double-tap ← gesture", () => { expect(showAgentHub).not.toHaveBeenCalled(); }); }); + +describe("InputController Up-on-empty undo-send wiring", () => { + it("restores queued user work and consumes Up", () => { + const { ctx, editor, spies } = createContext(); + spies.getQueuedMessages.mockReturnValue({ steering: ["queued line"], followUp: [] }); + spies.clearQueue.mockReturnValue({ steering: [{ text: "queued line" }], followUp: [] }); + const controller = new InputController(ctx); + controller.setupKeyHandlers(); + const consumed = editor.onUpWhenEmpty?.(); + expect(consumed).toBe(true); + expect(spies.clearQueue).toHaveBeenCalledTimes(1); + expect(editor.getText()).toBe("queued line"); + }); + + it("falls through to history when the display reports work the queue cannot drain", () => { + const { ctx, editor, spies } = createContext(); + spies.getQueuedMessages.mockReturnValue({ steering: ["ghost"], followUp: [] }); + spies.clearQueue.mockReturnValue({ steering: [], followUp: [] }); + const controller = new InputController(ctx); + controller.setupKeyHandlers(); + const consumed = editor.onUpWhenEmpty?.(); + expect(consumed).toBe(false); + expect(editor.getText()).toBe(""); + }); + + it("restores the focused subagent's queue, not the main session's, when a subagent is focused", () => { + const { ctx, editor, spies } = createContext(); + const viewClearQueue = vi.fn(() => ({ steering: [{ text: "focused line" }], followUp: [] })); + const viewGetQueuedMessages = vi.fn(() => ({ steering: ["focused line"], followUp: [] })); + Object.assign(ctx, { + viewSession: { + getQueuedMessages: viewGetQueuedMessages, + clearQueue: viewClearQueue, + abort: vi.fn(), + }, + }); + const controller = new InputController(ctx); + controller.setupKeyHandlers(); + const consumed = editor.onUpWhenEmpty?.(); + expect(consumed).toBe(true); + expect(viewClearQueue).toHaveBeenCalledTimes(1); + expect(spies.clearQueue).not.toHaveBeenCalled(); + expect(editor.getText()).toBe("focused line"); + }); + + it("clears only the restored messages' signatures, preserving main-session queued signatures", () => { + const { ctx } = createContext(); + ctx.locallySubmittedUserSignatures.add("main queued steer\u00000"); + ctx.locallySubmittedUserSignatures.add("focused line\u00000"); + Object.assign(ctx, { + viewSession: { + getQueuedMessages: vi.fn(() => ({ steering: ["focused line"], followUp: [] })), + clearQueue: vi.fn(() => ({ steering: [{ text: "focused line" }], followUp: [] })), + abort: vi.fn(), + }, + }); + const controller = new InputController(ctx); + controller.restoreQueuedMessagesToEditor(); + expect(ctx.locallySubmittedUserSignatures.has("focused line\u00000")).toBe(false); + expect(ctx.locallySubmittedUserSignatures.has("main queued steer\u00000")).toBe(true); + }); +}); diff --git a/packages/coding-agent/test/input-controller-orphan-submit.test.ts b/packages/coding-agent/test/input-controller-orphan-submit.test.ts index c71c52fe8d7..de92687c530 100644 --- a/packages/coding-agent/test/input-controller-orphan-submit.test.ts +++ b/packages/coding-agent/test/input-controller-orphan-submit.test.ts @@ -81,6 +81,7 @@ function createContext(sessionOverride?: InteractiveModeContext["session"]) { editor: editor as unknown as InteractiveModeContext["editor"], ui: { requestRender } as unknown as InteractiveModeContext["ui"], session, + viewSession: session, sessionManager: { getSessionName: () => "named-session" } as InteractiveModeContext["sessionManager"], compactionQueuedMessages: [] as InteractiveModeContext["compactionQueuedMessages"], fileSlashCommands: new Set(), diff --git a/packages/coding-agent/test/input-controller-skill-queue.test.ts b/packages/coding-agent/test/input-controller-skill-queue.test.ts index 6e4bfc07b3b..3454f1f1212 100644 --- a/packages/coding-agent/test/input-controller-skill-queue.test.ts +++ b/packages/coding-agent/test/input-controller-skill-queue.test.ts @@ -25,6 +25,8 @@ import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manage import { Container } from "@oh-my-pi/pi-tui"; import { TempDir } from "@oh-my-pi/pi-utils"; +const stripAnsi = (s: string): string => s.replace(/\u001b\[[0-9;]*m/g, ""); + type StubEditor = { setText: (text: string) => void; getText: () => string; @@ -655,13 +657,16 @@ function createStubInteractiveModeContextForUiHelpers(session: AgentSession) { const ctx = { editor, - ui: { requestRender }, + ui: { requestRender, terminal: { columns: 80 } }, pendingMessagesContainer, session, viewSession: session, compactionQueuedMessages: [], keybindings: { - getDisplayString: (_action: string) => "Alt+Up", + getDisplayString: (action: string) => (action === "app.message.expandQueue" ? "Alt+O" : "Alt+Up"), + }, + settings: { + get: (path: string) => (path === "pendingQueueCollapseLines" ? 3 : undefined), }, updatePendingMessagesDisplay, locallySubmittedUserSignatures: new Set(), @@ -689,7 +694,7 @@ describe("UiHelpers / InputController against derived queued custom display", () vi.restoreAllMocks(); }); - it("renders the compact slash form for queued skills", async () => { + it("renders queued skills inside the compact steering box", async () => { fixture = await createRealSession(); const { session } = fixture; queueCustomSteer(session, "/skill:test-skill arg1 arg2"); @@ -698,8 +703,66 @@ describe("UiHelpers / InputController against derived queued custom display", () const uiHelpers = new UiHelpers(ctx); uiHelpers.updatePendingMessagesDisplay(); - const rendered = pendingMessagesContainer.render(120).join("\n"); - expect(rendered).toMatch(/Steer: \/skill:test-skill arg1 arg2/); + const rendered = stripAnsi(pendingMessagesContainer.render(120).join("\n")); + expect(rendered).toContain("Steer"); + expect(rendered).toContain("└─ /skill:test-skill arg1 arg2"); + expect(rendered).toContain("Alt+Up (or Up) to edit"); + }); + + it("renders a coalesced image/text queue entry as one compact box", async () => { + fixture = await createRealSession(); + const { session } = fixture; + session.agent.steer({ + role: "user", + content: [ + { type: "text", text: "[Image #1] describe\nmore context" }, + { type: "image", data: "QUJD", mimeType: "image/png" }, + ], + steering: true, + attribution: "user", + timestamp: Date.now(), + }); + + const { ctx, pendingMessagesContainer } = createStubInteractiveModeContextForUiHelpers(session); + const uiHelpers = new UiHelpers(ctx); + uiHelpers.updatePendingMessagesDisplay(); + + const rendered = stripAnsi(pendingMessagesContainer.render(120).join("\n")); + expect(rendered.match(/ Steer /g)?.length).toBe(1); + expect(rendered).toContain("├─ [Image #1] describe"); + expect(rendered).toContain("└─ more context"); + expect(rendered).toContain("Alt+Up (or Up) to edit"); + }); + + it("collapses queued steering text with a row and character footer", async () => { + fixture = await createRealSession(); + const { session } = fixture; + queueCustomSteer(session, "one\ntwo\nthree\nfour"); + + const { ctx, pendingMessagesContainer } = createStubInteractiveModeContextForUiHelpers(session); + const uiHelpers = new UiHelpers(ctx); + uiHelpers.updatePendingMessagesDisplay(); + + const rendered = stripAnsi(pendingMessagesContainer.render(120).join("\n")); + expect(rendered).toContain("├─ one"); + expect(rendered).toContain("├─ two"); + expect(rendered).toContain("├─ three"); + expect(rendered).not.toContain("four"); + expect(rendered).toContain("Alt+Up (or Up) to edit, Alt+O to expand"); + expect(rendered).toContain("+1 rows · 4 chars"); + }); + + it("shows the expand hint when a queued line wraps past the collapsed row limit", async () => { + fixture = await createRealSession(); + const { session } = fixture; + queueCustomSteer(session, "wrapped ".repeat(80).trim()); + + const { ctx, pendingMessagesContainer } = createStubInteractiveModeContextForUiHelpers(session); + const uiHelpers = new UiHelpers(ctx); + uiHelpers.updatePendingMessagesDisplay(); + + const rendered = stripAnsi(pendingMessagesContainer.render(80).join("\n")); + expect(rendered).toContain("Alt+Up (or Up) to edit, Alt+O to expand"); }); it("restores the compact slash form into the editor and clears the queue", async () => { diff --git a/packages/coding-agent/test/modes/components/custom-editor-up-undo-send.test.ts b/packages/coding-agent/test/modes/components/custom-editor-up-undo-send.test.ts new file mode 100644 index 00000000000..bcc223034e0 --- /dev/null +++ b/packages/coding-agent/test/modes/components/custom-editor-up-undo-send.test.ts @@ -0,0 +1,66 @@ +/** + * Contract: Up-arrow on an EMPTY editor is offered to the host as an "undo send" + * gesture (`onUpWhenEmpty`). The host returns true to consume it (e.g. it pulled + * queued messages back into the editor) or false to fall through to normal + * input-history navigation. Up on a non-empty editor is never diverted, so + * multi-line cursor movement and history recall keep working untouched. + */ +import { afterAll, beforeAll, describe, expect, it } from "bun:test"; +import { resetSettingsForTest, Settings } from "@oh-my-pi/pi-coding-agent/config/settings"; +import { CustomEditor } from "@oh-my-pi/pi-coding-agent/modes/components/custom-editor"; +import { getEditorTheme, initTheme } from "@oh-my-pi/pi-coding-agent/modes/theme/theme"; + +const UP = "\x1b[A"; + +beforeAll(async () => { + resetSettingsForTest(); + await Settings.init({ inMemory: true }); + await initTheme(false); +}); + +afterAll(() => { + resetSettingsForTest(); +}); + +describe("CustomEditor Up-on-empty undo-send", () => { + it("invokes onUpWhenEmpty and consumes Up when it returns true (history not recalled)", () => { + const editor = new CustomEditor(getEditorTheme()); + editor.addToHistory("remembered draft"); + let called = 0; + editor.onUpWhenEmpty = () => { + called++; + return true; + }; + editor.handleInput(UP); + expect(called).toBe(1); + // Consumed → the key never reached base input-history navigation. + expect(editor.getText()).toBe(""); + }); + + it("falls through to input-history navigation when onUpWhenEmpty returns false", () => { + const editor = new CustomEditor(getEditorTheme()); + editor.addToHistory("remembered draft"); + let called = 0; + editor.onUpWhenEmpty = () => { + called++; + return false; + }; + editor.handleInput(UP); + expect(called).toBe(1); + // Not consumed → base editor recalled the previous history entry. + expect(editor.getText()).toBe("remembered draft"); + }); + + it("does not invoke onUpWhenEmpty when the editor holds a draft", () => { + const editor = new CustomEditor(getEditorTheme()); + editor.setText("a draft in progress"); + let called = 0; + editor.onUpWhenEmpty = () => { + called++; + return true; + }; + editor.handleInput(UP); + expect(called).toBe(0); + expect(editor.getText()).toBe("a draft in progress"); + }); +}); diff --git a/packages/coding-agent/test/modes/components/wrapper-selector-mouse-offset.test.ts b/packages/coding-agent/test/modes/components/wrapper-selector-mouse-offset.test.ts index 76805d3e914..8f27a0b94f7 100644 --- a/packages/coding-agent/test/modes/components/wrapper-selector-mouse-offset.test.ts +++ b/packages/coding-agent/test/modes/components/wrapper-selector-mouse-offset.test.ts @@ -5,6 +5,7 @@ import { QueueModeSelectorComponent } from "@oh-my-pi/pi-coding-agent/modes/comp import { ThemeSelectorComponent } from "@oh-my-pi/pi-coding-agent/modes/components/theme-selector"; import { ThinkingSelectorComponent } from "@oh-my-pi/pi-coding-agent/modes/components/thinking-selector"; import { initTheme } from "@oh-my-pi/pi-coding-agent/modes/theme/theme"; +import type { QueueMode } from "@oh-my-pi/pi-coding-agent/session/agent-session"; import type { SgrMouseEvent } from "@oh-my-pi/pi-tui"; beforeAll(async () => { @@ -63,9 +64,9 @@ describe("inline-picker wrapper routeMouse offset", () => { }); it("QueueModeSelectorComponent ignores the border row and selects the first mode below it", () => { - let selected: "all" | "one-at-a-time" | undefined; + let selected: QueueMode | undefined; const component = new QueueModeSelectorComponent( - "all", + "one-at-a-time", value => { selected = value; },