diff --git a/AGENTS.md b/AGENTS.md index 5d69f990..9531c240 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -84,7 +84,8 @@ That makes the session key **derived, not secret**: anyone who learns the artifa A `status: "ended"` response carries `ended_by`; the final `status: "feedback"` batch delivered right before a session ends carries the same signal via `session_ended: true` plus `ended_by`, so that last `next_step` also skips the reopen instruction. Default no-timeout polls stream whitespace heartbeat bytes before the final JSON response and always write the one-shot wait banner to stderr - it is the "not hung" signal an agent needs while stdout stays empty - but write the recurring per-minute wait ticks only when stderr is an interactive terminal (`shouldNarratePollWaitTicks`), so agent harnesses with piped, non-TTY stdio get no unbounded tick noise in their merged capture; stdout is always reserved for the final JSON/TOON response, and `--timeout-ms` is a non-streaming test/debug escape hatch. If SIGINT or SIGTERM interrupts a no-timeout poll, the CLI writes re-run guidance to stderr and exits with the conventional signal code; queued feedback persists, so re-running the same poll is safe. -8. The `/events/:key` SSE stream emits `agent-presence` states: `waiting` before any poll has attached, `listening` while one is active, and `working` after a poll has delivered feedback and released; the chrome allows queued feedback while waiting or listening and blocks sends only while working. An agent reply (`POST /api/:key/agent-reply`, the CLI's `--agent-reply`) concludes the working state and returns presence to `waiting`, so sends re-enable as soon as the agent answers instead of staying blocked until another poll attaches. +8. The `/events/:key` SSE stream emits `agent-presence` states: `waiting` before any poll has attached, `listening` while one is active, and `working` after a poll has delivered feedback and released; the chrome keeps all feedback actions available because the server queues them for the next poll. An agent reply (`POST /api/:key/agent-reply`, the CLI's `--agent-reply`) concludes the working state and returns presence to `waiting`, and a batch delivered with `session_ended` releases it immediately because no later poll or reply can. + Every poll exit path has to undo the presence it set: `/api/poll` subscribes to the request's `close` before its first `await` and re-checks it after arming the listeners, so a client that disconnects while the immediate-feedback take is in flight never strands `listening`. `--agent-reply` posts a chat message into the session before polling, rendered in the browser conversation panel via the same stream. ### Passive layout-warning inbox diff --git a/README.md b/README.md index 01331043..2d678e4e 100644 --- a/README.md +++ b/README.md @@ -195,7 +195,7 @@ pnpm link - **Keyboard shortcuts** - In the chrome composer, Enter sends queued prompts and Shift+Enter inserts a newline. In the annotation card, Enter queues the annotation, Shift+Enter inserts a newline, and Ctrl+Enter (Cmd+Enter on macOS) queues it and sends all queued prompts immediately. Cmd+I or Ctrl+I toggles between annotate and explore mode from either the browser chrome or the artifact iframe, including while focus is in a textarea or control. -- **Agent presence** - The browser shows when no agent is listening, keeps queued feedback for the next successful `lavish-axi poll` send even across reloads, and only blocks human sends while the agent is working on delivered feedback; the agent's reply (`--agent-reply`) concludes that work and re-enables sends. +- **Agent presence** - The browser shows when no agent is listening, keeps queued feedback for the next successful `lavish-axi poll` send even across reloads, and keeps human feedback actions available while the agent is working because the server queues them for the next poll. The agent's reply (`--agent-reply`) concludes delivered work and returns presence to waiting. The no-timeout poll always writes an immediate stderr banner so it is visibly not hung; it adds the periodic stderr wait ticks only in an interactive terminal, so when stderr is piped (as under agent harnesses) the captured output carries no tick noise. Stdout always stays reserved for the final response; if the poll is interrupted or times out, re-run it because queued feedback is never lost. Codex-specific guidance keeps that poll attached to the active turn instead of hiding it in a background task, because completed background tasks may not resume the agent. - **Session end etiquette** - Lavish tracks who ended a session: a human clicking **End session** (or **Send & end session**) in the browser is a user-initiated end, while `lavish-axi end ` is agent-initiated. @@ -262,7 +262,7 @@ For flows, architecture, state, or sequence diagrams, open the diagram playbook | `lavish-axi export` | `--out ` | Write the export to a specific path instead of `.export.html` next to the source. | | `lavish-axi share` | `--password ` | Make the third-party ht-ml.app page private; viewers must supply the password. | | `lavish-axi share` | `--token ` | Attach an optional bearer token (`LAVISH_AXI_HTML_APP_TOKEN`); never required to publish. | -| `lavish-axi poll` | `--agent-reply "..."` | Show the agent's reply in the existing browser chat and re-enable human sends before polling again. | +| `lavish-axi poll` | `--agent-reply "..."` | Show the agent's reply in the existing browser chat, conclude delivered work, and return presence to waiting before polling again. | | `lavish-axi poll` | `--timeout-ms ` | Test/debug escape hatch only; agents should normally omit it and leave the long poll running. | | `lavish-axi stop` | `--port ` | Shut down a server running on a non-default port. | | `lavish-axi server` | `--verbose` | Log session and watcher events to stderr; can also be enabled with `LAVISH_AXI_DEBUG=1`. Detached server output is appended to `~/.lavish-axi/server.log` (or `LAVISH_AXI_STATE_DIR/server.log`) for startup and crash diagnostics. | diff --git a/src/chrome-client.js b/src/chrome-client.js index acd478b9..158af78d 100644 --- a/src/chrome-client.js +++ b/src/chrome-client.js @@ -324,7 +324,7 @@ function render() { } function updateSendState() { - sendButton.disabled = ended || agentPresence === "working"; + sendButton.disabled = ended; sendAndEndButton.disabled = sendButton.disabled; if (warningsQueueButton) updateWarningSelectionState(); } @@ -554,7 +554,7 @@ function requestSnapshot(action) { } function sendQueued(endAfter) { - if (ended || agentPresence === "working") return; + if (ended) return; closeMenus(); const text = chatInput.value.trim(); @@ -982,7 +982,7 @@ function updateWarningSelectionState() { warningsSelectAll.checked = selectable.length > 0 && selectedCount === selectable.length; warningsSelectAll.indeterminate = selectedCount > 0 && selectedCount < selectable.length; warningsSelected.textContent = selectedCount === 0 ? "None selected" : selectedCount + " selected"; - warningsQueueButton.disabled = selectedCount === 0 || ended || agentPresence === "working"; + warningsQueueButton.disabled = selectedCount === 0 || ended; } function toggleSelectAllWarnings() { @@ -1039,7 +1039,7 @@ async function dismissWarning(id) { // One queued batch = one ordinary queued prompt. The CLI cannot tell it apart from any other // feedback, which is exactly the point: no parallel agent protocol. async function queueSelectedWarningFixes() { - if (ended || agentPresence === "working") return; + if (ended) return; const ids = [...selectedWarningIds]; if (ids.length === 0) return; warningsQueueButton.disabled = true; diff --git a/src/server.js b/src/server.js index aeca3c99..e43874c2 100644 --- a/src/server.js +++ b/src/server.js @@ -249,6 +249,51 @@ export async function serve({ const logEvent = verbose ? (line) => writeLog(`[lavish] ${line}`) : null; let publicPort = port; + function finishFeedbackDelivery(key, result) { + if (result.status !== "feedback") return; + const chat = result.chat; + delete result.chat; + markFeedbackDelivered(key, activePolls, deliveredFeedback, events); + // A batch flagged `session_ended` is the last one this session will ever deliver, so no + // later poll or agent reply can retire the working state markFeedbackDelivered just set: + // release it here or presence reports an agent still working on a session that is over. + if (result.session_ended) clearFeedbackDelivery(key, activePolls, deliveredFeedback, events); + if (Array.isArray(chat)) events.emit("chat-sync", key, chat); + } + + async function restoreClosedFeedback(key, result) { + if (result.status !== "feedback") return; + const prompts = Array.isArray(result.prompts) ? result.prompts : []; + const session = await store.queuePrompts( + key, + { + dom_snapshot: result.dom_snapshot || "", + prompts, + ...(Array.isArray(result.artifact_failures) ? { artifact_failures: result.artifact_failures } : {}), + }, + { + restore: true, + resolveAttachment: (sessionKeyValue, id) => resolveAttachment(attachmentStateRoot, sessionKeyValue, id), + maxPerPrompt: attachmentConfig.maxPerPrompt, + maxPromptBytes: attachmentConfig.maxPromptBytes, + }, + ); + const restoredPrompts = + prompts.length === 0 + ? [] + : session && !session.rejected && !session.conflict && Array.isArray(session.prompts) + ? session.prompts.slice(-prompts.length) + : null; + const restoredFailures = session && Array.isArray(session.artifact_failures) ? session.artifact_failures : null; + if ( + !restoredPrompts || + JSON.stringify(restoredPrompts) !== JSON.stringify(prompts) || + (Array.isArray(result.artifact_failures) && + JSON.stringify(restoredFailures) !== JSON.stringify(result.artifact_failures)) + ) { + writeLog("[lavish] closed poll feedback restore was incomplete; delivery was not marked"); + } + } // Whiteboard sidecar files live next to state.json, keyed by session + diagram. const whiteboardStateRoot = path.dirname(stateFile); @@ -376,6 +421,18 @@ export async function serve({ }); app.get("/api/poll", async (req, res, next) => { + // `close` is subscribed before the first `await` and re-checked after the listeners are armed, + // because a client that disconnects while `takeFeedback` is in flight would otherwise arrive + // too late for its own cleanup: the handler marks the poll active afterwards and nothing left + // would clear it, leaving presence stuck on "listening" for an agent that is already gone. + let requestClosed = Boolean(req.destroyed); + let cleanupPoll = null; + const onRequestClose = () => { + requestClosed = true; + cleanupPoll?.(); + }; + const detachRequestClose = () => req.off("close", onRequestClose); + req.on("close", onRequestClose); try { const file = await canonicalFile(String(req.query.file || "")); const key = sessionKey(file); @@ -383,10 +440,20 @@ export async function serve({ req.query.timeoutMs === undefined ? null : Math.max(0, Math.min(Number(req.query.timeoutMs || 0), 2147483647)); const immediate = await store.takeFeedback(key); if (immediate.status !== "waiting") { - if (immediate.status === "feedback") markFeedbackDelivered(key, activePolls, deliveredFeedback, events); + if (requestClosed || req.destroyed || res.writableEnded) { + await restoreClosedFeedback(key, immediate); + detachRequestClose(); + return; + } + finishFeedbackDelivery(key, immediate); + detachRequestClose(); res.json(immediate); return; } + if (requestClosed || req.destroyed || res.writableEnded) { + detachRequestClose(); + return; + } const streamHeartbeat = timeoutMs === null; let heartbeat = null; if (streamHeartbeat) { @@ -399,7 +466,7 @@ export async function serve({ } setPollActive(key, activePolls, deliveredFeedback, events, true); refreshIdleTimer(); - const timer = timeoutMs === null ? null : setTimeout(() => respond().catch(handleRespondError), timeoutMs); + let timer = null; let cleaned = false; let responding = false; const cleanup = () => { @@ -411,13 +478,15 @@ export async function serve({ events.off("ended", onFeedback); setPollActive(key, activePolls, deliveredFeedback, events, false); refreshIdleTimer(); + cleanupPoll = null; + detachRequestClose(); }; const respond = async () => { if (responding || res.writableEnded) return; responding = true; try { const result = await store.takeFeedback(key); - if (result.status === "feedback") markFeedbackDelivered(key, activePolls, deliveredFeedback, events); + finishFeedbackDelivery(key, result); if (streamHeartbeat) { res.end(JSON.stringify(result)); } else { @@ -443,8 +512,15 @@ export async function serve({ }; events.on("feedback", onFeedback); events.on("ended", onFeedback); - req.on("close", cleanup); + cleanupPoll = cleanup; + if (requestClosed || req.destroyed || res.writableEnded) { + cleanup(); + return; + } + timer = timeoutMs === null ? null : setTimeout(() => respond().catch(handleRespondError), timeoutMs); } catch (error) { + cleanupPoll?.(); + detachRequestClose(); next(error); } }); @@ -620,9 +696,9 @@ export async function serve({ } events.emit("agent-reply", req.params.key, text); // The reply concludes the delivered-feedback "working" state. Without this, a poll that - // drains feedback and then releases leaves presence stuck on "working" — the chrome keeps - // Send disabled — until some future poll happens to attach, even though the agent already - // answered. See "SSE agent-presence returns to waiting after an agent reply". + // drains feedback and then releases leaves presence stuck on "working" even after the agent + // answers. Human sends remain available while working because the server queues them for the + // next poll. See "SSE agent-presence returns to waiting after an agent reply". clearFeedbackDelivery(req.params.key, activePolls, deliveredFeedback, events); res.json({ status: "sent" }); } catch (error) { diff --git a/src/session-store.js b/src/session-store.js index b591ece0..46569214 100644 --- a/src/session-store.js +++ b/src/session-store.js @@ -142,6 +142,7 @@ export class SessionStore { } const prompts = Array.isArray(payload.prompts) ? payload.prompts : []; const shouldEndSession = Boolean(payload.endSession || payload.end_session); + const restoring = options.restore === true; const alreadyEnded = session.status === "ended"; const normalized = prompts.map(normalizePrompt); const normalizedPrompts = normalized.map((entry) => entry.prompt); @@ -172,52 +173,70 @@ export class SessionStore { const revision = normalizeRevision(session.artifact_revision); const at = new Date().toISOString(); let warnings = normalizeStoredWarnings(session.layout_warnings); - const layoutPlans = []; - const conflicts = new Set(); - for (const prompt of normalizedPrompts) { - const warningIds = layoutWarningPromptIds(prompt); - if (warningIds === null) { - layoutPlans.push({ - prompt, - warningIds: null, - expectedRevision: null, - conflicts: [], - queueIds: [], - hadKnownWarning: false, - }); - continue; - } - const plan = planLayoutWarningPrompt(warnings, prompt, revision); - for (const id of plan.conflicts) conflicts.add(id); - layoutPlans.push({ prompt, ...plan }); - } - if (conflicts.size > 0) { - return { - conflict: true, - session, - warning_ids: [...conflicts], - warnings: serializeLayoutWarnings(warnings), - }; - } - const acceptedPrompts = []; - for (const plan of layoutPlans) { - if (plan.warningIds === null) { - acceptedPrompts.push(plan.prompt); - continue; - } - const result = queueWarningRecords(warnings, plan.queueIds, { revision, at }); - warnings = result.warnings; - if (result.queued.length > 0 || !plan.hadKnownWarning) acceptedPrompts.push(plan.prompt); + let acceptedPrompts; + if (restoring) { + acceptedPrompts = normalizedPrompts; + } else { + const layoutPlans = []; + const conflicts = new Set(); + for (const prompt of normalizedPrompts) { + const warningIds = layoutWarningPromptIds(prompt); + if (warningIds === null) { + layoutPlans.push({ + prompt, + warningIds: null, + expectedRevision: null, + conflicts: [], + queueIds: [], + hadKnownWarning: false, + }); + continue; + } + const plan = planLayoutWarningPrompt(warnings, prompt, revision); + for (const id of plan.conflicts) conflicts.add(id); + layoutPlans.push({ prompt, ...plan }); + } + if (conflicts.size > 0) { + return { + conflict: true, + session, + warning_ids: [...conflicts], + warnings: serializeLayoutWarnings(warnings), + }; + } + acceptedPrompts = []; + for (const plan of layoutPlans) { + if (plan.warningIds === null) { + acceptedPrompts.push(plan.prompt); + continue; + } + const result = queueWarningRecords(warnings, plan.queueIds, { revision, at }); + warnings = result.warnings; + if (result.queued.length > 0 || !plan.hadKnownWarning) acceptedPrompts.push(plan.prompt); + } } session.layout_warnings = warnings; - const userMessages = acceptedPrompts - .filter((prompt) => prompt.tag === "message" && prompt.prompt) - .map((prompt) => ({ role: "user", text: prompt.prompt, at: new Date().toISOString() })); + const userMessages = restoring + ? [] + : acceptedPrompts + .filter((prompt) => prompt.tag === "message" && prompt.prompt) + .map((prompt) => ({ role: "user", text: prompt.prompt, at: new Date().toISOString() })); session.prompts = [...(session.prompts || []), ...acceptedPrompts]; session.chat = [...(session.chat || []), ...userMessages]; + if (restoring) { + session.artifact_failures = Array.isArray(payload.artifact_failures) + ? JSON.parse(JSON.stringify(payload.artifact_failures)) + : []; + } session.pending_prompts = session.prompts.length; session.dom_snapshot = String(payload.domSnapshot || payload.dom_snapshot || ""); - session.status = shouldEndSession || alreadyEnded ? "ended" : session.prompts.length > 0 ? "feedback" : "open"; + session.status = + shouldEndSession || alreadyEnded + ? "ended" + : session.prompts.length > 0 || + (restoring && Array.isArray(session.artifact_failures) && session.artifact_failures.length > 0) + ? "feedback" + : "open"; if (shouldEndSession) session.ended_by = "user"; session.updated_at = new Date().toISOString(); await this.writeState(state); diff --git a/test/chrome-client-queue.test.js b/test/chrome-client-queue.test.js index 33db80fa..d8c1b7a7 100644 --- a/test/chrome-client-queue.test.js +++ b/test/chrome-client-queue.test.js @@ -878,6 +878,26 @@ test("nothing is selected by default and Select all is an explicit action", asyn assert.equal(chrome.element("warningsQueueButton").disabled, false); }); +test("warning fixes stay queueable while the agent is working", async () => { + const posts = []; + const chrome = await createChromeHarness({ + fetchImpl: async (url, init) => { + posts.push({ url, body: init && init.body ? JSON.parse(init.body) : null }); + return { ok: true, json: async () => ({ warnings: [warningPayload()], prompt: null }) }; + }, + }); + chrome.eventSource().listeners.get("agent-presence")({ data: JSON.stringify({ state: "working" }) }); + chrome.eventSource().listeners.get("layout-warnings")({ data: JSON.stringify({ warnings: [warningPayload()] }) }); + + const [row] = chrome.warningRows(); + row.children[0].checked = true; + row.children[0].dispatch("change"); + assert.equal(chrome.element("warningsQueueButton").disabled, false); + + await chrome.element("warningsQueueButton").onclick(); + assert.ok(posts.some((post) => post.url === "/api/abc/layout-warnings/queue")); +}); + test("queueing a selected subset produces exactly one ordinary prompt with only those warnings", async () => { const posts = []; const queuedWarnings = [ @@ -1768,6 +1788,60 @@ test("chrome client strips the internal queue key before posting prompts", async assert.equal(chrome.queued().length, 0); }); +test("chrome client sends queued prompts while the agent is working", async () => { + const posts = []; + const chrome = await createChromeHarness({ + fetchImpl: async (url, init) => { + posts.push({ url, body: JSON.parse(init.body) }); + return { ok: true }; + }, + }); + + chrome.eventSource().listeners.get("agent-presence")({ data: JSON.stringify({ state: "working" }) }); + chrome.sendFrameMessage({ + type: "lavish:queuePrompt", + prompt: { prompt: "Follow up", selector: "button#follow-up", tag: "choice", text: "Follow up" }, + }); + chrome.element("send").onclick(); + assert.equal(chrome.postedToFrame.at(-1).type, "lavish:requestSnapshot"); + + chrome.sendFrameMessage({ type: "lavish:snapshot", snapshot: "uid=1 body" }); + await flushPromises(); + + const submitted = posts.filter((post) => post.url === "/api/abc/prompts"); + assert.equal(submitted.length, 1); + assert.deepEqual( + submitted[0].body.prompts.map((prompt) => prompt.prompt), + ["Follow up"], + ); + assert.equal(chrome.queued().length, 0); +}); + +test("send controls stay enabled while the agent works and lock only once the session ends", async () => { + const chrome = await createChromeHarness({ + fetchImpl: async () => ({ ok: true, json: async () => ({}) }), + }); + + assert.equal(chrome.element("send").disabled, false); + assert.equal(chrome.element("sendAndEnd").disabled, false); + + chrome.eventSource().listeners.get("agent-presence")({ data: JSON.stringify({ state: "working" }) }); + assert.equal(chrome.element("send").disabled, false); + assert.equal(chrome.element("sendAndEnd").disabled, false); + + chrome.sendFrameMessage({ + type: "lavish:queuePrompt", + prompt: { prompt: "Ship this", selector: "button#ship", tag: "choice", text: "Ship" }, + }); + chrome.element("sendAndEnd").onclick(); + chrome.sendFrameMessage({ type: "lavish:snapshot", snapshot: "uid=1 body" }); + await flushPromises(); + await flushPromises(); + + assert.equal(chrome.element("send").disabled, true); + assert.equal(chrome.element("sendAndEnd").disabled, true); +}); + test("chrome send and end carries the end intent with queued prompts", async () => { const posts = []; const chrome = await createChromeHarness({ @@ -2620,7 +2694,7 @@ test("a non-array attachments field cannot wedge the queue (E5)", async () => { assert.deepEqual(chrome.queued(), [{ prompt: "bad", selector: "h1", tag: "annotation", text: "" }]); }); -test("a poisoned prompt already in sessionStorage cannot wedge a reload (E5)", async () => { +test("a poisoned prompt already in the restored queue cannot wedge a reload (E5)", async () => { const chrome = await createChromeHarness({ storedQueue: [{ prompt: "old poison", selector: "h1", tag: "annotation", text: "", attachments: [null] }], }); @@ -2681,8 +2755,9 @@ test("a queued attachment ref is projected to primitives, not kept by reference chrome.sendFrameMessage({ type: "lavish:snapshot", snapshot: "uid=1 body" }); await flushPromises(); - assert.equal(posts.length, 1, "the queue is still sendable"); - assert.deepEqual(posts[0].body.prompts[0].attachments, [{ id, name: "ok.png" }]); + const submitted = posts.filter((post) => post.url === "/api/abc/prompts"); + assert.equal(submitted.length, 1, "the queue is still sendable"); + assert.deepEqual(submitted[0].body.prompts[0].attachments, [{ id, name: "ok.png" }]); }); test("a non-string attachment name is dropped rather than carried (E5)", async () => { diff --git a/test/fixtures/layout-audit/real-editorial.html b/test/fixtures/layout-audit/real-editorial.html index 53159eb4..09c153f6 100644 --- a/test/fixtures/layout-audit/real-editorial.html +++ b/test/fixtures/layout-audit/real-editorial.html @@ -231,8 +231,7 @@

