From ae8b7b3bd88fec4a6a4453e0a8c25119b0c9c353 Mon Sep 17 00:00:00 2001 From: Thomas Mustier <6326440+tmustier@users.noreply.github.com> Date: Thu, 1 Oct 2026 13:48:38 +0100 Subject: [PATCH 1/4] fix: release steering when automatic compaction finishes --- README.md | 1 + index.ts | 32 ++++++++++++++++-- test/pi-integration.test.ts | 67 +++++++++++++++++++++++++++++++++++-- test/queue-state.test.ts | 61 +++++++++++++++++++++++++++++++++ 4 files changed, 155 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index f876d6e..39e8cc9 100644 --- a/README.md +++ b/README.md @@ -62,6 +62,7 @@ The extension follows your configured Pi action bindings. These are the default The extension keeps Pi’s 2 delivery classes: - steering reaches the current run at Pi’s next safe turn boundary +- automatic compaction holds delivery only while compaction is active; steering resumes at the next boundary even if the run continues making tool calls - follow-ups wait until the run finishes - the blue steering box remains above the yellow follow-up box - each lane keeps its own first-in, first-out order diff --git a/index.ts b/index.ts index d855803..8c20399 100644 --- a/index.ts +++ b/index.ts @@ -353,7 +353,8 @@ 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 || (nativeCompactionInputQueued && !nativeCompactionTurnStarted) + || queue.laneLength(lane) === 0 || laneIsHeld(lane)) return []; const isMessage = (item: QueuedMessage) => itemCommand(item) === undefined; if (queueModes()[lane] === "all") return queue.shiftWhile(lane, isMessage); const head = queue.peek(lane); @@ -470,7 +471,7 @@ export default function queueSteerExtension(pi: ExtensionAPI) { const dispatchFromIdle = (ctx: ExtensionContext): boolean => { activeContext = ctx; - if (blockingActivity) { + if (blockingActivity || nativeCompactionInputQueued) { renderQueue(ctx); return false; } @@ -510,6 +511,22 @@ export default function queueSteerExtension(pi: ExtensionAPI) { }, 0); }; + const finishAutomaticCompaction = (ctx: ExtensionContext): void => { + if (blockingActivity !== "auto-compact") return; + // Automatic compaction can finish between tool turns, long before the + // run settles. Release message delivery now, not at agent_settled. + blockingActivity = undefined; + renderQueue(ctx); + if (compactionFinishTimer) clearTimeout(compactionFinishTimer); + compactionFinishTimer = setTimeout(() => { + compactionFinishTimer = undefined; + const current = activeContext ?? ctx; + // Native input may still be in asynchronous preflight. dispatchFromIdle + // keeps command rows behind that input until its run has settled. + if (!paused && !editSession && queue.length > 0 && current.isIdle()) dispatchFromIdle(current); + }, 0); + }; + const startCompaction = (ctx: ExtensionContext, instructions: string | undefined): boolean => { blockingActivity = "compact"; nativeCompactionInputQueued = false; @@ -852,9 +869,12 @@ export default function queueSteerExtension(pi: ExtensionAPI) { renderQueue(ctx); }); + pi.on("session_compact", (_event, ctx) => finishAutomaticCompaction(ctx)); + pi.on("session_compact_failed", (_event, ctx) => finishAutomaticCompaction(ctx)); + pi.on("turn_start", (_event, ctx) => { activeContext = ctx; - if (isCompacting() && nativeCompactionInputQueued) nativeCompactionTurnStarted = true; + if (nativeCompactionInputQueued) nativeCompactionTurnStarted = true; }); pi.on("turn_end", async (event, ctx) => { @@ -907,6 +927,12 @@ export default function queueSteerExtension(pi: ExtensionAPI) { deferCompactionFinish(ctx, activity); return; } + if (nativeCompactionInputQueued && !nativeCompactionTurnStarted) { + renderQueue(ctx); + return; + } + nativeCompactionInputQueued = false; + nativeCompactionTurnStarted = false; renderQueue(ctx); if (!paused && !editSession && queue.length > 0 && ctx.isIdle() && !blockingActivity) dispatchFromIdle(ctx); }); diff --git a/test/pi-integration.test.ts b/test/pi-integration.test.ts index 7f6b5d1..d808d64 100644 --- a/test/pi-integration.test.ts +++ b/test/pi-integration.test.ts @@ -5,6 +5,7 @@ import { join } from "node:path"; import test from "node:test"; import { fauxAssistantMessage, + fauxToolCall, fauxProvider, type FauxProviderHandle, } from "@earendil-works/pi-ai/compat"; @@ -120,6 +121,8 @@ async function createIntegrationHarness(options: { maxTokens?: number; extraExtensions?: ExtensionFactory[]; retryEnabled?: boolean; + reserveTokens?: number; + tools?: string[]; } = {}): Promise { const cwd = mkdtempSync(join(tmpdir(), "pi-queue-integration-")); const agentDir = join(cwd, "agent"); @@ -133,7 +136,7 @@ async function createIntegrationHarness(options: { }); const model = faux.getModel(); const settingsManager = SettingsManager.inMemory({ - compaction: { enabled: options.compactionEnabled ?? true, keepRecentTokens: 1, reserveTokens: 0 }, + compaction: { enabled: options.compactionEnabled ?? true, keepRecentTokens: 1, reserveTokens: options.reserveTokens ?? 0 }, retry: options.retryEnabled ? { enabled: true, maxRetries: 2, baseDelayMs: 1 } : { enabled: false }, @@ -176,7 +179,8 @@ async function createIntegrationHarness(options: { settingsManager, sessionManager, resourceLoader, - noTools: "all", + tools: options.tools, + noTools: options.tools ? undefined : "all", }); await session.bindExtensions({ mode: "tui" }); return { @@ -193,7 +197,7 @@ async function createIntegrationHarness(options: { } function gatedResponse( - content: string, + content: Parameters[0], options?: Parameters[1], ): { step: () => Promise>; @@ -380,6 +384,63 @@ test("Pi 0.87 finishes settled handlers before starting queue-steer's requested } }); +test("steering reaches the continuing tool run after real threshold compaction, before agent_end", async () => { + let enterResumed: (() => void) | undefined; + const resumedEntered = new Promise((resolve) => { enterResumed = resolve; }); + const resumed = gatedResponse( + [fauxToolCall("bash", { command: "printf resumed-tool", timeout: 5 })], + { stopReason: "toolUse" }, + ); + const summaryExtension: ExtensionFactory = (pi) => { + pi.on("session_before_compact", (event) => ({ + compaction: { + summary: "short integration summary", + firstKeptEntryId: event.preparation.firstKeptEntryId, + tokensBefore: event.preparation.tokensBefore, + }, + })); + }; + const harness = await createIntegrationHarness({ + contextWindow: 100_000, + reserveTokens: 99_000, + tools: ["bash"], + extraExtensions: [summaryExtension], + }); + const trace: string[] = []; + harness.session.subscribe((event) => { + trace.push(event.type); + }); + harness.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("bash", { command: "printf initial-tool", timeout: 5 })], { stopReason: "toolUse" }), + async () => { + enterResumed?.(); + return resumed.step(); + }, + (context) => { + const steering = context.messages.filter((message) => message.role === "user").at(-1); + assert.ok(JSON.stringify(steering).includes("steer after threshold")); + assert.equal(trace.includes("agent_end"), false); + return fauxAssistantMessage("steering handled inside original run"); + }, + ]); + const prompt = harness.session.prompt("initial context ".repeat(1_000)); + try { + await within(resumedEntered, () => trace.join(", ")); + assert.ok(trace.includes("compaction_end")); + assert.equal(trace.includes("agent_settled"), false); + await harness.session.prompt("steer after threshold", { streamingBehavior: "steer" }); + resumed.release(); + await within(prompt, () => trace.join(", ")); + assert.equal(userTexts(harness.session).filter((text) => text === "steer after threshold").length, 1); + assert.equal(harness.session.getLastAssistantText(), "steering handled inside original run"); + assert.equal(trace.filter((type) => type === "agent_start").length, 1); + } finally { + resumed.release(); + await prompt.catch(() => undefined); + await harness.cleanup(); + } +}); + test("real public prompt path triggers overflow compaction and preserves a queued follow-up", async () => { const summaryExtension: ExtensionFactory = (pi) => { pi.on("session_before_compact", (event) => { diff --git a/test/queue-state.test.ts b/test/queue-state.test.ts index e129fdb..7eb3aa4 100644 --- a/test/queue-state.test.ts +++ b/test/queue-state.test.ts @@ -1026,6 +1026,67 @@ test("releases a length-stop hold at settle when Pi does not compact", async () assert.equal(harness.sent[0]?.content, "after full-length output"); }); +for (const [outcome, aborted] of [ + ["session_compact", false], + ["session_compact_failed", false], + ["session_compact_failed", true], +] as const) { + test(`releases steering at the next turn after automatic compaction: ${outcome}, aborted=${aborted}`, async () => { + const harness = createHarness(); + await harness.emit("session_start"); + await harness.emit("session_before_compact", { reason: "threshold" }); + await enqueue(harness, "steer", "steer after compact"); + await enqueue(harness, "followUp", "wait until run ends"); + await harness.emit("turn_end", { message: { role: "assistant", stopReason: "toolUse" } }); + assert.equal(harness.sent.length, 0); + + await harness.emit(outcome, { reason: "threshold", aborted }); + // No agent_settled event: the original run is still making tool calls. + await harness.emit("turn_start"); + await harness.emit("turn_end", { message: { role: "assistant", stopReason: "toolUse" } }); + assert.deepEqual(harness.sent, [{ content: "steer after compact", options: { deliverAs: "steer" } }]); + assert.match(renderWidget(harness), /wait until run ends/); + }); +} + +test("automatic compaction completion preserves native-input ordering but allows steering during that run", async () => { + const harness = createHarness(); + await harness.emit("session_start"); + harness.setIdle(true); + await harness.emit("session_before_compact", { reason: "threshold" }); + harness.editor.onSubmit?.("ordinary native input"); + harness.editor.onSubmit?.("/reload"); + await enqueue(harness, "steer", "steer the native run"); + await harness.emit("session_compact", { reason: "threshold" }); + await new Promise((resolve) => setTimeout(resolve, 5)); + assert.deepEqual(harness.submitted, ["ordinary native input"]); + assert.equal(harness.sent.length, 0); + + harness.setIdle(false); + await harness.emit("turn_start"); + await harness.emit("turn_end", { message: { role: "assistant", stopReason: "toolUse" } }); + assert.equal(harness.sent[0]?.content, "steer the native run"); + assert.deepEqual(harness.submitted, ["ordinary native input"]); + harness.setIdle(true); + await harness.emit("agent_settled"); + await waitFor(() => harness.submitted.length === 2); + assert.deepEqual(harness.submitted, ["ordinary native input", "/reload"]); +}); + +test("automatic compaction completion does not release a paused or edited steering head", async () => { + for (const hold of ["pause", "edit"] as const) { + const harness = createHarness(); + await harness.emit("session_start"); + await harness.emit("session_before_compact", { reason: "threshold" }); + await enqueue(harness, "steer", "held head"); + harness.editor.handleInput(hold === "pause" ? "escape" : "alt-up"); + await harness.emit("session_compact", { reason: "threshold" }); + await harness.emit("turn_end", { message: { role: "assistant", stopReason: "toolUse" } }); + assert.equal(harness.sent.length, 0); + assert.match(renderWidget(harness), hold === "pause" ? /paused/ : /held while editing/); + } +}); + test("holds reload through automatic compaction until the agent settles", async () => { const harness = createHarness(); await harness.emit("session_start"); From b46e3b14b1d64ce8708f3b5dec15d8603150ed6c Mon Sep 17 00:00:00 2001 From: Thomas Mustier <6326440+tmustier@users.noreply.github.com> Date: Thu, 1 Oct 2026 13:51:38 +0100 Subject: [PATCH 2/4] docs: record automatic-compaction steering verification --- docs/validation.md | 37 ++++++++++++++++++++++++++++++++++++- 1 file changed, 36 insertions(+), 1 deletion(-) diff --git a/docs/validation.md b/docs/validation.md index 5b638e8..7b9737e 100644 --- a/docs/validation.md +++ b/docs/validation.md @@ -25,7 +25,42 @@ 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, settled-handler launch ordering, repeated reload restoration, and compaction/native-input ordering. -Latest result with Pi 0.87.0: 82 tests passed. +Latest result for the automatic-compaction steering fix: 88 tests passed on both Pi 0.87.0 (the lockfile) and Pi 0.99.2 (the current installed release). + +## Automatic-compaction steering regression + +Validated on 1 October 2026 against code revision `ae8b7b3bd88fec4a6a4453e0a8c25119b0c9c353`. The later validation-record commit changes documentation only. + +The regression test makes Pi compact between tool turns, then queues steering while the next assistant response is gated. It checks that the following provider request sees the steering before `agent_end`, with one agent run and exactly one delivered steering message. The test uses actual `AgentSession` scheduling, real bash tools and a deterministic provider. + +Against the original `91e3a5f` implementation, all 5 targeted regressions failed: continuing-tool-run delivery, automatic compaction success, failure, cancellation, and native-input ordering with mid-run steering. They pass with the fix. Additional checks preserve paused and edited heads. The full suite passed with 88 tests on both Pi versions. + +The full `test/tui-evidence.sh` harness also passed on both versions at the tested revision with a clean working tree. It covers manual compaction success and failure, abort recovery, native-input-before-command ordering, repeated reloads, overflow recovery, resources and all-mode FIFO delivery. + +### Live-model proof + +A separate scratch Pi 0.99.2 TUI used provider `openai`, model `gpt-6.1-sol`, and medium thinking. It loaded the fixed extension and a scratch `queue_probe` tool. Normal Pi summarization performed automatic threshold compaction. The scratch compaction settings were `reserveTokens: 266000` and `keepRecentTokens: 200` (agent defaults for this test, not user rules). No production settings changed. + +The prompt was: + +```text +This is an isolated queue-steer regression test. Use queue_probe only. Call inflate first, then call hold in a separate later assistant turn, then call record with token ORIGINAL in a later turn, then finish. Never batch phases in one assistant turn. The filler is disposable and can be summarized very briefly. If a later steering message changes the token, record its token instead. Do not skip phases or ask questions. +``` + +The model called `inflate`, Pi compacted from 26,628 tokens, and the model called `hold`. While that tool waited, the TUI queued: + +```text +Steering update: when the hold tool returns, call queue_probe record with token STEERED instead of ORIGINAL, then finish. +``` + +After the tool was released, the steering left the visible queue. Pi compacted again, and the model called `record` with `token: "STEERED"` before its final response. Independent file and transcript assertions confirmed: + +- the scratch result file contained `{"phase":"record","token":"STEERED"}` +- the transcript contained exactly one steering user message, before the recording tool call +- the event log contained one `agent_start` and one `agent_end` +- the recording tool ran before `agent_end`, after the first compaction completed + +Terminal captures showed the queued steering row beside the waiting tool, then the delivered message and recording tool. This live run covers successful threshold compaction and mid-run steering. Automatic failure and cancellation use deterministic regressions; native-input ordering uses deterministic and real-TUI tests. Asynchronous public-API rejection remains outside the extension's acknowledgement contract. ## Real TUI evidence From afd4db006d72792d5605a1d026e809f3018ff975 Mon Sep 17 00:00:00 2001 From: Thomas Mustier <6326440+tmustier@users.noreply.github.com> Date: Thu, 1 Oct 2026 18:07:44 +0100 Subject: [PATCH 3/4] refactor: simplify compaction delivery and prepare v0.2.1 --- CHANGELOG.md | 14 ++++ README.md | 12 ++- index.ts | 153 +++++++++--------------------------- package-lock.json | 4 +- package.json | 2 +- test/pi-integration.test.ts | 134 +++++++++++++++++-------------- test/queue-state.test.ts | 87 ++------------------ 7 files changed, 142 insertions(+), 264 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff0653a..7d5c0ce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ ## Unreleased +## 0.2.1 - 2026-10-01 + +### Fixed + +- Deliver steering at the next turn after automatic compaction, even when the agent continues making tool calls. +- Release compaction holds on success, failure and cancellation while preserving native post-compaction input ordering. + +### Changed + +- Replace native-input booleans with explicit pending and running states, and share one compaction completion path. +- Inline single-use dispatch helpers and remove unreachable synchronous compaction recovery. +- Test success, failure and cancellation through real Pi sessions and tools instead of fabricated lifecycle events. +- Publish the package on npm as `@tmustier/pi-queue-steer`. + ## 0.2.0 - 2026-08-09 ### Added diff --git a/README.md b/README.md index 39e8cc9..a2d960a 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,13 @@ Move into any row to edit it. The selected row becomes the live Pi editor, with ## Install -Install the latest version from GitHub: +Install the latest published version: + +```bash +pi install npm:@tmustier/pi-queue-steer +``` + +Or install from GitHub: ```bash pi install git:github.com/tmustier/pi-queue-steer @@ -24,7 +30,7 @@ pi install git:github.com/tmustier/pi-queue-steer Pin the current release: ```bash -pi install git:github.com/tmustier/pi-queue-steer@v0.2.0 +pi install git:github.com/tmustier/pi-queue-steer@v0.2.1 ``` Then start a new Pi session or run `/reload`. @@ -83,7 +89,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; queued rows resume when compaction completes, fails or is cancelled - `/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 diff --git a/index.ts b/index.ts index 8c20399..4efb58d 100644 --- a/index.ts +++ b/index.ts @@ -249,15 +249,14 @@ export default function queueSteerExtension(pi: ExtensionAPI) { let settingsManager: SettingsManager | undefined; let blockingActivity: "compact" | "auto-compact" | "reload" | undefined; let compactionFinishTimer: ReturnType | undefined; - let nativeCompactionInputQueued = false; - let nativeCompactionTurnStarted = false; + let nativeCompactionInput: "none" | "pending" | "running" = "none"; const isCompacting = (): boolean => blockingActivity === "compact" || blockingActivity === "auto-compact"; const trackNativeCompactionSubmission = ( text: string, behavior: "submit" | "followUp" = "submit", ): void => { if (isCompacting() && queuesDuringCompaction(text, pi.getCommands(), behavior)) { - nativeCompactionInputQueued = true; + nativeCompactionInput = "pending"; } }; // Pi's own editor submit handler, captured by the submit guard. Replaying text @@ -349,26 +348,21 @@ export default function queueSteerExtension(pi: ExtensionAPI) { ); }; - // Message rows only; command rows never dispatch at active-run boundaries. - // 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 || (nativeCompactionInputQueued && !nativeCompactionTurnStarted) - || queue.laneLength(lane) === 0 || laneIsHeld(lane)) return []; - const isMessage = (item: QueuedMessage) => itemCommand(item) === undefined; - if (queueModes()[lane] === "all") return queue.shiftWhile(lane, isMessage); - const head = queue.peek(lane); - if (!head || !isMessage(head)) return []; - const item = queue.shift(lane); - return item ? [item] : []; - }; - - const deliverBatchToNativeQueue = async ( - ctx: ExtensionContext, - lane: QueueLane, - items: QueuedMessage[], - ): Promise => { - if (items.length === 0) return false; + const dispatchLaneAtBoundary = (ctx: ExtensionContext, lane: QueueLane): boolean => { + activeContext = ctx; + if (paused || blockingActivity || nativeCompactionInput === "pending" || laneIsHeld(lane)) { + renderQueue(ctx); + return false; + } + const mode = queueModes()[lane]; + const headId = queue.peek(lane)?.id; + const items = queue.shiftWhile(lane, (item) => ( + !itemCommand(item) && (mode === "all" || item.id === headId) + )); + if (items.length === 0) { + renderQueue(ctx); + return false; + } let prepared: QueuedMessage[]; try { const commands = pi.getCommands(); @@ -400,16 +394,6 @@ export default function queueSteerExtension(pi: ExtensionAPI) { } }; - const dispatchLaneAtBoundary = async (ctx: ExtensionContext, lane: QueueLane): Promise => { - activeContext = ctx; - const items = takeLaneBatch(lane); - if (items.length === 0) { - renderQueue(ctx); - return false; - } - return deliverBatchToNativeQueue(ctx, lane, items); - }; - // Execute the command row at the lane head. Only called when the agent is idle. const executeCommandRow = (ctx: ExtensionContext, lane: QueueLane): boolean => { const next = queue.peek(lane); @@ -427,11 +411,8 @@ export default function queueSteerExtension(pi: ExtensionAPI) { 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 true; } blockingActivity = "reload"; // Defer so the extension runtime is never torn down from inside this handler. @@ -471,7 +452,7 @@ export default function queueSteerExtension(pi: ExtensionAPI) { const dispatchFromIdle = (ctx: ExtensionContext): boolean => { activeContext = ctx; - if (blockingActivity || nativeCompactionInputQueued) { + if (blockingActivity || nativeCompactionInput !== "none") { renderQueue(ctx); return false; } @@ -489,67 +470,26 @@ export default function queueSteerExtension(pi: ExtensionAPI) { return sendHeadMessage(ctx, lane); }; - const deferCompactionFinish = ( - ctx: ExtensionContext, - activity: "compact" | "auto-compact", - ): void => { - compactionFinishTimer = setTimeout(() => { - compactionFinishTimer = undefined; - if (blockingActivity !== activity) return; - // Pi flushes ordinary TUI submissions after compaction without - // awaiting prompt preflight. Keep command rows behind that native run. - if (nativeCompactionInputQueued) { - renderQueue(activeContext ?? ctx); - return; - } - blockingActivity = undefined; - nativeCompactionInputQueued = false; - nativeCompactionTurnStarted = false; - const current = activeContext ?? ctx; - renderQueue(current); - if (!paused && !editSession && queue.length > 0 && current.isIdle()) dispatchFromIdle(current); - }, 0); - }; - - const finishAutomaticCompaction = (ctx: ExtensionContext): void => { - if (blockingActivity !== "auto-compact") return; - // Automatic compaction can finish between tool turns, long before the - // run settles. Release message delivery now, not at agent_settled. + const finishCompaction = (ctx: ExtensionContext): void => { + if (!isCompacting()) return; blockingActivity = undefined; renderQueue(ctx); if (compactionFinishTimer) clearTimeout(compactionFinishTimer); compactionFinishTimer = setTimeout(() => { compactionFinishTimer = undefined; const current = activeContext ?? ctx; - // Native input may still be in asynchronous preflight. dispatchFromIdle - // keeps command rows behind that input until its run has settled. - if (!paused && !editSession && queue.length > 0 && current.isIdle()) dispatchFromIdle(current); + if (!paused && !editSession && current.isIdle()) dispatchFromIdle(current); }, 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; - } + nativeCompactionInput = "none"; + ctx.compact({ + customInstructions: instructions, + onComplete: () => finishCompaction(ctx), + onError: () => finishCompaction(ctx), + }); }; const deferCommand = (ctx: ExtensionContext, text: string): void => { @@ -864,17 +804,16 @@ export default function queueSteerExtension(pi: ExtensionAPI) { activeContext = ctx; if (blockingActivity || event.reason === "manual") return; blockingActivity = "auto-compact"; - nativeCompactionInputQueued = false; - nativeCompactionTurnStarted = false; + nativeCompactionInput = "none"; renderQueue(ctx); }); - pi.on("session_compact", (_event, ctx) => finishAutomaticCompaction(ctx)); - pi.on("session_compact_failed", (_event, ctx) => finishAutomaticCompaction(ctx)); + pi.on("session_compact", (_event, ctx) => finishCompaction(ctx)); + pi.on("session_compact_failed", (_event, ctx) => finishCompaction(ctx)); pi.on("turn_start", (_event, ctx) => { activeContext = ctx; - if (nativeCompactionInputQueued) nativeCompactionTurnStarted = true; + if (nativeCompactionInput === "pending") nativeCompactionInput = "running"; }); pi.on("turn_end", async (event, ctx) => { @@ -916,25 +855,10 @@ export default function queueSteerExtension(pi: ExtensionAPI) { pi.on("agent_settled", (_event, ctx) => { activeContext = ctx; - if (blockingActivity === "compact" || blockingActivity === "auto-compact") { - const activity = blockingActivity; - if (nativeCompactionInputQueued && !nativeCompactionTurnStarted) { - renderQueue(ctx); - return; - } - // The ordinary post-compaction turn, if any, is now fully settled. - nativeCompactionInputQueued = false; - deferCompactionFinish(ctx, activity); - return; - } - if (nativeCompactionInputQueued && !nativeCompactionTurnStarted) { - renderQueue(ctx); - return; - } - nativeCompactionInputQueued = false; - nativeCompactionTurnStarted = false; + if (nativeCompactionInput === "pending") return; + nativeCompactionInput = "none"; renderQueue(ctx); - if (!paused && !editSession && queue.length > 0 && ctx.isIdle() && !blockingActivity) dispatchFromIdle(ctx); + if (!paused && !editSession && ctx.isIdle()) dispatchFromIdle(ctx); }); pi.on("session_shutdown", (event) => { @@ -969,8 +893,7 @@ export default function queueSteerExtension(pi: ExtensionAPI) { settingsManager = undefined; paused = false; blockingActivity = undefined; - nativeCompactionInputQueued = false; - nativeCompactionTurnStarted = false; + nativeCompactionInput = "none"; tuiSubmit = undefined; queue.clear(); }); diff --git a/package-lock.json b/package-lock.json index 8dd7775..10ae141 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@tmustier/pi-queue-steer", - "version": "0.2.0", + "version": "0.2.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@tmustier/pi-queue-steer", - "version": "0.2.0", + "version": "0.2.1", "license": "MIT", "devDependencies": { "@earendil-works/pi-ai": "*", diff --git a/package.json b/package.json index 2230c1e..f210dfd 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@tmustier/pi-queue-steer", - "version": "0.2.0", + "version": "0.2.1", "description": "Visible steering and follow-up queues with inline editing for Pi.", "type": "module", "license": "MIT", diff --git a/test/pi-integration.test.ts b/test/pi-integration.test.ts index d808d64..574dc4a 100644 --- a/test/pi-integration.test.ts +++ b/test/pi-integration.test.ts @@ -203,18 +203,14 @@ function gatedResponse( step: () => Promise>; release(): void; } { - let releaseGate: (() => void) | undefined; - const gate = new Promise((resolve) => { - releaseGate = resolve; - }); + let release!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); return { step: async () => { await gate; return fauxAssistantMessage(content, options); }, - release() { - releaseGate?.(); - }, + release, }; } @@ -384,62 +380,78 @@ test("Pi 0.87 finishes settled handlers before starting queue-steer's requested } }); -test("steering reaches the continuing tool run after real threshold compaction, before agent_end", async () => { - let enterResumed: (() => void) | undefined; - const resumedEntered = new Promise((resolve) => { enterResumed = resolve; }); - const resumed = gatedResponse( - [fauxToolCall("bash", { command: "printf resumed-tool", timeout: 5 })], - { stopReason: "toolUse" }, - ); - const summaryExtension: ExtensionFactory = (pi) => { - pi.on("session_before_compact", (event) => ({ - compaction: { - summary: "short integration summary", - firstKeptEntryId: event.preparation.firstKeptEntryId, - tokensBefore: event.preparation.tokensBefore, +for (const outcome of ["success", "failure", "cancelled"] as const) { + test(`steering reaches the continuing tool run after threshold compaction ${outcome}`, async () => { + let enterResumed!: () => void; + const resumedEntered = new Promise((resolve) => { enterResumed = resolve; }); + const resumed = gatedResponse( + [fauxToolCall("bash", { command: "printf resumed-tool", timeout: 5 })], + { stopReason: "toolUse" }, + ); + const summaryExtension: ExtensionFactory = (pi) => { + let first = true; + pi.on("session_before_compact", (event) => { + if (outcome === "cancelled") return { cancel: true }; + if (outcome === "failure" && first) { + first = false; + return; + } + return { + compaction: { + summary: "short integration summary", + firstKeptEntryId: event.preparation.firstKeptEntryId, + tokensBefore: event.preparation.tokensBefore, + }, + }; + }); + }; + const harness = await createIntegrationHarness({ + contextWindow: 100_000, + // agent-default (not a user rule): 2026-10-01, small test threshold. + reserveTokens: 99_000, + tools: ["bash"], + extraExtensions: [summaryExtension], + }); + const trace: string[] = []; + harness.session.subscribe((event) => { + trace.push(event.type); + }); + harness.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("bash", { command: "printf initial-tool", timeout: 5 })], { stopReason: "toolUse" }), + ...(outcome === "failure" ? [() => { throw new Error("threshold summary failed"); }] : []), + () => { + enterResumed(); + return resumed.step(); }, - })); - }; - const harness = await createIntegrationHarness({ - contextWindow: 100_000, - reserveTokens: 99_000, - tools: ["bash"], - extraExtensions: [summaryExtension], - }); - const trace: string[] = []; - harness.session.subscribe((event) => { - trace.push(event.type); + () => { + assert.equal(userTexts(harness.session).at(-1), "steer after threshold"); + assert.equal(trace.includes("agent_end"), false); + return fauxAssistantMessage("steering handled inside original run"); + }, + ]); + const compactionEnded = nextCompactionEnd(harness.session); + const prompt = harness.session.prompt("initial context ".repeat(1_000)); + try { + await within(resumedEntered, () => trace.join(", ")); + const compaction = await within(compactionEnded, () => trace.join(", ")); + assert.equal(compaction.reason, "threshold"); + assert.equal(compaction.aborted, outcome === "cancelled"); + if (outcome === "failure") assert.match(compaction.errorMessage ?? "", /threshold summary failed/); + assert.equal(!!compaction.result, outcome === "success"); + assert.equal(trace.includes("agent_settled"), false); + await harness.session.prompt("steer after threshold", { streamingBehavior: "steer" }); + resumed.release(); + await within(prompt, () => trace.join(", ")); + assert.equal(userTexts(harness.session).filter((text) => text === "steer after threshold").length, 1); + assert.equal(harness.session.getLastAssistantText(), "steering handled inside original run"); + assert.equal(trace.filter((type) => type === "agent_start").length, 1); + } finally { + resumed.release(); + await prompt.catch(() => undefined); + await harness.cleanup(); + } }); - harness.faux.setResponses([ - fauxAssistantMessage([fauxToolCall("bash", { command: "printf initial-tool", timeout: 5 })], { stopReason: "toolUse" }), - async () => { - enterResumed?.(); - return resumed.step(); - }, - (context) => { - const steering = context.messages.filter((message) => message.role === "user").at(-1); - assert.ok(JSON.stringify(steering).includes("steer after threshold")); - assert.equal(trace.includes("agent_end"), false); - return fauxAssistantMessage("steering handled inside original run"); - }, - ]); - const prompt = harness.session.prompt("initial context ".repeat(1_000)); - try { - await within(resumedEntered, () => trace.join(", ")); - assert.ok(trace.includes("compaction_end")); - assert.equal(trace.includes("agent_settled"), false); - await harness.session.prompt("steer after threshold", { streamingBehavior: "steer" }); - resumed.release(); - await within(prompt, () => trace.join(", ")); - assert.equal(userTexts(harness.session).filter((text) => text === "steer after threshold").length, 1); - assert.equal(harness.session.getLastAssistantText(), "steering handled inside original run"); - assert.equal(trace.filter((type) => type === "agent_start").length, 1); - } finally { - resumed.release(); - await prompt.catch(() => undefined); - await harness.cleanup(); - } -}); +} test("real public prompt path triggers overflow compaction and preserves a queued follow-up", async () => { const summaryExtension: ExtensionFactory = (pi) => { diff --git a/test/queue-state.test.ts b/test/queue-state.test.ts index 7eb3aa4..e030273 100644 --- a/test/queue-state.test.ts +++ b/test/queue-state.test.ts @@ -203,7 +203,6 @@ function createHarness(options: { commands?: SlashCommandInfo[]; mode?: "tui" | "rpc" | "json" | "print"; sendFailureAt?: number; - compactStartError?: Error; autocompleteVisible?: boolean; } = {}) { type Handler = (event: any, context: any) => any; @@ -268,7 +267,6 @@ function createHarness(options: { aborted = true; }, compact(compactOptions: CompactOptions = {}) { - if (options.compactStartError) throw options.compactStartError; compactCalls.push(compactOptions); }, }; @@ -949,22 +947,6 @@ test("owns busy manual compaction so its abort does not pause queued rows", asyn assert.equal(harness.sent[0]?.content, "continue after compact"); }); -test("restores and pauses a command row when compaction cannot start", async () => { - 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"); @@ -1008,6 +990,7 @@ test("holds a follow-up at a length stop so Pi can decide whether to auto-compac assert.equal(harness.sent.length, 0); await harness.emit("session_before_compact", { reason: "overflow" }); + await harness.emit("session_compact", { reason: "overflow" }); harness.setIdle(true); await harness.emit("agent_settled"); await waitFor(() => harness.sent.length === 1); @@ -1026,68 +1009,7 @@ test("releases a length-stop hold at settle when Pi does not compact", async () assert.equal(harness.sent[0]?.content, "after full-length output"); }); -for (const [outcome, aborted] of [ - ["session_compact", false], - ["session_compact_failed", false], - ["session_compact_failed", true], -] as const) { - test(`releases steering at the next turn after automatic compaction: ${outcome}, aborted=${aborted}`, async () => { - const harness = createHarness(); - await harness.emit("session_start"); - await harness.emit("session_before_compact", { reason: "threshold" }); - await enqueue(harness, "steer", "steer after compact"); - await enqueue(harness, "followUp", "wait until run ends"); - await harness.emit("turn_end", { message: { role: "assistant", stopReason: "toolUse" } }); - assert.equal(harness.sent.length, 0); - - await harness.emit(outcome, { reason: "threshold", aborted }); - // No agent_settled event: the original run is still making tool calls. - await harness.emit("turn_start"); - await harness.emit("turn_end", { message: { role: "assistant", stopReason: "toolUse" } }); - assert.deepEqual(harness.sent, [{ content: "steer after compact", options: { deliverAs: "steer" } }]); - assert.match(renderWidget(harness), /wait until run ends/); - }); -} - -test("automatic compaction completion preserves native-input ordering but allows steering during that run", async () => { - const harness = createHarness(); - await harness.emit("session_start"); - harness.setIdle(true); - await harness.emit("session_before_compact", { reason: "threshold" }); - harness.editor.onSubmit?.("ordinary native input"); - harness.editor.onSubmit?.("/reload"); - await enqueue(harness, "steer", "steer the native run"); - await harness.emit("session_compact", { reason: "threshold" }); - await new Promise((resolve) => setTimeout(resolve, 5)); - assert.deepEqual(harness.submitted, ["ordinary native input"]); - assert.equal(harness.sent.length, 0); - - harness.setIdle(false); - await harness.emit("turn_start"); - await harness.emit("turn_end", { message: { role: "assistant", stopReason: "toolUse" } }); - assert.equal(harness.sent[0]?.content, "steer the native run"); - assert.deepEqual(harness.submitted, ["ordinary native input"]); - harness.setIdle(true); - await harness.emit("agent_settled"); - await waitFor(() => harness.submitted.length === 2); - assert.deepEqual(harness.submitted, ["ordinary native input", "/reload"]); -}); - -test("automatic compaction completion does not release a paused or edited steering head", async () => { - for (const hold of ["pause", "edit"] as const) { - const harness = createHarness(); - await harness.emit("session_start"); - await harness.emit("session_before_compact", { reason: "threshold" }); - await enqueue(harness, "steer", "held head"); - harness.editor.handleInput(hold === "pause" ? "escape" : "alt-up"); - await harness.emit("session_compact", { reason: "threshold" }); - await harness.emit("turn_end", { message: { role: "assistant", stopReason: "toolUse" } }); - assert.equal(harness.sent.length, 0); - assert.match(renderWidget(harness), hold === "pause" ? /paused/ : /held while editing/); - } -}); - -test("holds reload through automatic compaction until the agent settles", async () => { +test("holds reload until automatic compaction completes", async () => { const harness = createHarness(); await harness.emit("session_start"); harness.setIdle(true); @@ -1095,7 +1017,7 @@ test("holds reload through automatic compaction until the agent settles", async harness.editor.onSubmit?.("/reload"); assert.equal(harness.submitted.length, 0); - await harness.emit("agent_settled"); + await harness.emit("session_compact", { reason: "overflow" }); await waitFor(() => harness.submitted.length === 1); assert.deepEqual(harness.submitted, ["/reload"]); }); @@ -1110,7 +1032,7 @@ test("captures a compaction command while slash autocomplete is visible", async assert.match(renderWidget(harness), /\/reload/); assert.deepEqual(harness.submitted, []); - await harness.emit("agent_settled"); + await harness.emit("session_compact", { reason: "threshold" }); await waitFor(() => harness.submitted.length === 1); assert.deepEqual(harness.submitted, ["/reload"]); }); @@ -1123,6 +1045,7 @@ test("holds automatic-compaction commands through ordinary native input", async harness.editor.onSubmit?.("ordinary after automatic compaction"); harness.editor.onSubmit?.("/reload"); + await harness.emit("session_compact", { reason: "threshold" }); await harness.emit("agent_settled"); await new Promise((resolve) => setTimeout(resolve, 0)); assert.deepEqual(harness.submitted, ["ordinary after automatic compaction"]); From 7fa59302f00776103c152aff22ba51cd4559e8b2 Mon Sep 17 00:00:00 2001 From: Thomas Mustier <6326440+tmustier@users.noreply.github.com> Date: Thu, 1 Oct 2026 18:11:28 +0100 Subject: [PATCH 4/4] docs: shorten and update v0.2.1 validation record --- docs/validation.md | 27 ++++++++++----------------- 1 file changed, 10 insertions(+), 17 deletions(-) diff --git a/docs/validation.md b/docs/validation.md index 7b9737e..4961129 100644 --- a/docs/validation.md +++ b/docs/validation.md @@ -25,42 +25,35 @@ 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, settled-handler launch ordering, repeated reload restoration, and compaction/native-input ordering. -Latest result for the automatic-compaction steering fix: 88 tests passed on both Pi 0.87.0 (the lockfile) and Pi 0.99.2 (the current installed release). +Release 0.2.1: 84 tests passed on both Pi 0.87.0 (the lockfile) and Pi 0.99.2 (the current installed release). ## Automatic-compaction steering regression -Validated on 1 October 2026 against code revision `ae8b7b3bd88fec4a6a4453e0a8c25119b0c9c353`. The later validation-record commit changes documentation only. +Validated on 1 October 2026 at `afd4db006d72792d5605a1d026e809f3018ff975` for release 0.2.1. -The regression test makes Pi compact between tool turns, then queues steering while the next assistant response is gated. It checks that the following provider request sees the steering before `agent_end`, with one agent run and exactly one delivered steering message. The test uses actual `AgentSession` scheduling, real bash tools and a deterministic provider. +The 3 regressions use real `AgentSession` scheduling and bash tools to exercise threshold compaction success, failure and cancellation. They queue steering while the resumed assistant response waits, then check delivery before `agent_end`, in the original run, exactly once. All 3 fail against the original `91e3a5f` implementation. -Against the original `91e3a5f` implementation, all 5 targeted regressions failed: continuing-tool-run delivery, automatic compaction success, failure, cancellation, and native-input ordering with mid-run steering. They pass with the fix. Additional checks preserve paused and edited heads. The full suite passed with 88 tests on both Pi versions. - -The full `test/tui-evidence.sh` harness also passed on both versions at the tested revision with a clean working tree. It covers manual compaction success and failure, abort recovery, native-input-before-command ordering, repeated reloads, overflow recovery, resources and all-mode FIFO delivery. +The release review replaced fabricated completion events with these integration tests. It also removed an impossible synchronous compaction-start failure test. The full suite has 84 passing tests on Pi 0.87.0 and 0.99.2. The full TUI harness passed on both versions at the tested revision with a clean working tree. ### Live-model proof -A separate scratch Pi 0.99.2 TUI used provider `openai`, model `gpt-6.1-sol`, and medium thinking. It loaded the fixed extension and a scratch `queue_probe` tool. Normal Pi summarization performed automatic threshold compaction. The scratch compaction settings were `reserveTokens: 266000` and `keepRecentTokens: 200` (agent defaults for this test, not user rules). No production settings changed. +A scratch Pi 0.99.2 TUI used `openai/gpt-6.1-sol` at medium thinking, the release extension and a `queue_probe` tool. Normal Pi summarization compacted between tool turns. Scratch settings were `reserveTokens: 266000` and `keepRecentTokens: 200` (agent defaults for this test, not user rules). Production settings were unchanged. -The prompt was: +Prompt: ```text This is an isolated queue-steer regression test. Use queue_probe only. Call inflate first, then call hold in a separate later assistant turn, then call record with token ORIGINAL in a later turn, then finish. Never batch phases in one assistant turn. The filler is disposable and can be summarized very briefly. If a later steering message changes the token, record its token instead. Do not skip phases or ask questions. ``` -The model called `inflate`, Pi compacted from 26,628 tokens, and the model called `hold`. While that tool waited, the TUI queued: +The model called `inflate`, Pi compacted, and the model called `hold`. While that tool waited, the TUI queued: ```text Steering update: when the hold tool returns, call queue_probe record with token STEERED instead of ORIGINAL, then finish. ``` -After the tool was released, the steering left the visible queue. Pi compacted again, and the model called `record` with `token: "STEERED"` before its final response. Independent file and transcript assertions confirmed: - -- the scratch result file contained `{"phase":"record","token":"STEERED"}` -- the transcript contained exactly one steering user message, before the recording tool call -- the event log contained one `agent_start` and one `agent_end` -- the recording tool ran before `agent_end`, after the first compaction completed +After release, the model called `record` with `token: "STEERED"` before its final response. Independent file, transcript and event-log checks confirmed one steering user message, one agent run and recording before `agent_end`. Terminal captures showed the waiting queue and the successful tool result. -Terminal captures showed the queued steering row beside the waiting tool, then the delivered message and recording tool. This live run covers successful threshold compaction and mid-run steering. Automatic failure and cancellation use deterministic regressions; native-input ordering uses deterministic and real-TUI tests. Asynchronous public-API rejection remains outside the extension's acknowledgement contract. +Live-model coverage is successful threshold compaction and mid-run steering. Failure and cancellation use real-session deterministic tests. Native-input ordering uses the real TUI harness. Later asynchronous send rejection remains outside Pi's public acknowledgement contract. ## Real TUI evidence @@ -74,7 +67,7 @@ Run: 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 full harness passed against Pi 0.87.0 for this compatibility update. The latest retained clean release-evidence run reported: +The full harness passed against Pi 0.87.0 and 0.99.2 for release 0.2.1. The historical 0.2.0 release-evidence run reported: ```text pi: 0.84.1