fix(cli): deliver an end-of-turn reply past the unseen-messages hold (TASK-535)

cli/__tests__/enforcement.test.mjs

@@ -7,7 +7,8 @@
  *  - createClaimKeeper: win/lose/fail-open, renewal loss, release semantics
  *  - splitForChat: short passthrough, boundary preference, fences atomic,
  *    never cuts content
- *  - deliverChatReply: single/split/attach/attach-fallback modes
+ *  - deliverChatReply: single/split/attach/attach-fallback modes, and the
+ *    end-of-turn unseen-messages hold (TASK-535)
  *
  * The run-loop wiring of these pieces is covered in run-loop.test.mjs.
  */
@@ -929,6 +930,199 @@ describe('deliverChatReply', () => {
   });
 });
 
+// TASK-535. A long turn ends after the pod has moved on, and the backend's
+// send-time freshness guard answers every post with 409 `unseen_messages`.
+// Before the fix the thread headline, the attach card and the split fallback
+// each hit that same hold in turn (rhea, 2026-10-25 10:01Z) and the reply was
+// lost. These pin the CTO ruling: on the first hold of a delivery, re-read the
+// pod tail ONCE, then post with continueAnyway — and every word still lands
+// exactly once.
+describe('deliverChatReply — the unseen-messages hold at end of turn (TASK-535)', () => {
+  const messagesPath = '/api/agents/runtime/pods/pod-1/messages';
+  // The shape api.js throws for a non-2xx response: status + parsed body.
+  const unseenHold = (unseenCount = 12) => Object.assign(
+    new Error(`There are ${unseenCount} unseen messages. Read the conversation and retry, or call again with continueAnyway:true to continue.`),
+    { status: 409, body: { code: 'unseen_messages', unseenCount, messages: [] } },
+  );
+  const prose = Array.from({ length: 8 }, (_, i) => `Point ${i}: ${'x'.repeat(300)}`).join('\n\n');
+  const fence = `\`\`\`js\n${'const x = 1;\n'.repeat(72)}\`\`\``; // one indivisible oversize unit
+  const rereads = (get) => get.mock.calls.filter(([p]) => p === messagesPath);
+  // A call that threw delivered nothing, so "posted" is every call whose
+  // promise resolved. Calls are settled in order, so pairing each call with
+  // whether it rejected is exact.
+  const deliveredBodies = async (post) => {
+    const settled = await Promise.allSettled(post.mock.results.map((r) => r.value));
+    return post.mock.calls
+      .filter((_, i) => settled[i].status === 'fulfilled')
+      .map(([, body]) => body);
+  };
+
+  test('thread rung: a held headline re-reads once, then the whole thread posts once with continueAnyway', async () => {
+    const post = jest.fn()
+      .mockRejectedValueOnce(unseenHold())
+      .mockResolvedValue({ message: { id: 'root-1' } });
+    const get = jest.fn().mockResolvedValue({ messages: [] });
+    const upload = jest.fn();
+    const log = jest.fn();
+
+    const res = await deliverChatReply({ client: { post, get, upload }, podId: 'pod-1', text: prose, log });
+
+    const chunks = splitForChat(prose);
+    expect(res).toMatchObject({ mode: 'thread', messages: chunks.length, threadRootId: 'root-1' });
+    expect(upload).not.toHaveBeenCalled();
+    // The held attempt carried no override; the re-read is the tail read that
+    // records a seen cursor (no before / threadRootId), at the route's cap.
+    expect(post.mock.calls[0][1]).toEqual({ content: chunks[0] });
+    expect(rereads(get)).toEqual([[messagesPath, { limit: 50 }]]);
+    // Exactly once: each chunk delivered one time, all under the override.
+    const delivered = await deliveredBodies(post);
+    expect(delivered.map((b) => b.content)).toEqual(chunks);
+    expect(delivered.every((b) => b.continueAnyway === true)).toBe(true);
+    expect(delivered.slice(1).every((b) => b.threadRootId === 'root-1')).toBe(true);
+    expect(log).toHaveBeenCalledWith(expect.stringContaining('12 unseen message(s) in pod pod-1'));
+  });
+
+  test('attach rung: a held file card re-reads once and posts once — the upload is not repeated', async () => {
+    const post = jest.fn()
+      .mockRejectedValueOnce(unseenHold())
+      .mockResolvedValue({});
+    const get = jest.fn().mockResolvedValue({ messages: [] });
+    const upload = jest.fn().mockResolvedValue({
+      fileName: 'srv.md', originalName: 'reply.md', size: 950, kind: 'document',
+    });
+
+    const res = await deliverChatReply({ client: { post, get, upload }, podId: 'pod-1', text: fence });
+
+    expect(res).toEqual({ mode: 'attach', messages: 1 });
+    expect(upload).toHaveBeenCalledTimes(1);
+    expect(rereads(get)).toHaveLength(1);
+    expect(post).toHaveBeenCalledTimes(2);
+    expect(post.mock.calls[0][1].continueAnyway).toBeUndefined();
+    const delivered = await deliveredBodies(post);
+    expect(delivered).toHaveLength(1);
+    expect(delivered[0]).toMatchObject({ continueAnyway: true });
+    expect(delivered[0].content).toContain('[[upload:srv.md');
+  });
+
+  test('split rung: a held first chunk re-reads once, then every chunk posts once with continueAnyway', async () => {
+    const post = jest.fn()
+      .mockRejectedValueOnce(unseenHold())
+      .mockResolvedValue({});
+    const get = jest.fn().mockResolvedValue({ messages: [] });
+    const upload = jest.fn().mockRejectedValue(new Error('older server'));
+    const text = `${'y'.repeat(350)}\n\n${fence}\n\n${'z'.repeat(350)}`;
+
+    const res = await deliverChatReply({ client: { post, get, upload }, podId: 'pod-1', text });
+
+    const chunks = splitForChat(text);
+    expect(res).toEqual({ mode: 'split-fallback', messages: chunks.length });
+    expect(rereads(get)).toHaveLength(1);
+    const delivered = await deliveredBodies(post);
+    expect(delivered.map((b) => b.content)).toEqual(chunks);
+    expect(delivered.every((b) => b.continueAnyway === true)).toBe(true);
+  });
+
+  test('the override is per DELIVERY: a later rung carries it without a second re-read', async () => {
+    // The rhea shape with the fix in place, made harder: the override retry
+    // on the thread headline dies for an unrelated reason, so the ladder
+    // falls through to attach. That rung must not re-read or wait for its own
+    // hold — it posts with the override straight away.
+    const post = jest.fn()
+      .mockRejectedValueOnce(unseenHold())
+      .mockRejectedValueOnce(Object.assign(new Error('upstream 503'), { status: 503 }))
+      .mockResolvedValue({});
+    const get = jest.fn().mockResolvedValue({ messages: [] });
+    const upload = jest.fn().mockResolvedValue({ fileName: 'srv.md', originalName: 'reply.md' });
+
+    const res = await deliverChatReply({ client: { post, get, upload }, podId: 'pod-1', text: prose });
+
+    expect(res).toEqual({ mode: 'attach', messages: 1 });
+    expect(rereads(get)).toHaveLength(1);
+    expect(post).toHaveBeenCalledTimes(3);
+    expect(post.mock.calls[2][1]).toMatchObject({ continueAnyway: true });
+    expect(await deliveredBodies(post)).toHaveLength(1);
+  });
+
+  test('a hold on a thread continuation also re-reads once, and no chunk posts twice', async () => {
+    let n = 0;
+    const post = jest.fn().mockImplementation(async () => {
+      n += 1;
+      if (n === 3) throw unseenHold(2); // a peer posted mid-thread
+      return { message: { id: 'root-1' } };
+    });
+    const get = jest.fn().mockResolvedValue({ messages: [] });
+
+    const res = await deliverChatReply({ client: { post, get, upload: jest.fn() }, podId: 'pod-1', text: prose });
+
+    const chunks = splitForChat(prose);
+    expect(res.mode).toBe('thread');
+    expect(rereads(get)).toHaveLength(1);
+    const delivered = await deliveredBodies(post);
+    expect(delivered.map((b) => b.content)).toEqual(chunks);
+    expect(delivered.slice(1).every((b) => b.threadRootId === 'root-1')).toBe(true);
+  });
+
+  test('a short single reply held at the end of a turn still posts exactly once', async () => {
+    const post = jest.fn()
+      .mockRejectedValueOnce(unseenHold(1))
+      .mockResolvedValue({});
+    const get = jest.fn().mockResolvedValue({ messages: [] });
+
+    const res = await deliverChatReply({ client: { post, get }, podId: 'pod-1', text: 'short answer' });
+
+    expect(res).toEqual({ mode: 'single', messages: 1 });
+    expect(post.mock.calls).toEqual([
+      [messagesPath, { content: 'short answer' }],
+      [messagesPath, { content: 'short answer', continueAnyway: true }],
+    ]);
+  });
+
+  test('a failed re-read does not block the override post', async () => {
+    const post = jest.fn()
+      .mockRejectedValueOnce(unseenHold())
+      .mockResolvedValue({});
+    const get = jest.fn().mockRejectedValue(new Error('read timeout'));
+    const log = jest.fn();
+
+    const res = await deliverChatReply({ client: { post, get }, podId: 'pod-1', text: 'short answer', log });
+
+    expect(res).toEqual({ mode: 'single', messages: 1 });
+    expect(post.mock.calls[1][1]).toEqual({ content: 'short answer', continueAnyway: true });
+    expect(log).toHaveBeenCalledWith(expect.stringContaining('read timeout'));
+  });
+
+  test('when every rung still fails it throws after ONE re-read — never a loop', async () => {
+    const post = jest.fn().mockRejectedValue(unseenHold());
+    const get = jest.fn().mockResolvedValue({ messages: [] });
+    const upload = jest.fn().mockResolvedValue({ fileName: 'srv.md', originalName: 'reply.md' });
+
+    await expect(deliverChatReply({ client: { post, get, upload }, podId: 'pod-1', text: prose }))
+      .rejects.toThrow(/unseen messages/);
+
+    expect(rereads(get)).toHaveLength(1);
+    // headline + its one override retry, the attach card, the first split
+    // chunk. Each later rung tries once, with the override, and stops.
+    expect(post).toHaveBeenCalledTimes(4);
+    expect(post.mock.calls.slice(1).every(([, b]) => b.continueAnyway === true)).toBe(true);
+  });
+
+  test.each([
+    ['a 409 with a different code', Object.assign(new Error('conflict'), { status: 409, body: { code: 'claim_held' } })],
+    ['a 500', Object.assign(new Error('boom'), { status: 500, body: { message: 'boom' } })],
+    ['a network error', new Error('fetch failed')],
+  ])('%s never triggers a re-read or continueAnyway', async (_label, error) => {
+    const post = jest.fn().mockRejectedValue(error);
+    const get = jest.fn();
+    const upload = jest.fn().mockRejectedValue(new Error('older server'));
+
+    await expect(deliverChatReply({ client: { post, get, upload }, podId: 'pod-1', text: prose }))
+      .rejects.toThrow(error.message);
+
+    expect(get).not.toHaveBeenCalled();
+    expect(post.mock.calls.some(([, b]) => 'continueAnyway' in b)).toBe(false);
+  });
+});
+
 describe('createClaimHandicap', () => {
   test('no handicap before any win; a win adds a jittered delay to the NEXT race', () => {
     let clock = 0;

cli/__tests__/run-loop.test.mjs

@@ -2035,6 +2035,101 @@ describe('performRun', () => {
     expect(errors[0].message).toMatch(/Ack failed.*ack transport offline/);
   });
 
+  // TASK-535: a long turn ends with the pod having moved on. The backend holds
+  // the reply with 409 `unseen_messages`; before the fix every rung hit the
+  // same hold, the reply was lost and the events re-ran (rhea, 10:01Z).
+  const unseenHold = (unseenCount) => Object.assign(
+    new Error(`There are ${unseenCount} unseen messages. Read the conversation and retry, or call again with continueAnyway:true to continue.`),
+    { status: 409, body: { code: 'unseen_messages', unseenCount, messages: [] } },
+  );
+
+  test('a turn that ends with 12 unseen messages still posts exactly once — and a redelivery does not post again', async () => {
+    const event = makeEvent({ _id: 'evt-long-turn-unseen' });
+    const get = jest.fn(async (route) => (route.endsWith('/messages') ? { messages: [] } : { events: [event] }));
+    let held = false;
+    let ackAttempts = 0;
+    const post = jest.fn(async (route, body) => {
+      if (route.endsWith('/messages') && !body?.continueAnyway) {
+        held = true;
+        throw unseenHold(12);
+      }
+      if (route.endsWith('/ack')) {
+        ackAttempts += 1;
+        // The success is recorded locally before this ack; losing the ack is
+        // what makes the kernel redeliver the event.
+        if (ackAttempts === 1) throw new Error('ack transport offline');
+      }
+      return {};
+    });
+    createClient.mockReturnValue({ get, post });
+    const spawn = jest.fn(async () => ({ text: 'Summary of 43 minutes of work.' }));
+    const adapter = { name: 'stub', detect: stubAdapter.detect, spawn };
+    const logs = [];
+
+    const first = performRun({
+      instanceUrl: 'http://localhost:5000', token: 'cm_agent_test', adapter, agentName: 'my-stub',
+      setTimeoutImpl: noopTimeout, log: (line) => logs.push(line), onError: () => {},
+    });
+    await drainMicrotasks();
+    first.stop();
+
+    expect(held).toBe(true);
+    expect(replyPosts(post)).toEqual([
+      ['/api/agents/runtime/pods/pod-abc/messages', { content: 'Summary of 43 minutes of work.' }],
+      ['/api/agents/runtime/pods/pod-abc/messages', { content: 'Summary of 43 minutes of work.', continueAnyway: true }],
+    ]);
+    expect(logs.some((l) => /posted .* as 1 message \(single\)/.test(l))).toBe(true);
+    expect(wasEventHandled('my-stub', event._id)).toBe(true);
+
+    // The kernel redelivers the event whose ack was lost. The reply already
+    // reached the pod, so it must not post a second time.
+    const second = performRun({
+      instanceUrl: 'http://localhost:5000', token: 'cm_agent_test', adapter, agentName: 'my-stub',
+      setTimeoutImpl: noopTimeout, onError: () => {},
+    });
+    await drainMicrotasks();
+    second.stop();
+
+    expect(spawn).toHaveBeenCalledTimes(1);
+    expect(replyPosts(post).filter(([, body]) => body.continueAnyway === true)).toHaveLength(1);
+    expect(replyPosts(post)).toHaveLength(2);
+    expect(ackPosts(post)[1][1]).toEqual({ result: { outcome: 'no_action', reason: 'duplicate-delivery' } });
+  });
+
+  test('a reply that no rung can deliver leaves the events unacked and writes one error line naming pod, count and reason', async () => {
+    const events = [
+      makeEvent({ _id: 'evt-lost-1' }),
+      makeEvent({ _id: 'evt-lost-2', payload: { content: 'and another thing' } }),
+    ];
+    const get = jest.fn(async (route) => (route.endsWith('/messages') ? { messages: [] } : { events }));
+    const post = jest.fn(async (route) => {
+      if (route.endsWith('/messages')) throw unseenHold(12);
+      return {};
+    });
+    createClient.mockReturnValue({ get, post });
+    const spawn = jest.fn(async () => ({ text: 'A summary nobody will see.' }));
+    const adapter = { name: 'stub', detect: stubAdapter.detect, spawn };
+    const logs = [];
+
+    const { stop } = performRun({
+      instanceUrl: 'http://localhost:5000', token: 'cm_agent_test', adapter, agentName: 'my-stub',
+      setTimeoutImpl: noopTimeout, log: (line) => logs.push(line), onError: () => {},
+    });
+    await drainMicrotasks();
+    stop();
+
+    expect(spawn).toHaveBeenCalledTimes(1);
+    // One held post, one override post, then stop: never a loop.
+    expect(replyPosts(post)).toHaveLength(2);
+    expect(ackPosts(post)).toHaveLength(0);
+    for (const event of events) expect(wasEventHandled('my-stub', event._id)).toBe(false);
+    const errorLines = logs.filter((l) => /reply NOT delivered/.test(l));
+    expect(errorLines).toHaveLength(1);
+    expect(errorLines[0]).toContain('pod pod-abc');
+    expect(errorLines[0]).toContain('2 events left unacked');
+    expect(errorLines[0]).toContain('12 unseen messages');
+  });
+
   test.each([false, true])('backlog catch-up reaches the model before one reply and ack (skills: %s)', async (skills) => {
     const event = makeEvent({ _id: 'backlog-event' });
     let caughtUp = false;

cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@commonlyai/cli",
-  "version": "0.1.155",
+  "version": "0.1.168",
   "license": "Apache-2.0",
   "description": "The Commonly CLI — connect agents, manage pods, iterate fast",
   "type": "module",

cli/src/commands/agent.js

@@ -2035,15 +2035,32 @@ export const performRun = ({
     } else {
       // ADR-018 length gate: the tone contract is enforced here, where the
       // wrapper is the one posting. Split, attach — never truncate.
-      const delivery = await deliverChatReply({
-        client,
-        podId: eventPodId,
-        text: replyText,
-        limit: chatCharLimit,
-        maxChunks: maxChatChunks,
-        uploadName: `${agentName}-reply-${event._id}.md`,
-        log: (line) => log(`[${event.type}] ${line}`),
-      });
+      let delivery;
+      try {
+        delivery = await deliverChatReply({
+          client,
+          podId: eventPodId,
+          text: replyText,
+          limit: chatCharLimit,
+          maxChunks: maxChatChunks,
+          uploadName: `${agentName}-reply-${event._id}.md`,
+          log: (line) => log(`[${event.type}] ${line}`),
+        });
+      } catch (deliveryError) {
+        // TASK-535: every rung threw, so the reply reached nobody. The throw
+        // still leaves the events unacked for redelivery (unchanged), but the
+        // retry report is about the BATCH; this line is about the reply, and
+        // it names where it was going so a lost summary is never silent.
+        const eventCount = Array.isArray(event.payload?.batchEventIds)
+          ? event.payload.batchEventIds.length
+          : 1;
+        log(
+          `[${event.type}] ERROR reply NOT delivered to pod ${eventPodId} — `
+          + `${Buffer.byteLength(replyText)} bytes lost, ${eventCount} event${eventCount === 1 ? '' : 's'} `
+          + `left unacked for retry: ${deliveryError.message}`,
+        );
+        throw deliveryError;
+      }
       if (delivery.skipped) {
         // The server answered 200 and created nothing: a duplicate inside its
         // dedupe window, a silent body, a heartbeat it keeps. Nothing reached