Make the structure do the work

--measure-compact: 48ch; --space-section: clamp(3rem, 8vw, 7rem); --tone-quiet: color-mix(in oklab, currentColor 58%, transparent); -} +}

The final test is not whether everything fits in a screenshot. It is whether the page remains composed when diff --git a/test/server.test.js b/test/server.test.js index 646dd34e..b60bb418 100644 --- a/test/server.test.js +++ b/test/server.test.js @@ -1,6 +1,7 @@ import assert from "node:assert/strict"; import { mkdtemp, mkdir, readFile, realpath, rm, symlink, writeFile } from "node:fs/promises"; import { createServer, request as httpRequest } from "node:http"; +import { connect as netConnect } from "node:net"; import { homedir, tmpdir } from "node:os"; import path from "node:path"; import { Readable } from "node:stream"; @@ -28,7 +29,7 @@ import { resolveWatchTarget, serve, } from "../src/server.js"; -import { canonicalFile, sessionKey } from "../src/session-store.js"; +import { canonicalFile, sessionKey, SessionStore } from "../src/session-store.js"; async function chromeClientSource() { return readFile(new URL("../src/chrome-client.js", import.meta.url), "utf8"); @@ -738,16 +739,6 @@ test("chrome shows agent working state when a previous poll has released", async assert.match(js, /spinner/); }); -test("chrome disables sending only while working or ended", async () => { - const js = await chromeClientSource(); - - assert.match(js, /let agentPresence = "waiting"/); - assert.match(js, /function updateSendState\(\)/); - assert.match(js, /sendButton\.disabled = ended \|\| agentPresence === "working"/); - assert.match(js, /sendAndEndButton\.disabled = sendButton\.disabled/); - assert.doesNotMatch(js, /hasContent/); -}); - test("sending with an empty composer nudges instead of blocking", async () => { const html = createChromeHtml({ key: "abc", file: "/tmp/artifact.html" }); const js = await chromeClientSource(); @@ -3478,6 +3469,7 @@ test("send-and-end prompt submissions wake active polls with ended attribution", assert.equal(feedback.session_ended, true); assert.equal(feedback.ended_by, "user"); assert.equal(feedback.prompts.length, 1); + assert.equal(await presence.next(), "waiting"); const ended = await fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}&timeoutMs=0`); const endedBody = await ended.json(); @@ -3492,6 +3484,44 @@ test("send-and-end prompt submissions wake active polls with ended attribution", } }); +test("ending an active poll without final feedback leaves presence waiting", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "lavish-serve-")); + const artifact = path.join(dir, "artifact.html"); + const keepAlive = path.join(dir, "keep-alive.html"); + await writeFile(artifact, ""); + await writeFile(keepAlive, ""); + const server = await serve({ port: 0, stateFile: path.join(dir, "state.json"), version: "9.9.9-test" }); + try { + const base = `http://127.0.0.1:${server.port}`; + await fetch(`${base}/api/sessions`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ file: keepAlive }), + }); + const open = await fetch(`${base}/api/sessions`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ file: artifact }), + }); + const { key } = await open.json(); + const presence = await startPresenceStream(base, key); + try { + assert.equal(await presence.next(), "waiting"); + const poll = fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}`).then((res) => res.json()); + assert.equal(await presence.next(), "listening"); + + await fetch(`${base}/api/${key}/end`, { method: "POST" }); + assert.equal((await poll).status, "ended"); + assert.equal(await presence.next(), "waiting"); + } finally { + await presence.close(); + } + } finally { + await server.close(); + await rm(dir, { recursive: true, force: true }); + } +}); + test("SSE agent-presence reflects waiting, listening, and working transitions", async () => { const dir = await mkdtemp(path.join(tmpdir(), "lavish-serve-")); const artifact = path.join(dir, "artifact.html"); @@ -3720,10 +3750,10 @@ test("heartbeat long-poll errors close the stream without Express error handling assert.match(source, /respond\(\)\.catch\(handleRespondError\)/); }); -test("SSE agent-presence switches to working when poll immediately takes queued feedback", async () => { +test("a poll dropped before it arms never leaves presence listening", async () => { const dir = await mkdtemp(path.join(tmpdir(), "lavish-serve-")); const artifact = path.join(dir, "artifact.html"); - await (await import("node:fs/promises")).writeFile(artifact, ""); + await writeFile(artifact, ""); const server = await serve({ port: 0, stateFile: path.join(dir, "state.json"), version: "9.9.9-test" }); try { const base = `http://127.0.0.1:${server.port}`; @@ -3734,67 +3764,197 @@ test("SSE agent-presence switches to working when poll immediately takes queued }); const { key } = await open.json(); - const presenceEvents = []; - const presenceWaiters = []; - const presenceController = new AbortController(); - const presenceFetch = fetch(`${base}/events/${key}`, { signal: presenceController.signal }).then(async (res) => { - const reader = res.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; - while (true) { - const { value, done } = await reader.read(); - if (done) break; - buffer += decoder.decode(value, { stream: true }); - let lines; - while ((lines = buffer.match(/^event: agent-presence\ndata: (.+)\n\n/m))) { - const data = JSON.parse(lines[1]); - presenceEvents.push(data.state); - buffer = buffer.replace(lines[0], ""); - const waiter = presenceWaiters.shift(); - if (waiter) waiter(data.state); - } - } - }); - presenceFetch.catch(() => {}); - - const waitForPresence = () => - new Promise((resolve, reject) => { - const timer = setTimeout(() => reject(new Error("timed out waiting for agent presence event")), 500); - if (presenceEvents.length > waitForPresence.lastIndex) { - waitForPresence.lastIndex++; - clearTimeout(timer); - resolve(presenceEvents[waitForPresence.lastIndex - 1]); - return; - } - presenceWaiters.push((state) => { - waitForPresence.lastIndex = presenceEvents.length; - clearTimeout(timer); - resolve(state); + // Send a real poll request, then drop the socket while the handler is still inside its + // startup awaits - before it can register the long poll. A cleanup hook attached after + // that point never runs, so the poll would arm "listening" with nobody left to release it. + await new Promise((resolve, reject) => { + const socket = netConnect(server.port, "127.0.0.1", () => { + const target = `/api/poll?file=${encodeURIComponent(artifact)}`; + socket.write(`GET ${target} HTTP/1.1\r\nHost: 127.0.0.1:${server.port}\r\n\r\n`, () => { + socket.destroy(); + resolve(); }); }); - waitForPresence.lastIndex = 0; + socket.on("error", reject); + }); + await new Promise((resolve) => setTimeout(resolve, 150)); - const initial = await waitForPresence(); + const presence = await startPresenceStream(base, key); + try { + assert.equal(await presence.next(), "waiting"); + } finally { + await presence.close(); + } + + // The abandoned poll also must not have consumed anything: a fresh poll still gets the + // feedback queued after it. + await fetch(`${base}/api/${key}/prompts`, { + method: "POST", + headers: { "content-type": "application/json", origin: base }, + body: JSON.stringify({ prompts: [{ prompt: "still here", tag: "message" }] }), + }); + const next = await fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}&timeoutMs=0`); + const feedback = await next.json(); + assert.equal(feedback.status, "feedback"); + assert.deepEqual( + feedback.prompts.map((prompt) => prompt.prompt), + ["still here"], + ); + } finally { + await server.close(); + await rm(dir, { recursive: true, force: true }); + } +}); + +test("immediate poll delivery leaves presence working and preserves the next send", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "lavish-serve-")); + const artifact = path.join(dir, "artifact.html"); + await (await import("node:fs/promises")).writeFile(artifact, ""); + const server = await serve({ port: 0, stateFile: path.join(dir, "state.json"), version: "9.9.9-test" }); + try { + const base = `http://127.0.0.1:${server.port}`; + const open = await fetch(`${base}/api/sessions`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ file: artifact }), + }); + const { key } = await open.json(); + + const initialPresence = await startPresenceStream(base, key); + const initial = await initialPresence.next(); assert.equal(initial, "waiting"); + await initialPresence.close(); await fetch(`${base}/api/${key}/prompts`, { method: "POST", headers: { "content-type": "application/json", origin: base }, body: JSON.stringify({ prompts: [{ prompt: "hello", tag: "message" }] }), }); - await fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}`); + const immediate = await fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}`); + assert.deepEqual( + (await immediate.json()).prompts.map((prompt) => prompt.prompt), + ["hello"], + ); - const working = await waitForPresence(); - assert.equal(working, "working"); + const afterImmediatePresence = await startPresenceStream(base, key); + try { + assert.equal(await afterImmediatePresence.next(), "working"); + } finally { + await afterImmediatePresence.close(); + } - presenceController.abort(); - await presenceFetch.catch(() => {}); + const submitted = await fetch(`${base}/api/${key}/prompts`, { + method: "POST", + headers: { "content-type": "application/json", origin: base }, + body: JSON.stringify({ prompts: [{ prompt: "follow-up", tag: "message" }] }), + }); + assert.equal(submitted.status, 200); + + const nextPoll = await fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}&timeoutMs=0`); + const nextFeedback = await nextPoll.json(); + assert.equal(nextFeedback.status, "feedback"); + assert.deepEqual( + nextFeedback.prompts.map((prompt) => prompt.prompt), + ["follow-up"], + ); } finally { await server.close(); await rm(dir, { recursive: true, force: true }); } }); +test("a disconnect during immediate feedback take requeues the batch without working presence", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "lavish-serve-")); + const artifact = path.join(dir, "artifact.html"); + const stateFile = path.join(dir, "state.json"); + await writeFile(artifact, ""); + const server = await serve({ port: 0, stateFile, version: "9.9.9-test" }); + const originalTakeFeedback = SessionStore.prototype.takeFeedback; + try { + const base = `http://127.0.0.1:${server.port}`; + const open = await fetch(`${base}/api/sessions`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ file: artifact }), + }); + const { key } = await open.json(); + const queued = { + domSnapshot: 'uid=1 body "review"', + prompts: [ + { + uid: "choice-1", + prompt: "Use the compact layout", + selector: "#compact", + tag: "choice", + text: "Compact", + target: { type: "text-range", text: "Compact", commonAncestorSelector: "#options" }, + }, + { uid: "message-1", prompt: "Looks good", selector: "body", tag: "message", text: "" }, + ], + }; + const submitted = await fetch(`${base}/api/${key}/prompts`, { + method: "POST", + headers: { "content-type": "application/json", origin: base }, + body: JSON.stringify(queued), + }); + assert.equal(submitted.status, 200); + const beforeState = JSON.parse(await readFile(stateFile, "utf8")).sessions[key]; + const before = beforeState.prompts; + + /** @type {() => void} */ + let releaseTake = () => {}; + const takeReleased = new Promise((resolve) => { + releaseTake = () => resolve(); + }); + let takeStarted; + const takePending = new Promise((resolve) => { + takeStarted = resolve; + }); + let delayed = true; + SessionStore.prototype.takeFeedback = async function (sessionKey) { + if (delayed && sessionKey === key) { + delayed = false; + takeStarted(); + await takeReleased; + } + return originalTakeFeedback.call(this, sessionKey); + }; + + const socket = await new Promise((resolve, reject) => { + const client = netConnect(server.port, "127.0.0.1", () => { + client.write( + `GET /api/poll?file=${encodeURIComponent(artifact)} HTTP/1.1\r\nHost: 127.0.0.1:${server.port}\r\n\r\n`, + () => resolve(client), + ); + }); + client.on("error", reject); + }); + await takePending; + socket.on("error", () => {}); + socket.destroy(); + releaseTake(); + await new Promise((resolve) => setTimeout(resolve, 50)); + + const presence = await startPresenceStream(base, key); + try { + assert.equal(await presence.next(), "waiting"); + } finally { + await presence.close(); + } + + const next = await fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}&timeoutMs=0`); + const feedback = await next.json(); + assert.equal(feedback.status, "feedback"); + assert.deepEqual(feedback.dom_snapshot, queued.domSnapshot); + assert.deepEqual(feedback.prompts, before); + assert.deepEqual(JSON.parse(await readFile(stateFile, "utf8")).sessions[key].chat, beforeState.chat); + } finally { + SessionStore.prototype.takeFeedback = originalTakeFeedback; + await server.close(); + await rm(dir, { recursive: true, force: true }); + } +}); + test("SSE agent-presence resets to waiting after ending and reopening a session", async () => { const dir = await mkdtemp(path.join(tmpdir(), "lavish-serve-")); const artifact = path.join(dir, "artifact.html"); @@ -3818,7 +3978,6 @@ test("SSE agent-presence resets to waiting after ending and reopening a session" body: JSON.stringify({ prompts: [{ prompt: "hello", tag: "message" }] }), }); await fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}`); - assert.equal(await presence.next(), "working"); await fetch(`${base}/api/${key}/end`, { method: "POST" }); // The browser end above is user-initiated, so reopening requires the explicit opt-in. @@ -3843,6 +4002,48 @@ test("SSE agent-presence resets to waiting after ending and reopening a session" } }); +test("immediate send-and-end delivery clears working presence without an active poll", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "lavish-serve-")); + const artifact = path.join(dir, "artifact.html"); + await writeFile(artifact, ""); + const server = await serve({ port: 0, stateFile: path.join(dir, "state.json"), version: "9.9.9-test" }); + try { + const base = `http://127.0.0.1:${server.port}`; + const open = await fetch(`${base}/api/sessions`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ file: artifact }), + }); + const { key } = await open.json(); + const presence = await startPresenceStream(base, key); + try { + assert.equal(await presence.next(), "waiting"); + + const submitted = await fetch(`${base}/api/${key}/prompts`, { + method: "POST", + headers: { "content-type": "application/json", origin: base }, + body: JSON.stringify({ + endSession: true, + prompts: [{ prompt: "bye", tag: "message" }], + }), + }); + assert.equal(submitted.status, 200); + + const immediate = await fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}`); + const feedback = await immediate.json(); + assert.equal(feedback.status, "feedback"); + assert.equal(feedback.session_ended, true); + assert.equal(await presence.next(), "working"); + assert.equal(await presence.next(), "waiting"); + } finally { + await presence.close(); + } + } finally { + await server.close(); + await rm(dir, { recursive: true, force: true }); + } +}); + test("SSE agent-presence returns to waiting after an agent reply", async () => { const dir = await mkdtemp(path.join(tmpdir(), "lavish-serve-")); const artifact = path.join(dir, "artifact.html"); @@ -3860,17 +4061,19 @@ test("SSE agent-presence returns to waiting after an agent reply", async () => { try { assert.equal(await presence.next(), "waiting"); + const poll = fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}`).then((response) => response.json()); + assert.equal(await presence.next(), "listening"); await fetch(`${base}/api/${key}/prompts`, { method: "POST", headers: { "content-type": "application/json", origin: base }, body: JSON.stringify({ prompts: [{ prompt: "hello", tag: "message" }] }), }); - // A poll that drains the feedback and releases leaves presence "working". - await fetch(`${base}/api/poll?file=${encodeURIComponent(artifact)}`); + await poll; + // An armed poll that drains the feedback and releases leaves presence "working". assert.equal(await presence.next(), "working"); - // The reply concludes that work. Without a clear here, presence stays "working" - // forever (the chrome disables Send) until some future poll happens to attach. + // The reply concludes that work. Without a clear here, presence stays "working" forever + // even though the agent has answered. await fetch(`${base}/api/${key}/agent-reply`, { method: "POST", headers: { "content-type": "application/json" }, @@ -3886,7 +4089,7 @@ test("SSE agent-presence returns to waiting after an agent reply", async () => { } }); -test("SSE agent-presence stays working when resuming an open session", async () => { +test("SSE agent-presence stays working when resuming after immediate feedback", async () => { const dir = await mkdtemp(path.join(tmpdir(), "lavish-serve-")); const artifact = path.join(dir, "artifact.html"); await writeFile(artifact, "");