diff --git a/AGENTS.md b/AGENTS.md index 9f594a0a1..a699216da 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -276,6 +276,10 @@ Direct `plannotator annotate` invocations may add `--require-approval` and/or `- Strict decisions use one newline-terminated JSON record on stdout and, when requested, identical bytes in the result file. Exit codes follow the grep convention: approval exits `0`; with `--require-approval`, annotated and dismissed decisions are published before exiting `1` (negative human outcome); usage/startup/validation failures — bad flag combinations, strict flags outside `annotate --gate --json`, a missing `--result-file` parent, a pre-existing or dangling-symlink destination, and every annotate startup failure (missing path, unreachable URL, empty folder, ambiguous name, missing file, oversized file) — exit `2` (the gate itself was misconfigured or could not start). Those startup sites exit `1` as before for non-strict invocations; under a strict flag `1` is reserved for "the reviewer did not approve", so a typo'd path must never masquerade as a rejection. Post-decision publication failures (destination appears between validation and publish, hard links unavailable) also exit `2`: the result *file* was not published, so they present as environment errors — "the gate could not publish its result" — never as a reviewer outcome, and never as approval (still fail-closed, since only `0` means approved). The stdout decision record is written **before** result-file publication and is still emitted whenever the decision itself completed; only a stdout write failure leaves no record anywhere. Signal deaths keep `128+n`. Result paths resolve from the invocation working directory, require an existing parent and absent destination, and publish via a flushed/closed `0600` same-directory temporary file plus an atomic no-clobber hard link—never copy or overwrite fallback (the `0600` mode is a no-op on Windows, and the atomic link/rename is not followed by a parent-directory fsync, so publication is atomic but not crash-durable). Keep reviewed sources at stable project paths; unique result and diagnostic log files may use a narrow temporary directory. Explicit Close emits `dismissed`; missing results or process/browser failures are recovery cases, never approval. +### Abandoned strict gate sessions + +Local direct structured gates (`--gate --json`, not `--hook`, not remote) advertise a client lease in `/api/plan` and serve `/api/annotate/client-lease` (SSE, `ANNOTATE_CLIENT_LEASE_STREAM_PATH`). Each open stream is one connected review surface; the server heartbeats every 5s and, once at least one client has connected, starts a 30s reconnect grace when the last one disconnects. A reconnect inside the grace continues the same review; expiry resolves the gate as the same `dismissed` decision an explicit Close produces, except that it keeps the saved annotation draft so an abandoned review can still be recovered. Approve, feedback, explicit exit, and server stop all cancel a pending expiry. Whichever producer settles the session first wins: every one of them (each connected surface and the expiry itself) goes through a single one-shot settlement, so a decision arriving after the session already resolved is rejected with `409` rather than deleting the draft and reporting success for an outcome the caller never received. Page lifecycle events are deliberately not used: `pagehide`/`beforeunload` also fire on reload and navigation, so they cannot distinguish abandonment from a reconnect. A session that never receives its first client never auto-dismisses, so browser-launch failures still need a caller-side timeout, and remote/shared sessions keep the capability off because tunnel disconnects would read as abandonment. + ## Archive Flow ``` @@ -400,6 +404,7 @@ During normal plan review, an Archive sidebar tab provides the same browsing via | `/api/doc` | GET | Serve linked .md/.mdx/.html file or code file (`?path=&base=`) | | `/api/doc/exists` | POST | Batch-validate code-file paths (body: `{ paths: string[], base?: string }`) | | `/api/draft` | GET/POST/DELETE | Auto-save annotation drafts to survive server crashes | +| `/api/annotate/client-lease` | GET (SSE) | Client lease for local direct structured gates: each open stream is one connected review surface. 404 when the capability is not advertised. | | `/api/agent-terminal/pty/` | WebSocket | Tokenized PTY bridge for the optional annotate-mode agent terminal | | `/api/ai/capabilities` | GET | Check if AI features are available | | `/api/ai/session` | POST | Create or fork an AI session | diff --git a/apps/hook/server/annotate-output.test.ts b/apps/hook/server/annotate-output.test.ts index 2129fdecf..0b820c6cb 100644 --- a/apps/hook/server/annotate-output.test.ts +++ b/apps/hook/server/annotate-output.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test"; import { formatAnnotateOutcome, supportsAnnotateApprovalNotes, + supportsAnnotateClientLease, } from "./annotate-output"; describe("annotate stdout", () => { @@ -48,4 +49,12 @@ describe("annotate stdout", () => { expect(supportsAnnotateApprovalNotes({ gate: true, json: false, hook: false })).toBe(false); expect(supportsAnnotateApprovalNotes({ gate: true, json: true, hook: true })).toBe(false); }); + + test("advertises client-lease only for gated direct JSON, local sessions", () => { + expect(supportsAnnotateClientLease({ gate: true, json: true, hook: false, isRemote: false })).toBe(true); + expect(supportsAnnotateClientLease({ gate: false, json: true, hook: false, isRemote: false })).toBe(false); + expect(supportsAnnotateClientLease({ gate: true, json: false, hook: false, isRemote: false })).toBe(false); + expect(supportsAnnotateClientLease({ gate: true, json: true, hook: true, isRemote: false })).toBe(false); + expect(supportsAnnotateClientLease({ gate: true, json: true, hook: false, isRemote: true })).toBe(false); + }); }); diff --git a/apps/hook/server/annotate-output.ts b/apps/hook/server/annotate-output.ts index a9c4a3fc4..41d44d6fc 100644 --- a/apps/hook/server/annotate-output.ts +++ b/apps/hook/server/annotate-output.ts @@ -11,6 +11,11 @@ export interface AnnotateApprovalCapabilityOptions extends AnnotateOutputOptions gate: boolean; } +export interface AnnotateClientLeaseCapabilityOptions extends AnnotateApprovalCapabilityOptions { + /** True for remote/shared sessions, where a lost tab connection is expected and not abandonment. */ + isRemote: boolean; +} + const APPROVED_PLAINTEXT_MARKER = "The user approved."; export function supportsAnnotateApprovalNotes( @@ -19,6 +24,19 @@ export function supportsAnnotateApprovalNotes( return options.gate && options.json && !options.hook; } +/** + * Local direct structured annotate gates (`--gate --json`, not `--hook`, not + * a remote/shared session) are the only transport where a tab's abandonment + * can be safely resolved automatically — the caller is already blocked on a + * structured decision and no other protocol (hook JSON, plaintext) depends on + * the exact timing of the response. + */ +export function supportsAnnotateClientLease( + options: AnnotateClientLeaseCapabilityOptions, +): boolean { + return options.gate && options.json && !options.hook && !options.isRemote; +} + export function formatAnnotateOutcome( result: AnnotateOutcome, options: AnnotateOutputOptions, diff --git a/apps/hook/server/index.ts b/apps/hook/server/index.ts index 7f1298e6b..ca6aa8a5e 100644 --- a/apps/hook/server/index.ts +++ b/apps/hook/server/index.ts @@ -78,6 +78,7 @@ import { import { startAnnotateServer, handleAnnotateServerReady, + isRemoteSession, } from "@plannotator/server/annotate"; import { startGoalSetupServer, @@ -157,6 +158,7 @@ import { buildLocalWorkspaceReview, type WorkspaceDiffType } from "@plannotator/ import { createAnnotateOutcomeEmitter, supportsAnnotateApprovalNotes, + supportsAnnotateClientLease, } from "./annotate-output"; // Embed the built HTML at compile time @@ -1067,6 +1069,12 @@ if (args[0] === "sessions") { json: jsonFlag, hook: hookFlag, }), + clientLeaseSupported: supportsAnnotateClientLease({ + gate: gateFlag, + json: jsonFlag, + hook: hookFlag, + isRemote: isRemoteSession(), + }), rawHtml, renderHtml: !!rawHtml, convertHtml: renderMarkdownFlag, @@ -1271,6 +1279,12 @@ if (args[0] === "sessions") { json: jsonFlag, hook: hookFlag, }), + clientLeaseSupported: supportsAnnotateClientLease({ + gate: gateFlag, + json: jsonFlag, + hook: hookFlag, + isRemote: isRemoteSession(), + }), htmlContent: planHtmlContent, recentMessages: pickerMessages, onReady: async (url, isRemote, port) => { @@ -1772,6 +1786,12 @@ if (args[0] === "sessions") { json: jsonFlag, hook: hookFlag, }), + clientLeaseSupported: supportsAnnotateClientLease({ + gate: gateFlag, + json: jsonFlag, + hook: hookFlag, + isRemote: isRemoteSession(), + }), htmlContent: planHtmlContent, onReady: async (url, isRemote, port) => { handleAnnotateServerReady(url, isRemote, port); diff --git a/apps/marketing/src/content/docs/guides/annotate-gates-and-json-responses.md b/apps/marketing/src/content/docs/guides/annotate-gates-and-json-responses.md index 61905542a..c40514759 100644 --- a/apps/marketing/src/content/docs/guides/annotate-gates-and-json-responses.md +++ b/apps/marketing/src/content/docs/guides/annotate-gates-and-json-responses.md @@ -133,7 +133,13 @@ Two caveats on the publication guarantees: Keep the reviewed source at a stable project path so revisions and version history continue to refer to the same artifact. Result and diagnostic log files can instead live in a narrowly scoped temporary directory. -Clicking Close publishes `{"decision":"dismissed"}`. Closing or crashing the browser outside that explicit action is not guaranteed to produce a decision; callers should treat a missing result or failed process as a recovery case, never as approval. +Clicking Close publishes `{"decision":"dismissed"}`. + +Abandoning the review publishes the same `dismissed` decision. A local direct structured gate tracks its connected review surfaces: once at least one has connected, losing the last one starts a 30-second reconnect grace period, and expiry resolves the gate as `dismissed`. Reloading the page, navigating away and back, or closing one of several tabs all reconnect or leave another surface connected, so none of them dismiss the review. Approve, Send Annotations, and Close still win over a pending expiry. + +An abandoned review keeps its saved annotation draft, so nothing you wrote is lost. If a stale tab comes back after the gate already resolved, its Approve, Send Annotations, or Close reports an error instead of pretending to apply: the decision the caller received is the one that counts. + +Two cases remain caller-side recovery, never approval: a session where no review client ever connected (a browser that failed to launch keeps waiting, so pass your own startup timeout), and a half-open transport loss, where the connection is only proven dead by a failing heartbeat write and can take longer than the grace period to notice. Remote and shared sessions keep the behavior off entirely, because a tunnel or proxy disconnect is not abandonment. ## Primary use cases diff --git a/apps/pi-extension/plannotator-browser.ts b/apps/pi-extension/plannotator-browser.ts index a58b73d0f..2805f3a3d 100644 --- a/apps/pi-extension/plannotator-browser.ts +++ b/apps/pi-extension/plannotator-browser.ts @@ -590,6 +590,7 @@ export async function startMarkdownAnnotationSession( sourceConverted, gate, approvalNotesSupported: true, + clientLeaseSupported: gate === true && !isRemoteSession(), rawHtml, renderHtml, convertHtml, diff --git a/apps/pi-extension/server.test.ts b/apps/pi-extension/server.test.ts index b927eabf6..218026c62 100644 --- a/apps/pi-extension/server.test.ts +++ b/apps/pi-extension/server.test.ts @@ -471,6 +471,276 @@ describe("pi annotate approval notes", () => { }); }); +describe("pi annotate client lease", () => { + /** + * Connect to the client-lease SSE stream and wait for the ready comment. + * Returns a `disconnect()` that aborts the underlying request — a plain + * `reader.cancel()` only stops local reads and does not close the + * connection the server observes, whereas aborting the fetch closes the + * socket the way an abandoned browser tab actually would (triggering the + * node:http response's "close" event). + */ + async function connectClientLease(url: string): Promise<{ disconnect: () => Promise }> { + const controller = new AbortController(); + const response = await fetch(`${url}/api/annotate/client-lease`, { signal: controller.signal }); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + const first = await Promise.race([ + reader.read(), + new Promise((_, reject) => { + setTimeout(() => reject(new Error("Timed out waiting for ready comment")), 1000); + }), + ]); + expect(first.done).toBe(false); + return { + disconnect: async () => { + controller.abort(); + await reader.cancel().catch(() => {}); + }, + }; + } + + /** Track whether a promise has settled, without racing it against a timer. */ + function trackSettled(promise: Promise): () => boolean { + let settled = false; + promise.then(() => { + settled = true; + }); + return () => settled; + } + + test("advertises the effective client-lease capability in /api/plan", async () => { + for (const clientLeaseSupported of [true, false]) { + delete process.env.PLANNOTATOR_PORT; + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(makeTempDir("plannotator-pi-client-lease-capability-"), "test.md"), + htmlContent: "annotate", + origin: "pi", + gate: true, + approvalNotesSupported: true, + clientLeaseSupported, + }); + + try { + const response = await fetch(`${server.url}/api/plan`); + const plan = await response.json() as { clientLease?: { enabled: boolean; reconnectGraceMs?: number } }; + if (clientLeaseSupported) { + expect(plan.clientLease).toEqual({ enabled: true, reconnectGraceMs: 30_000 }); + } else { + expect(plan.clientLease).toEqual({ enabled: false }); + } + } finally { + server.stop(); + } + } + }); + + test("returns 404 for the client-lease stream when the capability is disabled", async () => { + delete process.env.PLANNOTATOR_PORT; + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(makeTempDir("plannotator-pi-client-lease-disabled-"), "test.md"), + htmlContent: "annotate", + origin: "pi", + }); + + try { + const response = await fetch(`${server.url}/api/annotate/client-lease`); + expect(response.status).toBe(404); + } finally { + server.stop(); + } + }); + + test("resolves the decision as dismissed after the last client disconnects and the grace period elapses", async () => { + delete process.env.PLANNOTATOR_PORT; + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(makeTempDir("plannotator-pi-client-lease-expiry-"), "test.md"), + htmlContent: "annotate", + origin: "pi", + gate: true, + approvalNotesSupported: true, + clientLeaseSupported: true, + clientLeaseTestOverrides: { graceMs: 50 }, + }); + + try { + const decision = server.waitForDecision(); + const isSettled = trackSettled(decision); + const client = await connectClientLease(server.url); + + // Still connected — no expiry. + await Bun.sleep(20); + expect(isSettled()).toBe(false); + + await client.disconnect(); + + expect(await decision).toEqual({ feedback: "", annotations: [], exit: true }); + } finally { + server.stop(); + } + }); + + test("a reconnect before the grace deadline cancels the pending expiry", async () => { + delete process.env.PLANNOTATOR_PORT; + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(makeTempDir("plannotator-pi-client-lease-reconnect-"), "test.md"), + htmlContent: "annotate", + origin: "pi", + gate: true, + approvalNotesSupported: true, + clientLeaseSupported: true, + clientLeaseTestOverrides: { graceMs: 80 }, + }); + + try { + const decision = server.waitForDecision(); + const isSettled = trackSettled(decision); + + const firstClient = await connectClientLease(server.url); + await firstClient.disconnect(); + + // Reconnect well before the 80ms grace deadline. + await Bun.sleep(20); + const secondClient = await connectClientLease(server.url); + + // Even past the original deadline, the reconnect cancelled the pending expiry. + await Bun.sleep(100); + expect(isSettled()).toBe(false); + + // A fresh disconnect starts its own full grace window. + await secondClient.disconnect(); + expect(await decision).toEqual({ feedback: "", annotations: [], exit: true }); + } finally { + server.stop(); + } + }); + + test("an explicit approval wins over a later client-lease expiry", async () => { + delete process.env.PLANNOTATOR_PORT; + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(makeTempDir("plannotator-pi-client-lease-explicit-"), "test.md"), + htmlContent: "annotate", + origin: "pi", + gate: true, + approvalNotesSupported: true, + clientLeaseSupported: true, + clientLeaseTestOverrides: { graceMs: 60 }, + }); + + try { + const client = await connectClientLease(server.url); + + const approve = await fetch(`${server.url}/api/approve`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ feedback: "Looks good.", annotations: [] }), + }); + expect(approve.status).toBe(200); + expect(await server.waitForDecision()).toEqual({ + approved: true, + feedback: "Looks good.", + annotations: [], + }); + + // Disconnecting after the explicit decision must not overwrite it once + // the grace period elapses — the approval already cancelled tracking. + await client.disconnect(); + await Bun.sleep(120); + expect(await server.waitForDecision()).toEqual({ + approved: true, + feedback: "Looks good.", + annotations: [], + }); + } finally { + server.stop(); + } + }); + + test("a decision arriving after the lease expired is rejected instead of reported as applied", async () => { + delete process.env.PLANNOTATOR_PORT; + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(makeTempDir("plannotator-pi-client-lease-late-"), "test.md"), + htmlContent: "annotate", + origin: "pi", + gate: true, + approvalNotesSupported: true, + clientLeaseSupported: true, + clientLeaseTestOverrides: { graceMs: 30 }, + }); + + try { + const client = await connectClientLease(server.url); + await client.disconnect(); + expect(await server.waitForDecision()).toEqual({ + feedback: "", + annotations: [], + exit: true, + }); + + for (const [path, init] of [ + [ + "/api/approve", + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ feedback: "Looks good.", annotations: [] }), + }, + ], + [ + "/api/feedback", + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ feedback: "Please change this.", annotations: [] }), + }, + ], + ["/api/exit", { method: "POST" }], + ] as const) { + const response = await fetch(`${server.url}${path}`, init); + expect(response.status).toBe(409); + } + + expect(await server.waitForDecision()).toEqual({ + feedback: "", + annotations: [], + exit: true, + }); + } finally { + server.stop(); + } + }); + + test("stopping the server ends live lease streams", async () => { + delete process.env.PLANNOTATOR_PORT; + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(makeTempDir("plannotator-pi-client-lease-stop-"), "test.md"), + htmlContent: "annotate", + origin: "pi", + gate: true, + approvalNotesSupported: true, + clientLeaseSupported: true, + }); + + const response = await fetch(`${server.url}/api/annotate/client-lease`); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + const first = await reader.read(); + expect(new TextDecoder().decode(first.value)).toBe(": ready\n\n"); + + server.stop(); + + const next = await reader.read(); + expect(next.done).toBe(true); + }); +}); + describe("pi review server", () => { const testIfJj = hasJj() ? test : test.skip; const testIfUnix = process.platform === "win32" ? test.skip : test; diff --git a/apps/pi-extension/server/serverAnnotate.ts b/apps/pi-extension/server/serverAnnotate.ts index 53d004d6a..d0995b12c 100644 --- a/apps/pi-extension/server/serverAnnotate.ts +++ b/apps/pi-extension/server/serverAnnotate.ts @@ -63,6 +63,14 @@ import { supportsAnnotateAgentTerminalMode, type AgentTerminalCapability, } from "../generated/agent-terminal.ts"; +import { + ANNOTATE_CLIENT_LEASE_GRACE_MS, + ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS, + ANNOTATE_CLIENT_LEASE_STREAM_PATH, + createAnnotateClientLeaseStreamSession, + createAnnotateClientLeaseTracker, +} from "../generated/annotate-client-lease.ts"; +import { createAnnotateDecisionSettler } from "../generated/annotate-decision.ts"; export interface AnnotateServerResult { port: number; @@ -204,6 +212,9 @@ export async function startAnnotateServer(options: { sourceConverted?: boolean; gate?: boolean; approvalNotesSupported?: boolean; + clientLeaseSupported?: boolean; + /** @internal Test-only timing overrides for the client-lease grace/heartbeat intervals — never sleeps real 30s in tests. */ + clientLeaseTestOverrides?: { graceMs?: number; heartbeatMs?: number }; rawHtml?: string; renderHtml?: boolean; convertHtml?: boolean; @@ -238,6 +249,29 @@ export async function startAnnotateServer(options: { resolveDecision = r; }); + // Every decision producer goes through this: connected tabs and the client + // lease below race, and a producer that loses must not delete the reviewer's + // draft or report success for an outcome the caller never received. + const decision = createAnnotateDecisionSettler(resolveDecision); + const sendAlreadyDecided = (res: import("node:http").ServerResponse): void => { + json(res, { error: "This review session has already been decided." }, 409); + }; + + // Last-client abandonment lease: once the tab's client-lease stream + // disconnects (as reported by the transport) and stays disconnected for the + // grace period with no reconnect, the decision resolves as dismissed + // instead of hanging the Pi orchestration forever. Grace timing is bounded + // only for clean disconnects; abrupt/half-open connection loss is detected + // on a best-effort basis by the transport and can take longer than + // graceMs to be noticed at all. See packages/shared/annotate-client-lease.ts + // (vendored to generated/annotate-client-lease.ts by vendor.sh). + const clientLeaseGraceMs = options.clientLeaseTestOverrides?.graceMs ?? ANNOTATE_CLIENT_LEASE_GRACE_MS; + const clientLeaseHeartbeatMs = options.clientLeaseTestOverrides?.heartbeatMs ?? ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS; + const clientLease = createAnnotateClientLeaseTracker( + () => decision.settle({ feedback: "", annotations: [], exit: true }), + { graceMs: clientLeaseGraceMs }, + ); + // Folder annotation has no stable markdown body, so key drafts by folder path instead. const draftSource = options.mode === "annotate-folder" && options.folderPath @@ -413,6 +447,42 @@ export async function startAnnotateServer(options: { if (await externalAnnotations.handle(req, res, url)) return; if (url.pathname.startsWith("/api/ai/") && await handlePiAIRequest(req, res, url, aiRuntime)) return; + // API: Client-lease SSE — see generated/annotate-client-lease.ts. + // Only local direct structured annotate gates advertise and serve + // this; other sessions get a 404. + if (url.pathname === ANNOTATE_CLIENT_LEASE_STREAM_PATH && req.method === "GET") { + if (!options.clientLeaseSupported) { + res.writeHead(404); + res.end("Client lease unavailable"); + return; + } + + res.writeHead(200, { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }); + res.setTimeout(0); + + const session = createAnnotateClientLeaseStreamSession({ + tracker: clientLease, + heartbeatMs: clientLeaseHeartbeatMs, + write: (chunk) => { + res.write(chunk); + }, + endStream: () => res.end(), + }); + + // `req.on("close")` reliably fires on client disconnect; `res.on("close")` + // is registered too as defense-in-depth for Node runtimes/versions where + // only the response object emits it. `session.close()` is idempotent so a + // double-fire is harmless. + req.on("close", () => session.close()); + res.on("close", () => session.close()); + + return; + } + if (url.pathname === "/api/plan" && req.method === "GET") { const displayRawHtml = options.renderHtml && options.rawHtml ? htmlAssets.rewriteHtml(options.rawHtml, options.filePath) @@ -435,6 +505,9 @@ export async function startAnnotateServer(options: { sourceSave: primarySource.sourceSave, gate: options.gate ?? false, approvalNotesSupported: options.approvalNotesSupported ?? false, + clientLease: options.clientLeaseSupported + ? { enabled: true as const, reconnectGraceMs: clientLeaseGraceMs } + : { enabled: false as const }, renderAs: displayRawHtml ? 'html' : 'markdown', ...(displayRawHtml ? { rawHtml: displayRawHtml } : {}), ...(diffHtml ? { diffHtml } : {}), @@ -692,8 +765,12 @@ export async function startAnnotateServer(options: { } else if (url.pathname === "/favicon.png") { handleFavicon(res); } else if (url.pathname === "/api/exit" && req.method === "POST") { + if (!decision.settle({ feedback: "", annotations: [], exit: true })) { + sendAlreadyDecided(res); + return; + } deleteDraft(draftKey, readDraftGenerationFromUrl(req)); - resolveDecision({ feedback: "", annotations: [], exit: true }); + clientLease.cancel(); json(res, { ok: true }); } else if (url.pathname === "/api/approve" && req.method === "POST") { try { @@ -708,8 +785,7 @@ export async function startAnnotateServer(options: { return; } - deleteDraft(draftKey, readDraftGenerationFromBody(body)); - resolveDecision({ + const approvalWon = decision.settle({ feedback: (body.feedback as string | undefined) || "", annotations: (body.annotations as unknown[] | undefined) || [], approved: true, @@ -720,6 +796,12 @@ export async function startAnnotateServer(options: { selectedMessageId: typeof body.selectedMessageId === "string" ? body.selectedMessageId : undefined, feedbackScope: body.feedbackScope === "messages" ? "messages" : body.feedbackScope === "message" ? "message" : undefined, }); + if (!approvalWon) { + sendAlreadyDecided(res); + return; + } + deleteDraft(draftKey, readDraftGenerationFromBody(body)); + clientLease.cancel(); json(res, { ok: true }); } catch (err) { json(res, { error: err instanceof Error ? err.message : "Invalid JSON body." }, 400); @@ -727,13 +809,18 @@ export async function startAnnotateServer(options: { } else if (url.pathname === "/api/feedback" && req.method === "POST") { try { const body = await parseBody(req); - deleteDraft(draftKey, readDraftGenerationFromBody(body)); - resolveDecision({ + const feedbackWon = decision.settle({ feedback: (body.feedback as string) || "", annotations: (body.annotations as unknown[]) || [], selectedMessageId: typeof body.selectedMessageId === "string" ? body.selectedMessageId : undefined, feedbackScope: body.feedbackScope === "messages" ? "messages" : body.feedbackScope === "message" ? "message" : undefined, }); + if (!feedbackWon) { + sendAlreadyDecided(res); + return; + } + deleteDraft(draftKey, readDraftGenerationFromBody(body)); + clientLease.cancel(); json(res, { ok: true }); } catch (err) { const message = err instanceof Error ? err.message : "Failed to process feedback"; @@ -765,6 +852,11 @@ export async function startAnnotateServer(options: { url: `http://localhost:${port}`, waitForDecision: () => decisionPromise, stop: () => { + clientLease.cancel(); + // Long-lived host process: an unclosed lease stream would keep its + // heartbeat timer and socket alive past the session, and would keep + // server.close() from ever completing. + clientLease.closeSessions(); aiRuntime?.dispose(); agentTerminal.dispose(); server.close(); diff --git a/apps/pi-extension/vendor.sh b/apps/pi-extension/vendor.sh index 845197618..14e937e46 100755 --- a/apps/pi-extension/vendor.sh +++ b/apps/pi-extension/vendor.sh @@ -29,7 +29,7 @@ for f in config-types storage-types workspace-status-types; do done # Everything else in the original flat list stays sourced from packages/shared. -for f in prompts review-core diff-paths cli-pagination jj-core gitbutler-core vcs-core review-args draft annotate-history pr-types pr-context-live pr-artifact-document pr-provider pr-stack pr-github pr-gitlab checklist integrations-common repo reference-common resolve-file annotate-reference-roots-node worktree worktree-pool html-to-markdown html-diff html-assets html-assets-node url-to-markdown tour annotate-args at-reference review-workspace-node review-workspace pfm-reminder improvement-hooks code-nav data-dir semantic-diff-types semantic-diff single-flight source-save-node review-profiles guide guide-store commit-avatars commit-history port-range; do +for f in prompts review-core diff-paths cli-pagination jj-core gitbutler-core vcs-core review-args draft annotate-history pr-types pr-context-live pr-artifact-document pr-provider pr-stack pr-github pr-gitlab checklist integrations-common repo reference-common resolve-file annotate-reference-roots-node worktree worktree-pool html-to-markdown html-diff html-assets html-assets-node url-to-markdown tour annotate-args at-reference review-workspace-node review-workspace pfm-reminder improvement-hooks code-nav data-dir semantic-diff-types semantic-diff single-flight source-save-node review-profiles guide guide-store commit-avatars commit-history port-range annotate-client-lease annotate-decision; do src="../../packages/shared/$f.ts" printf '// @generated — DO NOT EDIT. Source: packages/shared/%s.ts\n' "$f" | cat - "$src" > "generated/$f.ts" done diff --git a/packages/editor/App.tsx b/packages/editor/App.tsx index 073374ebe..0de3d5d12 100644 --- a/packages/editor/App.tsx +++ b/packages/editor/App.tsx @@ -146,6 +146,11 @@ import { buildCompleteAnnotateFeedback, getAnnotateApprovalPolicy, } from './annotateSubmission'; +import { + openAnnotateClientLeaseStream, + shouldConnectAnnotateClientLease, + type AnnotateClientLeaseConfig, +} from './annotateClientLease'; import { editableDocumentKey, useEditableDocuments, @@ -398,6 +403,7 @@ const App: React.FC = () => { const [annotateMode, setAnnotateMode] = useState(false); const [gate, setGate] = useState(false); const [approvalNotesSupported, setApprovalNotesSupported] = useState(false); + const [clientLease, setClientLease] = useState(null); const [annotateSource, setAnnotateSource] = useState<'file' | 'message' | 'folder' | null>(null); const [recentMessages, setRecentMessages] = useState([]); const [selectedMessageId, setSelectedMessageId] = useState(null); @@ -2346,7 +2352,7 @@ const App: React.FC = () => { if (!res.ok) throw new Error('Not in API mode'); return res.json(); }) - .then((data: { plan: string; origin?: Origin; mode?: 'annotate' | 'annotate-last' | 'annotate-folder' | 'archive' | 'goal-setup'; goalSetup?: GoalSetupBundle; filePath?: string; sourceInfo?: string; sourceConverted?: boolean; sourceSave?: SourceSaveCapability; gate?: boolean; approvalNotesSupported?: boolean; renderAs?: 'html' | 'markdown'; rawHtml?: string; shareHtml?: string; diffHtml?: string; convertHtml?: boolean; sharingEnabled?: boolean; shareBaseUrl?: string; pasteApiUrl?: string; repoInfo?: { display: string; branch?: string; host?: string }; previousPlan?: string | null; versionInfo?: { version: number; totalVersions: number; project: string }; archivePlans?: ArchivedPlan[]; projectRoot?: string; isWSL?: boolean; serverConfig?: { displayName?: string; gitUser?: string }; recentMessages?: PickerMessage[]; agentTerminal?: AgentTerminalCapability; feedbackTemplates?: AnnotateFeedbackTemplates }) => { + .then((data: { plan: string; origin?: Origin; mode?: 'annotate' | 'annotate-last' | 'annotate-folder' | 'archive' | 'goal-setup'; goalSetup?: GoalSetupBundle; filePath?: string; sourceInfo?: string; sourceConverted?: boolean; sourceSave?: SourceSaveCapability; gate?: boolean; approvalNotesSupported?: boolean; clientLease?: AnnotateClientLeaseConfig; renderAs?: 'html' | 'markdown'; rawHtml?: string; shareHtml?: string; diffHtml?: string; convertHtml?: boolean; sharingEnabled?: boolean; shareBaseUrl?: string; pasteApiUrl?: string; repoInfo?: { display: string; branch?: string; host?: string }; previousPlan?: string | null; versionInfo?: { version: number; totalVersions: number; project: string }; archivePlans?: ArchivedPlan[]; projectRoot?: string; isWSL?: boolean; serverConfig?: { displayName?: string; gitUser?: string }; recentMessages?: PickerMessage[]; agentTerminal?: AgentTerminalCapability; feedbackTemplates?: AnnotateFeedbackTemplates }) => { // Initialize config store with server-provided values (config file > cookie > default) configStore.init(data.serverConfig); // Session-level force-markdown preference (--markdown); threaded into folder/linked @@ -2391,6 +2397,7 @@ const App: React.FC = () => { setAnnotateMode(true); setGate(data.gate ?? false); setApprovalNotesSupported(data.approvalNotesSupported ?? false); + setClientLease(data.clientLease ?? null); } if (data.mode === 'annotate-folder') { sidebar.open('files'); @@ -2465,6 +2472,21 @@ const App: React.FC = () => { .finally(() => setIsLoading(false)); }, [isLoadingShared, isSharedSession]); + // Client-lease: while a local direct structured annotate gate is open, keep + // exactly one EventSource open so the server can detect this tab going away + // and auto-dismiss the gate after its grace period instead of hanging the + // CLI/hook caller forever. Grace only starts once the transport reports a + // disconnect; abrupt/half-open connection loss is best-effort and not + // bounded by the grace period. Only ever a presence signal — no message + // payload is read from the stream. + useEffect(() => { + if (typeof EventSource === 'undefined') return; + if (!shouldConnectAnnotateClientLease({ annotateMode, isSharedSession, submitted, clientLease })) return; + + const stream = openAnnotateClientLeaseStream(EventSource); + return () => stream.close(); + }, [annotateMode, isSharedSession, submitted, clientLease]); + useEffect(() => { if (!aiSessionEnabled || !isApiMode || isSharedSession) { setAiAvailable(false); diff --git a/packages/editor/annotateClientLease.test.ts b/packages/editor/annotateClientLease.test.ts new file mode 100644 index 000000000..76bf768aa --- /dev/null +++ b/packages/editor/annotateClientLease.test.ts @@ -0,0 +1,139 @@ +import { describe, expect, test } from 'bun:test'; +import { + ANNOTATE_CLIENT_LEASE_STREAM_PATH, + shouldConnectAnnotateClientLease, + openAnnotateClientLeaseStream, + type AnnotateClientLeaseConfig, +} from './annotateClientLease'; + +/** Minimal fake EventSource — just enough to observe construction and close(). */ +class FakeEventSource { + static instances: FakeEventSource[] = []; + url: string; + closed = false; + constructor(url: string) { + this.url = url; + FakeEventSource.instances.push(this); + } + close(): void { + this.closed = true; + } +} + +describe('shouldConnectAnnotateClientLease', () => { + const enabled: AnnotateClientLeaseConfig = { enabled: true, reconnectGraceMs: 30_000 }; + const disabled: AnnotateClientLeaseConfig = { enabled: false }; + + test('connects when annotate mode is active, not shared, not yet submitted, and the server enabled it', () => { + expect( + shouldConnectAnnotateClientLease({ + annotateMode: true, + isSharedSession: false, + submitted: null, + clientLease: enabled, + }), + ).toBe(true); + }); + + test('treats an undefined decision the same as null: the session is still open', () => { + expect( + shouldConnectAnnotateClientLease({ + annotateMode: true, + isSharedSession: false, + submitted: undefined, + clientLease: enabled, + }), + ).toBe(true); + }); + + test('does not connect when the server capability is disabled', () => { + expect( + shouldConnectAnnotateClientLease({ + annotateMode: true, + isSharedSession: false, + submitted: null, + clientLease: disabled, + }), + ).toBe(false); + }); + + test('does not connect when clientLease is unavailable (undefined/null — e.g. plan review mode, or /api/plan fetch failed)', () => { + expect( + shouldConnectAnnotateClientLease({ + annotateMode: true, + isSharedSession: false, + submitted: null, + clientLease: undefined, + }), + ).toBe(false); + expect( + shouldConnectAnnotateClientLease({ + annotateMode: true, + isSharedSession: false, + submitted: null, + clientLease: null, + }), + ).toBe(false); + }); + + test('does not connect outside annotate mode (e.g. plan review)', () => { + expect( + shouldConnectAnnotateClientLease({ + annotateMode: false, + isSharedSession: false, + submitted: null, + clientLease: enabled, + }), + ).toBe(false); + }); + + test('does not connect for a shared/static session (no live server to lease against)', () => { + expect( + shouldConnectAnnotateClientLease({ + annotateMode: true, + isSharedSession: true, + submitted: null, + clientLease: enabled, + }), + ).toBe(false); + }); + + test('does not connect once a decision has already been submitted', () => { + for (const submitted of ['approved', 'denied', 'exited'] as const) { + expect( + shouldConnectAnnotateClientLease({ + annotateMode: true, + isSharedSession: false, + submitted, + clientLease: enabled, + }), + ).toBe(false); + } + }); +}); + +describe('openAnnotateClientLeaseStream', () => { + test('opens exactly one EventSource against the shared client-lease path', () => { + FakeEventSource.instances = []; + const stream = openAnnotateClientLeaseStream(FakeEventSource as unknown as typeof EventSource); + + expect(FakeEventSource.instances).toHaveLength(1); + expect(FakeEventSource.instances[0]!.url).toBe(ANNOTATE_CLIENT_LEASE_STREAM_PATH); + + stream.close(); + }); + + test('close() is idempotent and actually closes the underlying EventSource', () => { + FakeEventSource.instances = []; + const stream = openAnnotateClientLeaseStream(FakeEventSource as unknown as typeof EventSource); + const instance = FakeEventSource.instances[0]!; + + expect(instance.closed).toBe(false); + stream.close(); + expect(instance.closed).toBe(true); + + // Calling close() again must not throw or reopen/reconstruct anything. + expect(() => stream.close()).not.toThrow(); + expect(FakeEventSource.instances).toHaveLength(1); + }); +}); diff --git a/packages/editor/annotateClientLease.ts b/packages/editor/annotateClientLease.ts new file mode 100644 index 000000000..cc313fd61 --- /dev/null +++ b/packages/editor/annotateClientLease.ts @@ -0,0 +1,68 @@ +/** + * Editor-side client-lease helper. + * + * The annotate server advertises a last-client abandonment lease (see + * packages/shared/annotate-client-lease.ts) via `/api/plan`'s `clientLease` + * field for local direct structured annotate gates. When enabled, this tab + * opens a single EventSource against the shared stream path so the server + * can detect the tab going away and dismiss the pending gate after a grace + * period — no pagehide/beforeunload/sendBeacon involved; presence is + * inferred purely from the open connection. + */ + +import { ANNOTATE_CLIENT_LEASE_STREAM_PATH } from '@plannotator/shared/annotate-client-lease'; + +export { ANNOTATE_CLIENT_LEASE_STREAM_PATH }; + +export interface AnnotateClientLeaseConfig { + enabled: boolean; + reconnectGraceMs?: number; +} + +export interface ShouldConnectAnnotateClientLeaseInput { + annotateMode: boolean; + isSharedSession: boolean; + /** Decision already taken, if any. Nullish means the session is still open. */ + submitted: string | null | undefined; + clientLease: AnnotateClientLeaseConfig | null | undefined; +} + +/** + * Whether this tab should open the client-lease stream right now. Only one + * live annotate session — not shared/static, not already decided — with the + * server-advertised capability enabled should ever connect. + */ +export function shouldConnectAnnotateClientLease( + input: ShouldConnectAnnotateClientLeaseInput, +): boolean { + return ( + input.annotateMode && + !input.isSharedSession && + input.submitted == null && + !!input.clientLease?.enabled + ); +} + +export interface AnnotateClientLeaseStream { + /** Idempotent — safe to call more than once (e.g. cleanup + explicit completion both fire it). */ + close: () => void; +} + +/** + * Open the single client-lease EventSource for this tab. No message payload + * is read — the connection itself (open vs. closed) is the entire signal; + * the server's ready/heartbeat comments only keep the stream alive. + */ +export function openAnnotateClientLeaseStream( + EventSourceCtor: typeof EventSource, +): AnnotateClientLeaseStream { + const source = new EventSourceCtor(ANNOTATE_CLIENT_LEASE_STREAM_PATH); + let closed = false; + return { + close: () => { + if (closed) return; + closed = true; + source.close(); + }, + }; +} diff --git a/packages/server/annotate.test.ts b/packages/server/annotate.test.ts index 821d112c7..354eb76ac 100644 --- a/packages/server/annotate.test.ts +++ b/packages/server/annotate.test.ts @@ -1138,3 +1138,284 @@ describe("annotate server: approval notes", () => { } }); }); + +describe("annotate server: client lease", () => { + let savedPort: string | undefined; + let savedRemote: string | undefined; + + beforeEach(() => { + savedPort = process.env.PLANNOTATOR_PORT; + savedRemote = process.env.PLANNOTATOR_REMOTE; + delete process.env.PLANNOTATOR_PORT; + process.env.PLANNOTATOR_REMOTE = "0"; + }); + + afterEach(() => { + if (savedPort === undefined) delete process.env.PLANNOTATOR_PORT; + else process.env.PLANNOTATOR_PORT = savedPort; + if (savedRemote === undefined) delete process.env.PLANNOTATOR_REMOTE; + else process.env.PLANNOTATOR_REMOTE = savedRemote; + }); + + /** + * Connect to the client-lease stream and wait for its first byte (the + * ready comment). Returns a `disconnect()` that aborts the underlying + * fetch — plain `reader.cancel()` only stops local reads and does not + * propagate a close to the server's `ReadableStream.cancel()`, whereas + * aborting the request closes the connection the way an abandoned browser + * tab actually would. + */ + async function connectClientLease(url: string): Promise<{ disconnect: () => Promise }> { + const controller = new AbortController(); + const response = await fetch(`${url}/api/annotate/client-lease`, { signal: controller.signal }); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + const first = await Promise.race([ + reader.read(), + new Promise((_, reject) => { + setTimeout(() => reject(new Error("Timed out waiting for ready comment")), 1000); + }), + ]); + expect(first.done).toBe(false); + return { + disconnect: async () => { + controller.abort(); + await reader.cancel().catch(() => {}); + }, + }; + } + + /** + * Track whether a promise has settled without racing it against a timer — + * a `Promise.race` between an already-resolved sentinel and a promise that + * may or may not have settled is nondeterministic. Attaching `.then` up + * front and reading a flag afterward is reliable regardless of timing. + */ + function trackSettled(promise: Promise): () => boolean { + let settled = false; + promise.then(() => { + settled = true; + }); + return () => settled; + } + + test("advertises the effective client-lease capability in /api/plan", async () => { + for (const clientLeaseSupported of [true, false]) { + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(tmpdir(), "client-lease-capability.md"), + htmlContent: MINIMAL_HTML, + gate: true, + approvalNotesSupported: true, + clientLeaseSupported, + }); + + try { + const response = await fetch(`${server.url}/api/plan`); + const plan = await response.json() as { clientLease?: { enabled: boolean; reconnectGraceMs?: number } }; + if (clientLeaseSupported) { + expect(plan.clientLease).toEqual({ enabled: true, reconnectGraceMs: 30_000 }); + } else { + expect(plan.clientLease).toEqual({ enabled: false }); + } + } finally { + server.stop(); + } + } + }); + + test("returns 404 for the client-lease stream when the capability is disabled", async () => { + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(tmpdir(), "client-lease-disabled.md"), + htmlContent: MINIMAL_HTML, + }); + + try { + const response = await fetch(`${server.url}/api/annotate/client-lease`); + expect(response.status).toBe(404); + } finally { + server.stop(); + } + }); + + test("resolves the decision as dismissed after the last client disconnects and the grace period elapses", async () => { + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(tmpdir(), "client-lease-expiry.md"), + htmlContent: MINIMAL_HTML, + gate: true, + approvalNotesSupported: true, + clientLeaseSupported: true, + clientLeaseTestOverrides: { graceMs: 50 }, + }); + + try { + const decision = server.waitForDecision(); + const isSettled = trackSettled(decision); + const client = await connectClientLease(server.url); + + // Still connected — no expiry. + await Bun.sleep(20); + expect(isSettled()).toBe(false); + + await client.disconnect(); + + expect(await decision).toEqual({ feedback: "", annotations: [], exit: true }); + } finally { + server.stop(); + } + }); + + test("a reconnect before the grace deadline cancels the pending expiry", async () => { + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(tmpdir(), "client-lease-reconnect.md"), + htmlContent: MINIMAL_HTML, + gate: true, + approvalNotesSupported: true, + clientLeaseSupported: true, + clientLeaseTestOverrides: { graceMs: 80 }, + }); + + try { + const decision = server.waitForDecision(); + const isSettled = trackSettled(decision); + + const firstClient = await connectClientLease(server.url); + await firstClient.disconnect(); + + // Reconnect well before the 80ms grace deadline. + await Bun.sleep(20); + const secondClient = await connectClientLease(server.url); + + // Even past the original deadline, the reconnect cancelled the pending expiry. + await Bun.sleep(100); + expect(isSettled()).toBe(false); + + // A fresh disconnect starts its own full grace window. + await secondClient.disconnect(); + expect(await decision).toEqual({ feedback: "", annotations: [], exit: true }); + } finally { + server.stop(); + } + }); + + test("an explicit approval wins over a later client-lease expiry", async () => { + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(tmpdir(), "client-lease-explicit-decision.md"), + htmlContent: MINIMAL_HTML, + gate: true, + approvalNotesSupported: true, + clientLeaseSupported: true, + clientLeaseTestOverrides: { graceMs: 60 }, + }); + + try { + const client = await connectClientLease(server.url); + + const approve = await fetch(`${server.url}/api/approve`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ feedback: "Looks good.", annotations: [] }), + }); + expect(approve.status).toBe(200); + expect(await server.waitForDecision()).toEqual({ + approved: true, + feedback: "Looks good.", + annotations: [], + }); + + // Disconnecting after the explicit decision must not overwrite it once + // the grace period elapses — the approval already cancelled tracking. + await client.disconnect(); + await Bun.sleep(120); + expect(await server.waitForDecision()).toEqual({ + approved: true, + feedback: "Looks good.", + annotations: [], + }); + } finally { + server.stop(); + } + }); + + test("a decision arriving after the lease expired is rejected instead of reported as applied", async () => { + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(tmpdir(), "client-lease-late-decision.md"), + htmlContent: MINIMAL_HTML, + gate: true, + approvalNotesSupported: true, + clientLeaseSupported: true, + clientLeaseTestOverrides: { graceMs: 30 }, + }); + + try { + const client = await connectClientLease(server.url); + await client.disconnect(); + expect(await server.waitForDecision()).toEqual({ + feedback: "", + annotations: [], + exit: true, + }); + + // A tab that never saw the dismissal must not be told its decision was + // applied: the caller already received `dismissed`. + for (const [path, init] of [ + [ + "/api/approve", + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ feedback: "Looks good.", annotations: [] }), + }, + ], + [ + "/api/feedback", + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ feedback: "Please change this.", annotations: [] }), + }, + ], + ["/api/exit", { method: "POST" }], + ] as const) { + const response = await fetch(`${server.url}${path}`, init); + expect(response.status).toBe(409); + } + + expect(await server.waitForDecision()).toEqual({ + feedback: "", + annotations: [], + exit: true, + }); + } finally { + server.stop(); + } + }); + + test("stopping the server ends live lease streams", async () => { + const server = await startAnnotateServer({ + markdown: "# Test", + filePath: join(tmpdir(), "client-lease-stop.md"), + htmlContent: MINIMAL_HTML, + gate: true, + approvalNotesSupported: true, + clientLeaseSupported: true, + }); + + const response = await fetch(`${server.url}/api/annotate/client-lease`); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + const first = await reader.read(); + expect(new TextDecoder().decode(first.value)).toBe(": ready\n\n"); + + server.stop(); + + // The stream must complete rather than stay open on a server that is gone. + const next = await reader.read(); + expect(next.done).toBe(true); + }); +}); diff --git a/packages/server/annotate.ts b/packages/server/annotate.ts index 56095d81c..1d2a228d7 100644 --- a/packages/server/annotate.ts +++ b/packages/server/annotate.ts @@ -34,6 +34,15 @@ import { saveSourceFileAtomic, } from "@plannotator/shared/source-save-node"; import { createExternalAnnotationHandler } from "./external-annotations"; +import { + ANNOTATE_CLIENT_LEASE_GRACE_MS, + ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS, + ANNOTATE_CLIENT_LEASE_STREAM_PATH, + createAnnotateClientLeaseStreamSession, + createAnnotateClientLeaseTracker, + type AnnotateClientLeaseStreamSession, +} from "@plannotator/shared/annotate-client-lease"; +import { createAnnotateDecisionSettler } from "@plannotator/shared/annotate-decision"; import { saveConfig, detectGitUser, getServerConfig, loadConfig, resolveAIEnabled, resolveAnnotateHistory } from "./config"; import { existsSync } from "fs"; import { dirname, resolve as resolvePath } from "path"; @@ -88,6 +97,19 @@ export interface AnnotateServerOptions { gate?: boolean; /** Whether this transport can deliver feedback attached to an approval. */ approvalNotesSupported?: boolean; + /** + * Whether this transport can safely resolve an abandoned gate automatically. + * Only local direct structured annotate gates (`--gate --json`, not `--hook`, + * not remote/shared) qualify — see supportsAnnotateClientLease in + * apps/hook/server/annotate-output.ts. + */ + clientLeaseSupported?: boolean; + /** + * @internal Test-only timing overrides for the client-lease grace/heartbeat + * period. Production always uses the real 30s/5s defaults; tests inject + * short values so they don't have to sleep for the real grace period. + */ + clientLeaseTestOverrides?: { graceMs?: number; heartbeatMs?: number }; /** Raw HTML content for direct iframe rendering. */ rawHtml?: string; /** Render HTML as-is in an iframe. */ @@ -151,6 +173,8 @@ export async function startAnnotateServer( pasteApiUrl, gate = false, approvalNotesSupported = false, + clientLeaseSupported = false, + clientLeaseTestOverrides, rawHtml, renderHtml = false, convertHtml = false, @@ -338,6 +362,29 @@ export async function startAnnotateServer( resolveDecision = resolve; }); + // Every decision producer goes through this: connected tabs and the client + // lease below race, and a producer that loses must not delete the reviewer's + // draft or report success for an outcome the caller never received. + const decision = createAnnotateDecisionSettler(resolveDecision!); + const alreadyDecided = () => + Response.json({ error: "This review session has already been decided." }, { status: 409 }); + + // Last-client abandonment lease: once the tab's client-lease stream + // disconnects (as reported by the transport) and stays disconnected for the + // grace period with no reconnect, the decision resolves as dismissed + // instead of hanging the CLI/hook caller forever. Only meaningful once at + // least one client connects. Grace timing is bounded only for clean + // disconnects; abrupt/half-open connection loss is detected on a + // best-effort basis by the transport (e.g. a failing heartbeat write) and + // can take longer than graceMs to be noticed at all — see + // packages/shared/annotate-client-lease.ts. + const clientLeaseGraceMs = clientLeaseTestOverrides?.graceMs ?? ANNOTATE_CLIENT_LEASE_GRACE_MS; + const clientLeaseHeartbeatMs = clientLeaseTestOverrides?.heartbeatMs ?? ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS; + const clientLease = createAnnotateClientLeaseTracker( + () => decision.settle({ feedback: "", annotations: [], exit: true }), + { graceMs: clientLeaseGraceMs }, + ); + const server = await startBunServerOnAvailablePort((port) => Bun.serve({ hostname: getServerHostname(), @@ -380,6 +427,9 @@ export async function startAnnotateServer( sourceSave: primarySource.sourceSave, gate, approvalNotesSupported, + clientLease: clientLeaseSupported + ? { enabled: true as const, reconnectGraceMs: clientLeaseGraceMs } + : { enabled: false as const }, renderAs: displayRawHtml ? 'html' as const : 'markdown' as const, ...(displayRawHtml ? { rawHtml: displayRawHtml } : {}), ...(diffHtml ? { diffHtml } : {}), @@ -668,6 +718,41 @@ export async function startAnnotateServer( return handleDraftLoad(draftKey); } + // API: Client-lease SSE — see packages/shared/annotate-client-lease.ts. + // Only local direct structured annotate gates (--gate --json) advertise + // and serve this; other transports get a 404 (idleTimeout is already 0 + // for the whole server above, so no per-connection opt-out is needed). + if (url.pathname === ANNOTATE_CLIENT_LEASE_STREAM_PATH && req.method === "GET") { + if (!clientLeaseSupported) { + return new Response("Client lease unavailable", { status: 404 }); + } + + const encoder = new TextEncoder(); + let session: AnnotateClientLeaseStreamSession | null = null; + + const stream = new ReadableStream({ + start(controller) { + session = createAnnotateClientLeaseStreamSession({ + tracker: clientLease, + heartbeatMs: clientLeaseHeartbeatMs, + write: (chunk) => controller.enqueue(encoder.encode(chunk)), + endStream: () => controller.close(), + }); + }, + cancel() { + session?.close(); + }, + }); + + return new Response(stream, { + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }, + }); + } + // API: External annotations (SSE-based, for any external tool) const externalResponse = await externalAnnotations.handle(req, url, { disableIdleTimeout: () => server.timeout(req, 0), @@ -696,8 +781,11 @@ export async function startAnnotateServer( // API: Exit annotation session without feedback if (url.pathname === "/api/exit" && req.method === "POST") { + if (!decision.settle({ feedback: "", annotations: [], exit: true })) { + return alreadyDecided(); + } deleteDraft(draftKey, readDraftGenerationFromUrl(req)); - resolveDecision({ feedback: "", annotations: [], exit: true }); + clientLease.cancel(); return Response.json({ ok: true }); } @@ -728,8 +816,7 @@ export async function startAnnotateServer( return Response.json({ error: "Invalid approval body." }, { status: 400 }); } - deleteDraft(draftKey, readDraftGenerationFromBody(body)); - resolveDecision({ + const approvalWon = decision.settle({ feedback: (body.feedback as string | undefined) || "", annotations: (body.annotations as unknown[] | undefined) || [], approved: true, @@ -746,6 +833,9 @@ export async function startAnnotateServer( ? "message" : undefined, }); + if (!approvalWon) return alreadyDecided(); + deleteDraft(draftKey, readDraftGenerationFromBody(body)); + clientLease.cancel(); return Response.json({ ok: true }); } @@ -760,13 +850,15 @@ export async function startAnnotateServer( draftGeneration?: number; }; - deleteDraft(draftKey, readDraftGenerationFromBody(body)); - resolveDecision({ + const feedbackWon = decision.settle({ feedback: body.feedback || "", annotations: body.annotations || [], selectedMessageId: body.selectedMessageId, feedbackScope: body.feedbackScope, }); + if (!feedbackWon) return alreadyDecided(); + deleteDraft(draftKey, readDraftGenerationFromBody(body)); + clientLease.cancel(); return Response.json({ ok: true }); } catch (err) { @@ -826,6 +918,8 @@ export async function startAnnotateServer( isRemote, waitForDecision: () => decisionPromise, stop: () => { + clientLease.cancel(); + clientLease.closeSessions(); aiRuntime?.dispose(); agentTerminal.dispose(); server.stop(); diff --git a/packages/shared/annotate-client-lease.test.ts b/packages/shared/annotate-client-lease.test.ts new file mode 100644 index 000000000..fc01d90bb --- /dev/null +++ b/packages/shared/annotate-client-lease.test.ts @@ -0,0 +1,391 @@ +import { describe, expect, test } from "bun:test"; + +import { + ANNOTATE_CLIENT_LEASE_GRACE_MS, + ANNOTATE_CLIENT_LEASE_HEARTBEAT_COMMENT, + ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS, + ANNOTATE_CLIENT_LEASE_READY_COMMENT, + createAnnotateClientLeaseStreamSession, + createAnnotateClientLeaseTracker, +} from "./annotate-client-lease"; + +/** + * Deterministic virtual clock — tests drive expiry by advancing simulated + * time instead of sleeping on the real 30s grace period. Timers fire in + * scheduled order when `advance()` crosses their deadline. + */ +function createFakeScheduler() { + let nextId = 1; + let now = 0; + const timers = new Map void }>(); + + return { + setTimer: (callback: () => void, ms: number): number => { + const id = nextId++; + timers.set(id, { at: now + ms, callback }); + return id; + }, + clearTimer: (id: number): void => { + timers.delete(id); + }, + advance(ms: number): void { + now += ms; + const due = [...timers.entries()] + .filter(([, timer]) => timer.at <= now) + .sort((a, b) => a[1].at - b[1].at); + for (const [id, timer] of due) { + timers.delete(id); + timer.callback(); + } + }, + }; +} + +describe("annotate client-lease tracker: defaults", () => { + test("exposes the documented heartbeat and grace constants", () => { + expect(ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS).toBe(5_000); + expect(ANNOTATE_CLIENT_LEASE_GRACE_MS).toBe(30_000); + }); +}); + +describe("annotate client-lease tracker: never-connected", () => { + test("never expires when no client ever connects", () => { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer }); + + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS * 10); + + expect(expireCalls).toBe(0); + expect(tracker.activeCount()).toBe(0); + expect(tracker.isExpired()).toBe(false); + }); +}); + +describe("annotate client-lease tracker: last disconnect expiry", () => { + test("expires exactly at the grace deadline after the only client disconnects", () => { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer }); + + const release = tracker.acquire(); + expect(tracker.activeCount()).toBe(1); + + release(); + expect(tracker.activeCount()).toBe(0); + + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS - 1); + expect(expireCalls).toBe(0); + expect(tracker.isExpired()).toBe(false); + + scheduler.advance(1); + expect(expireCalls).toBe(1); + expect(tracker.isExpired()).toBe(true); + }); + + test("honors a custom graceMs override", () => { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer, graceMs: 1_000 }); + + tracker.acquire()(); + scheduler.advance(999); + expect(expireCalls).toBe(0); + scheduler.advance(1); + expect(expireCalls).toBe(1); + }); +}); + +describe("annotate client-lease tracker: reconnect cancel", () => { + test("a reconnect before the grace deadline cancels the pending expiry", () => { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer }); + + tracker.acquire()(); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS / 2); + expect(expireCalls).toBe(0); + + // Reconnect cancels the pending timer — advancing past the original + // deadline must not fire it. + const release = tracker.acquire(); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS); + expect(expireCalls).toBe(0); + expect(tracker.activeCount()).toBe(1); + + // A fresh disconnect starts its own full grace window. + release(); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS - 1); + expect(expireCalls).toBe(0); + scheduler.advance(1); + expect(expireCalls).toBe(1); + }); +}); + +describe("annotate client-lease tracker: multiple clients", () => { + test("only expires once every active client has released", () => { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer }); + + const releaseFirst = tracker.acquire(); + const releaseSecond = tracker.acquire(); + expect(tracker.activeCount()).toBe(2); + + releaseFirst(); + expect(tracker.activeCount()).toBe(1); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS * 2); + expect(expireCalls).toBe(0); + + releaseSecond(); + expect(tracker.activeCount()).toBe(0); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS - 1); + expect(expireCalls).toBe(0); + scheduler.advance(1); + expect(expireCalls).toBe(1); + }); + + test("release is idempotent — calling it twice does not double-decrement", () => { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer }); + + const releaseFirst = tracker.acquire(); + tracker.acquire(); + expect(tracker.activeCount()).toBe(2); + + releaseFirst(); + releaseFirst(); + releaseFirst(); + expect(tracker.activeCount()).toBe(1); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS); + expect(expireCalls).toBe(0); + }); +}); + +describe("annotate client-lease tracker: cancel", () => { + test("cancel() permanently stops tracking, even mid-grace", () => { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer }); + + tracker.acquire()(); + tracker.cancel(); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS * 5); + expect(expireCalls).toBe(0); + + // Acquiring after cancel is a safe no-op — it must not resurrect tracking. + const release = tracker.acquire(); + expect(tracker.activeCount()).toBe(0); + release(); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS * 5); + expect(expireCalls).toBe(0); + }); + + test("activeCount() stays truthful when a client disconnects after cancel()", () => { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer }); + + const release = tracker.acquire(); + expect(tracker.activeCount()).toBe(1); + + tracker.cancel(); + release(); + + expect(tracker.activeCount()).toBe(0); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS * 5); + expect(expireCalls).toBe(0); + }); + + test("cancel() is idempotent and safe before any client ever connects", () => { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer }); + + tracker.cancel(); + tracker.cancel(); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS * 5); + expect(expireCalls).toBe(0); + expect(tracker.isExpired()).toBe(false); + }); +}); + +describe("annotate client-lease tracker: once", () => { + test("the expiry callback fires at most once", () => { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer }); + + tracker.acquire()(); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS); + expect(expireCalls).toBe(1); + + // Further time passing, or redundant releases, must not refire it. + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS * 10); + expect(expireCalls).toBe(1); + expect(tracker.isExpired()).toBe(true); + }); +}); + +describe("annotate client-lease stream session", () => { + function setup(options: { failReady?: boolean; failHeartbeatAfter?: number } = {}) { + const scheduler = createFakeScheduler(); + let expireCalls = 0; + const tracker = createAnnotateClientLeaseTracker(() => { + expireCalls += 1; + }, { setTimer: scheduler.setTimer, clearTimer: scheduler.clearTimer }); + + const written: string[] = []; + let heartbeats = 0; + const session = createAnnotateClientLeaseStreamSession({ + tracker, + heartbeatMs: ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS, + // The fake scheduler is one-shot, so re-arm to emulate setInterval. + setHeartbeat: (callback, ms) => { + const repeat = () => { + scheduler.setTimer(repeat, ms); + callback(); + }; + return scheduler.setTimer(repeat, ms); + }, + clearHeartbeat: (handle) => scheduler.clearTimer(handle as number), + write: (chunk) => { + if (chunk === ANNOTATE_CLIENT_LEASE_READY_COMMENT && options.failReady) { + throw new Error("peer gone"); + } + if (chunk === ANNOTATE_CLIENT_LEASE_HEARTBEAT_COMMENT) { + heartbeats += 1; + if (options.failHeartbeatAfter !== undefined && heartbeats > options.failHeartbeatAfter) { + throw new Error("peer gone"); + } + } + written.push(chunk); + }, + }); + + return { scheduler, tracker, session, written, expireCalls: () => expireCalls }; + } + + test("acquires the slot and writes the ready comment on connect", () => { + const { tracker, written, session } = setup(); + + expect(written).toEqual([ANNOTATE_CLIENT_LEASE_READY_COMMENT]); + expect(tracker.activeCount()).toBe(1); + expect(session.isClosed()).toBe(false); + }); + + test("heartbeats on the configured interval while connected", () => { + const { scheduler, written, tracker, expireCalls } = setup(); + + // One advance per interval: the fake scheduler fires only timers that were + // already due when it was called, so a re-armed heartbeat needs its own tick. + scheduler.advance(ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS); + scheduler.advance(ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS); + scheduler.advance(ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS); + + expect(written.filter((c) => c === ANNOTATE_CLIENT_LEASE_HEARTBEAT_COMMENT)).toHaveLength(3); + expect(tracker.activeCount()).toBe(1); + expect(expireCalls()).toBe(0); + }); + + test("close() releases the slot once and starts the grace period", () => { + const { scheduler, session, tracker, expireCalls } = setup(); + + session.close(); + session.close(); + + expect(session.isClosed()).toBe(true); + expect(tracker.activeCount()).toBe(0); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS); + expect(expireCalls()).toBe(1); + }); + + test("a failed ready write closes the session so the gate stays dismissable", () => { + const { scheduler, session, tracker, expireCalls } = setup({ failReady: true }); + + expect(session.isClosed()).toBe(true); + expect(tracker.activeCount()).toBe(0); + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS); + expect(expireCalls()).toBe(1); + }); + + test("a failed heartbeat write releases the slot instead of holding it forever", () => { + const { scheduler, session, tracker, expireCalls } = setup({ failHeartbeatAfter: 1 }); + + scheduler.advance(ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS); + expect(tracker.activeCount()).toBe(1); + + scheduler.advance(ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS); + expect(session.isClosed()).toBe(true); + expect(tracker.activeCount()).toBe(0); + + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS); + expect(expireCalls()).toBe(1); + }); + + test("closeSessions() ends every live session, so shutdown leaves nothing running", () => { + const { scheduler, tracker, session, written, expireCalls } = setup(); + + tracker.cancel(); + tracker.closeSessions(); + + expect(session.isClosed()).toBe(true); + expect(tracker.activeCount()).toBe(0); + scheduler.advance(ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS * 5); + expect(written).toEqual([ANNOTATE_CLIENT_LEASE_READY_COMMENT]); + // Cancelled means abandonment no longer matters: no expiry may fire. + scheduler.advance(ANNOTATE_CLIENT_LEASE_GRACE_MS); + expect(expireCalls()).toBe(0); + }); + + test("close() ends the underlying stream exactly once", () => { + const scheduler = createFakeScheduler(); + const tracker = createAnnotateClientLeaseTracker(() => {}, { + setTimer: scheduler.setTimer, + clearTimer: scheduler.clearTimer, + }); + let ended = 0; + const session = createAnnotateClientLeaseStreamSession({ + tracker, + write: () => {}, + endStream: () => { + ended += 1; + }, + }); + + session.close(); + session.close(); + + expect(ended).toBe(1); + }); + + test("no heartbeat is written after close", () => { + const { scheduler, session, written } = setup(); + + session.close(); + scheduler.advance(ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS * 5); + + expect(written).toEqual([ANNOTATE_CLIENT_LEASE_READY_COMMENT]); + }); +}); diff --git a/packages/shared/annotate-client-lease.ts b/packages/shared/annotate-client-lease.ts new file mode 100644 index 000000000..49c4083e3 --- /dev/null +++ b/packages/shared/annotate-client-lease.ts @@ -0,0 +1,247 @@ +/** + * Annotate client-lease tracker — pure last-client abandonment detector. + * + * Local direct structured annotate gates (`plannotator annotate --gate --json`) + * block a CLI/hook caller on `waitForDecision()`. If the browser tab that owns + * the decision goes away without ever sending `/api/exit`/`/api/approve` + * (closed tab, killed terminal) the gate would otherwise hang forever. + * + * This tracker gives the server a safe, connection-based signal instead: it + * counts concurrently connected SSE clients (normally exactly one — the tab's + * own client-lease stream), and once the *last* one disconnects, waits a grace + * period (default 30s) for a reconnect (tab refresh) before firing a one-shot + * expiry callback. A tab that never connects at all never expires — there is + * nothing to abandon yet. + * + * The grace clock only starts once the server's transport actually reports a + * disconnect; this tracker has no notion of the network itself. A clean close + * (tab closed, navigated away) reports promptly. An abrupt loss — killed + * process, unplugged network, a half-open TCP connection with no traffic — + * is detected on a best-effort basis by the transport (e.g. failing + * heartbeat writes) and is not bounded by `graceMs`: it can take longer than + * the grace period for the transport to notice the peer is gone at all, + * during which this tracker still believes a client is connected. + * + * Deliberately dependency-free: no fetch, no DOM, no framework. Bun and Pi + * servers each wire this to their own SSE transport; the editor never talks + * to it directly (see packages/editor/annotateClientLease.ts for the client + * side of the wire protocol). + */ + +/** Default reconnect grace period after the last client disconnects. */ +export const ANNOTATE_CLIENT_LEASE_GRACE_MS = 30_000; + +/** Default SSE heartbeat interval used by server transports. */ +export const ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS = 5_000; + +/** SSE route path shared by the Bun and Pi annotate servers and the editor client. */ +export const ANNOTATE_CLIENT_LEASE_STREAM_PATH = "/api/annotate/client-lease"; + +/** First byte written once a client-lease stream is open and has acquired the tracker. */ +export const ANNOTATE_CLIENT_LEASE_READY_COMMENT = ": ready\n\n"; + +/** Keep-alive comment written every heartbeat interval. */ +export const ANNOTATE_CLIENT_LEASE_HEARTBEAT_COMMENT = ": heartbeat\n\n"; + +export type AnnotateClientLeaseTimerHandle = unknown; + +export interface AnnotateClientLeaseTrackerOptions { + /** Milliseconds to wait after the last client disconnects before expiring. Default 30_000. */ + graceMs?: number; + /** Injectable scheduler — production uses real timers, tests supply a virtual clock. */ + setTimer?: (callback: () => void, ms: number) => AnnotateClientLeaseTimerHandle; + /** Injectable timer cancellation matching `setTimer`. */ + clearTimer?: (handle: AnnotateClientLeaseTimerHandle) => void; +} + +export interface AnnotateClientLeaseTracker { + /** + * Register a newly connected client. Returns a release callback to call on + * disconnect — idempotent, safe to call more than once for the same client. + */ + acquire: () => () => void; + /** + * Permanently stop tracking — e.g. an explicit decision (approve/feedback/ + * exit) already resolved the gate, so abandonment no longer matters. Safe + * to call repeatedly, and safe to call before any client ever connects. + */ + cancel: () => void; + /** + * Close every live stream session created against this tracker. Servers call + * this while shutting down: releasing the slot is not enough, because each + * session also owns a heartbeat timer and an open response that would + * otherwise outlive the session (in a long-lived host process, forever). + */ + closeSessions: () => void; + /** Number of currently connected (not yet released) clients. */ + activeCount: () => number; + /** Whether the one-shot expiry callback has already fired. */ + isExpired: () => boolean; + /** @internal Session bookkeeping for `closeSessions`. */ + registerSession: (session: { close: () => void }) => () => void; +} + +/** + * Create a tracker that invokes `onExpire` at most once, after the last + * connected client has stayed disconnected for the full grace period. + */ +export function createAnnotateClientLeaseTracker( + onExpire: () => void, + options: AnnotateClientLeaseTrackerOptions = {}, +): AnnotateClientLeaseTracker { + const graceMs = options.graceMs ?? ANNOTATE_CLIENT_LEASE_GRACE_MS; + const setTimer = options.setTimer ?? ((callback, ms) => setTimeout(callback, ms)); + const clearTimer = options.clearTimer ?? ((handle) => clearTimeout(handle as ReturnType)); + + let active = 0; + let cancelled = false; + let expired = false; + let graceTimer: AnnotateClientLeaseTimerHandle | null = null; + const sessions = new Set<{ close: () => void }>(); + + function clearGraceTimer(): void { + if (graceTimer !== null) { + clearTimer(graceTimer); + graceTimer = null; + } + } + + function scheduleExpiry(): void { + clearGraceTimer(); + graceTimer = setTimer(() => { + graceTimer = null; + if (cancelled || expired || active > 0) return; + expired = true; + onExpire(); + }, graceMs); + } + + return { + acquire() { + if (cancelled || expired) return () => {}; + + active += 1; + // A (re)connect always cancels a pending expiry — including the very + // first connect, which is a no-op here since none is scheduled yet. + clearGraceTimer(); + + let released = false; + return () => { + if (released) return; + released = true; + // Always account for the disconnect, so activeCount() stays truthful + // after cancellation; only the expiry scheduling is suppressed. + active = Math.max(0, active - 1); + if (cancelled || expired) return; + if (active === 0) scheduleExpiry(); + }; + }, + cancel() { + cancelled = true; + clearGraceTimer(); + }, + closeSessions() { + // close() unregisters, so iterate a snapshot. + for (const session of [...sessions]) session.close(); + }, + activeCount() { + return active; + }, + isExpired() { + return expired; + }, + registerSession(session) { + sessions.add(session); + return () => sessions.delete(session); + }, + }; +} + +export interface AnnotateClientLeaseStreamSessionOptions { + /** Tracker owning this session's presence slot. */ + tracker: AnnotateClientLeaseTracker; + /** + * End the underlying response, if the transport can. Called on close so a + * server-side shutdown actually finishes the stream instead of leaving the + * client hanging on a connection nobody heartbeats any more. + */ + endStream?: () => void; + /** Transport write. Must throw (or be replaced) when the peer is gone. */ + write: (chunk: string) => void; + /** Heartbeat interval. Defaults to `ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS`. */ + heartbeatMs?: number; + /** Injectable repeating scheduler; production uses real timers. */ + setHeartbeat?: (callback: () => void, ms: number) => AnnotateClientLeaseTimerHandle; + /** Injectable cancellation matching `setHeartbeat`. */ + clearHeartbeat?: (handle: AnnotateClientLeaseTimerHandle) => void; +} + +export interface AnnotateClientLeaseStreamSession { + /** Stop heartbeating and release the presence slot. Idempotent. */ + close: () => void; + /** Whether the session has already been closed. */ + isClosed: () => boolean; +} + +/** + * Wire one connected client-lease stream to a tracker. + * + * Runtime-agnostic on purpose: the Bun server passes a `ReadableStream` + * controller enqueue, the Pi server passes `res.write`, and tests pass a + * writer that throws. Keeping the acquire/ready/heartbeat/release sequence in + * one place is what makes the two runtimes provably identical, including the + * case that is otherwise unreachable from an integration test: a write that + * fails must close the session, because a stream that can no longer be written + * to is a client that is no longer present. Leaving the slot held there would + * make the gate un-dismissable for the rest of the session. + */ +export function createAnnotateClientLeaseStreamSession( + options: AnnotateClientLeaseStreamSessionOptions, +): AnnotateClientLeaseStreamSession { + const heartbeatMs = options.heartbeatMs ?? ANNOTATE_CLIENT_LEASE_HEARTBEAT_MS; + const setHeartbeat = options.setHeartbeat ?? ((callback, ms) => setInterval(callback, ms)); + const clearHeartbeat = + options.clearHeartbeat ?? ((handle) => clearInterval(handle as ReturnType)); + + const release = options.tracker.acquire(); + let closed = false; + let heartbeat: AnnotateClientLeaseTimerHandle | null = null; + + function close(): void { + if (closed) return; + closed = true; + if (heartbeat !== null) { + clearHeartbeat(heartbeat); + heartbeat = null; + } + unregister(); + release(); + // Ending an already-dead stream throws in both runtimes; that is exactly + // the case where there is nothing left to end. + try { + options.endStream?.(); + } catch { + // ignore + } + } + + const unregister = options.tracker.registerSession({ close: () => close() }); + + try { + options.write(ANNOTATE_CLIENT_LEASE_READY_COMMENT); + } catch { + close(); + return { close, isClosed: () => closed }; + } + + heartbeat = setHeartbeat(() => { + if (closed) return; + try { + options.write(ANNOTATE_CLIENT_LEASE_HEARTBEAT_COMMENT); + } catch { + close(); + } + }, heartbeatMs); + + return { close, isClosed: () => closed }; +} diff --git a/packages/shared/annotate-decision.test.ts b/packages/shared/annotate-decision.test.ts new file mode 100644 index 000000000..77f08b2f0 --- /dev/null +++ b/packages/shared/annotate-decision.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, test } from "bun:test"; +import { createAnnotateDecisionSettler } from "./annotate-decision"; + +describe("annotate decision settler", () => { + test("the first producer wins and resolves exactly once", () => { + const resolved: string[] = []; + const decision = createAnnotateDecisionSettler((d) => resolved.push(d)); + + expect(decision.settle("approved")).toBe(true); + expect(decision.isSettled()).toBe(true); + expect(resolved).toEqual(["approved"]); + }); + + test("a later producer loses and cannot resolve", () => { + const resolved: string[] = []; + const decision = createAnnotateDecisionSettler((d) => resolved.push(d)); + + decision.settle("dismissed"); + + expect(decision.settle("approved")).toBe(false); + expect(decision.settle("annotated")).toBe(false); + expect(resolved).toEqual(["dismissed"]); + }); + + test("nothing is settled before the first producer", () => { + const decision = createAnnotateDecisionSettler(() => {}); + expect(decision.isSettled()).toBe(false); + }); +}); diff --git a/packages/shared/annotate-decision.ts b/packages/shared/annotate-decision.ts new file mode 100644 index 000000000..9121aa940 --- /dev/null +++ b/packages/shared/annotate-decision.ts @@ -0,0 +1,37 @@ +/** + * One-shot settlement for an annotate session's decision. + * + * An annotate session has several independent decision producers: any connected + * tab can approve, send feedback, or close, and the client lease can expire on + * its own (see annotate-client-lease.ts). The awaited promise already ignores a + * second resolve, which silently hides the problem: a late producer still runs + * its side effects (deleting the reviewer's draft) and still answers `ok`, so a + * tab reports success for a decision the caller never received. + * + * Routing every producer through one settler makes the winner explicit. A + * producer that loses must run no side effect and must tell its caller it lost, + * rather than claiming an outcome that did not happen. + */ +export interface AnnotateDecisionSettler { + /** Resolve the session with this decision. Returns false if one already won. */ + settle: (decision: TDecision) => boolean; + /** Whether some producer has already won. */ + isSettled: () => boolean; +} + +export function createAnnotateDecisionSettler( + resolve: (decision: TDecision) => void, +): AnnotateDecisionSettler { + let settled = false; + return { + settle(decision) { + if (settled) return false; + settled = true; + resolve(decision); + return true; + }, + isSettled() { + return settled; + }, + }; +} diff --git a/packages/shared/package.json b/packages/shared/package.json index a769f59e1..4622a0867 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -69,7 +69,9 @@ "./port-range": "./port-range.ts", "./plan-review-lifecycle": "./plan-review-lifecycle.ts", "./review-profiles": "./review-profiles.ts", - "./commit-avatars": "./commit-avatars.ts" + "./commit-avatars": "./commit-avatars.ts", + "./annotate-client-lease": "./annotate-client-lease.ts", + "./annotate-decision": "./annotate-decision.ts" }, "dependencies": { "@plannotator/core": "workspace:*",