From 8d7e2017e6b2927cbcf560d95d7c89ea00128791 Mon Sep 17 00:00:00 2001 From: Thomas Mustier <6326440+tmustier@users.noreply.github.com> Date: Sun, 9 Aug 2026 16:30:18 +0100 Subject: [PATCH] refactor: trim defensive queue code --- CHANGELOG.md | 5 ++ README.md | 8 +- docs/validation.md | 71 ++-------------- index.ts | 163 +++++++++++++----------------------- queue-state.ts | 36 +------- test/pi-integration.test.ts | 42 +++++----- test/queue-state.test.ts | 113 +++++-------------------- test/queued-input.test.ts | 44 +++++----- 8 files changed, 142 insertions(+), 340 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff0653a..fb90968 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,11 @@ ## Unreleased +### Changed + +- Simplify dispatch around Pi's fire-and-forget message handoff and callback-based compaction APIs, while preserving restoration for prompt and Skill expansion failures. +- Keep validation documentation focused on reproducible commands and generated evidence rather than stale run output. + ## 0.2.0 - 2026-08-09 ### Added diff --git a/README.md b/README.md index f876d6e..5d56235 100644 --- a/README.md +++ b/README.md @@ -82,7 +82,7 @@ Text-only rows whose text is exactly `/compact`, `/compact ` or `/ - `Option+Enter` while the agent works queues the command in follow-up order - a command row executes only once the agent is idle; rows behind it wait — so `/compact` followed by `continue` compacts first and delivers `continue` after compaction completes - `/reload` runs Pi’s built-in reload; committed rows queued behind it retain their IDs, lanes, attachments and pause state across the runtime swap -- idle `/compact` uses Pi’s public compaction API so queued rows resume when compaction finishes; a start failure restores and pauses the command row +- idle `/compact` uses Pi’s public compaction API so queued rows resume when compaction completes or reports an error - `/reload` submitted while the agent works or tracked compaction runs stays queued instead of showing Pi’s built-in wait warning - `Enter` on `/compact` while the agent works uses Pi’s public compaction API and holds visible rows until compaction settles - ordinary messages submitted during compaction remain in Pi’s native queue and can run before extension-owned command rows after compaction finishes @@ -109,13 +109,13 @@ A touched head row is pinned until you save or cancel. In `one-at-a-time` mode, Aborting a run pauses both visible lanes. This prevents a follow-up from starting immediately after the abort. -Press `Enter` on the empty composer to resume. A synchronous handoff or preflight failure returns the affected batch to the front of its lane. +Press `Enter` on the empty composer to resume. A prompt or Skill expansion failure returns the affected batch to the front of its lane. Queue state, pause state and edit drafts are session-local. They never enter the Pi transcript or persistent session data. A `/reload` runtime swap carries committed rows and pause state through a short in-process handoff; unsaved edit drafts do not cross the swap. ## Public API limits -Pi’s public `sendUserMessage` API is fire-and-forget. The extension restores synchronous dispatch failures and preflight/expansion failures without reordering, but Pi does not expose later asynchronous input rejection to extensions. Inferring rejection from queue timing could duplicate a delayed successful handoff, so the extension does not do that. +Pi’s public `sendUserMessage` API is fire-and-forget. The extension restores prompt and Skill expansion failures before handoff, but Pi does not expose asynchronous input rejection to extensions. Inferring rejection from queue timing could duplicate a delayed successful handoff, so the extension does not do that. Pi also exposes queued `/reload` only through the TUI editor’s `void` submit callback. The extension prevents known busy and compaction conflicts and restores trailing rows on a successful runtime swap, but Pi cannot acknowledge or reject that submit back to the extension. @@ -140,7 +140,7 @@ pi -e ./index.ts The automated suite covers delivery, editing, command rows, resource expansion, recovery, images, editor composition, repeated reloads, real retry ordering, real manual compaction success/failure and real automatic overflow compaction. The tmux harness exercises the same paths through Pi's real TUI, including actual runtime reloads and native post-compaction input. -The Pi package ranges are intentionally unpinned. The full suite and real-TUI harness are verified against the current resolved Pi release; see [the validation record](docs/validation.md) for exact commands and evidence. +The Pi package ranges are intentionally unpinned. The full suite and real-TUI harness are verified against the current resolved Pi release; see [the validation guide](docs/validation.md) for the commands and generated evidence. ## Security diff --git a/docs/validation.md b/docs/validation.md index 2839ce7..55a4414 100644 --- a/docs/validation.md +++ b/docs/validation.md @@ -1,6 +1,6 @@ # Compaction and reload validation -This document records the deterministic validation matrix for compaction-aware command rows. The implementation remains extension-only and uses public Pi extension APIs. +This document records the deterministic validation workflow for compaction-aware command rows. The implementation remains extension-only and uses public Pi extension APIs. ## Automated suite @@ -23,13 +23,11 @@ npm update --ignore-scripts \ npm run ci ``` -The suite covers queue/edit invariants, command classification, images, one-at-a-time and all-mode delivery, synchronous partial handoff restoration, non-TUI pass-through, prompt and Skill expansion, manual compaction success/failure, automatic overflow compaction, retry ordering, repeated reload restoration, and compaction/native-input ordering. - -Latest result with Pi 0.84.1: 81 tests passed. +The suite covers queue and edit invariants, command classification, images, one-at-a-time and all-mode delivery, preparation-failure restoration, non-TUI pass-through, prompt and Skill expansion, manual compaction success and failure, automatic overflow compaction, retry ordering, reload restoration, and compaction/native-input ordering. ## Real TUI evidence -`test/tui-evidence.sh` starts the real Pi 0.84.1 TUI under tmux with a deterministic faux provider. It uses actual terminal key sequences, public compaction lifecycle events, public provider registration, actual runtime reloads, and Pi's real native compaction queue. +`test/tui-evidence.sh` starts the real Pi TUI under tmux with a deterministic faux provider. It uses actual terminal key sequences, public compaction lifecycle events, public provider registration, runtime reloads, and Pi's native compaction queue. Run: @@ -37,65 +35,12 @@ Run: ./test/tui-evidence.sh /tmp/pi-queue-tui-evidence ``` -The output directory contains plain terminal captures, provider-call logs, lifecycle-event logs, and runtime-initialization logs. Run it immediately before review so `summary.txt` records the exact Pi version, commit and working-tree state under test. A release evidence run should report `working tree: clean`. - -The latest complete run reported: - -```text -pi: 0.84.1 -commit: 37fcd1433b8960f13c030d9ba1a5e8cc36535e05 -working tree: clean -manual events: {"event":"session_before_compact","reason":"manual"} {"event":"session_before_compact","reason":"manual"} -overflow events: {"event":"session_before_compact","reason":"overflow"} {"event":"session_before_compact","reason":"threshold"} -runtime initializations across two queued reloads: 3 -captures: abort-paused, manual-reload-resources, native-before-command, automatic-overflow, all-mode -``` - -The three runtime initializations are the initial load plus two queued `/reload` rows. The final queued message ran after both reloads. - -The semantic capture excerpts were: - -```text -[compaction] -Compacted from 798 tokens -FAUX RESPONSE: after manual compaction - -Error: Compaction failed: Summarization failed: synthetic TUI summary failure -FAUX RESPONSE: after failed compaction - -Operation aborted -follow-ups (1) · paused -enter resume · option+up edit · escape keep paused -FAUX RESPONSE: after abort resume - -Reloaded keybindings, extensions, skills, prompts, themes, and context files -FAUX RESPONSE: after repeated reload - -PROMPT EXPANDED: first=alpha all=alpha beta default=fallback -[skill] bro -FAUX RESPONSE: -``` - -The native post-compaction ordering capture showed the ordinary message submitted during manual `/compact` entering Pi's native queue, finishing before the extension-owned command row, and `/reload` never reaching the model: - -```text -[compaction] -Compacted from 785 tokens -ordinary native during compaction -FAUX RESPONSE: ordinary native during compaction -Reloaded keybindings, extensions, skills, prompts, themes, and context files -``` - -The overflow event log recorded `reason: "overflow"`, the TUI rendered a compaction entry, and `overflow-provider-calls.jsonl` proved the queued follow-up completed exactly once. `all-mode-provider-calls.jsonl` proved all three rows reached Pi exactly once in FIFO order; the all-mode capture rendered them together before the final response. - -## Normal-Pi adversarial evidence - -PR [#9](https://github.com/tmustier/pi-queue-steer/pull/9) was also exercised at commit `37fcd1433b8960f13c030d9ba1a5e8cc36535e05` through normal `pi` execution under tmux, using the installed extension and Pi 0.84.1 rather than extension-selection or test-fixture flags. The three captures cover automatic compaction above 200k tokens followed by queued `/reload`, manual `/compact` with Pi-native queued input ahead of an extension-owned `/reload`, and abort recovery with repeated queued reloads. +The output directory contains terminal captures, provider-call logs, lifecycle-event logs, and runtime-initialization logs. Run it immediately before review so `summary.txt` records the exact Pi version, commit, and working-tree state under test. A release evidence run should report `working tree: clean`. -The [public evidence comment](https://github.com/tmustier/pi-queue-steer/pull/9#issuecomment-5231404721) embeds the replacement Menlo-rendered videos and screenshots. Its [reproducible evidence bundle](https://github.com/user-attachments/files/30873175/pi-queue-steer-normal-pi-evidence-37fcd14.zip) has SHA-256 `c6b1150f13fccc195eb5747aa6af63c4589b73df116e3d36d27e92fb85a45e98`. The archive contains machine-checked assertions, tapes, captures, deterministic-suite output and bounded sanitized session proof slices. - -This evidence confirms the public API boundary: ordinary input submitted while manual compaction is active belongs to Pi's native post-compaction queue and can execute before extension-owned command rows resume. +The harness verifies manual compaction success and failure, abort pause and resume, queued reload restoration, resource expansion, native post-compaction ordering, overflow handling, and all-mode FIFO delivery. ## Public API boundary -`ExtensionAPI.sendUserMessage` and the TUI editor submit callback return `void`. The extension can restore synchronous handoff failures and preflight/expansion failures, but it cannot prove every later asynchronous acceptance or rejection without risking duplicate delivery. Queued `/reload` likewise has no result channel. These limits are documented in the README and are not hidden by timing heuristics. +`ExtensionAPI.sendUserMessage` and the TUI editor submit callback return `void`. The extension restores rows when prompt or Skill expansion fails before handoff, but it cannot observe later asynchronous acceptance or rejection without risking duplicate delivery. Queued `/reload` likewise has no result channel. + +`ExtensionContext.compact` reports completion and failure through `onComplete` and `onError`. Command rows remain blocked until one of those callbacks releases the queue. diff --git a/index.ts b/index.ts index d855803..a7c3f74 100644 --- a/index.ts +++ b/index.ts @@ -230,9 +230,8 @@ function userContent(item: QueuedMessage): string | (TextContent | } function itemCommand(item: Pick, "text" | "images">): QueuedCommand | undefined { - // Treat an image-bearing row as a message so executing a command can never - // silently discard its attachments. - return item.images.length === 0 ? parseQueuedCommand(item.text) : undefined; + if (item.images.length > 0) return undefined; + return parseQueuedCommand(item.text); } export default function queueSteerExtension(pi: ExtensionAPI) { @@ -269,15 +268,6 @@ export default function queueSteerExtension(pi: ExtensionAPI) { followUp: settingsManager?.getFollowUpMode() ?? "one-at-a-time", }); - const pauseAfterPreparationFailure = (ctx: ExtensionContext, lane: QueueLane, error: unknown): void => { - paused = true; - renderQueue(ctx); - ctx.ui.notify( - `Could not prepare queued ${laneLabel(lane)}; queue paused: ${error instanceof Error ? error.message : String(error)}`, - "error", - ); - }; - const laneIsHeld = (lane: QueueLane): boolean => { if (!editSession) return false; const mode = queueModes()[lane]; @@ -353,7 +343,7 @@ export default function queueSteerExtension(pi: ExtensionAPI) { // A command row at the lane head holds everything behind it (FIFO) until the // agent settles and dispatchFromIdle executes it. const takeLaneBatch = (lane: QueueLane): QueuedMessage[] => { - if (paused || blockingActivity || queue.laneLength(lane) === 0 || laneIsHeld(lane)) return []; + if (paused || blockingActivity || laneIsHeld(lane)) return []; const isMessage = (item: QueuedMessage) => itemCommand(item) === undefined; if (queueModes()[lane] === "all") return queue.shiftWhile(lane, isMessage); const head = queue.peek(lane); @@ -362,75 +352,61 @@ export default function queueSteerExtension(pi: ExtensionAPI) { return item ? [item] : []; }; - const deliverBatchToNativeQueue = async ( + const deliverMessages = ( ctx: ExtensionContext, lane: QueueLane, - items: QueuedMessage[], - ): Promise => { - if (items.length === 0) return false; + items: readonly QueuedMessage[], + deliverAs?: QueueLane, + ): void => { let prepared: QueuedMessage[]; try { const commands = pi.getCommands(); prepared = items.map((item) => ({ ...item, text: expandQueuedInput(item.text, commands) })); } catch (error) { queue.prependMany(items); - pauseAfterPreparationFailure(ctx, lane, error); - return false; - } - renderQueue(ctx); - let submitted = 0; - try { - for (const item of prepared) { - pi.sendUserMessage(userContent(item), { deliverAs: lane }); - submitted += 1; - } - // The public send API is fire-and-forget. Once invoked, do not infer - // rejection from aggregate queue timing: a delayed preflight could - // otherwise accept the original after we restored and duplicate it. - return true; - } catch (error) { - queue.prependMany(items.slice(submitted)); + paused = true; renderQueue(ctx); ctx.ui.notify( - `Could not deliver queued ${laneLabel(lane)}: ${error instanceof Error ? error.message : String(error)}`, + `Could not prepare queued ${laneLabel(lane)}; queue paused: ${error instanceof Error ? error.message : String(error)}`, "error", ); - return false; + return; + } + renderQueue(ctx); + for (const item of prepared) { + pi.sendUserMessage(userContent(item), deliverAs ? { deliverAs } : undefined); } }; - const dispatchLaneAtBoundary = async (ctx: ExtensionContext, lane: QueueLane): Promise => { + const dispatchLaneAtBoundary = (ctx: ExtensionContext, lane: QueueLane): void => { activeContext = ctx; const items = takeLaneBatch(lane); if (items.length === 0) { renderQueue(ctx); - return false; + return; } - return deliverBatchToNativeQueue(ctx, lane, items); + deliverMessages(ctx, lane, items, lane); }; // Execute the command row at the lane head. Only called when the agent is idle. - const executeCommandRow = (ctx: ExtensionContext, lane: QueueLane): boolean => { + const executeCommandRow = (ctx: ExtensionContext, lane: QueueLane): void => { const next = queue.peek(lane); - if (!next) return false; + if (!next) return; const command = itemCommand(next); - if (!command) return false; + if (!command) return; const submit = tuiSubmit; if (command.kind === "reload" && !submit) { paused = true; renderQueue(ctx); ctx.ui.notify("Could not run queued /reload; queue paused because no interactive submit handler is available", "error"); - return false; + return; } queue.shift(lane); paused = false; renderQueue(ctx); if (command.kind === "compact") { - if (startCompaction(ctx, command.instructions)) return true; - queue.prepend(next); - paused = true; - renderQueue(ctx); - return false; + startCompaction(ctx, command.instructions); + return; } blockingActivity = "reload"; // Defer so the extension runtime is never torn down from inside this handler. @@ -438,41 +414,20 @@ export default function queueSteerExtension(pi: ExtensionAPI) { reloadSubmitTimer = undefined; submit?.("/reload"); }, 0); - return true; }; - const sendHeadMessage = (ctx: ExtensionContext, lane: QueueLane, deliverAs?: QueueLane): boolean => { - const head = queue.peek(lane); - if (!head) return false; - let prepared: QueuedMessage; - try { - prepared = { ...head, text: expandQueuedInput(head.text, pi.getCommands()) }; - } catch (error) { - pauseAfterPreparationFailure(ctx, lane, error); - return false; - } - queue.shift(lane); + const sendHeadMessage = (ctx: ExtensionContext, lane: QueueLane, deliverAs?: QueueLane): void => { + const head = queue.shift(lane); + if (!head) return; paused = false; - renderQueue(ctx); - try { - pi.sendUserMessage(userContent(prepared), deliverAs ? { deliverAs } : undefined); - return true; - } catch (error) { - queue.prepend(head); - renderQueue(ctx); - ctx.ui.notify( - `Could not send queued ${laneLabel(lane)}: ${error instanceof Error ? error.message : String(error)}`, - "error", - ); - return false; - } + deliverMessages(ctx, lane, [head], deliverAs); }; - const dispatchFromIdle = (ctx: ExtensionContext): boolean => { + const dispatchFromIdle = (ctx: ExtensionContext): void => { activeContext = ctx; if (blockingActivity) { renderQueue(ctx); - return false; + return; } const lane: QueueLane | undefined = queue.laneLength("steer") > 0 ? "steer" @@ -481,11 +436,14 @@ export default function queueSteerExtension(pi: ExtensionAPI) { : undefined; if (!lane || laneIsHeld(lane)) { renderQueue(ctx); - return false; + return; } const head = queue.peek(lane); - if (head && itemCommand(head)) return executeCommandRow(ctx, lane); - return sendHeadMessage(ctx, lane); + if (head && itemCommand(head)) { + executeCommandRow(ctx, lane); + return; + } + sendHeadMessage(ctx, lane); }; const deferCompactionFinish = ( @@ -510,29 +468,19 @@ export default function queueSteerExtension(pi: ExtensionAPI) { }, 0); }; - const startCompaction = (ctx: ExtensionContext, instructions: string | undefined): boolean => { + const startCompaction = (ctx: ExtensionContext, instructions: string | undefined): void => { blockingActivity = "compact"; nativeCompactionInputQueued = false; nativeCompactionTurnStarted = false; - try { - ctx.compact({ - customInstructions: instructions, - onComplete: () => { - if (!nativeCompactionInputQueued) deferCompactionFinish(ctx, "compact"); - }, - onError: () => { - if (!nativeCompactionInputQueued) deferCompactionFinish(ctx, "compact"); - }, - }); - return true; - } catch (error) { - blockingActivity = undefined; - ctx.ui.notify( - `Could not start compaction: ${error instanceof Error ? error.message : String(error)}`, - "error", - ); - return false; - } + ctx.compact({ + customInstructions: instructions, + onComplete: () => { + if (!nativeCompactionInputQueued) deferCompactionFinish(ctx, "compact"); + }, + onError: () => { + if (!nativeCompactionInputQueued) deferCompactionFinish(ctx, "compact"); + }, + }); }; const deferCommand = (ctx: ExtensionContext, text: string): void => { @@ -541,18 +489,19 @@ export default function queueSteerExtension(pi: ExtensionAPI) { renderQueue(ctx); }; - const sendFollowUpNow = (ctx: ExtensionContext): boolean => { + const sendFollowUpNow = (ctx: ExtensionContext): void => { const head = queue.peek("followUp"); - if (!head) return false; + if (!head) return; const headCommand = itemCommand(head); if (headCommand) { if (blockingActivity === "reload" || !ctx.isIdle()) { ctx.ui.notify(`Queued /${headCommand.kind} runs when the agent is idle`, "info"); - return false; + return; } - return executeCommandRow(ctx, "followUp"); + executeCommandRow(ctx, "followUp"); + return; } - return sendHeadMessage(ctx, "followUp", ctx.isIdle() ? undefined : "steer"); + sendHeadMessage(ctx, "followUp", ctx.isIdle() ? undefined : "steer"); }; const finishEditing = ( @@ -857,7 +806,7 @@ export default function queueSteerExtension(pi: ExtensionAPI) { if (isCompacting() && nativeCompactionInputQueued) nativeCompactionTurnStarted = true; }); - pi.on("turn_end", async (event, ctx) => { + pi.on("turn_end", (event, ctx) => { activeContext = ctx; if (event.message.role === "assistant" && event.message.stopReason === "aborted") { if (queue.length > 0 && blockingActivity !== "compact") paused = true; @@ -865,13 +814,13 @@ export default function queueSteerExtension(pi: ExtensionAPI) { return; } if (paused) return; - await dispatchLaneAtBoundary(ctx, "steer"); + dispatchLaneAtBoundary(ctx, "steer"); }); // Pi checks its native queues again after extension agent_end handlers. // Feeding one item (or an all-mode batch) here preserves native follow-up // continuation semantics without relinquishing later editable rows early. - pi.on("agent_end", async (event, ctx) => { + pi.on("agent_end", (event, ctx) => { activeContext = ctx; if (paused) return; const lastMessage = event.messages.at(-1); @@ -888,10 +837,10 @@ export default function queueSteerExtension(pi: ExtensionAPI) { return; } if (queue.laneLength("steer") > 0) { - await dispatchLaneAtBoundary(ctx, "steer"); + dispatchLaneAtBoundary(ctx, "steer"); return; } - await dispatchLaneAtBoundary(ctx, "followUp"); + dispatchLaneAtBoundary(ctx, "followUp"); }); pi.on("agent_settled", (_event, ctx) => { diff --git a/queue-state.ts b/queue-state.ts index 8c95023..138bbb9 100644 --- a/queue-state.ts +++ b/queue-state.ts @@ -62,8 +62,7 @@ export class DeliveryQueue { prependMany(items: readonly QueuedMessage[]): void { for (let index = items.length - 1; index >= 0; index -= 1) { - const item = items[index]; - if (item) this.prepend(item); + this.prepend(items[index]); } } @@ -80,7 +79,6 @@ export class DeliveryQueue { const index = this.items.findIndex((item) => item.id === id); if (index === -1) return false; const [item] = this.items.splice(index, 1); - if (!item) return false; item.lane = lane; this.items.push(item); return true; @@ -90,7 +88,7 @@ export class DeliveryQueue { const index = this.items.findIndex((item) => item.id === id); if (index === -1) return undefined; const [item] = this.items.splice(index, 1); - return item ? this.copy(item) : undefined; + return this.copy(item); } peek(lane: QueueLane): QueuedMessage | undefined { @@ -102,13 +100,7 @@ export class DeliveryQueue { const index = this.items.findIndex((item) => item.lane === lane); if (index === -1) return undefined; const [item] = this.items.splice(index, 1); - return item ? this.copy(item) : undefined; - } - - shiftAll(lane: QueueLane): QueuedMessage[] { - const removed = this.items.filter((item) => item.lane === lane).map((item) => this.copy(item)); - this.items = this.items.filter((item) => item.lane !== lane); - return removed; + return this.copy(item); } /** @@ -122,7 +114,6 @@ export class DeliveryQueue { const index = this.items.findIndex((item) => item.lane === lane); if (index === -1 || !accept(this.items[index])) break; const [item] = this.items.splice(index, 1); - if (!item) break; taken.push(this.copy(item)); } return taken; @@ -133,24 +124,6 @@ export class DeliveryQueue { return item ? this.copy(item) : undefined; } - previousId(currentId?: string): string | undefined { - const ordered = this.snapshot(); - if (ordered.length === 0) return undefined; - if (!currentId) return this.mostRecentId(); - const index = ordered.findIndex((item) => item.id === currentId); - if (index <= 0) return ordered.at(-1)?.id; - return ordered[index - 1]?.id; - } - - nextId(currentId?: string): string | undefined { - const ordered = this.snapshot(); - if (ordered.length === 0) return undefined; - if (!currentId) return this.mostRecentId(); - const index = ordered.findIndex((item) => item.id === currentId); - if (index === -1 || index === ordered.length - 1) return ordered[0]?.id; - return ordered[index + 1]?.id; - } - mostRecentId(): string | undefined { let newest: QueuedMessage | undefined; for (const item of this.items) { @@ -177,13 +150,10 @@ export class DeliveryQueue { /** Restore an in-memory queue snapshot without changing row identity or recency. */ restore(items: readonly QueuedMessage[]): void { - const ids = new Set(); let highestIdNumber = 0; let highestSequence = 0; const restored: QueuedMessage[] = []; for (const item of items) { - if (ids.has(item.id)) throw new Error(`Duplicate queued row ID: ${item.id}`); - ids.add(item.id); const idNumber = /-(\d+)$/.exec(item.id)?.[1]; if (idNumber) highestIdNumber = Math.max(highestIdNumber, Number.parseInt(idNumber, 10)); highestSequence = Math.max(highestSequence, item.sequence); diff --git a/test/pi-integration.test.ts b/test/pi-integration.test.ts index 5073f92..f4b7bef 100644 --- a/test/pi-integration.test.ts +++ b/test/pi-integration.test.ts @@ -66,28 +66,6 @@ function nextAgentRun(session: AgentSession): Promise { }); } -function nextAgentRunForUser(session: AgentSession, expected: string): Promise { - return new Promise((resolve) => { - let matched = false; - let unsubscribe: (() => void) | undefined; - unsubscribe = session.subscribe((event) => { - if (event.type === "message_start" && event.message.role === "user") { - const text = typeof event.message.content === "string" - ? event.message.content - : event.message.content - .filter((part) => part.type === "text") - .map((part) => part.text) - .join("\n"); - if (text === expected) matched = true; - return; - } - if (event.type !== "agent_settled" || !matched) return; - unsubscribe?.(); - resolve(); - }); - }); -} - async function within(promise: Promise, detail: () => string): Promise { let timer: ReturnType | undefined; try { @@ -358,7 +336,25 @@ test("real public prompt path triggers overflow compaction and preserves a queue const prompt = harness.session.prompt("x".repeat(20_000)); await within(activeStarted, () => trace.join(", ")); await harness.session.prompt("queued across overflow", { streamingBehavior: "followUp" }); - const queuedRun = nextAgentRunForUser(harness.session, "queued across overflow"); + const queuedRun = new Promise((resolve) => { + let matched = false; + let unsubscribe: (() => void) | undefined; + unsubscribe = harness.session.subscribe((event) => { + if (event.type === "message_start" && event.message.role === "user") { + const text = typeof event.message.content === "string" + ? event.message.content + : event.message.content + .filter((part) => part.type === "text") + .map((part) => part.text) + .join("\n"); + if (text === "queued across overflow") matched = true; + return; + } + if (event.type !== "agent_settled" || !matched) return; + unsubscribe?.(); + resolve(); + }); + }); active.release(); await within(prompt, () => trace.join(", ")); diff --git a/test/queue-state.test.ts b/test/queue-state.test.ts index e129fdb..2fef923 100644 --- a/test/queue-state.test.ts +++ b/test/queue-state.test.ts @@ -29,17 +29,13 @@ test("keeps steering and follow-ups in independent FIFOs", () => { assert.equal(queue.shift("followUp")?.text, "later one"); }); -test("selects the globally most recent item before navigating spatially", () => { +test("identifies the globally most recent item across lanes", () => { const queue = new DeliveryQueue(); - const firstSteer = queue.enqueue("steer", "steer one"); - const latestFollowUp = queue.enqueue("followUp", "later"); - const latestSteer = queue.enqueue("steer", "steer two"); - - assert.equal(queue.mostRecentId(), latestSteer.id); - assert.equal(queue.previousId(), latestSteer.id); - assert.equal(queue.previousId(latestSteer.id), firstSteer.id); - assert.equal(queue.nextId(latestSteer.id), latestFollowUp.id); - assert.equal(queue.nextId(latestFollowUp.id), firstSteer.id); + queue.enqueue("steer", "steer one"); + queue.enqueue("followUp", "later"); + const latest = queue.enqueue("steer", "steer two"); + + assert.equal(queue.mostRecentId(), latest.id); }); test("edits a row without changing its stable lane position", () => { @@ -55,7 +51,7 @@ test("restores failed batches at the front in their original order", () => { const queue = new DeliveryQueue(); queue.enqueue("followUp", "first"); queue.enqueue("followUp", "second"); - const failed = queue.shiftAll("followUp"); + const failed = queue.shiftWhile("followUp", () => true); queue.enqueue("followUp", "third"); queue.prependMany(failed); @@ -78,12 +74,6 @@ test("restores queue snapshots with stable IDs, recency, images, and collision-f assert.ok(next.sequence > mostRecent.sequence); }); -test("rejects duplicate row IDs in restored snapshots", () => { - const queue = new DeliveryQueue(); - const row = queue.enqueue("steer", "one"); - assert.throws(() => queue.restore([row, { ...row, text: "duplicate" }]), /Duplicate queued row ID/); -}); - test("edit sessions keep cross-lane drafts private until commit", () => { const queue = new DeliveryQueue(); const steer = queue.enqueue("steer", "steer original"); @@ -199,11 +189,8 @@ test.after(() => rmSync(DEFAULT_TEST_CWD, { recursive: true, force: true })); function createHarness(options: { cwd?: string; - projectTrusted?: boolean; commands?: SlashCommandInfo[]; mode?: "tui" | "rpc" | "json" | "print"; - sendFailureAt?: number; - compactStartError?: Error; autocompleteVisible?: boolean; } = {}) { type Handler = (event: any, context: any) => any; @@ -213,7 +200,6 @@ function createHarness(options: { const compactCalls: CompactOptions[] = []; const notifications: Array<{ message: string; level: string }> = []; let idle = false; - let pending = false; let aborted = false; const createDefaultEditor = (): MockEditor => { const editor = new MockEditor(); @@ -262,13 +248,11 @@ function createHarness(options: { cwd: options.cwd ?? DEFAULT_TEST_CWD, ui, isIdle: () => idle, - isProjectTrusted: () => options.projectTrusted ?? true, - hasPendingMessages: () => pending, + isProjectTrusted: () => true, abort() { aborted = true; }, compact(compactOptions: CompactOptions = {}) { - if (options.compactStartError) throw options.compactStartError; compactCalls.push(compactOptions); }, }; @@ -280,9 +264,7 @@ function createHarness(options: { handlers.set(name, registered); }, sendUserMessage(content: unknown, sendOptions?: unknown) { - if (options.sendFailureAt === sent.length + 1) throw new Error("synthetic send failure"); sent.push({ content, options: sendOptions }); - if (sendOptions) pending = true; }, getCommands: () => options.commands ?? [], }; @@ -324,9 +306,6 @@ function createHarness(options: { setIdle(value: boolean) { idle = value; }, - clearPending() { - pending = false; - }, replaceEditor(editor = new MockEditor()) { ui.setEditorComponent(() => editor); }, @@ -455,41 +434,6 @@ test("injects follow-ups through Pi's native continuation queue at agent_end", a assert.match(renderWidget(harness), /later two/); }); -test("restores only the unsent tail after a synchronous all-mode batch failure", async () => { - const cwd = mkdtempSync(join(tmpdir(), "pi-queue-partial-send-")); - mkdirSync(join(cwd, ".pi")); - writeFileSync(join(cwd, ".pi", "settings.json"), JSON.stringify({ followUpMode: "all" })); - const harness = createHarness({ cwd, projectTrusted: true, sendFailureAt: 2 }); - try { - await harness.emit("session_start"); - await enqueue(harness, "followUp", "accepted first"); - await enqueue(harness, "followUp", "restore second"); - await enqueue(harness, "followUp", "restore third"); - - await harness.emit("agent_end"); - assert.deepEqual(harness.sent.map((item) => item.content), ["accepted first"]); - const rendered = renderWidget(harness); - assert.doesNotMatch(rendered, /accepted first/); - assert.match(rendered, /restore second/); - assert.match(rendered, /restore third/); - assert.match(harness.notifications.at(-1)?.message ?? "", /synthetic send failure/); - } finally { - rmSync(cwd, { recursive: true, force: true }); - } -}); - -test("restores an idle row after a synchronous send failure", async () => { - const harness = createHarness({ sendFailureAt: 1 }); - await harness.emit("session_start"); - harness.setIdle(true); - await enqueue(harness, "followUp", "retry me"); - - await harness.emit("agent_settled"); - assert.equal(harness.sent.length, 0); - assert.match(renderWidget(harness), /retry me/); - assert.match(harness.notifications.at(-1)?.message ?? "", /synthetic send failure/); -}); - test("delivers image-bearing command text as a message without dropping attachments", async () => { const harness = createHarness(); const image: ImageContent = { type: "image", data: "AA==", mimeType: "image/png" }; @@ -510,13 +454,20 @@ test("delivers image-bearing command text as a message without dropping attachme assert.deepEqual(harness.submitted, []); }); -test("does not take ownership of interactive-source input outside TUI mode", async () => { - const modes: ("rpc" | "json" | "print")[] = ["rpc", "json", "print"]; - for (const mode of modes) { +test("does not take ownership of input outside TUI mode", async () => { + const inputs: Array<{ + mode: "print" | "json" | "rpc"; + source: "interactive" | "rpc"; + }> = [ + { mode: "print", source: "interactive" }, + { mode: "json", source: "interactive" }, + { mode: "rpc", source: "rpc" }, + ]; + for (const { mode, source } of inputs) { const harness = createHarness({ mode }); await harness.emit("session_start", { reason: "startup" }); const results = await harness.emit("input", { - source: "interactive", + source, text: "/reload", streamingBehavior: "followUp", }); @@ -537,7 +488,7 @@ test("honours Pi all-mode settings and pins the whole edited lane", async () => JSON.stringify({ steeringMode: "all", followUpMode: "all" }), ); try { - const steering = createHarness({ cwd, projectTrusted: true }); + const steering = createHarness({ cwd }); await steering.emit("session_start"); await enqueue(steering, "steer", "steer one"); await enqueue(steering, "steer", "steer two"); @@ -548,7 +499,7 @@ test("honours Pi all-mode settings and pins the whole edited lane", async () => await steering.emit("turn_end", { message: { role: "assistant", stopReason: "toolUse" } }); assert.deepEqual(steering.sent.map((item) => item.content), ["steer one", "steer two"]); - const followUps = createHarness({ cwd, projectTrusted: true }); + const followUps = createHarness({ cwd }); await followUps.emit("session_start"); await enqueue(followUps, "followUp", "later one"); await enqueue(followUps, "followUp", "later two"); @@ -564,7 +515,7 @@ test("restarts an all-mode lane in FIFO order after it stays pinned through sett mkdirSync(join(cwd, ".pi")); writeFileSync(join(cwd, ".pi", "settings.json"), JSON.stringify({ followUpMode: "all" })); try { - const harness = createHarness({ cwd, projectTrusted: true }); + const harness = createHarness({ cwd }); await harness.emit("session_start"); await enqueue(harness, "followUp", "restart one"); await enqueue(harness, "followUp", "restart two"); @@ -865,7 +816,6 @@ test("expands queued prompt templates and short Agent Skill commands at delivery assert.match(String(harness.sent[0]?.content), / { - const harness = createHarness({ compactStartError: new Error("cannot start") }); - await harness.emit("session_start"); - harness.setIdle(true); - await enqueue(harness, "followUp", "/compact"); - await enqueue(harness, "followUp", "after compact"); - - await harness.emit("agent_settled"); - const rendered = renderWidget(harness); - assert.match(rendered, /\/compact/); - assert.match(rendered, /after compact/); - assert.match(rendered, /paused/); - assert.match(harness.notifications.at(-1)?.message ?? "", /Could not start compaction: cannot start/); - assert.equal(harness.sent.length, 0); -}); - test("leaves ordinary compaction input native and waits for its full run", async () => { const harness = createHarness(); await harness.emit("session_start"); @@ -1127,7 +1060,7 @@ test("preserves a paused queue and attachments across direct runtime reload", as }); test("survives repeated queued reloads without expiry, reordering, or duplication", async () => { - const reloadCount = 25; + const reloadCount = 2; let runtime = createHarness(); await runtime.emit("session_start", { reason: "startup" }); runtime.setIdle(true); diff --git a/test/queued-input.test.ts b/test/queued-input.test.ts index 4ef38df..5510368 100644 --- a/test/queued-input.test.ts +++ b/test/queued-input.test.ts @@ -87,27 +87,31 @@ test("uses defaults for empty all-argument prompt placeholders", () => { } }); -test("classifies only native post-compaction TUI submissions", () => { - const extension = command("deploy", "extension", "/extension.ts"); - const prompt = command("review", "prompt", "/review.md"); - assert.equal(queuesDuringCompaction("ordinary message", [extension, prompt]), true); - assert.equal(queuesDuringCompaction("/unknown as text", [extension, prompt]), true); - assert.equal(queuesDuringCompaction("/review now", [extension, prompt]), true); - assert.equal(queuesDuringCompaction("/skill:bro now", [extension, prompt]), true); - assert.equal(queuesDuringCompaction("/deploy prod", [extension, prompt]), false); - assert.equal(queuesDuringCompaction("/model small", [extension, prompt]), false); - assert.equal(queuesDuringCompaction(" /model small ", [extension, prompt]), false); - assert.equal(queuesDuringCompaction("/debug", [extension, prompt]), false); - assert.equal(queuesDuringCompaction("!echo now", [extension, prompt]), false); - assert.equal(queuesDuringCompaction(" !echo now ", [extension, prompt]), false); - assert.equal(queuesDuringCompaction("", [extension, prompt]), false); - assert.equal(queuesDuringCompaction(" ", [extension, prompt]), false); +test("classifies native post-compaction TUI submissions", () => { + const commands = [ + command("deploy", "extension", "/extension.ts"), + command("review", "prompt", "/review.md"), + ]; + const cases: Array<{ + text: string; + behavior?: "followUp"; + expected: boolean; + }> = [ + { text: "ordinary message", expected: true }, + { text: "/review now", expected: true }, + { text: "/deploy prod", expected: false }, + { text: "/model small", expected: false }, + { text: "!echo now", expected: false }, + { text: " ", expected: false }, + { text: "ordinary follow-up", behavior: "followUp", expected: true }, + { text: "/model small", behavior: "followUp", expected: true }, + { text: "/deploy prod", behavior: "followUp", expected: false }, + { text: " ", behavior: "followUp", expected: false }, + ]; - assert.equal(queuesDuringCompaction("ordinary follow-up", [extension, prompt], "followUp"), true); - assert.equal(queuesDuringCompaction("/model small", [extension, prompt], "followUp"), true); - assert.equal(queuesDuringCompaction("!echo now", [extension, prompt], "followUp"), true); - assert.equal(queuesDuringCompaction("/deploy prod", [extension, prompt], "followUp"), false); - assert.equal(queuesDuringCompaction(" ", [extension, prompt], "followUp"), false); + for (const { text, behavior, expected } of cases) { + assert.equal(queuesDuringCompaction(text, commands, behavior), expected, text); + } }); test("rejects discovered extension commands", () => {