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 f876d6e..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`. @@ -62,6 +68,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 @@ -82,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/docs/validation.md b/docs/validation.md index 5b638e8..4961129 100644 --- a/docs/validation.md +++ b/docs/validation.md @@ -25,7 +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 with Pi 0.87.0: 82 tests passed. +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 at `afd4db006d72792d5605a1d026e809f3018ff975` for release 0.2.1. + +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. + +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 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. + +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, 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 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. + +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 @@ -39,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 diff --git a/index.ts b/index.ts index d855803..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,25 +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 || 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(); @@ -399,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); @@ -426,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. @@ -470,7 +452,7 @@ export default function queueSteerExtension(pi: ExtensionAPI) { const dispatchFromIdle = (ctx: ExtensionContext): boolean => { activeContext = ctx; - if (blockingActivity) { + if (blockingActivity || nativeCompactionInput !== "none") { renderQueue(ctx); return false; } @@ -488,51 +470,26 @@ export default function queueSteerExtension(pi: ExtensionAPI) { return sendHeadMessage(ctx, lane); }; - const deferCompactionFinish = ( - ctx: ExtensionContext, - activity: "compact" | "auto-compact", - ): void => { + const finishCompaction = (ctx: ExtensionContext): void => { + if (!isCompacting()) return; + blockingActivity = undefined; + renderQueue(ctx); + if (compactionFinishTimer) clearTimeout(compactionFinishTimer); 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); + 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 => { @@ -847,14 +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) => finishCompaction(ctx)); + pi.on("session_compact_failed", (_event, ctx) => finishCompaction(ctx)); + pi.on("turn_start", (_event, ctx) => { activeContext = ctx; - if (isCompacting() && nativeCompactionInputQueued) nativeCompactionTurnStarted = true; + if (nativeCompactionInput === "pending") nativeCompactionInput = "running"; }); pi.on("turn_end", async (event, ctx) => { @@ -896,19 +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 (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) => { @@ -943,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 7f6b5d1..574dc4a 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,24 +197,20 @@ async function createIntegrationHarness(options: { } function gatedResponse( - content: string, + content: Parameters[0], options?: Parameters[1], ): { 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, }; } @@ -380,6 +380,79 @@ test("Pi 0.87 finishes settled handlers before starting queue-steer's requested } }); +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(); + }, + () => { + 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(); + } + }); +} + 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..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,7 +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"); }); -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); @@ -1034,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"]); }); @@ -1049,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"]); }); @@ -1062,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"]);