poll: return closed when the review ended before the poll started

src/server.js

@@ -43,6 +43,13 @@ const MAX_LOCAL_REDIRECTS = 5;
 /** Generous enough for a dev server's cold compile, but a wedged one can't hang us forever. */
 const LOCAL_FETCH_TIMEOUT_MS = 30000;
 const MAX_LOCAL_PAGE_BYTES = 24 * 1024 * 1024;
+/** What a poll gets once the human has ended the review. One wording, two delivery paths. */
+const SESSION_CLOSED = {
+  status: "closed",
+  next_step:
+    "The user ended this review session. Stop polling — do not run the poll command again. " +
+    "Any unsent feedback is kept and will ship the next time this target is reviewed.",
+};
 
 const hash = (text) => crypto.createHash("sha1").update(text).digest("hex");
 const uid = (prefix) => `${prefix}_${crypto.randomBytes(6).toString("hex")}`;
@@ -325,19 +332,17 @@ export function createServer() {
     session.clients.clear();
     // Another window on the same target keeps its agent connection alive.
     if (sessionsForEntry(session.entryKey).length > 0) return;
+    // Leave a mark for a poll that has not started yet. Draining the pollers below
+    // only reaches agents already waiting, and the documented loop is apply-then-poll,
+    // so the poll that follows a delivered batch usually arrives after this point.
+    store.markEnded(session.entryKey);
     const set = pollers.get(session.entryKey);
     if (!set) return;
     for (const poller of [...set]) {
       clearInterval(poller.timer);
       set.delete(poller);
-      poller.res.end(
-        JSON.stringify({
-          status: "closed",
-          next_step:
-            "The user ended this review session. Stop polling — do not run the poll command again. " +
-            "Any unsent feedback is kept and will ship the next time this target is reviewed.",
-        })
-      );
+      store.takeEnded(session.entryKey);
+      poller.res.end(JSON.stringify(SESSION_CLOSED));
     }
   }
 
@@ -843,6 +848,9 @@ export function createServer() {
           return json(res, 200, pending.batch);
         }
 
+        // Feedback first: an end mark must never hide a batch the human already sent.
+        if (store.takeEnded(entryKey)) return json(res, 200, SESSION_CLOSED);
+
         res.writeHead(200, { "content-type": "application/json; charset=utf-8" });
         res.write(" ");
         const set = pollers.get(entryKey) || new Set();

src/state.js

