From 86a619f10d67bba802e7e025e87570a29ee1825f Mon Sep 17 00:00:00 2001 From: 4ndreello <4ndreello@users.noreply.github.com> Date: Tue, 8 Sep 2026 12:00:09 -0300 Subject: [PATCH 1/3] feat(send): one-slot queued message dispatched as next turn Busy session.send persists pendingMessage/pendingAt (last-wins) and returns queued:true instead of SESSION_BUSY; the stream tail dispatches it via lock-guarded tryDispatch. stop/release clear the slot on every exit; open sessions never queue. Adds message.queued event, daemon-side validation (INVALID), web/CLI surfacing, migration for existing DBs. --- .specs/features/send-queue/spec.md | 198 +++++++++++++++++ src/cli/commands/send.ts | 1 + src/cli/commands/web.ts | 2 +- src/core/events.ts | 8 + src/core/session.ts | 4 + src/daemon/daemon.ts | 331 ++++++++++++++++++++--------- src/store/database.ts | 4 + src/store/sessions.ts | 13 +- src/web/canvas-page.ts | 24 ++- tests/power-send.test.ts | 10 +- tests/send-queue.test.ts | 105 +++++++++ 11 files changed, 595 insertions(+), 105 deletions(-) create mode 100644 .specs/features/send-queue/spec.md create mode 100644 tests/send-queue.test.ts diff --git a/.specs/features/send-queue/spec.md b/.specs/features/send-queue/spec.md new file mode 100644 index 0000000..3d45e92 --- /dev/null +++ b/.specs/features/send-queue/spec.md @@ -0,0 +1,198 @@ +# Send queue (one-slot pending message) + +## Goal + +Let the user type the next message while a headless session is still +working, instead of waiting staring at "aguarde o turno atual terminar". +The message waits in one visible slot and auto-dispatches as the next +turn when the current turn ends. Interactive sessions (`origin=open`) +stay out of scope: the daemon has no wire to their PTY. + +## Current behavior (evidence, corrected per review) + +- Headless `send` is a new process, not stdin: `SessionDriver.send()` + (`src/drivers/session-driver.ts:126`) calls `start({ resumeSessionId, + prompt })`, appending to the same log files. Native id resolves as + `session.nativeSessionId || runtime.nativeSessionId` — dispatch must + use the same resolution, not require the column alone. +- Between-turn state is terminal, not idle: every harness ends a turn + with terminal `session.completed`, and `session.send` + (`src/daemon/daemon.ts:560-605`) has NO terminal rejection — the + non-`interrupted` branch only returns `SESSION_BUSY` (lock held, + `starting`, `working`+live pid, runtime not done/drained) or + `CAPABILITY_NOT_SUPPORTED`, otherwise it resumes. Canvas agrees: + `paintSendState()` (`src/web/canvas-page.ts:862`) enables send on + `completed`. So "resumable terminal" IS today's success path. +- The daemon never sets `idle`/`needs_input`: `updateSessionFromEvent` + only sets `completed`/`failed`; `needs_input` is canvas-local + (set on `permission.requested` in `canvas-page.ts:570`). A + permission-parked turn still has a live runtime → counts as busy, + dispatches only at stream end, never on `permission.resolved`. +- Restart states (`recover()`, `daemon.ts:145-290`): `working` reattaches + (tail dispatch covers it); dead/pid-reused rows become + `failed`/`orphaned`; `interrupted` rows are skipped via `continue`. + There is no "idle-but-queued" state after boot. +- Web mirrors busy: `paintSendState()` disables input on + `working/starting` and on `origin=open` always. +- `interrupted` is terminal (`src/core/session.ts:87`), produced only by + `markInterrupted()` (`src/daemon/daemon.ts:1300`). Archiving it + (`session.release`, `daemon.ts:483`) is a separate, smaller change and + explicitly out of scope here. + +## Design + +### Slot + +- One slot per session: `pendingMessage: string | null` + `pendingAt: string | null`. +- Persisted in SQLite, not in-memory: the queue must survive a daemon + restart and be visible in `session.get`. Four touch points, all required: + 1. `CREATE TABLE` column list in `src/store/database.ts`, + 2. `addMissingColumns()` in the same file (existing installs throw + "no such column" without it), + 3. explicit `INSERT` column list in `SessionStore.create()`, + 4. column map in `SessionStore.update()` + `rowToSession()` + (`src/store/sessions.ts`). +- Single slot, last-wins: a second `send` while the slot is occupied + overwrites. Rationale: no unbounded growth, no stuck queue behind a + stale message, no concat-runaway prompts. The overwrite is observable + (new `pendingAt`, new queued event). + +### Admission (`session.send`, `src/daemon/daemon.ts:560`) + +Precedence (first match wins): + +1. Empty message → validation error (new daemon contract, see below). + Checked before origin so the error is stable regardless of session kind. +2. `origin === "open"` → `CAPABILITY_NOT_SUPPORTED`, never queued. +3. `interrupted` with live PID identity (triple + `pid && pidStartTime && processAlive && processStartTime equal`, same + as the `stop` path) → `SESSION_BUSY` (stop first), not queued. +4. Busy (today's `SESSION_BUSY` conditions: lock held, `starting`, + `working`+live identity, runtime not done/drained — permission-parked + included) → enqueue (see locking), return `{ ok: true, queued: true }`. + No `turn.started`, no status change, no driver call. +5. Otherwise (resumable: terminal `completed`/`failed`/`stopped`/`orphaned`, + or `interrupted` without live identity, with `resume` capable driver) → + start now. If the slot is occupied (stale leftover), the new message + wins: it starts now and the slot is cleared. Returns + `{ ok: true, queued: false }`. + +Enqueue runs under `sessionLocks`: acquire the session lock around +busy-check + slot persist + queued-event append (same bounded discipline +as `markInterrupted`), otherwise a message landing between dispatch's +slot read and clear is stranded or silently destroyed. Double-enqueue +stays safe (last-wins); event order follows lock order. + +### Dispatch (`tryDispatch`) + +- One function owns starting a queued turn; two call sites: + 1. tail of `attachDriverEvents()` (`daemon.ts:1054`) when the stream + ends with the session back in a resumable state (turn completed, + or failed — a queued message retries as the next turn); + 2. `recover()` post-restart for rows that actually exist with a + surviving slot: reattached-`working` is covered by (1); for + `failed`/`orphaned`-with-pending the boot site dispatches only + under an explicit policy (see acceptance). No `send`-fast-path + call (it already started a turn). +- Dispatch precondition (rechecked under `sessionLocks`): slot non-empty, + lock free, not `starting`, no live runtime (`done && drained`), no live + PID identity per the triple above (never `processAlive` alone — a + recycled PID must not block or permit), `origin !== "open"`, driver + `resume` capable, native id resolvable per + `session.nativeSessionId || runtime.nativeSessionId`. Permission-parked + (live runtime) counts as busy. If unmet, the slot stays; nothing is lost. +- Dispatch reuses the existing send body factored out, with one change: + failure handling. There is no IPC caller at tail/boot, so a dispatch + failure (including stale native id → `SessionDriver.send` resume throw, + or a crash between clear and `turn.started`) must NOT vanish the + message: restore the slot and append + broadcast a dispatch-failed event + (keeping the failure classification honest) instead of `SEND_FAILED`-to-nobody. +- Clearing policy: `session.stop` and `session.release` clear the slot on + ALL exits including error/early-return paths (`SESSION_BUSY`, + `SESSION_NOT_RUNNING`, `STOP_UNSAFE`, release's terminal early return), + otherwise a phantom "1 na fila" survives with no tail left to consume + it. Daemon shutdown (`markInterrupted`) keeps it by design. + +### Validation (new daemon contract) + +- `parseSendBody` (`src/cli/commands/web.ts:63`) only trims/rejects + empty; the 64 KiB cap is `MAX_SEND_BODY_BYTES` on raw body bytes at the + route (`web.ts:60,149-170`). The daemon and `codedeck send` validate + nothing today. +- This spec adds daemon-side validation with a named error code, units + (chars vs bytes — bytes, to match the route), and web/CLI status + mapping. Precedence: empty → validation error even for `origin=open` + (documented flip from `CAPABILITY_NOT_SUPPORTED`); oversize → same + rejection idle or busy (never queued). + +### Events and reads + +- Enqueue appends + broadcasts a queued event. `AgentEventType` + (`src/core/events.ts:4-18`) is a closed union and the canvas + (`selectSession` transcript branches, `onEvent` live handler) plus + `updateSessionFromEvent` silently drop unknown types — so adding + `message.queued` REQUIRES explicit branches in all three (transcript + render "você (na fila)", live handler, `lastEvent` update) or the + render claim is dropped and only `pendingMessage`-in-`get` drives the + "na fila" display. `EventStore.append` itself is type-agnostic, so + persistence works either way. Dispatch emits the normal `turn.started`. +- `session.get` / `session.logs` expose `pendingMessage` (or preview) + + `pendingAt` so the canvas and `ps` can show "1 na fila" without a new + round trip. + +### Surfaces + +- Web (`src/web/canvas-page.ts`): input stays enabled while + `working/starting` for non-`open` sessions; disabled only for + `origin=open` and while `sendInFlight`. Helper text: "mensagem na fila + — envia quando o turno terminar" after a queued send; detail shows the + queued text until dispatch. `POST /send` maps `queued:true` to 200 with + `{ ok, queued }`, preserving existing error codes otherwise, plus the + new validation code mapping. +- CLI (`codedeck send`): prints `queued — sends when the current turn + ends` vs `sent` today. `--json` passes `queued` through. + +## Acceptance criteria + +- Send while `working` returns `{ queued: true }`, changes no status, + emits the queued event, and the message appears as pending in `get`. +- When the turn ends (completed or failed), the pending message starts + exactly one new turn with the normal `turn.started` flow; no duplicate + turns under overlapping completions (lock-guarded dispatch AND enqueue). +- Second send while occupied overwrites: one queued event per send, one + pending message, newest text wins. +- `stop`/`release` cancel pending on every exit path (slot empty after, + no phantom "na fila"); shutdown keeps it (slot present after restart). +- Boot policy (explicit): reattached-`working`-with-pending dispatches via + the normal tail; `failed`/`orphaned`/`interrupted`-with-pending do NOT + auto-spend tokens on boot — the slot stays visible and the next manual + `send` (admission rule 5) or an explicit opt-in resumes it. If product + wants boot auto-dispatch for terminal-with-pending, state it here; it + is new token-spending behavior, not a bug fix. +- `origin=open` never queues under any status; error stays + `CAPABILITY_NOT_SUPPORTED` (except empty message → validation error + per precedence). +- Oversize/empty messages are rejected the same way whether the session + is busy or resumable; nothing oversize is ever queued. +- Dispatch failure restores the slot + emits a dispatch-failed event; + no silent pending-vanish, no double-send after crash. + +## Out of scope + +- Interrupting the live turn to send now (`send --interrupt`); this spec + never kills a running harness. +- PTY/stdin injection into headless sessions; no per-session pty, no log + format change, parsers untouched. +- Archiving `interrupted` sessions (`release` in canvas/CLI); tracked + separately. +- Multi-message FIFO, priorities, per-run queues (overwrite covers edit). + +## Risks + +- Dispatch races overlapping stream tails → hold `sessionLocks` across + check-clear-start AND busy-check-enqueue; the existing terminal-frame + discard under lock (`daemon.ts:1059`) is the pattern to follow. +- Failed-turn auto-dispatch could loop user-visible retries → acceptable: + one queued message yields at most one extra turn, then the slot is empty. +- Stale `nativeSessionId` at dispatch → slot restored + dispatch-failed + event, never silent loss. diff --git a/src/cli/commands/send.ts b/src/cli/commands/send.ts index 013d440..525e660 100644 --- a/src/cli/commands/send.ts +++ b/src/cli/commands/send.ts @@ -19,6 +19,7 @@ export function registerSendCommand(program: Command): void { process.exit(1); } if (opts.json) console.log(JSON.stringify(result, null, 2)); + else if (result && result.queued) console.log(`Message queued for ${id} — sends when the current turn ends`); else console.log(`Message sent to ${id}`); }); } diff --git a/src/cli/commands/web.ts b/src/cli/commands/web.ts index 2df7ff8..8b65e16 100644 --- a/src/cli/commands/web.ts +++ b/src/cli/commands/web.ts @@ -68,11 +68,11 @@ export function parseSendBody(body: unknown): string | null { return trimmed === "" ? null : trimmed; } -/** Daemon error code → HTTP status for the send proxy. */ function sendErrorStatus(code: string | undefined): number { if (code === "SESSION_NOT_FOUND") return 404; if (code === "SESSION_BUSY") return 409; if (code === "CAPABILITY_NOT_SUPPORTED") return 400; + if (code === "INVALID") return 400; return 502; } diff --git a/src/core/events.ts b/src/core/events.ts index c1778b3..53e6a32 100644 --- a/src/core/events.ts +++ b/src/core/events.ts @@ -6,6 +6,7 @@ export type AgentEventType = | "turn.started" | "text.delta" | "message" + | "message.queued" | "tool.started" | "tool.completed" | "file.changed" @@ -62,6 +63,12 @@ export interface ToolStartedEvent extends BaseAgentEvent { }; } +export interface MessageQueuedEvent extends BaseAgentEvent { + type: "message.queued"; + prompt: string; + pendingAt: string; +} + export interface ToolCompletedEvent extends BaseAgentEvent { type: "tool.completed"; tool: { @@ -137,6 +144,7 @@ export type AgentEvent = | TurnStartedEvent | TextDeltaEvent | MessageEvent + | MessageQueuedEvent | ToolStartedEvent | ToolCompletedEvent | FileChangedEvent diff --git a/src/core/session.ts b/src/core/session.ts index cc809d9..b743030 100644 --- a/src/core/session.ts +++ b/src/core/session.ts @@ -65,6 +65,10 @@ export interface Session { usage?: SessionUsage; lastEvent?: string; failure?: FailureInfo; + // One-slot send queue: message waiting for the next turn, set by + // session.send while busy (last-wins), cleared on dispatch/stop/release. + pendingMessage?: string | null; + pendingAt?: string | null; // Byte offsets into the session's log files (see drivers/tailer.ts) after // the last fully persisted line, so a reattaching daemon does not replay // events already in the store. diff --git a/src/daemon/daemon.ts b/src/daemon/daemon.ts index e7a6b23..d01fd97 100644 --- a/src/daemon/daemon.ts +++ b/src/daemon/daemon.ts @@ -53,6 +53,30 @@ function withLiveStatus(s: Session): Session { return status === s.status ? s : { ...s, status: status as Session["status"] }; } +// Send-queue cap: bytes, matching MAX_SEND_BODY_BYTES on the web route. +const MAX_SEND_MESSAGE_BYTES = 64 * 1024; + +function validateSendMessage(raw: unknown): { message: string } | { code: string; error: string } { + if (typeof raw !== "string") return { code: "INVALID", error: "message required" }; + const message = raw.trim(); + if (!message) return { code: "INVALID", error: "message required" }; + if (Buffer.byteLength(message, "utf8") > MAX_SEND_MESSAGE_BYTES) { + return { code: "INVALID", error: "message too large" }; + } + return { message }; +} + +// PID identity triple, same rule as the stop path: a recycled PID must +// neither block a legitimate resume nor permit killing a stranger. +function livePidIdentity(s: Session): boolean { + return ( + s.pid != null && + s.pidStartTime != null && + processAlive(s.pid) && + processStartTime(s.pid) === s.pidStartTime + ); +} + class Daemon { private db: Database; private sessions: SessionStore; @@ -487,9 +511,12 @@ class Daemon { send({ error: { code: "SESSION_NOT_FOUND", message: `Session ${p.id} not found` } }); return; } - + // Release owns the terminal outcome: a queued message must not + // survive as a phantom "na fila" on every exit, including the + // terminal early return below. + this.clearPending(s.id); if (isTerminalStatus(s.status)) { - send({ result: { session: s } }); + send({ result: { session: this.sessions.get(s.id)! } }); return; } @@ -559,6 +586,12 @@ class Daemon { case "session.send": { const p = params as { id: string; message: string }; + const validated = validateSendMessage(p.message); + if ("code" in validated) { + send({ error: { code: validated.code, message: validated.error } }); + return; + } + const message = validated.message; const s = this.sessions.get(p.id); if (!s) { send({ error: { code: "SESSION_NOT_FOUND", message: `Session ${p.id} not found` } }); return; } if (s.origin === "open") { @@ -566,113 +599,75 @@ class Daemon { return; } const driver = this.registry.get(s.agent); - // Do not start a second harness while the current one is still live: - // both runtimes would tail the same per-session file and duplicate - // every event from the follow-up turn. - if (this.sessionLocks.has(s.id)) { - send({ error: { code: "SESSION_BUSY", message: `Session ${s.id} has another lifecycle operation in progress` } }); + if (!driver.capabilities().resume) { + send({ error: { code: "CAPABILITY_NOT_SUPPORTED", message: `Agent ${s.agent} does not support resume` } }); return; } - if (s.status === "interrupted") { - // Resume-turn admission for power-interrupted sessions: capability - // BEFORE liveness, and liveness by process identity (same rule as - // stop). A recycled PID must not block a legitimate resume. - if (!s.nativeSessionId || !driver.capabilities().resume) { - send({ error: { code: "CAPABILITY_NOT_SUPPORTED", message: `Session ${s.id} cannot resume (no native session id or agent ${s.agent} does not support resume)` } }); return; - } - const liveIdentity = - s.pid != null && - s.pidStartTime != null && - processAlive(s.pid) && - processStartTime(s.pid) === s.pidStartTime; - if (liveIdentity) { - send({ error: { code: "SESSION_BUSY", message: `Session ${s.id} is still running (stop it first)` } }); - return; - } - } else { - const handle = driver.getHandle?.(s.id); - const runtimeState = handle && typeof handle === "object" - ? handle as { done?: boolean; drained?: boolean } - : undefined; - const runtimeDraining = handle !== undefined && (runtimeState?.done !== true || runtimeState?.drained !== true); - if (s.status === "starting" || runtimeDraining || (s.status === "working" && s.pid != null && processAlive(s.pid))) { - send({ error: { code: "SESSION_BUSY", message: `Session ${s.id} is still running` } }); - return; - } - if (!driver.capabilities().resume) { - send({ error: { code: "CAPABILITY_NOT_SUPPORTED", message: `Agent ${s.agent} does not support resume` } }); return; - } + if (s.status === "interrupted" && !s.nativeSessionId) { + send({ error: { code: "CAPABILITY_NOT_SUPPORTED", message: `Session ${s.id} cannot resume (no native session id or agent ${s.agent} does not support resume)` } }); + return; } - - const drvSession: DriverSession = { - id: s.id, - nativeSessionId: s.nativeSessionId, - cwd: s.worktree || s.cwd, - model: s.model, - effort: s.effort, - fast: s.fast, - sandbox: s.sandbox, - dangerouslyBypassApprovalsAndSandbox: s.dangerouslyBypassApprovalsAndSandbox, - pid: s.pid, - pidStartTime: s.pidStartTime, - }; - this.sessionLocks.add(s.id); - try { - if (this.shuttingDown) { - send({ error: { code: "SERVICE_UNAVAILABLE", message: "daemon is shutting down" } }); + // Power-interrupted sessions with a live process keep the old rule: + // stop first, never queue behind a running harness. + if (s.status === "interrupted" && livePidIdentity(s)) { + send({ error: { code: "SESSION_BUSY", message: `Session ${s.id} is still running (stop it first)` } }); + return; + } + // Do not start a second harness while the current one is still live: + // both runtimes would tail the same per-session file and duplicate + // every event from the follow-up turn. + const busyHandle = driver.getHandle?.(s.id); + const busyState = busyHandle && typeof busyHandle === "object" + ? busyHandle as { done?: boolean; drained?: boolean } + : undefined; + const runtimeBusy = busyHandle !== undefined && (busyState?.done !== true || busyState?.drained !== true); + const busy = this.sessionLocks.has(s.id) || s.status === "starting" || runtimeBusy || livePidIdentity(s); + if (busy) { + // One-slot queue, last-wins: persist without holding the lifecycle + // lock, then converge — the turn may have ended between the busy + // read and this persist, in which case tryDispatch starts now. + const pendingAt = new Date().toISOString(); + try { + this.sessions.update(s.id, { pendingMessage: message, pendingAt }); + } catch (e) { + send({ error: { code: "SEND_FAILED", message: e instanceof Error ? e.message : String(e) } }); return; } - this.sessions.setStatus(s.id, "working", { lastEvent: `send: ${p.message.slice(0, 80)}` }); - - const turnEvent: AgentEvent = { - type: "turn.started", + const queuedEvent: AgentEvent = { + type: "message.queued", sessionId: s.id, - timestamp: new Date().toISOString(), - prompt: p.message, + timestamp: pendingAt, + prompt: message, + pendingAt, }; - this.events.append(s.id, turnEvent); - this.broadcast(s.id, turnEvent); - - await driver.send(drvSession, p.message); - if (this.shuttingDown) { - try { await driver.stop(drvSession); } catch {} - send({ error: { code: "SERVICE_UNAVAILABLE", message: "daemon is shutting down" } }); - return; - } - - const handle = driver.getHandle?.(s.id); - const handleNativeId = - handle && - typeof handle === "object" && - "nativeSessionId" in handle && - typeof handle.nativeSessionId === "string" - ? handle.nativeSessionId - : undefined; - const newNative = handleNativeId || drvSession.nativeSessionId; - if (newNative && newNative !== s.nativeSessionId) { - this.sessions.update(s.id, { nativeSessionId: newNative }); - } - if (drvSession.pid) { - this.sessions.update(s.id, { - pid: drvSession.pid, - pidStartTime: processStartTime(drvSession.pid), - }); - } - this.sessions.update(s.id, { status: "working" }); + try { this.events.append(s.id, queuedEvent); } catch {} + this.broadcast(s.id, queuedEvent); + try { this.sessions.update(s.id, { lastEvent: `queued: ${message.slice(0, 80)}` }); } catch {} + void this.tryDispatch(s.id); + send({ result: { ok: true, queued: true } }); + return; + } - // Attach event loop for new turn - this.attachDriverEvents(s.id, driver, drvSession).catch(() => {}); - send({ result: { ok: true } }); + // Resumable now (terminal completed/failed/stopped/orphaned, or + // interrupted without a live process): a stale queued slot loses to + // the new message, which starts immediately. + this.sessionLocks.add(s.id); + try { + try { this.sessions.update(s.id, { pendingMessage: null, pendingAt: null }); } catch {} + await this.runResumeTurn(s, message); + send({ result: { ok: true, queued: false } }); } catch (e) { - const message = e instanceof Error ? e.message : String(e); - if (this.shuttingDown) { - try { await driver.stop(drvSession); } catch {} - send({ error: { code: "SERVICE_UNAVAILABLE", message: "daemon is shutting down" } }); + const errText = e instanceof Error ? e.message : String(e); + const code = e instanceof Error && (e as NodeJS.ErrnoException).code === "SERVICE_UNAVAILABLE" + ? "SERVICE_UNAVAILABLE" + : this.shuttingDown ? "SERVICE_UNAVAILABLE" : "SEND_FAILED"; + if (code === "SERVICE_UNAVAILABLE") { + send({ error: { code, message: errText } }); return; } try { this.sessions.setStatus(s.id, "failed"); } catch {} - send({ error: { code: "SEND_FAILED", message } }); + send({ error: { code: "SEND_FAILED", message: errText } }); } finally { if (!this.shuttingDown) this.sessionLocks.delete(s.id); } @@ -683,6 +678,9 @@ class Daemon { const p = params as { id: string }; const s = this.sessions.get(p.id); if (!s) { send({ error: { code: "SESSION_NOT_FOUND", message: `Session ${p.id} not found` } }); return; } + // Stop cancels a queued message on every exit (explicit cancel), + // including the BUSY/NOT_RUNNING/UNSAFE early returns below. + this.clearPending(s.id); const driver = this.registry.get(s.agent); if (this.sessionLocks.has(s.id)) { send({ error: { code: "SESSION_BUSY", message: `Session ${s.id} has another lifecycle operation in progress` } }); @@ -1051,6 +1049,141 @@ class Daemon { await this.attachDriverEvents(sessionId, driver, drvSession); } + /** Best-effort queue cancel; stop/release own the terminal outcome. */ + private clearPending(sessionId: string): void { + try { this.sessions.update(sessionId, { pendingMessage: null, pendingAt: null }); } catch {} + } + + /** + + * Starts a resumed turn; caller MUST hold sessionLocks for the session. + + * Shared by session.send (immediate) and tryDispatch (queued) so both + + * paths emit the same frames in the same order. + + */ + private async runResumeTurn(s: Session, message: string): Promise { + const driver = this.registry.get(s.agent); + if (this.shuttingDown) { + const early = new Error("daemon is shutting down") as NodeJS.ErrnoException; + early.code = "SERVICE_UNAVAILABLE"; + throw early; + } + this.sessions.setStatus(s.id, "working", { lastEvent: `send: ${message.slice(0, 80)}` }); + const turnEvent: AgentEvent = { + type: "turn.started", + sessionId: s.id, + timestamp: new Date().toISOString(), + prompt: message, + }; + this.events.append(s.id, turnEvent); + this.broadcast(s.id, turnEvent); + const drvSession: DriverSession = { + id: s.id, + nativeSessionId: s.nativeSessionId, + cwd: s.worktree || s.cwd, + model: s.model, + effort: s.effort, + fast: s.fast, + sandbox: s.sandbox, + dangerouslyBypassApprovalsAndSandbox: s.dangerouslyBypassApprovalsAndSandbox, + pid: s.pid, + pidStartTime: s.pidStartTime, + }; + try { + await driver.send(drvSession, message); + } catch (e) { + if (this.shuttingDown) { try { await driver.stop(drvSession); } catch {} } + throw e; + } + if (this.shuttingDown) { + try { await driver.stop(drvSession); } catch {} + const late = new Error("daemon is shutting down") as NodeJS.ErrnoException; + late.code = "SERVICE_UNAVAILABLE"; + throw late; + } + const handle = driver.getHandle?.(s.id); + const handleNativeId = + handle && + typeof handle === "object" && + "nativeSessionId" in handle && + typeof handle.nativeSessionId === "string" + ? handle.nativeSessionId + : undefined; + const newNative = handleNativeId || drvSession.nativeSessionId; + if (newNative && newNative !== s.nativeSessionId) { + this.sessions.update(s.id, { nativeSessionId: newNative }); + } + if (drvSession.pid) { + this.sessions.update(s.id, { + pid: drvSession.pid, + pidStartTime: processStartTime(drvSession.pid), + }); + } + this.sessions.update(s.id, { status: "working" }); + // Attach event loop for new turn + this.attachDriverEvents(s.id, driver, drvSession).catch(() => {}); + } + + /** + + * Starts the queued message as a new turn when the session is resumable. + + * Never blocks: returns false leaving the slot intact when busy, + + * unresumable, or shutting down. The stream tail calls this, and the + + * busy send path calls it to converge on races (enqueue landing as the + + * turn ends starts immediately instead of stranding). + + */ + private async tryDispatch(sessionId: string): Promise { + if (this.shuttingDown || this.sessionLocks.has(sessionId)) return false; + const queued = this.sessions.get(sessionId); + if (!queued || !queued.pendingMessage || queued.origin === "open") return false; + this.sessionLocks.add(sessionId); + let message: string | undefined; + let pendingAt: string | null | undefined; + try { + const s = this.sessions.get(sessionId); + if (!s || !s.pendingMessage || s.origin === "open") return false; + const driver = this.registry.get(s.agent); + if (!driver.capabilities().resume) return false; + const handle = driver.getHandle?.(s.id); + const hs = handle && typeof handle === "object" + ? handle as { done?: boolean; drained?: boolean } + : undefined; + if (s.status === "starting") return false; + if (handle !== undefined && (hs?.done !== true || hs?.drained !== true)) return false; + if (livePidIdentity(s)) return false; + message = s.pendingMessage; + pendingAt = s.pendingAt; + // Clear first so a crash mid-start cannot double-send; restored below + // when the start fails. + this.sessions.update(s.id, { pendingMessage: null, pendingAt: null }); + await this.runResumeTurn(s, message); + return true; + } catch (e) { + const errText = e instanceof Error ? e.message : String(e); + const unavailable = (e as NodeJS.ErrnoException)?.code === "SERVICE_UNAVAILABLE" || this.shuttingDown; + if (!unavailable) { + const failure = classifyFailure(errText); + const errEv: AgentEvent = { + type: "session.failed", + sessionId, + timestamp: new Date().toISOString(), + error: errText, + failure, + }; + try { this.events.append(sessionId, errEv); } catch {} + this.broadcast(sessionId, errEv); + try { this.sessions.setStatus(sessionId, "failed", { lastEvent: errText.slice(0, 200), failure }); } catch {} + } + // Restore the slot so a failed dispatch never vanishes silently; the + // next manual send (or a future tail) retries it. + try { + if (message !== undefined) { + this.sessions.update(sessionId, { pendingMessage: message, pendingAt: pendingAt ?? new Date().toISOString() }); + } + } catch {} + return false; + } finally { + if (!this.shuttingDown) this.sessionLocks.delete(sessionId); + } + } + private async attachDriverEvents(sessionId: string, driver: AgentDriver, drvSession: DriverSession): Promise { try { for await (const ev of driver.events(drvSession)) { @@ -1131,6 +1264,10 @@ class Daemon { this.sessions.setStatus(sessionId, "failed", { lastEvent: error.slice(0, 200), failure }); } } + // A message queued while the turn ran starts now as the next turn. + // tryDispatch rechecks resumability under the lifecycle lock; a stale + // or unresumable slot stays put for a manual send. + if (!this.shuttingDown) void this.tryDispatch(sessionId); } private updateSessionFromEvent(sessionId: string, ev: AgentEvent): void { @@ -1147,6 +1284,8 @@ class Daemon { this.sessions.setStatus(sessionId, "failed", { lastEvent: ev.error.slice(0, 200), failure: ev.failure }); } else if (ev.type === "tool.started") { this.sessions.update(sessionId, { lastEvent: `tool: ${ev.tool.name}` }); + } else if (ev.type === "message.queued") { + this.sessions.update(sessionId, { lastEvent: `queued: ${ev.prompt.slice(0, 80)}` }); } else if (ev.type === "message") { this.sessions.update(sessionId, { lastEvent: ev.content.slice(0, 80) }); } else if (ev.type === "usage.updated") { diff --git a/src/store/database.ts b/src/store/database.ts index f83652a..aa379ae 100644 --- a/src/store/database.ts +++ b/src/store/database.ts @@ -64,6 +64,8 @@ export class Database { usage_cached_tokens INTEGER, usage_cost REAL, last_event TEXT, + pending_message TEXT, + pending_at TEXT, effort TEXT, fast INTEGER NOT NULL DEFAULT 0, sandbox TEXT, @@ -120,6 +122,8 @@ export class Database { ["pid_start_time", "TEXT"], ["run_id", "TEXT"], ["origin", "TEXT"], + ["pending_message", "TEXT"], + ["pending_at", "TEXT"], ]; for (const [name, type] of additions) { if (!existing.has(name)) this.db.exec(`ALTER TABLE sessions ADD COLUMN ${name} ${type}`); diff --git a/src/store/sessions.ts b/src/store/sessions.ts index 56bbf82..398cd7c 100644 --- a/src/store/sessions.ts +++ b/src/store/sessions.ts @@ -35,6 +35,8 @@ export interface SessionRow { log_offset: number | null; stderr_offset: number | null; origin: string | null; + pending_message: string | null; + pending_at: string | null; } @@ -87,6 +89,8 @@ function rowToSession(row: SessionRow): Session { : undefined, lastEvent: row.last_event ?? undefined, failure, + pendingMessage: row.pending_message ?? undefined, + pendingAt: row.pending_at ?? undefined, logOffset: row.log_offset ?? undefined, stderrOffset: row.stderr_offset ?? undefined, }; @@ -106,11 +110,12 @@ export class SessionStore { pid_start_time, created_at, updated_at, completed_at, usage_input_tokens, usage_output_tokens, usage_cached_tokens, usage_cost, last_event, effort, fast, sandbox, dangerously_bypass_approvals_and_sandbox, failure, log_offset, stderr_offset, - run_id, origin + run_id, origin, pending_message, pending_at ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, - ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, + ?, ? ) `); stmt.run( @@ -144,6 +149,8 @@ export class SessionStore { session.stderrOffset ?? null, session.runId ?? null, session.origin ?? null, + session.pendingMessage ?? null, + session.pendingAt ?? null, ); } @@ -232,6 +239,8 @@ export class SessionStore { stderr_offset: patch.stderrOffset, failure: patch.failure === undefined ? undefined : JSON.stringify(patch.failure), origin: patch.origin, + pending_message: patch.pendingMessage === undefined ? undefined : (patch.pendingMessage ?? null), + pending_at: patch.pendingAt === undefined ? undefined : (patch.pendingAt ?? null), }; for (const [col, val] of Object.entries(map)) { diff --git a/src/web/canvas-page.ts b/src/web/canvas-page.ts index 8371d9b..04e4198 100644 --- a/src/web/canvas-page.ts +++ b/src/web/canvas-page.ts @@ -562,6 +562,12 @@ function onEvent(n, ev) { touch(n, "respondendo"); feed(agentName(n.agent), String(ev.content || "").slice(0, 90)); } + } else if (ev.type === "message.queued") { + touch(n, "mensagem na fila"); + feed(agentName(n.agent), "mensagem na fila"); + if (n.id && n.id === detailId) { detailPending = ev.prompt || true; paintSendState(); } + } else if (ev.type === "turn.started") { + if (n.id && n.id === detailId && detailPending) { detailPending = null; paintSendState(); } } else if (ev.type === "turn.completed") { feed(agentName(n.agent), "etapa concluída"); } else if (ev.type === "file.changed") { @@ -839,7 +845,7 @@ document.getElementById("btnHide").title = hideDone ? "Mostrar concluídas" : "O (o daemon anexa no turno inicial e a cada send), a resposta em message(role=assistant) com content completo. Ferramentas viram linhas compactas para a conversa continuar legível. */ -var detailId = null, detailStatus = null, detailOrigin = null; +var detailId = null, detailStatus = null, detailOrigin = null, detailPending = null; var sendInFlight = false, sendBlockedNote = false; function chatMsg(who, text, cls) { var d = document.createElement("div"); @@ -864,12 +870,14 @@ function paintSendState() { var btn = document.getElementById("dSendBtn"); var msg = document.getElementById("dSendMsg"); if (!detailId) return; + // origin=open (TUI interativa) trava sempre; working/starting entra na + // fila em vez de travar. Mensagem de sucesso não é apagada aqui. var reason = ""; if (detailOrigin === "open") reason = "sessão interativa não aceita mensagens"; - else if (detailStatus === "working" || detailStatus === "starting") reason = "aguarde o turno atual terminar"; input.disabled = sendInFlight || reason !== ""; btn.disabled = sendInFlight || reason !== ""; if (reason) { msg.textContent = reason; sendBlockedNote = true; } + else if (detailPending) { msg.textContent = "1 mensagem na fila — envia quando o turno terminar"; sendBlockedNote = true; } else if (sendBlockedNote) { msg.textContent = ""; sendBlockedNote = false; } } function row(dt, dd) { @@ -883,6 +891,7 @@ function closeDetail() { detailId = null; detailStatus = null; detailOrigin = null; + detailPending = null; document.getElementById("detail").classList.remove("open"); } document.getElementById("detailClose").onclick = closeDetail; @@ -894,6 +903,7 @@ function selectSession(id) { detailId = s.id; detailStatus = s.status; detailOrigin = s.origin || null; + detailPending = s.pendingMessage || null; var title = document.getElementById("dTitle"); title.innerHTML = ""; var logo = document.createElement("span"); @@ -936,6 +946,8 @@ function selectSession(id) { if (!ev || !ev.type) continue; if (ev.type === "turn.started" && ev.prompt) { chat.appendChild(chatMsg("você", String(ev.prompt), "user")); + } else if (ev.type === "message.queued" && ev.prompt) { + chat.appendChild(chatMsg("você", String(ev.prompt) + " (na fila)", "user")); } else if (ev.type === "message" && ev.role === "assistant" && ev.content) { chat.appendChild(chatMsg(agentName(s.agent), String(ev.content), "assistant")); } else if (ev.type === "tool.started") { @@ -981,7 +993,13 @@ document.getElementById("dSend").addEventListener("submit", function (ev) { }).then(function (out) { if (out.ok) { input.value = ""; - msg.textContent = "mensagem enviada — nova etapa começou"; + if (out.body && out.body.queued) { + detailPending = text; + msg.textContent = "mensagem na fila — envia quando o turno terminar"; + } else { + detailPending = null; + msg.textContent = "mensagem enviada — nova etapa começou"; + } poll(); } else { msg.textContent = (out.body && out.body.error) || "não deu para enviar"; diff --git a/tests/power-send.test.ts b/tests/power-send.test.ts index 81f39b0..105e280 100644 --- a/tests/power-send.test.ts +++ b/tests/power-send.test.ts @@ -57,6 +57,7 @@ interface SendCase { message: string; error?: string; sent: number; + queued?: boolean; then?: (daemon: Daemon, sent: SentCall[]) => void; } @@ -106,14 +107,17 @@ const cases: SendCase[] = [ }, }, { - title: "keeps the working busy check for non-interrupted sessions", + title: "queues behind a live turn for non-interrupted sessions", id: "s-working", status: "working", resume: true, extra: { nativeSessionId: "n-4", ...livePid }, message: "more work", - error: "SESSION_BUSY", sent: 0, + queued: true, + then: (daemon) => { + expect(seam(daemon).sessions.get("s-working")?.pendingMessage).toBe("more work"); + }, }, ]; @@ -125,10 +129,10 @@ describe("power send admission for interrupted", () => { seed(daemon, c.id, c.status ?? "interrupted", c.extra ?? {}); const res = await sendMessage(daemon, c.id, c.message); - if (c.error === undefined) expect(res.error).toBeUndefined(); else expect(res.error?.code).toBe(c.error); expect(sent).toHaveLength(c.sent); + if (c.queued !== undefined) expect((res.result as { queued?: boolean } | undefined)?.queued).toBe(c.queued); c.then?.(daemon, sent); }); }); diff --git a/tests/send-queue.test.ts b/tests/send-queue.test.ts new file mode 100644 index 0000000..14448b9 --- /dev/null +++ b/tests/send-queue.test.ts @@ -0,0 +1,105 @@ +import { describe, expect, it, beforeEach, afterEach } from "vitest"; +import { Daemon } from "../src/daemon/daemon.js"; +import { processStartTime } from "../src/utils/process.js"; +import { makeTempDir, removeTempDir, seam, seed, fakeSocket } from "./helpers/daemon-seam.js"; + +let dir: string; + +beforeEach(() => { + dir = makeTempDir("send-queue-"); + process.env.RUN_AGENT_DIR = dir; +}); + +afterEach(() => { + delete process.env.RUN_AGENT_DIR; + removeTempDir(dir); +}); + +async function send(daemon: Daemon, id: string, message: unknown): Promise { + const { writes, socket } = fakeSocket(); + await seam(daemon).handleRequest({ id: "r1", method: "session.send", params: { id, message } }, socket); + return JSON.parse(writes[0]); +} + +async function stop(daemon: Daemon, id: string): Promise { + const { writes, socket } = fakeSocket(); + await seam(daemon).handleRequest({ id: "r1", method: "session.stop", params: { id } }, socket); + return JSON.parse(writes[0]); +} + +describe("send queue", () => { + it("queues while starting without changing status", async () => { + const daemon = new Daemon(); + seed(daemon, "s-busy", "starting", { origin: "run" }); + const body = await send(daemon, "s-busy", "second thought"); + expect(body.result).toEqual({ ok: true, queued: true }); + const stored = seam(daemon).sessions.get("s-busy")!; + expect(stored.status).toBe("starting"); + expect(stored.pendingMessage).toBe("second thought"); + expect(stored.pendingAt).toBeTruthy(); + const types = seam(daemon).events.list("s-busy", 10).map((e) => e.type); + expect(types).toContain("message.queued"); + expect(stored.lastEvent).toMatch(/^queued: /); + }); + + it("last send wins the single slot", async () => { + const daemon = new Daemon(); + seed(daemon, "s-busy", "starting", { origin: "run" }); + await send(daemon, "s-busy", "first"); + await send(daemon, "s-busy", "second"); + expect(seam(daemon).sessions.get("s-busy")?.pendingMessage).toBe("second"); + }); + + it("rejects empty and oversize messages before admission", async () => { + const daemon = new Daemon(); + seed(daemon, "s-busy", "starting", { origin: "run" }); + expect((await send(daemon, "s-busy", " ")).error?.code).toBe("INVALID"); + expect((await send(daemon, "s-busy", "x".repeat(65 * 1024 + 1))).error?.code).toBe("INVALID"); + expect(seam(daemon).sessions.get("s-busy")?.pendingMessage).toBeUndefined(); + }); + + it("never queues interactive sessions", async () => { + const daemon = new Daemon(); + seed(daemon, "s-open", "starting", { origin: "open" }); + const body = await send(daemon, "s-open", "hello?"); + expect(body.error?.code).toBe("CAPABILITY_NOT_SUPPORTED"); + expect(seam(daemon).sessions.get("s-open")?.pendingMessage).toBeUndefined(); + }); + + it("keeps stop-first for interrupted sessions with a live process", async () => { + const daemon = new Daemon(); + seed(daemon, "s-int", "interrupted", { + origin: "run", + nativeSessionId: "n-live", + pid: process.pid, + pidStartTime: processStartTime(process.pid), + }); + const body = await send(daemon, "s-int", "hello?"); + expect(body.error?.code).toBe("SESSION_BUSY"); + expect(seam(daemon).sessions.get("s-int")?.pendingMessage).toBeUndefined(); + }); + + it("stop cancels pending even when there is nothing to stop", async () => { + const daemon = new Daemon(); + seed(daemon, "s-done", "completed", { + origin: "run", + pendingMessage: "stale", + pendingAt: new Date().toISOString(), + }); + const body = await stop(daemon, "s-done"); + expect(body.error?.code).toBe("SESSION_NOT_RUNNING"); + const stored = seam(daemon).sessions.get("s-done")!; + expect(stored.pendingMessage).toBeUndefined(); + expect(stored.pendingAt).toBeUndefined(); + }); + + it("exposes pending through session.get", async () => { + const daemon = new Daemon(); + seed(daemon, "s-busy", "starting", { origin: "run" }); + await send(daemon, "s-busy", "queued hello"); + const { writes, socket } = fakeSocket(); + await seam(daemon).handleRequest({ id: "r2", method: "session.get", params: { id: "s-busy" } }, socket); + const body = JSON.parse(writes[0]) as { result?: { session?: { pendingMessage?: string } } }; + expect(body.result?.session?.pendingMessage).toBe("queued hello"); + }); +}); From 3a248e581ed6a0dc117bf99db487ec0d0b57b498 Mon Sep 17 00:00:00 2001 From: 4ndreello <4ndreello@users.noreply.github.com> Date: Tue, 8 Sep 2026 12:07:17 -0300 Subject: [PATCH 2/3] =?UTF-8?q?fix(send-queue):=20address=20PR=20review=20?= =?UTF-8?q?=E2=80=94=20native=20gate,=20lock=20race,=20canvas=20hint,=20ta?= =?UTF-8?q?il=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tryDispatch returns with slot intact when no native id resolves - session.send returns SESSION_BUSY behind lifecycle locks instead of queueing; stop re-clears pending on success exit - canvas clears detailPending on terminal session events - tail-dispatch coverage: exactly-once turn start, quiet no-native leave, failure restore with honest event --- src/daemon/daemon.ts | 22 +++++++- src/web/canvas-page.ts | 4 +- tests/send-queue.test.ts | 115 +++++++++++++++++++++++++++++++++++++-- 3 files changed, 133 insertions(+), 8 deletions(-) diff --git a/src/daemon/daemon.ts b/src/daemon/daemon.ts index d01fd97..ba48d8c 100644 --- a/src/daemon/daemon.ts +++ b/src/daemon/daemon.ts @@ -613,15 +613,22 @@ class Daemon { send({ error: { code: "SESSION_BUSY", message: `Session ${s.id} is still running (stop it first)` } }); return; } + // A lifecycle operation owns the session: retry instead of queueing, + // so a send racing stop cannot strand a phantom slot with no stream + // tail left to consume it (stop cancels the queue on exit). + if (this.sessionLocks.has(s.id)) { + send({ error: { code: "SESSION_BUSY", message: `Session ${s.id} has another lifecycle operation in progress` } }); + return; + } // Do not start a second harness while the current one is still live: // both runtimes would tail the same per-session file and duplicate - // every event from the follow-up turn. + // every event from the follow-up turn. Queue behind a live turn only. const busyHandle = driver.getHandle?.(s.id); const busyState = busyHandle && typeof busyHandle === "object" ? busyHandle as { done?: boolean; drained?: boolean } : undefined; const runtimeBusy = busyHandle !== undefined && (busyState?.done !== true || busyState?.drained !== true); - const busy = this.sessionLocks.has(s.id) || s.status === "starting" || runtimeBusy || livePidIdentity(s); + const busy = s.status === "starting" || runtimeBusy || livePidIdentity(s); if (busy) { // One-slot queue, last-wins: persist without holding the lifecycle // lock, then converge — the turn may have ended between the busy @@ -743,6 +750,9 @@ class Daemon { this.events.append(s.id, ev); this.broadcast(s.id, ev); } + // A send may have persisted a slot after the entry clear (it saw a + // live turn before this stop took the lock): stop wins, clear again. + this.clearPending(s.id); send({ result: { ok: true } }); } catch (e) { if (this.shuttingDown) { @@ -1143,8 +1153,14 @@ class Daemon { if (!driver.capabilities().resume) return false; const handle = driver.getHandle?.(s.id); const hs = handle && typeof handle === "object" - ? handle as { done?: boolean; drained?: boolean } + ? handle as { done?: boolean; drained?: boolean; nativeSessionId?: unknown } : undefined; + // Same resolution as SessionDriver.send: without a native id the start + // would only throw after emitting turn.started and flipping status, so + // leave the slot quietly for a manual send instead. + const nativeId = s.nativeSessionId + || (typeof hs?.nativeSessionId === "string" ? hs.nativeSessionId : undefined); + if (!nativeId) return false; if (s.status === "starting") return false; if (handle !== undefined && (hs?.done !== true || hs?.drained !== true)) return false; if (livePidIdentity(s)) return false; diff --git a/src/web/canvas-page.ts b/src/web/canvas-page.ts index 04e4198..4758fa9 100644 --- a/src/web/canvas-page.ts +++ b/src/web/canvas-page.ts @@ -577,12 +577,14 @@ function onEvent(n, ev) { touch(n, "quer sua aprovação para continuar"); paintSummary(); spawnRing(n, "#ff9f0a"); - feed(agentName(n.agent), "pedindo sua aprovação"); } else if (ev.type === "session.completed" || ev.type === "session.failed") { n.status = ev.type === "session.failed" ? "failed" : "completed"; touch(n, ev.type === "session.failed" ? ("falhou: " + (ev.error || "")) : "concluída"); paintSummary(); feed(agentName(n.agent), ev.type === "session.failed" ? "falhou" : "concluída"); + // Stop/release cancels the queue server-side: drop a stale "na fila" + // hint on the open detail, mirroring the turn.started branch. + if (n.id && n.id === detailId && detailPending) { detailPending = null; paintSendState(); } if (n.es) { try { n.es.close(); } catch (e) {} n.es = null; } } } diff --git a/tests/send-queue.test.ts b/tests/send-queue.test.ts index 14448b9..dcecb85 100644 --- a/tests/send-queue.test.ts +++ b/tests/send-queue.test.ts @@ -1,5 +1,7 @@ import { describe, expect, it, beforeEach, afterEach } from "vitest"; import { Daemon } from "../src/daemon/daemon.js"; +import type { AgentDriver, DriverSession } from "../src/core/driver.js"; +import type { AgentEvent } from "../src/core/events.js"; import { processStartTime } from "../src/utils/process.js"; import { makeTempDir, removeTempDir, seam, seed, fakeSocket } from "./helpers/daemon-seam.js"; @@ -15,16 +17,21 @@ afterEach(() => { removeTempDir(dir); }); -async function send(daemon: Daemon, id: string, message: unknown): Promise { +interface IpcOutcome { + result?: { ok?: boolean; queued?: boolean }; + error?: { code?: string; message?: string }; +} + +async function send(daemon: Daemon, id: string, message: unknown): Promise { const { writes, socket } = fakeSocket(); await seam(daemon).handleRequest({ id: "r1", method: "session.send", params: { id, message } }, socket); - return JSON.parse(writes[0]); + return JSON.parse(writes[0]) as IpcOutcome; } -async function stop(daemon: Daemon, id: string): Promise { +async function stop(daemon: Daemon, id: string): Promise { const { writes, socket } = fakeSocket(); await seam(daemon).handleRequest({ id: "r1", method: "session.stop", params: { id } }, socket); - return JSON.parse(writes[0]); + return JSON.parse(writes[0]) as IpcOutcome; } describe("send queue", () => { @@ -103,3 +110,103 @@ describe("send queue", () => { expect(body.result?.session?.pendingMessage).toBe("queued hello"); }); }); + +interface TailTestSeam { + attachDriverEvents(sessionId: string, driver: AgentDriver, drvSession: DriverSession): Promise; +} + +function tailSeam(daemon: Daemon): TailTestSeam { + // Tests drive private lifecycle methods directly (same seam pattern). + return seam(daemon) as unknown as TailTestSeam; +} + +describe("tail dispatch", () => { + // The tail fires tryDispatch fire-and-forget; the whole chain is + // microtasks plus synchronous sqlite, so drain the queue instead of + // sleeping on the wall clock. + async function flushWhile(more: () => boolean, budget = 500): Promise { + for (let i = 0; i < budget && more(); i++) await Promise.resolve(); + } + + function completedEvent(sessionId: string): AgentEvent { + return { type: "session.completed", sessionId, timestamp: new Date().toISOString(), reason: "done" }; + } + + function installQueueDriver(daemon: Daemon, sent: { message: string }[], failSend = false): AgentDriver { + const driver = { + id: "claude", + capabilities: () => ({ streaming: true, resume: true }), + send: async (_session: unknown, message: string) => { + if (failSend) throw new Error("resume gone"); + sent.push({ message }); + }, + getHandle: () => undefined, + events: async function* () { + yield completedEvent("x"); + }, + }; + seam(daemon).registry.register(driver); + // Structurally the daemon only touches the fields above in this flow. + return driver as unknown as AgentDriver; + } + + function drvSession(id: string, nativeSessionId?: string): DriverSession { + return { id, nativeSessionId, cwd: "/tmp" }; + } + + it("starts the queued message as exactly one turn when the stream ends", async () => { + const daemon = new Daemon(); + const sent: { message: string }[] = []; + const driver = installQueueDriver(daemon, sent); + seed(daemon, "s-tail", "working", { + origin: "run", + nativeSessionId: "n-1", + pendingMessage: "queued hello", + pendingAt: new Date().toISOString(), + }); + await tailSeam(daemon).attachDriverEvents("s-tail", driver, drvSession("s-tail", "n-1")); + await flushWhile(() => sent.length === 0); + expect(sent).toHaveLength(1); + expect(sent[0].message).toBe("queued hello"); + await flushWhile(() => seam(daemon).sessions.get("s-tail")?.pendingMessage != null); + const all = seam(daemon).events.list("s-tail", 20); + const starts = all.filter((e) => e.type === "turn.started"); + expect(starts).toHaveLength(1); + expect(starts[0].type === "turn.started" ? starts[0].prompt : undefined).toBe("queued hello"); + }); + + it("leaves the slot quietly when no native id can be resolved", async () => { + const daemon = new Daemon(); + const sent: { message: string }[] = []; + const driver = installQueueDriver(daemon, sent); + seed(daemon, "s-nonative", "working", { + origin: "run", + pendingMessage: "waits for manual send", + pendingAt: new Date().toISOString(), + }); + await tailSeam(daemon).attachDriverEvents("s-nonative", driver, drvSession("s-nonative")); + await flushWhile(() => false); + expect(sent).toHaveLength(0); + expect(seam(daemon).sessions.get("s-nonative")?.pendingMessage).toBe("waits for manual send"); + const types = seam(daemon).events.list("s-nonative", 20).map((e) => e.type); + expect(types).not.toContain("turn.started"); + }); + + it("restores the slot with an honest event when dispatch fails", async () => { + const daemon = new Daemon(); + const sent: { message: string }[] = []; + const driver = installQueueDriver(daemon, sent, true); + seed(daemon, "s-fail", "working", { + origin: "run", + nativeSessionId: "n-9", + pendingMessage: "doomed", + pendingAt: new Date().toISOString(), + }); + await tailSeam(daemon).attachDriverEvents("s-fail", driver, drvSession("s-fail", "n-9")); + await flushWhile( + () => !seam(daemon).events.list("s-fail", 20).some((e) => e.type === "session.failed" && e.error === "resume gone"), + ); + expect(sent).toHaveLength(0); + expect(seam(daemon).sessions.get("s-fail")?.pendingMessage).toBe("doomed"); + }); +}); From fe06d8ac8cf7a8ac2bc611f5c36dba82a7c81d8c Mon Sep 17 00:00:00 2001 From: 4ndreello <4ndreello@users.noreply.github.com> Date: Tue, 8 Sep 2026 12:10:34 -0300 Subject: [PATCH 3/3] fix(send-queue): restore approval feed line, drain quiet tail test --- src/web/canvas-page.ts | 1 + tests/send-queue.test.ts | 4 +++- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/src/web/canvas-page.ts b/src/web/canvas-page.ts index 4758fa9..991d04a 100644 --- a/src/web/canvas-page.ts +++ b/src/web/canvas-page.ts @@ -577,6 +577,7 @@ function onEvent(n, ev) { touch(n, "quer sua aprovação para continuar"); paintSummary(); spawnRing(n, "#ff9f0a"); + feed(agentName(n.agent), "pedindo sua aprovação"); } else if (ev.type === "session.completed" || ev.type === "session.failed") { n.status = ev.type === "session.failed" ? "failed" : "completed"; touch(n, ev.type === "session.failed" ? ("falhou: " + (ev.error || "")) : "concluída"); diff --git a/tests/send-queue.test.ts b/tests/send-queue.test.ts index dcecb85..03d3c31 100644 --- a/tests/send-queue.test.ts +++ b/tests/send-queue.test.ts @@ -185,7 +185,9 @@ describe("tail dispatch", () => { pendingAt: new Date().toISOString(), }); await tailSeam(daemon).attachDriverEvents("s-nonative", driver, drvSession("s-nonative")); - await flushWhile(() => false); + // The no-native gate returns before any await, but drain a few ticks so + // the assertion covers the settled chain rather than a pending one. + await flushWhile(() => sent.length !== 0, 10); expect(sent).toHaveLength(0); expect(seam(daemon).sessions.get("s-nonative")?.pendingMessage).toBe("waits for manual send"); const types = seam(daemon).events.list("s-nonative", 20).map((e) => e.type);