diff --git a/docs/LOCAL_SHELL_SESSION.md b/docs/LOCAL_SHELL_SESSION.md index 0d5639d..60b071a 100644 --- a/docs/LOCAL_SHELL_SESSION.md +++ b/docs/LOCAL_SHELL_SESSION.md @@ -58,12 +58,73 @@ tools. Online account-agent and ATS chats remain separate surfaces. - `!` submissions made while a model turn is busy retain their shell type and wait until that turn completes. All local tools share a FIFO execution slot. Ctrl+C cancels the active command/turn and discards its queued follow-ups; - typing ahead retains the newer composer draft. + typing ahead retains the newer composer draft. See [Queued input](#queued-input). - Shell commands and results are excluded from chat history and hosted prompts. `/shell-result` explicitly shares up to 8 KiB of the latest user result as untrusted data. Reset clears that result. Ordinary chat history still honors `AETHER_NO_HISTORY=1`. +## Queued input + +Anything submitted while a turn, shell command or slash command is running +waits in one strictly ordered queue. Each entry gets a session-stable id +(`q1`, `q2`, … never reused) and a type: `chat`, `user shell`, `shell reset`, +`shell profile` or `shell-share`. The input that is running also has an id and is shown +separately; it cannot be edited (Ctrl+C cancels it). + +| Command | Effect | +|---|---| +| `/queue` or `/queue list` | Running entry, pending entries in order, and the bound in use | +| `/queue ` | Queue a task (runs at once when idle) | +| `/queue edit ` | Replace a pending chat or user shell entry in place | +| `/queue remove ` | Discard one pending entry | +| `/queue clear` | Discard every pending entry | +| `/queue resume` | Run entries that were kept after a failure | + +- Queue commands are local bookkeeping. They work mid-stream and while a model + switch is pending, never call a model, never start a process, and never + enter chat history (an edit can carry shell text). A removed or cleared + entry is never run. +- An edit is classified exactly like typed input and must keep the entry's + type: a chat entry needs chat text (`\!` for a leading `!`; slash commands + are not queued), a user shell entry needs `!`. Shell-reset, + shell-profile and shell-share actions have no editable text: remove and + queue again. A + rejected edit leaves the entry unchanged. +- Management keywords match exactly, so `/queue clear the cache` queues a + task. Edit and remove act only on `q` ids. +- Bound: 32 entries and 64 KiB of queued text. A rejected entry is not queued + and its draft stays in the composer. +- Slash commands typed mid-turn other than `/steer`, `/btw` and `/queue` are + not queued; the console says so and ↑ recalls them. + +Shell-share binding is explicit. A queued `/shell-result send` binds, when it +is queued, to the preview you reviewed (its command id). At execution it sends +only that preview, and it refuses and sends nothing if the preview was cancelled +or replaced by a different command's. It is not queued at all if nothing is +staged. Queued preview and edit actions (`lines`, `drop`, `replace`, `mask`, +`redact`) act on the staged preview, or stage the **latest shell result at +execution** if none is staged. A queued `cancel` discards whatever is staged +when it runs. + +What happens to pending entries: + +| Event | Pending entries | +|---|---| +| Turn or command completes | Next entry runs | +| Chat turn fails (including hosted 401) | Kept and listed as **paused**; nothing runs until `/queue resume`. A new submission runs alone, and later type-ahead is marked paused | +| Ctrl+C / turn or slash command cancelled | Discarded and listed | +| Local shell state lost, shell action failed, `/shell-reset`, or a shell profile change | Discarded and listed; queued commands never run against a different shell | +| `/auth new`, session exit | Discarded and listed | + +The console has no reconnect path that replays work: a hosted turn that drops +is a failed turn (paused above), never a silent resume. + +Line mode (non-TTY stdin, pipes, CI) reads the next line only after the +previous one finishes, so it has no pending queue. `/queue ` runs the +task as the next line. The management commands print that there is nothing +to manage. + ## Interactive Linux terminal Use `/terminal ` in the local coding console when a program needs diff --git a/docs/generated/commands.md b/docs/generated/commands.md index a1f2a74..796b38a 100644 --- a/docs/generated/commands.md +++ b/docs/generated/commands.md @@ -1,5 +1,5 @@ - + # Generated command reference This reference is generated from the validated, versioned command manifest. Availability is evaluated at runtime; a listed command may still require authentication, a hosted capability, or local tooling. @@ -525,9 +525,9 @@ show the current local terminal ID and state Permission: `unknown` · Availability: `runtime-dependent` · Telemetry: `slash.terminal-status` -#### `/queue ` +#### `/queue [|list|edit |remove |clear|resume]` -queue a task \(runs when current finishes\) +queue a task, or inspect/edit/remove/clear pending entries Permission: `unknown` · Availability: `runtime-dependent` · Telemetry: `slash.queue` diff --git a/src/commands/chat.ts b/src/commands/chat.ts index 4574fe1..056323f 100644 --- a/src/commands/chat.ts +++ b/src/commands/chat.ts @@ -57,6 +57,7 @@ import { SteerChannel, formatSteerAck } from "../core/steer_channel.js"; import type { RunOptions, ToolResult } from "../core/tool_executor.js"; import { ToolExecutor } from "../core/tool_executor.js"; import { ConsoleShell, classifyConsoleInput, type ConsoleInput } from "./console_input.js"; +import { ConsoleQueue, describeEntry, entryKind, parseQueueCommand, renderDisposition, type QueueCommand, type QueueEntry, type QueueableInput } from "./console_queue.js"; import { HostRenderer } from "../ui/host_render.js"; import type { TaskCommand } from "../core/brain.js"; import { getRegistry } from "../core/context_registry.js"; @@ -1131,7 +1132,7 @@ async function currentConsoleModel(ctx: AppContext): Promise { function switchDisposition(target: ModelTarget, brief: string, queueCount: number, draftSaved = false): string { return `Model switch pending: ${target.label} (${target.destination}).\n` + - (queueCount ? `Queued entries: ${queueCount}; they stay queued and will not replay across this switch. Cancel and finish them first.\n` : "Queued entries: none.\n") + + (queueCount ? `Queued entries: ${queueCount}; they stay queued and will not replay across this switch. Review them with /queue; remove them with /queue remove or /queue clear.\n` : "Queued entries: none.\n") + `Unsent draft: ${draftSaved ? "saved; restored after continue, fresh, or cancel; never sent automatically" : "none"}.\n` + (target.contextWindow !== null && target.contextWindow < 2_048 ? "This target's context window is too small for continuation; choose fresh or cancel.\n" : "") + `Exact continuation brief for review:\n${brief}\n` + @@ -1186,11 +1187,6 @@ export function ctrlCDecision(s: { return s.armed ? "exit" : "arm-exit"; } -/** Truncate a queued-prompt preview to 55 chars for the "⏳ Queued" echo lines. */ -function previewLine(s: string): string { - return s.length > 55 ? s.slice(0, 55) + "…" : s; -} - /** Stable machine terminal record. The prompt itself is deliberately omitted: * automation needs correlation/outcome/retry facts, not a second copy of user * content in every captured JSON log. */ @@ -1354,7 +1350,8 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P let pasting = false; let pasteAcc = ""; let carry = ""; // partial escape sequence held across chunk boundaries - const queue: ConsoleInput[] = []; + const queue = new ConsoleQueue(); + let runningSlash: string | null = null; // names a running slash command for /queue // /steer during a turn goes to that turn's brain when it can acknowledge // it; everything else waits for the next turn and is reported as such. const steerChannel = new SteerChannel((ack) => { @@ -1394,8 +1391,25 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P process.stdin.removeListener("data", onData); process.stdout.removeListener("resize", onResize); }; + /** Every pending entry is dropped and listed; none of them will resume. */ + const discardQueue = (reason: string): void => { + const removed = queue.clear(); + process.stdout.write(renderDisposition( + `\nQueue discarded (${reason}): ${removed.length} pending ${removed.length === 1 ? "entry" : "entries"} removed; none ran and none will resume.`, + removed, + )); + }; + /** After a failed entry the rest stay listed but wait for /queue resume. */ + const holdQueue = (failed: QueueEntry): void => { + if (!queue.hold(`${failed.id} failed`)) return; + process.stdout.write(renderDisposition( + `\nQueue paused (${failed.id} failed): ${queue.length} pending ${queue.length === 1 ? "entry" : "entries"} kept; none ran.`, + queue.pending, + "/queue resume runs them in order; /queue edit|remove or /queue clear changes them.", + )); + }; const finish = (code: number): void => { - turnAbort?.abort(); slashAbort?.abort(); queue.length = 0; + turnAbort?.abort(); slashAbort?.abort(); discardQueue("session ended"); cleanup(); process.stdout.write("\n"); resolve(code); @@ -1410,7 +1424,7 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P if (input.kind === "profile") { if (consoleShell.profileCommand(input)) { skillOpts = { ...skillOpts, exec: consoleShell.exec }; - queue.length = 0; + discardQueue("shell profile changed"); } return "completed"; } @@ -1418,7 +1432,11 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P turnAbort = new AbortController(); try { const result = await consoleShell.run(input, turnAbort.signal); - if (result !== "completed" || input.kind === "reset-shell") queue.length = 0; + // Queued shell commands assumed the old shell state: never run them + // against a fresh one, and never after the user said stop. + if (result === "aborted") discardQueue("cancelled"); + else if (result === "failed") discardQueue(consoleShell.session.state === "lost" ? "shell state lost" : "shell action failed"); + else if (input.kind === "reset-shell") discardQueue("shell reset"); return result; } finally { turnAbort = null; } } @@ -1512,9 +1530,9 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P continuation.recordTurn(text, toolResults, afterChanges.filter(change => !beforeChanges.includes(change)), outcome.state, !sharedShellResult && !authContinuation); if (ctx.flags.json) process.stdout.write(turnOutcomeJson(outcome) + "\n"); if (outcome.state === "cancelled") { - queue.length = 0; if (!ctx.flags.json) process.stdout.write("\n" + theme.dim("✗ turn aborted") + "\n"); steerEnd = "cancelled"; + discardQueue("turn cancelled"); return "aborted"; } return "completed"; @@ -1522,12 +1540,12 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P const afterChanges = observedWorkspaceChanges(ctx.flags.cwd); continuation.recordTurn(text, toolResults, afterChanges.filter(change => !beforeChanges.includes(change)), "failed", !sharedShellResult && !authContinuation); if (isAbortError(err)) { - // User said stop: drop the queued follow-ups too. - queue.length = 0; + // User said stop: drop the queued follow-ups too, and say which. const outcome = turnOutcomeForError(err); if (ctx.flags.json && outcome) process.stdout.write(turnOutcomeJson(outcome) + "\n"); else process.stdout.write("\n" + theme.dim("✗ turn aborted") + "\n"); steerEnd = "cancelled"; + discardQueue("turn cancelled"); return "aborted"; } // ChatTurnError means the Renderer already painted "✗ " for the @@ -1574,22 +1592,91 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P } }; - const runAndDrain = async (input: ConsoleInput, authContinuation = false): Promise => { + const runEntry = async (entry: QueueEntry, authContinuation = false): Promise<"completed" | "aborted" | "failed"> => { + queue.setRunning(entry); + try { return await runQueuedTurn(entry.input, authContinuation); } + finally { queue.setRunning(null); } + }; + + /** Run one entry, then drain pending entries in strict order. A failure + * pauses the rest; a cancellation discards them (see runQueuedTurn). */ + const runAndDrain = async (first: QueueEntry, authContinuation = false): Promise => { try { getRegistry().startAgentTimer(); - let result = await runQueuedTurn(input, authContinuation); - while (result === "completed" && queue.length > 0) { - const next = queue.shift()!; - const preview = next.kind === "chat" ? next.text : next.kind === "shell" ? "!" + next.command : next.kind === "share" ? "/shell-result" : next.kind === "profile" ? "/shell-profile" : "/shell-reset"; - process.stdout.write(`\n→ Queued: "${previewLine(preview)}"\n`); - result = await runQueuedTurn(next); + let entry: QueueEntry | undefined = first; + let result = await runEntry(first, authContinuation); + while (result === "completed") { + entry = queue.shift(); + if (!entry) break; + process.stdout.write(`\n→ Running queued ${entry.id} (${entryKind(entry.input)}): ${describeEntry(entry.input)}\n`); + result = await runEntry(entry); } + if (result === "failed" && entry) holdQueue(entry); } finally { busy = false; getRegistry().startUserTimer(); } }; + /** Queue one input behind the active operation. Returns false (and says + * why) when it was not queued, so the caller keeps the draft. */ + const enqueueInput = (input: QueueableInput): boolean => { + let toQueue = input; + if (input.kind === "share" && input.action === "send") { + const bound = consoleShell.stagedCommandId(); + if (!bound) { + process.stdout.write("\nNot queued: a queued /shell-result send must bind to a preview you have reviewed, and none is staged. Draft kept.\n"); + return false; + } + toQueue = { ...input, boundCommandId: bound }; + } + const result = queue.enqueue(toQueue); + if (!result.ok) { process.stdout.write(`\n${result.message} Draft kept.\n`); return false; } + const paused = queue.held ? "; queue PAUSED, /queue resume runs it" : ""; + process.stdout.write(`\n⏳ Queued ${result.entry.id} (${entryKind(toQueue)}, ${queue.length} pending${paused}): ${describeEntry(toQueue)}\n`); + return true; + }; + + /** /queue management is local bookkeeping: it never calls a model or + * starts a process (except /queue resume, which hands entries back to the + * ordinary drain). */ + const handleQueueCommand = async (command: QueueCommand): Promise => { + const write = (text: string): void => { process.stdout.write(text); }; + switch (command.op) { + case "usage": write(command.message + "\n"); return; + case "list": write(queue.render(runningSlash)); return; + case "clear": { + const removed = queue.clear(); + write(removed.length + ? renderDisposition(`Cleared ${removed.length} pending ${removed.length === 1 ? "entry" : "entries"}; none ran:`, removed) + : "Queue is already empty.\n"); + return; + } + case "remove": { + const result = queue.remove(command.id); + write(result.ok ? `Removed ${result.entry.id} (${entryKind(result.entry.input)}); it will not run.\n` : result.message + "\n"); + return; + } + case "edit": { + const result = queue.edit(command.id, command.text); + write(result.ok ? `Edited ${result.entry.id} (${entryKind(result.entry.input)}, position kept): ${describeEntry(result.entry.input)}\n` : result.message + "\n"); + return; + } + case "resume": { + if (continuation.pending) { write("Resolve the pending model switch with /switch first; the queue stays as listed.\n"); return; } + if (!queue.resume()) { write(queue.length ? "Queue is not paused.\n" : "Queue is empty; nothing to resume.\n"); return; } + write(`Queue resumed: ${queue.length} pending ${queue.length === 1 ? "entry runs" : "entries run"} in order${busy ? " after the current operation" : ""}.\n`); + if (busy) return; + const next = queue.shift(); + if (!next) return; + busy = true; + await runAndDrain(next); + renderHudLine(); + return; + } + } + }; + const onCtrlC = (): void => { const now = Date.now(); const armed = now - ctrlCArmedAt <= CTRL_C_WINDOW_MS && ctrlCArmedAt > 0; @@ -1628,6 +1715,17 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P const onSubmit = async (): Promise => { const raw = buf.value; + // Queue management works mid-turn and during a pending switch (whose + // disposition asks the user to clear entries). It stays out of history: + // an edit can carry local shell text. + const queueCommand = parseQueueCommand(raw); + if (queueCommand) { + buf.clear(); + process.stdout.write("\n"); + await handleQueueCommand(queueCommand); + repaint(); + return; + } if (continuation.pending && !raw.trim().startsWith("/switch")) { process.stdout.write("\nChoose /switch continue, fresh, cancel, brief, or edit first. Draft preserved.\n"); repaint(); @@ -1650,16 +1748,23 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P }; let t = input.kind === "chat" ? input.text : ""; if (t && heldDraft === t && !t.startsWith("/auth")) heldDraft = null; + if (input.kind === "error") { + buf.clear(); // a usage error is reported now, never queued + process.stdout.write("\n" + input.message + "\n"); + repaint(); + return; + } if (input.kind !== "chat" && input.kind !== "empty") { - buf.clear(); // shell commands never enter chat history or prompt context + // Shell commands never enter chat history or prompt context. A + // rejected enqueue keeps the draft in the composer. if (busy) { - queue.push(input); - process.stdout.write(`\n⏳ Local shell queued (${queue.length}).\n`); + if (enqueueInput(input)) buf.clear(); return; } + buf.clear(); process.stdout.write("\n"); busy = true; - await runAndDrain(input); + await runAndDrain(queue.allocate(input)); renderHudLine(); repaint(); return; } @@ -1684,11 +1789,15 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P return; } if (t.startsWith("/queue ")) t = t.slice(7).trim(); + if (!t || t.startsWith("/")) { + remember(buf.value); + buf.commit(buf.value); + if (t) process.stdout.write("\nSlash commands are not queued; run it after the current operation (↑ recalls it).\n"); + return; + } + if (!enqueueInput({ kind: "chat", text: t })) return; remember(buf.value); buf.commit(buf.value); - if (!t || t.startsWith("/")) return; // other slashes wait for the turn - queue.push({ kind: "chat", text: t }); - process.stdout.write(`\n⏳ Queued (${queue.length}): "${previewLine(t)}"\n`); return; } @@ -1721,7 +1830,7 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P } } else if (command === "continue" || command === "fresh") { if (queue.length) { - process.stdout.write(`Switch paused: ${queue.length} queued entries remain. Cancel and finish them first.\n`); + process.stdout.write(`Switch paused: ${queue.length} queued entries remain. Review them with /queue; remove them with /queue remove or /queue clear.\n`); } else if (process.env["AETHER_BACKEND"] && (await resolveBackend(ctx)) !== pending.target.destination) { process.stdout.write("Switch blocked: AETHER_BACKEND pins a different destination.\n"); } else { @@ -1779,6 +1888,7 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P busy = true; if (t.startsWith("/")) { slashAbort = new AbortController(); + runningSlash = `slash command ${sanitizeServerText(t.split(/\s/, 1)[0] ?? "")}`; if (t === "/auth" || t.startsWith("/auth ")) { const sub = t.slice(5).trim().toLowerCase() || "status"; try { @@ -1788,11 +1898,11 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P const pending = authRepair.takeContinuation(); if (pending) { process.stdout.write(`Continuing saved task ${pending.turnId} after explicit request.\n`); - await runAndDrain({ kind: "chat", text: pending.instruction }, true); + await runAndDrain(queue.allocate({ kind: "chat", text: pending.instruction }), true); } else process.stdout.write("No safely rejected task is ready. Use /auth status for details.\n"); } else if (sub === "new") { process.stdout.write(authRepair.startNewConversation()); - queue.length = 0; steerChannel.clearNextTurn(); btwNotes.length = 0; + discardQueue("new conversation"); steerChannel.clearNextTurn(); btwNotes.length = 0; } else if (sub === "draft") { if (buf.value) process.stdout.write("Current draft is still in the input line; clear it before restoring the earlier draft.\n"); else if (!heldDraft) process.stdout.write("No earlier type-ahead draft is saved.\n"); @@ -1802,7 +1912,7 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P if (isAbortError(err)) process.stdout.write("Login cancelled. Task and draft preserved.\n"); else printError(err, ctx.cfg.baseUrl); } finally { - slashAbort = null; busy = false; + slashAbort = null; busy = false; runningSlash = null; } if (!buf.value && heldDraft && sub !== "draft") buf.insert(heldDraft); renderHudLine(); repaint(); @@ -1813,6 +1923,7 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P try { const res = await handleSlash(ctx, t, process.stdout, slashAbort.signal); if (res.exit) { + discardQueue("session ended"); // entries held after a failure are listed, not lost silently cleanup(); resolve(0); return; @@ -1834,8 +1945,8 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P } } catch (err) { if (isAbortError(err)) { - queue.length = 0; process.stdout.write(theme.dim("✗ canceled") + "\n"); + discardQueue("command cancelled"); } else { printError(err, ctx.cfg.baseUrl); } @@ -1844,13 +1955,15 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P setupOwnsInput = false; busy = false; slashAbort = null; + runningSlash = null; } - if (queue.length && !continuation.pending) { busy = true; await runAndDrain(queue.shift()!); } + const next = continuation.pending ? undefined : queue.shift(); + if (next) { busy = true; await runAndDrain(next); } renderHudLine(); repaint(); return; } - await runAndDrain({ kind: "chat", text: t }); + await runAndDrain(queue.allocate({ kind: "chat", text: t })); renderHudLine(); repaint(); }; @@ -2089,6 +2202,8 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P }); } +const LINE_MODE_QUEUE_NOTE = "Line mode has no pending queue: each input line runs in order after the previous one finishes, so there is nothing to list, edit, remove, clear, or resume. /queue runs the task as the next line.\n"; + /** Non-TTY fallback (pipes / CI): a line-oriented readline loop — no raw-mode * key decoding since there's no real terminal to own. `inflight` still wires * Ctrl+C to cancel the current turn/slash-command rather than killing the @@ -2104,7 +2219,16 @@ export async function replLines(ctx: AppContext, skillOpts: TurnSkillOptions = { process.on("SIGINT", onSigint); try { if (p) process.stdout.write(p + consoleShell.prompt()); - for await (const line of rl) { + for await (const rawLine of rl) { + // Line mode reads the next line only after the previous one finished, so + // there is never a pending queue: `/queue ` simply runs in order. + if (parseQueueCommand(rawLine)) { + process.stdout.write(LINE_MODE_QUEUE_NOTE); + if (p) process.stdout.write(p + consoleShell.prompt()); + continue; + } + const queuePrefix = /^\s*\/queue[ \t]+/.exec(rawLine); + const line = queuePrefix ? rawLine.slice(queuePrefix[0].length) : rawLine; if (continuation.pending && !line.trim().startsWith("/switch")) { process.stdout.write("Choose /switch continue, fresh, cancel, brief, or edit first; input was not sent.\n"); if (p) process.stdout.write(p + consoleShell.prompt()); diff --git a/src/commands/command_manifest_data.ts b/src/commands/command_manifest_data.ts index e89ede8..31bfd5a 100644 --- a/src/commands/command_manifest_data.ts +++ b/src/commands/command_manifest_data.ts @@ -4330,9 +4330,9 @@ export const COMMAND_MANIFEST_SOURCE: readonly CommandManifestEntry[] = [ "aliases": [], "compatibilityAliases": [], "deprecatedAliases": [], - "args": "", - "summary": "queue a task (runs when current finishes)", - "detailedHelp": "/queue \nqueue a task (runs when current finishes)", + "args": "[|list|edit |remove |clear|resume]", + "summary": "queue a task, or inspect/edit/remove/clear pending entries", + "detailedHelp": "/queue \nqueue a task (runs when current finishes)\n/queue | /queue list\nshow the running entry and pending entries with ids, types and the queue bound\n/queue edit \nreplace a pending chat entry, or a user shell entry with !; position and type are kept\n/queue remove | /queue clear\ndiscard pending entries; they never run\n/queue resume\nrun entries kept after a failed turn, in order\nQueue commands are local and never call a model or start a process.", "section": "Steering", "hidden": false, "permissionClass": "unknown", @@ -4354,7 +4354,7 @@ export const COMMAND_MANIFEST_SOURCE: readonly CommandManifestEntry[] = [ "module": "src/commands/command_manifest_data.ts", "symbol": "COMMAND_MANIFEST_SOURCE", "target": "queue", - "usage": "/queue ", + "usage": "/queue [|list|edit |remove |clear|resume]", "visible": true, "disposition": "generated" }, diff --git a/src/commands/console_input.ts b/src/commands/console_input.ts index d69526a..ba60983 100644 --- a/src/commands/console_input.ts +++ b/src/commands/console_input.ts @@ -9,7 +9,9 @@ import { stripAnsi } from "../ui/text.js"; import { discoverShellProfiles, type ShellProfile } from "../core/shell_profiles.js"; type ShareAction = "preview" | "lines" | "drop" | "replace" | "mask" | "redact" | "send" | "cancel"; -type ShareInput = { kind: "share"; action: ShareAction; first?: number; last?: number; value?: string }; +/** `boundCommandId` is set when a send is queued: it may only send the preview + * the user reviewed for that command, never a later replacement. */ +type ShareInput = { kind: "share"; action: ShareAction; first?: number; last?: number; value?: string; boundCommandId?: string }; interface ShellCapture { sessionId: string; @@ -288,10 +290,18 @@ export class ConsoleShell { } if (input.action === "send") { const stage = this.staged ?? (scripted ? this.stageLatest() : null); + if (!stage && input.boundCommandId !== undefined) { + this.shareNotice(`Queued send was bound to the preview of command ${input.boundCommandId}, which is no longer staged. Nothing was sent.`, "rebound"); + return { kind: "empty" }; + } if (!stage) { this.shareNotice(scripted ? "No local shell result to share." : "Preview the shell result before sending: /shell-result", "missing"); return { kind: "empty" }; } + if (input.boundCommandId !== undefined && stage.capture.commandId !== input.boundCommandId) { + this.shareNotice(`Queued send was bound to the preview of command ${input.boundCommandId}, but the staged preview is now command ${stage.capture.commandId}. Nothing was sent; review it and send again.`, "rebound"); + return { kind: "empty" }; + } if (!stage.editable.trim()) { this.shareNotice("The staged shell selection is empty. Edit it or cancel; nothing was sent.", "empty"); return { kind: "empty" }; @@ -351,6 +361,8 @@ export class ConsoleShell { this.showPreview(stage); return { kind: "empty" }; } + /** Command whose preview is staged for review, if any. */ + stagedCommandId(): string | null { return this.staged?.capture.commandId ?? null; } prompt(): string { return `[${this.profile} ${sanitizeServerText(this.session.cwd)}${this.session.state === "lost" ? "; shell lost" : ""}] `; } close(): void { this.latest = null; this.staged = null; this.terminal?.stop(); this.session.close(); } } diff --git a/src/commands/console_queue.ts b/src/commands/console_queue.ts new file mode 100644 index 0000000..120b9a5 --- /dev/null +++ b/src/commands/console_queue.ts @@ -0,0 +1,215 @@ +import { classifyConsoleInput, type ConsoleInput } from "./console_input.js"; +import { sanitizeServerText } from "../core/transport.js"; + +/** Inputs that may wait behind an active turn. Errors and empty lines never queue. */ +export type QueueableInput = Extract; + +export interface QueueEntry { + /** Stable for the session and never reused: q1, q2, … */ + readonly id: string; + readonly input: QueueableInput; +} + +export const QUEUE_MAX_ENTRIES = 32; +export const QUEUE_MAX_BYTES = 64 * 1024; + +export type QueueCommand = + | { op: "list" } + | { op: "clear" } + | { op: "resume" } + | { op: "remove"; id: string } + | { op: "edit"; id: string; text: string } + | { op: "usage"; message: string }; + +export type QueueResult = { ok: true; entry: QueueEntry } | { ok: false; message: string }; + +/** Management commands are matched strictly so ordinary tasks still queue: + * `/queue clear the cache` is a task, `/queue clear` is the command, and + * edit/remove only act on a `q` id. */ +export function parseQueueCommand(raw: string): QueueCommand | null { + const text = raw.trim(); + if (text === "/queue" || text === "/queue list") return { op: "list" }; + if (text === "/queue clear") return { op: "clear" }; + if (text === "/queue resume") return { op: "resume" }; + const remove = /^\/queue[ \t]+(?:remove|rm)[ \t]+(q\d+)$/i.exec(text); + if (remove) return { op: "remove", id: remove[1]!.toLowerCase() }; + const edit = /^\/queue[ \t]+edit[ \t]+(q\d+)(?:[ \t]+([\s\S]*))?$/i.exec(text); + if (edit) { + const replacement = (edit[2] ?? "").trim(); + return replacement ? { op: "edit", id: edit[1]!.toLowerCase(), text: replacement } : { op: "usage", message: "usage: /queue edit " }; + } + return null; +} + +/** Human-facing type of an entry. */ +export function entryKind(input: QueueableInput): "chat" | "user shell" | "shell reset" | "shell profile" | "shell-share" { + if (input.kind === "chat") return "chat"; + if (input.kind === "shell") return "user shell"; + if (input.kind === "reset-shell") return "shell reset"; + if (input.kind === "profile") return "shell profile"; + return "shell-share"; +} + +function inputBytes(input: QueueableInput): number { + if (input.kind === "chat") return Buffer.byteLength(input.text, "utf8"); + if (input.kind === "shell") return Buffer.byteLength(input.command, "utf8"); + if (input.kind === "share") return Buffer.byteLength(input.value ?? "", "utf8"); + return 0; +} + +function clip(text: string, max = 55): string { + const flat = sanitizeServerText(text).replace(/\s*\n\s*/g, " ⏎ "); + return flat.length > max ? flat.slice(0, max) + "…" : flat; +} + +/** States the shell-share binding explicitly so nobody has to guess which + * result a queued action will touch when it finally runs. */ +export function shareBinding(input: Extract): string { + if (input.action === "send") { + return input.boundCommandId + ? `sends the preview you reviewed of command ${input.boundCommandId.slice(0, 8)}; refused if that preview is gone or replaced` + : "sends the staged preview"; + } + if (input.action === "cancel") return "discards whatever preview is staged at execution"; + return "acts on the staged preview, or stages the latest shell result at execution"; +} + +export function describeEntry(input: QueueableInput): string { + if (input.kind === "chat") return `"${clip(input.text)}"`; + // Commands often share a long prefix; keep enough to tell entries apart. + if (input.kind === "shell") return `!${clip(input.command, 120)} (local only; never sent to the model)`; + if (input.kind === "reset-shell") return "/shell-reset"; + if (input.kind === "profile") return `/shell-profile ${input.action}${input.profile ? " " + input.profile : ""}`; + const detail = input.action === "drop" ? ` ${input.first}-${input.last}` + : input.action === "replace" ? ` ${input.first}` + : input.action === "mask" ? " " + : ""; + return `${input.action}${detail}: ${shareBinding(input)}`; +} + +export function entryLine(entry: QueueEntry): string { + return `${entry.id.padEnd(4)} ${entryKind(entry.input).padEnd(11)} ${describeEntry(entry.input)}`; +} + +/** Ordered, bounded, local-only queue of follow-ups behind the active turn. + * Nothing here talks to a model or a process: callers decide when to run. */ +export class ConsoleQueue { + private entries: readonly QueueEntry[] = []; + private nextId = 1; + private heldReason: string | null = null; + private runningEntry: QueueEntry | null = null; + + get pending(): readonly QueueEntry[] { return this.entries; } + get length(): number { return this.entries.length; } + get held(): string | null { return this.heldReason; } + get running(): QueueEntry | null { return this.runningEntry; } + get bytes(): number { return this.entries.reduce((sum, e) => sum + inputBytes(e.input), 0); } + + /** An id for an input that runs immediately, so the running entry is named too. */ + allocate(input: QueueableInput): QueueEntry { + return { id: `q${this.nextId++}`, input }; + } + + enqueue(input: QueueableInput): QueueResult { + if (this.entries.length >= QUEUE_MAX_ENTRIES) { + return { ok: false, message: `Queue full (${QUEUE_MAX_ENTRIES} entries); not queued. Remove entries with /queue remove or /queue clear.` }; + } + if (this.bytes + inputBytes(input) > QUEUE_MAX_BYTES) { + return { ok: false, message: `Queue full (${QUEUE_MAX_BYTES / 1024} KiB of queued text); not queued. Remove entries with /queue remove or /queue clear.` }; + } + const entry = this.allocate(input); + this.entries = [...this.entries, entry]; + return { ok: true, entry }; + } + + /** Next entry to run, or undefined while empty or held after a failure. */ + shift(): QueueEntry | undefined { + if (this.heldReason !== null) return undefined; + const [next, ...rest] = this.entries; + this.entries = rest; + return next; + } + + setRunning(entry: QueueEntry | null): void { this.runningEntry = entry; } + + /** Keep entries but stop draining until the user explicitly resumes. */ + hold(reason: string): boolean { + if (!this.entries.length) return false; + this.heldReason = reason; + return true; + } + + resume(): boolean { + if (this.heldReason === null) return false; + this.heldReason = null; + return true; + } + + /** Remove every pending entry; returns exactly what was discarded. */ + clear(): readonly QueueEntry[] { + const removed = this.entries; + this.entries = []; + this.heldReason = null; + return removed; + } + + remove(id: string): QueueResult { + const entry = this.entries.find(e => e.id === id); + if (!entry) return { ok: false, message: this.missing(id) }; + this.entries = this.entries.filter(e => e.id !== id); + if (!this.entries.length) this.heldReason = null; + return { ok: true, entry }; + } + + /** Replace an entry's text in place. The replacement is classified exactly as + * typed input would be and must produce the same entry type. */ + edit(id: string, raw: string): QueueResult { + const index = this.entries.findIndex(e => e.id === id); + if (index < 0) return { ok: false, message: this.missing(id) }; + const current = this.entries[index]!; + const next = classifyConsoleInput(raw); + let input: QueueableInput; + if (current.input.kind === "chat") { + if (next.kind !== "chat") { + return { ok: false, message: `${id} is a chat entry and the replacement is not chat text. Not changed (use \\! for chat text starting with !).` }; + } + if (next.text.startsWith("/")) return { ok: false, message: `${id} is a chat entry; slash commands are not queued. Not changed.` }; + input = { kind: "chat", text: next.text }; + } else if (current.input.kind === "shell") { + if (next.kind !== "shell") return { ok: false, message: `${id} is a user shell entry; the replacement must be !. Not changed.` }; + input = { kind: "shell", command: next.command }; + } else { + return { ok: false, message: `${id} is a ${entryKind(current.input)} action with no editable text. Remove it and queue a new one.` }; + } + if (this.bytes - inputBytes(current.input) + inputBytes(input) > QUEUE_MAX_BYTES) { + return { ok: false, message: `Edit would exceed the ${QUEUE_MAX_BYTES / 1024} KiB queue bound. Not changed.` }; + } + const entry: QueueEntry = { id: current.id, input }; + this.entries = this.entries.map((e, i) => (i === index ? entry : e)); + return { ok: true, entry }; + } + + private missing(id: string): string { + if (this.runningEntry?.id === id) return `${id} is running and cannot be changed; Ctrl+C cancels it.`; + return `No pending entry ${id} (it already ran, was removed, or never existed). /queue lists pending entries.`; + } + + /** `runningLabel` names non-entry work (a slash command) when nothing queued runs. */ + render(runningLabel: string | null = null): string { + const running = this.runningEntry ? entryLine(this.runningEntry) : runningLabel; + const paused = this.heldReason ? ` — PAUSED (${this.heldReason}); nothing runs until /queue resume` : ""; + const lines = [ + `Queue: ${this.entries.length} pending${paused}. Bound: ${QUEUE_MAX_ENTRIES} entries, ${QUEUE_MAX_BYTES / 1024} KiB (${this.bytes} B used).`, + ` running ${running ?? "nothing"}`, + ...this.entries.map((e, i) => ` ${String(i + 1).padStart(2)}. ${entryLine(e)}`), + ]; + if (this.entries.length) lines.push("Edit: /queue edit · remove: /queue remove · clear: /queue clear"); + return lines.join("\n") + "\n"; + } +} + +/** Disposition text after failure, cancellation or shell-state loss. */ +export function renderDisposition(header: string, entries: readonly QueueEntry[], footer = ""): string { + if (!entries.length) return ""; + return [header, ...entries.map(e => " " + entryLine(e)), ...(footer ? [footer] : [])].join("\n") + "\n"; +} diff --git a/test/console_input.test.ts b/test/console_input.test.ts index 3706059..fc8a8ac 100644 --- a/test/console_input.test.ts +++ b/test/console_input.test.ts @@ -215,7 +215,7 @@ for (const tty of [false, true]) { submit('hello'); await until(() => calls === 1 && releaseModel !== null, 'model turn running'); submit('!echo QUEUED_SHELL'); - ${tty ? "await until(() => observed.includes('Local shell queued'), 'shell queued during model turn');" : ""} + ${tty ? "await until(() => observed.includes('(user shell, 1 pending)'), 'shell queued during model turn');" : ""} await releaseTurn(1); await until(() => completed() === 3, 'queued shell completion'); submit(${JSON.stringify(`!"${process.execPath}" -e "process.exit(7)"`)}); @@ -277,8 +277,8 @@ for (const tty of [false, true]) { test("console submit routing remains before history/rewrite and uses typed queue", () => { const code = readFileSync("src/commands/chat.ts", "utf8"); - assert.ok(code.includes("const queue: ConsoleInput[]")); + assert.ok(code.includes("const queue = new ConsoleQueue()")); assert.ok(code.indexOf("classifyConsoleInput(queuePrefix") < code.indexOf("const commit =")); - assert.ok(code.includes("result = await runQueuedTurn(next)")); + assert.ok(code.includes("result = await runEntry(entry)")); assert.ok(code.includes("appendHistory(line.trim(), historyPath(ctx.flags.cwd))"), "save the original escaped input, not its model prompt"); }); diff --git a/test/console_queue.test.ts b/test/console_queue.test.ts new file mode 100644 index 0000000..8eabc45 --- /dev/null +++ b/test/console_queue.test.ts @@ -0,0 +1,164 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { + ConsoleQueue, QUEUE_MAX_BYTES, QUEUE_MAX_ENTRIES, describeEntry, parseQueueCommand, renderDisposition, +} from "../src/commands/console_queue.js"; + +test("queue commands match strictly so ordinary tasks still queue", () => { + assert.deepEqual(parseQueueCommand("/queue"), { op: "list" }); + assert.deepEqual(parseQueueCommand(" /queue list "), { op: "list" }); + assert.deepEqual(parseQueueCommand("/queue clear"), { op: "clear" }); + assert.deepEqual(parseQueueCommand("/queue resume"), { op: "resume" }); + assert.deepEqual(parseQueueCommand("/queue remove q3"), { op: "remove", id: "q3" }); + assert.deepEqual(parseQueueCommand("/queue rm Q12"), { op: "remove", id: "q12" }); + assert.deepEqual(parseQueueCommand("/queue edit q2 new text here"), { op: "edit", id: "q2", text: "new text here" }); + assert.deepEqual(parseQueueCommand("/queue edit q2 !ls -la"), { op: "edit", id: "q2", text: "!ls -la" }); + assert.equal(parseQueueCommand("/queue edit q2")?.op, "usage"); + // Plausible tasks are not commands. + assert.equal(parseQueueCommand("/queue clear the cache"), null); + assert.equal(parseQueueCommand("/queue edit the readme"), null); + assert.equal(parseQueueCommand("/queue remove 2 stale files"), null); + assert.equal(parseQueueCommand("/queue fix login bug"), null); + assert.equal(parseQueueCommand("hello"), null); +}); + +test("entries get stable, never-reused ids and drain in strict order", () => { + const queue = new ConsoleQueue(); + const running = queue.allocate({ kind: "chat", text: "first" }); + assert.equal(running.id, "q1"); + const a = queue.enqueue({ kind: "chat", text: "a" }); + const b = queue.enqueue({ kind: "shell", command: "make test" }); + const c = queue.enqueue({ kind: "reset-shell" }); + assert.ok(a.ok && b.ok && c.ok); + assert.deepEqual(queue.pending.map(e => e.id), ["q2", "q3", "q4"]); + assert.ok(queue.remove("q3").ok); + const d = queue.enqueue({ kind: "chat", text: "d" }); + assert.ok(d.ok && d.entry.id === "q5", "removed ids are never reused"); + assert.deepEqual([queue.shift()?.id, queue.shift()?.id, queue.shift()?.id, queue.shift()], ["q2", "q4", "q5", undefined]); +}); + +test("edits keep id, position and type; reclassifying edits are rejected unchanged", () => { + const queue = new ConsoleQueue(); + queue.enqueue({ kind: "chat", text: "one" }); + queue.enqueue({ kind: "shell", command: "echo two" }); + queue.enqueue({ kind: "chat", text: "three" }); + queue.enqueue({ kind: "share", action: "preview" }); + queue.enqueue({ kind: "reset-shell" }); + const snapshot = queue.pending; + + const chat = queue.edit("q1", "one, revised"); + assert.ok(chat.ok); + assert.deepEqual(queue.pending.map(e => e.id), ["q1", "q2", "q3", "q4", "q5"]); + assert.deepEqual(queue.pending[0]!.input, { kind: "chat", text: "one, revised" }); + assert.deepEqual(snapshot[0]!.input, { kind: "chat", text: "one" }, "entries are replaced, not mutated"); + + const shell = queue.edit("q2", "!echo TWO"); + assert.ok(shell.ok); + assert.deepEqual(queue.pending[1]!.input, { kind: "shell", command: "echo TWO" }); + + const before = queue.pending; + const rejections = [ + queue.edit("q3", "!rm -rf build"), // chat → shell + queue.edit("q3", "/exit"), // chat → slash command + queue.edit("q3", "/shell-reset"), // chat → shell reset + queue.edit("q2", "plain words"), // shell → chat + queue.edit("q2", "!"), // shell → usage error + queue.edit("q4", "/shell-result send"), // share has no text + queue.edit("q5", "!echo hi"), // reset has no text + queue.edit("q99", "whatever"), // missing + ]; + for (const result of rejections) assert.equal(result.ok, false); + assert.match((rejections[0] as { message: string }).message, /q3 is a chat entry.*Not changed/); + assert.match((rejections[3] as { message: string }).message, /q2 is a user shell entry; the replacement must be !/); + assert.match((rejections[5] as { message: string }).message, /shell-share action with no editable text/); + assert.match((rejections[7] as { message: string }).message, /No pending entry q99/); + assert.equal(queue.pending, before, "rejected edits change nothing"); + + // \! keeps a leading ! as chat text. + const escaped = queue.edit("q3", "\\!important: three"); + assert.ok(escaped.ok); + assert.deepEqual(queue.pending[2]!.input, { kind: "chat", text: "!important: three" }); +}); + +test("the running entry is named but cannot be edited or removed", () => { + const queue = new ConsoleQueue(); + const running = queue.allocate({ kind: "chat", text: "streaming now" }); + queue.setRunning(running); + queue.enqueue({ kind: "chat", text: "next" }); + const edit = queue.edit(running.id, "changed"); + const remove = queue.remove(running.id); + assert.equal(edit.ok, false); + assert.equal(remove.ok, false); + assert.match((edit as { message: string }).message, /q1 is running and cannot be changed/); + const listing = queue.render(); + assert.match(listing, /running {2}q1 +chat +"streaming now"/); + assert.match(listing, /1\. +q2 +chat +"next"/); + assert.match(new ConsoleQueue().render("slash command /doctor"), /running {2}slash command \/doctor/); +}); + +test("bounds reject with useful feedback and leave the queue unchanged", () => { + const queue = new ConsoleQueue(); + for (let i = 0; i < QUEUE_MAX_ENTRIES; i++) assert.ok(queue.enqueue({ kind: "chat", text: `t${i}` }).ok); + const full = queue.enqueue({ kind: "chat", text: "one more" }); + assert.equal(full.ok, false); + assert.match((full as { message: string }).message, /Queue full \(32 entries\); not queued/); + assert.equal(queue.length, QUEUE_MAX_ENTRIES); + + const bytes = new ConsoleQueue(); + assert.ok(bytes.enqueue({ kind: "chat", text: "x".repeat(QUEUE_MAX_BYTES - 10) }).ok); + const big = bytes.enqueue({ kind: "shell", command: "y".repeat(11) }); + assert.equal(big.ok, false); + assert.match((big as { message: string }).message, /64 KiB/); + assert.ok(bytes.enqueue({ kind: "chat", text: "é".repeat(5) }).ok, "UTF-8 bytes, not code units, are counted"); + const edit = bytes.edit("q2", "z".repeat(20)); + assert.equal(edit.ok, false); + assert.match((edit as { message: string }).message, /exceed the 64 KiB queue bound/); +}); + +test("a hold keeps entries but stops draining until resume; clear and remove report exactly what went", () => { + const queue = new ConsoleQueue(); + assert.equal(queue.hold("q1 failed"), false, "an empty queue is never held"); + queue.enqueue({ kind: "chat", text: "a" }); + queue.enqueue({ kind: "shell", command: "echo b" }); + assert.equal(queue.hold("q0 failed"), true); + assert.equal(queue.shift(), undefined); + assert.equal(queue.length, 2); + assert.match(queue.render(), /PAUSED \(q0 failed\); nothing runs until \/queue resume/); + assert.equal(queue.resume(), true); + assert.equal(queue.resume(), false); + assert.equal(queue.shift()?.id, "q1"); + queue.hold("q1 failed"); + assert.equal(queue.remove("q2").ok, true); + assert.equal(queue.held, null, "removing the last held entry ends the hold"); + queue.enqueue({ kind: "chat", text: "c" }); + queue.enqueue({ kind: "chat", text: "d" }); + queue.hold("q3 failed"); + const removed = queue.clear(); + assert.deepEqual(removed.map(e => e.id), ["q3", "q4"]); + assert.equal(queue.length, 0); + assert.equal(queue.held, null); + const text = renderDisposition("Queue discarded (turn cancelled):", removed); + assert.match(text, /q3 +chat +"c"\n {2}q4 +chat +"d"/); + assert.equal(renderDisposition("nothing", []), ""); +}); + +test("each entry shows its type and the shell-share binding explicitly", () => { + assert.match(describeEntry({ kind: "shell", command: "npm test" }), /^!npm test \(local only; never sent to the model\)$/); + assert.equal(describeEntry({ kind: "reset-shell" }), "/shell-reset"); + assert.match(describeEntry({ kind: "share", action: "preview" }), /stages the latest shell result at execution/); + assert.match(describeEntry({ kind: "share", action: "drop", first: 2, last: 4 }), /^drop 2-4: acts on the staged preview/); + assert.match(describeEntry({ kind: "share", action: "send", boundCommandId: "0123456789abcdef" }), + /sends the preview you reviewed of command 01234567; refused if that preview is gone or replaced/); + assert.match(describeEntry({ kind: "share", action: "cancel" }), /discards whatever preview is staged at execution/); + // Terminal controls in queued text never reach the console verbatim. + assert.doesNotMatch(describeEntry({ kind: "chat", text: "hi\x1b]52;c;payload\x07 there" }), /\x1b|\x07/); + const queue = new ConsoleQueue(); + queue.enqueue({ kind: "chat", text: "c" }); + queue.enqueue({ kind: "shell", command: "ls" }); + queue.enqueue({ kind: "share", action: "send", boundCommandId: "abcdef0123" }); + const listing = queue.render(); + assert.match(listing, /q1 +chat /); + assert.match(listing, /q2 +user shell +!ls/); + assert.match(listing, /q3 +shell-share +send: sends the preview you reviewed of command abcdef01/); + assert.match(listing, /Bound: 32 entries, 64 KiB \(3 B used\)/); +}); diff --git a/test/console_queue_tty.test.ts b/test/console_queue_tty.test.ts new file mode 100644 index 0000000..197e8a7 --- /dev/null +++ b/test/console_queue_tty.test.ts @@ -0,0 +1,310 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { PassThrough } from "node:stream"; +import { repl, replLines } from "../src/commands/chat.js"; +import { ConsoleShell } from "../src/commands/console_input.js"; +import { ApiClient } from "../src/core/transport.js"; +import { DEFAULT_CONFIG } from "../src/core/config.js"; +import { historyPath } from "../src/core/history_store.js"; +import type { AppContext } from "../src/core/context.js"; +import type { TokenStore } from "../src/core/auth.js"; + +// Raw-TTY queue behaviour (#284). Every queue operation here happens while a +// hosted turn is still streaming; fetch is the only model path and the file +// system is the only process evidence. + +const tokens = { get: async () => "fixture-token" } as unknown as TokenStore; +function context(root: string): AppContext { + return { + cfg: { ...DEFAULT_CONFIG, baseUrl: "https://stub.test", backend: "cloud", defaultModel: "", permissionMode: "ask", autoApply: false }, + flags: { cwd: root, json: true, yes: false, audit: false }, + tokens, api: new ApiClient("https://stub.test", tokens), confirm: async () => false, + } as AppContext; +} +const turnResponse = (): Response => new Response('data: {"type":"delta","text":"answer"}\n\ndata: {"type":"done","uvt":0,"cents":0}\n\n', { headers: { "content-type": "text/event-stream" } }); +/** A local shell command that leaves a file behind, valid in Bash and cmd. */ +const touch = (name: string): string => `!"${process.execPath}" -e "require('fs').writeFileSync('${name}','x')"`; + +interface Harness { + root: string; + bodies: string[]; + output: () => string; + submit: (text: string) => void; + type: (text: string) => void; + until: (predicate: () => boolean, label: string) => Promise; + /** Wait for model call `n` to be in flight, then answer it. */ + answer: (n: number, response?: Response) => Promise; + waitForCall: (n: number) => Promise; +} + +async function withConsole(run: (h: Harness) => Promise): Promise { + const root = mkdtempSync(join(tmpdir(), "aether-queue-tty-")); + const oldFetch = globalThis.fetch; + const oldWrite = process.stdout.write; + const oldErrWrite = process.stderr.write; + const tty = Object.getOwnPropertyDescriptor(process.stdin, "isTTY"); + const raw = Object.getOwnPropertyDescriptor(process.stdin, "setRawMode"); + const columns = Object.getOwnPropertyDescriptor(process.stdout, "columns"); + let output = ""; + const bodies: string[] = []; + const responders: Array<(response: Response) => void> = []; + let pending: Promise | null = null; + const until = async (predicate: () => boolean, label: string): Promise => { + const deadline = Date.now() + 8000; + while (!predicate()) { + if (Date.now() > deadline) throw new Error(`TTY test timed out waiting for ${label}: ` + output.slice(-1500)); + await new Promise(resolve => setTimeout(resolve, 5)); + } + }; + Object.defineProperty(process.stdin, "isTTY", { value: true, configurable: true }); + Object.defineProperty(process.stdin, "setRawMode", { value: () => process.stdin, configurable: true }); + // Wide enough that the composer never windows a draft behind a long temp-dir prompt. + Object.defineProperty(process.stdout, "columns", { value: 400, configurable: true }); + process.stdout.write =((text: string | Uint8Array) => { output += String(text); return true; }) as typeof process.stdout.write; + process.stderr.write = ((text: string | Uint8Array) => { output += String(text); return true; }) as typeof process.stderr.write; + globalThis.fetch = (async (url, init) => { + if (!String(url).includes("/agent/chat/stream")) return Response.json({ models: [] }); + bodies.push(String(init?.body ?? "")); + return await new Promise((resolve, reject) => { + responders[bodies.length - 1] = resolve; + init?.signal?.addEventListener("abort", () => reject(new DOMException("aborted", "AbortError")), { once: true }); + }); + }) as typeof fetch; + const submit = (text: string): void => { process.stdin.emit("data", Buffer.from(text + "\r")); }; + const type = (text: string): void => { process.stdin.emit("data", Buffer.from(text)); }; + const waitForCall = (n: number): Promise => until(() => responders[n - 1] !== undefined, `model call ${n}`); + const answer = async (n: number, response = turnResponse()): Promise => { + await waitForCall(n); + responders[n - 1]!(response); + }; + try { + pending = repl(context(root), { noSkills: true }); + await until(() => output.includes("\x1b[?2004h"), "console ready"); + await run({ root, bodies, output: () => output, submit, type, until, answer, waitForCall }); + process.stdin.emit("data", Buffer.from("\x03")); // clear any draft + submit("/exit"); + assert.equal(await Promise.race([pending, new Promise((_, reject) => setTimeout(() => reject(new Error("TTY exit timed out")), 2000))]), 0); + pending = null; + } finally { + for (const respond of responders) respond?.(turnResponse()); + if (pending) { + process.stdin.emit("data", Buffer.from("\x03\x03\x04")); + await Promise.race([pending.catch(() => {}), new Promise(resolve => setTimeout(resolve, 1000))]); + } + globalThis.fetch = oldFetch; process.stdout.write = oldWrite; process.stderr.write = oldErrWrite; + if (tty) Object.defineProperty(process.stdin, "isTTY", tty); else delete (process.stdin as unknown as { isTTY?: boolean }).isTTY; + if (raw) Object.defineProperty(process.stdin, "setRawMode", raw); else delete (process.stdin as unknown as { setRawMode?: unknown }).setRawMode; + if (columns) Object.defineProperty(process.stdout, "columns", columns); else delete (process.stdout as unknown as { columns?: number }).columns; + rmSync(historyPath(root), { force: true }); rmSync(root, { recursive: true, force: true }); + } +} + +const shellResults = (output: string): number => (output.match(/"type":"shell_result"/g) ?? []).length; + +test("TTY: list, edit and remove mixed queued entries while a turn streams; removed entries make zero calls", async () => { + await withConsole(async ({ root, bodies, output, submit, type, until, answer, waitForCall }) => { + submit("first question"); + await waitForCall(1); + + // Type-ahead of every entry type while q1 streams. + submit("second question"); + submit(touch("removed.txt")); + submit(touch("kept.txt")); + submit("third question"); + await until(() => output().includes("Queued q5 (chat"), "four entries queued"); + assert.match(output(), /⏳ Queued q2 \(chat, 1 pending\): "second question"/); + assert.match(output(), /⏳ Queued q3 \(user shell, 2 pending\): !.*removed\.txt.*\(local only; never sent to the model\)/); + + // A send cannot queue without a reviewed preview to bind to. + submit("/shell-result send"); + await until(() => output().includes("Not queued: a queued /shell-result send must bind"), "unbound send refused"); + assert.equal(bodies.length, 1); + type("\x15"); // the refused draft stays in the composer; Ctrl+U clears it without cancelling q1 + + submit("/queue"); + await until(() => output().includes("Queue: 4 pending"), "listing"); + const listing = output().slice(output().lastIndexOf("Queue: 4 pending")); + assert.match(listing, /running {2}q1 +chat +"first question"/); + assert.match(listing, /1\. +q2 +chat +"second question"/); + assert.match(listing, /2\. +q3 +user shell +!/); + assert.match(listing, /4\. +q5 +chat +"third question"/); + + // Invalid edits are rejected without reclassifying the entry. + submit("/queue edit q2 " + touch("smuggled.txt")); + await until(() => output().includes("q2 is a chat entry"), "chat→shell edit rejected"); + submit("/queue edit q4 plain words"); + await until(() => output().includes("q4 is a user shell entry; the replacement must be !"), "shell→chat edit rejected"); + submit("/queue edit q1 changed"); + await until(() => output().includes("q1 is running and cannot be changed"), "running entry protected"); + + submit("/queue edit q2 second question, revised"); + await until(() => output().includes("Edited q2 (chat, position kept)"), "chat edit"); + submit("/queue edit q4 " + touch("edited.txt")); + await until(() => output().includes("Edited q4 (user shell, position kept)"), "shell edit"); + submit("/queue remove q3"); + await until(() => output().includes("Removed q3 (user shell); it will not run."), "shell removal"); + submit("/queue remove q5"); + await until(() => output().includes("Removed q5 (chat); it will not run."), "chat removal"); + type("draft stays"); + + // None of that touched the model or a process. + assert.equal(bodies.length, 1); + assert.equal(shellResults(output()), 0); + + await answer(1); + await answer(2); + await until(() => shellResults(output()) === 1, "edited shell entry ran"); + await new Promise(resolve => setImmediate(resolve)); + + assert.equal(bodies.length, 2, "removed chat made zero model calls"); + assert.match(bodies[1]!, /second question, revised/); + for (const body of bodies) assert.doesNotMatch(body, /third question|removed\.txt|kept\.txt|edited\.txt|smuggled/); + assert.equal(existsSync(join(root, "removed.txt")), false, "removed shell made zero process calls"); + assert.equal(existsSync(join(root, "kept.txt")), false, "edited shell ran its replacement only"); + assert.equal(existsSync(join(root, "smuggled.txt")), false); + assert.equal(existsSync(join(root, "edited.txt")), true); + assert.match(output(), /→ Running queued q2 \(chat\): "second question, revised"/); + await until(() => output().includes("draft stays"), "type-ahead draft repainted after the drain"); + + // Queue management and shell text stay out of chat history. + const history = existsSync(historyPath(root)) ? readFileSync(historyPath(root), "utf8") : ""; + assert.doesNotMatch(history, /\/queue|removed\.txt|edited\.txt/); + assert.match(history, /second question/); + }); +}); + +test("TTY: a failed turn pauses pending entries until /queue resume; nothing silently resumes", async () => { + await withConsole(async ({ root, bodies, output, submit, until, answer, waitForCall }) => { + submit("will fail"); + await waitForCall(1); + submit("after failure"); + submit(touch("held.txt")); + await until(() => output().includes("Queued q3 (user shell"), "entries queued"); + await answer(1, Response.json({ detail: "insufficient UVT balance" }, { status: 402 })); + await until(() => output().includes("Queue paused (q1 failed): 2 pending entries kept; none ran."), "pause disposition"); + const disposition = output().slice(output().lastIndexOf("Queue paused")); + assert.match(disposition, /q2 +chat +"after failure"/); + assert.match(disposition, /q3 +user shell +!/); + assert.match(disposition, /\/queue resume runs them in order/); + await new Promise(resolve => setTimeout(resolve, 50)); + assert.equal(bodies.length, 1); + assert.equal(existsSync(join(root, "held.txt")), false); + + // A new submission runs alone; the held entries stay held. + process.stdin.emit("data", Buffer.from("\x03")); // clear the restored failed prompt + submit("fresh question"); + await answer(2); + await until(() => (output().match(/"type":"turn_outcome"/g) ?? []).length >= 1, "fresh turn finished"); + await new Promise(resolve => setTimeout(resolve, 50)); + assert.equal(bodies.length, 2); + assert.match(bodies[1]!, /fresh question/); + assert.equal(existsSync(join(root, "held.txt")), false); + submit("/queue list"); + await until(() => output().includes("Queue: 2 pending — PAUSED (q1 failed)"), "paused listing"); + + submit("/queue resume"); + await until(() => output().includes("Queue resumed: 2 pending entries run in order."), "resume"); + await answer(3); + await until(() => shellResults(output()) === 1, "held shell ran after resume"); + assert.match(bodies[2]!, /after failure/); + assert.equal(existsSync(join(root, "held.txt")), true); + }); +}); + +test("TTY: cancelling a streaming turn discards and lists pending entries; none run", async () => { + await withConsole(async ({ root, bodies, output, submit, until, waitForCall }) => { + submit("long task"); + await waitForCall(1); + submit("never sent"); + submit(touch("never.txt")); + await until(() => output().includes("Queued q3 (user shell"), "entries queued"); + process.stdin.emit("data", Buffer.from("\x03")); + await until(() => output().includes("Queue discarded (turn cancelled): 2 pending entries removed; none ran and none will resume."), "discard disposition"); + const disposition = output().slice(output().lastIndexOf("Queue discarded")); + assert.match(disposition, /q2 +chat +"never sent"/); + assert.match(disposition, /q3 +user shell +!/); + await new Promise(resolve => setTimeout(resolve, 50)); + submit("/queue"); + await until(() => output().includes("Queue: 0 pending."), "empty after cancel"); + assert.equal(bodies.length, 1); + assert.equal(existsSync(join(root, "never.txt")), false); + }); +}); + +test("a queued shell-share send only sends the preview it was bound to", async () => { + const root = mkdtempSync(join(tmpdir(), "aether-queue-share-")); + let output = ""; + const shell = new ConsoleShell(root, text => { output += text; }); + try { + await shell.run({ kind: "shell", command: "echo FIRST" }); + shell.share({ kind: "share", action: "preview" }); + const first = shell.stagedCommandId(); + assert.ok(first); + + // A different binding sends nothing and keeps the reviewed preview. + assert.deepEqual(shell.share({ kind: "share", action: "send", boundCommandId: "other-command" }), { kind: "empty" }); + assert.match(output, /bound to the preview of command other-command, but the staged preview is now command/); + assert.equal(shell.stagedCommandId(), first); + + // Cancelled before the queued send runs: still nothing is sent. + shell.share({ kind: "share", action: "cancel" }); + assert.deepEqual(shell.share({ kind: "share", action: "send", boundCommandId: first }), { kind: "empty" }); + assert.match(output, /which is no longer staged\. Nothing was sent\./); + + // A newer command re-staged: the old binding is refused. + await shell.run({ kind: "shell", command: "echo SECOND" }); + shell.share({ kind: "share", action: "preview" }); + assert.notEqual(shell.stagedCommandId(), first); + assert.deepEqual(shell.share({ kind: "share", action: "send", boundCommandId: first }), { kind: "empty" }); + + // The matching binding sends exactly the reviewed preview. + const sent = shell.share({ kind: "share", action: "send", boundCommandId: shell.stagedCommandId()! }); + assert.equal(sent.kind, "chat"); + assert.match((sent as { text: string }).text, /SECOND/); + assert.doesNotMatch((sent as { text: string }).text, /FIRST/); + } finally { shell.close(); rmSync(root, { recursive: true, force: true }); } +}); + +test("line mode has no pending queue: management explains itself and /queue runs in order", async () => { + const root = mkdtempSync(join(tmpdir(), "aether-queue-lines-")); + const oldFetch = globalThis.fetch; + const oldWrite = process.stdout.write; + let output = ""; + let modelCalls = 0; + process.stdout.write = ((text: string | Uint8Array) => { output += String(text); return true; }) as typeof process.stdout.write; + globalThis.fetch = (async (url) => { + if (String(url).includes("/agent/chat/stream")) modelCalls++; + return Response.json({ models: [] }); + }) as typeof fetch; + const input = new PassThrough(); + try { + const session = replLines(context(root), { noSkills: true }, new ConsoleShell(root, text => { output += text; }, true), input); + input.end(["/queue list", "/queue edit q1 something", "/queue clear", touch("ran.txt").replace("!", "/queue !"), ""].join("\n")); + assert.equal(await session, 0); + assert.equal((output.match(/Line mode has no pending queue/g) ?? []).length, 3); + assert.equal(existsSync(join(root, "ran.txt")), true, "/queue runs as the next line"); + assert.equal(modelCalls, 0); + } finally { + globalThis.fetch = oldFetch; process.stdout.write = oldWrite; + rmSync(historyPath(root), { force: true }); rmSync(root, { recursive: true, force: true }); + } +}); + +test("TTY: leaving the console with held entries lists them as discarded", async () => { + await withConsole(async ({ root, bodies, output, submit, until, answer, waitForCall }) => { + submit("will fail"); + await waitForCall(1); + submit(touch("abandoned.txt")); + await until(() => output().includes("Queued q2 (user shell"), "entry queued"); + await answer(1, Response.json({ detail: "insufficient UVT balance" }, { status: 402 })); + await until(() => output().includes("Queue paused (q1 failed)"), "pause disposition"); + process.stdin.emit("data", Buffer.from("\x03")); // clear the restored failed prompt + submit("/exit"); + await until(() => output().includes("Queue discarded (session ended): 1 pending entry removed; none ran and none will resume."), "exit disposition"); + assert.equal(bodies.length, 1); + assert.equal(existsSync(join(root, "abandoned.txt")), false); + }); +});