@@ -33,6 +33,7 @@ export function atomicWrite(file, data) {
  *   {
  *     pages:   { <key>: { key, file, pristine, comments[], edits[], updatedAt } },
  *     batches: { <entryKey>: { batch, cleanup, updatedAt } },
+ *     ended:   { <entryKey>: { updatedAt } },
  *   }
  *
  * Pages are fully independent: no page ever references another. Batches are
@@ -41,9 +42,11 @@ export function atomicWrite(file, data) {
  */
 export class Store {
   constructor() {
-    this.data = { pages: {}, batches: {} };
+    this.data = { pages: {}, batches: {}, ended: {} };
     /** Batches this process acked; save() must not resurrect them from disk. */
     this.clearedBatches = new Set();
+    /** End marks this process handed to a poller; same reason. */
+    this.clearedEnded = new Set();
     this.load();
   }
 
@@ -52,7 +55,7 @@ export class Store {
       const raw = fs.readFileSync(statePath(), "utf8");
       const parsed = JSON.parse(raw);
       if (parsed && typeof parsed === "object" && parsed.pages) {
-        this.data = { pages: parsed.pages, batches: parsed.batches || {} };
+        this.data = { pages: parsed.pages, batches: parsed.batches || {}, ended: parsed.ended || {} };
       }
     } catch {
       // Missing or unreadable state is not an error; start empty.
@@ -74,6 +77,9 @@ export class Store {
     for (const [key, batch] of Object.entries(this.data.batches)) {
       if (!fresh(batch, now)) delete this.data.batches[key];
     }
+    for (const [key, mark] of Object.entries(this.data.ended)) {
+      if (!fresh(mark, now)) delete this.data.ended[key];
+    }
   }
 
   /**
@@ -84,18 +90,20 @@ export class Store {
   save() {
     ensureStateDir();
     const target = statePath();
-    let onDisk = { pages: {}, batches: {} };
+    let onDisk = { pages: {}, batches: {}, ended: {} };
     try {
       const parsed = JSON.parse(fs.readFileSync(target, "utf8"));
-      if (parsed && parsed.pages) onDisk = { pages: parsed.pages, batches: parsed.batches || {} };
+      if (parsed && parsed.pages) onDisk = { pages: parsed.pages, batches: parsed.batches || {}, ended: parsed.ended || {} };
     } catch {
       // No readable state yet; ours becomes the file.
     }
     const merged = {
       pages: { ...onDisk.pages, ...this.data.pages },
       batches: { ...onDisk.batches, ...this.data.batches },
+      ended: { ...onDisk.ended, ...this.data.ended },
     };
     for (const key of this.clearedBatches) delete merged.batches[key];
+    for (const key of this.clearedEnded) delete merged.ended[key];
     // Age-prune the merged result too, so the file cannot grow without bound.
     const now = Date.now();
     for (const [key, page] of Object.entries(merged.pages)) {
@@ -104,9 +112,40 @@ export class Store {
     for (const [key, batch] of Object.entries(merged.batches)) {
       if (!fresh(batch, now)) delete merged.batches[key];
     }
+    for (const [key, mark] of Object.entries(merged.ended)) {
+      if (!fresh(mark, now)) delete merged.ended[key];
+    }
     atomicWrite(target, JSON.stringify(merged, null, 2));
   }
 
+  /**
+   * Remember that the human ended the review, for a poll that has not started yet.
+   * endSession can only hand `closed` to pollers already waiting, so an agent that
+   * applies a batch and then polls again — the loop SKILL.md prescribes — would
+   * otherwise register on a target nothing will ever close and burn its whole timeout.
+   */
+  markEnded(entryKey) {
+    this.clearedEnded.delete(entryKey);
+    this.data.ended[entryKey] = { updatedAt: Date.now() };
+    this.save();
+  }
+
+  /** Consume the end mark. True once per ended session, then false until the next one. */
+  takeEnded(entryKey) {
+    if (!this.data.ended[entryKey]) return false;
+    delete this.data.ended[entryKey];
+    this.clearedEnded.add(entryKey);
+    this.save();
+    return true;
+  }
+
+  /** A target being opened again is a live review, so any end mark is stale. */
+  clearEnded(entryKey) {
+    if (!this.data.ended[entryKey]) return;
+    delete this.data.ended[entryKey];
+    this.clearedEnded.add(entryKey);
+  }
+
   /** Register a file as a reviewable page, capturing the agent's version. */
   openPage(file, pristine) {
     const key = pageKey(file);
@@ -128,6 +167,7 @@ export class Store {
     }
     page.updatedAt = Date.now();
     this.data.pages[key] = page;
+    this.clearEnded(key);
     this.save();
     return page;
   }
@@ -152,6 +192,7 @@ export class Store {
     delete page.file;
     page.updatedAt = Date.now();
     this.data.pages[key] = page;
+    this.clearEnded(key);
     this.save();
     return page;
   }

test/poll-handoff.test.js

@@ -119,3 +119,57 @@ test("poll exits with the feedback batch when the user sends", async (t) => {
   assert.equal(batch.status, "feedback");
   assert.equal(batch.pages[0].comments[0].feedback, "Make this clearer.");
 });
+
+test("a poll started after the user ends the review exits at once instead of waiting", async (t) => {
+  // Its own state dir: the test above removes tmp in its teardown.
+  const own = fs.mkdtempSync(path.join(os.tmpdir(), "human-review-ended-"));
+  process.env.HUMAN_REVIEW_STATE_DIR = path.join(own, "state");
+  const file = path.join(own, "review.html");
+  fs.writeFileSync(file, "<p>Original</p>");
+
+  const reviewServer = spawn(process.execPath, ["src/server-entry.js"], {
+    cwd: project,
+    env: { ...process.env, HUMAN_REVIEW_STATE_DIR: process.env.HUMAN_REVIEW_STATE_DIR },
+    stdio: "ignore",
+  });
+
+  t.after(async () => {
+    if (reviewServer.exitCode === null) {
+      reviewServer.kill();
+      await once(reviewServer, "exit");
+    }
+    fs.rmSync(own, { recursive: true, force: true });
+  });
+
+  const server = await waitForServer();
+  const opened = await request(server, "POST", "/api/session", { file });
+  assert.equal(opened.status, 200);
+
+  // End the review with no poll in flight. This is the ordinary case: the agent
+  // applies a delivered batch, and only then polls again.
+  const ended = await request(server, "POST", `/api/session/${opened.body.sessionId}/end`);
+  assert.equal(ended.status, 200);
+
+  const started = Date.now();
+  const child = spawn(process.execPath, ["src/cli.js", "poll", file, "--timeout", "20"], {
+    cwd: project,
+    env: { ...process.env, HUMAN_REVIEW_STATE_DIR: process.env.HUMAN_REVIEW_STATE_DIR },
+    stdio: ["ignore", "pipe", "pipe"],
+  });
+  const result = await collect(child);
+  assert.equal(result.code, 0, result.stderr);
+  assert.equal(JSON.parse(result.stdout).status, "closed");
+  // Without the end mark this poll burns the full 20s and reports a timeout.
+  assert.ok(Date.now() - started < 10000, `poll took ${Date.now() - started}ms`);
+
+  // The mark is consumed, and reopening the target is a live review again, so a
+  // second poll must wait for real feedback rather than reporting closed twice.
+  const reopened = await request(server, "POST", "/api/session", { file });
+  assert.equal(reopened.status, 200);
+  const again = spawn(process.execPath, ["src/cli.js", "poll", file, "--timeout", "2"], {
+    cwd: project,
+    env: { ...process.env, HUMAN_REVIEW_STATE_DIR: process.env.HUMAN_REVIEW_STATE_DIR },
+    stdio: ["ignore", "pipe", "pipe"],
+  });
+  assert.equal(JSON.parse((await collect(again)).stdout).status, "timeout");
+});