diff --git a/cli/__tests__/enforcement.test.mjs b/cli/__tests__/enforcement.test.mjs index 830dfb34b..31a338b7d 100644 --- a/cli/__tests__/enforcement.test.mjs +++ b/cli/__tests__/enforcement.test.mjs @@ -450,6 +450,29 @@ describe('deliverChatReply', () => { expect(post).toHaveBeenCalledWith(messagesPath, { content: 'short answer' }); }); + test('a normal-return refusal is not recorded as a single-message delivery', async () => { + // The runtime route returns HTTP 200 for this policy refusal. The resolved + // promise proves only that the server decided, not that it created a row. + const post = jest.fn().mockResolvedValue({ + success: false, + refused: true, + reason: 'consecutive_run_cap', + consecutive: 3, + guidance: 'Do not retry this message unchanged.', + }); + const res = await deliverChatReply({ client: { post }, podId: 'pod-1', text: 'short answer' }); + + expect(res).toEqual({ + mode: 'refused', + messages: 0, + attemptedMessages: 1, + refused: true, + reason: 'consecutive_run_cap', + consecutive: 3, + guidance: 'Do not retry this message unchanged.', + }); + }); + test('a split-sized reply posts its chunks in order', async () => { const post = jest.fn().mockResolvedValue({}); const text = `${'a'.repeat(390)}\n\n${'b'.repeat(390)}`; @@ -459,8 +482,171 @@ describe('deliverChatReply', () => { expect(post.mock.calls[1][1].content).toBe('b'.repeat(390)); }); - test('a document-sized reply uploads whole and posts one lead message with the file card', async () => { - const post = jest.fn().mockResolvedValue({}); + test('a split reply stops at a normal-return refusal and reports only delivered chunks', async () => { + const refusal = { + success: false, + refused: true, + reason: 'consecutive_run_cap', + consecutive: 3, + guidance: 'Wait for someone else to speak.', + }; + const post = jest.fn() + .mockResolvedValueOnce({ success: true }) + .mockResolvedValueOnce({ success: true }) + .mockResolvedValueOnce(refusal); + const text = `${'a'.repeat(390)}\n\n${'b'.repeat(390)}\n\n${'c'.repeat(390)}`; + + const res = await deliverChatReply({ client: { post }, podId: 'pod-1', text }); + + expect(res).toEqual({ + mode: 'refused', + messages: 2, + attemptedMessages: 3, + refused: true, + reason: 'consecutive_run_cap', + consecutive: 3, + guidance: 'Wait for someone else to speak.', + }); + expect(post).toHaveBeenCalledTimes(3); + expect(post.mock.calls.map(([, body]) => body.content)).toEqual([ + 'a'.repeat(390), 'b'.repeat(390), 'c'.repeat(390), + ]); + }); + + // Prose overflow is a THREAD now, not an attachment (Sam 57691). + // + // These pin the rung itself, not just its return value. The old behaviour + // and the new one both post a first message containing the reply's opening + // line, so asserting "posted something that starts with Point 0" stays + // green across the regression — the distinguishing evidence is whether the + // CONTINUATIONS carry a threadRootId, and whether the full text survives + // without an upload at all. + describe('prose overflow continues in a thread', () => { + const prose = Array.from({ length: 8 }, (_, i) => `Point ${i}: ${'x'.repeat(300)}`).join('\n\n'); + + test('posts the headline to the channel and the rest under it', async () => { + const post = jest.fn() + .mockResolvedValueOnce({ message: { id: 4242 } }) + .mockResolvedValue({}); + const upload = jest.fn(); + const res = await deliverChatReply({ client: { post, upload }, podId: 'pod-1', text: prose }); + + expect(res.mode).toBe('thread'); + expect(res.threadRootId).toBe('4242'); + expect(upload).not.toHaveBeenCalled(); + + // First call is top-level: no threadRootId at all, not a null one. + expect(post.mock.calls[0][1]).toEqual({ content: expect.stringContaining('Point 0:') }); + // Every continuation is rooted at the headline. + for (const call of post.mock.calls.slice(1)) { + expect(call[1].threadRootId).toBe('4242'); + } + expect(post.mock.calls.length).toBeGreaterThan(1); + }); + + test('no word is lost — the chunks reassemble to the original', async () => { + const post = jest.fn() + .mockResolvedValueOnce({ message: { id: 7 } }) + .mockResolvedValue({}); + await deliverChatReply({ client: { post, upload: jest.fn() }, podId: 'pod-1', text: prose }); + const posted = post.mock.calls.map((c) => c[1].content).join('\n\n'); + expect(posted).toBe(prose); + }); + + test('accepts _id as well as id, since the two shapes are both live', async () => { + const post = jest.fn() + .mockResolvedValueOnce({ message: { _id: 99 } }) + .mockResolvedValue({}); + const res = await deliverChatReply({ client: { post, upload: jest.fn() }, podId: 'pod-1', text: prose }); + expect(res.threadRootId).toBe('99'); + }); + + test('a refused thread headline stops without falling through to an attachment', async () => { + const refusal = { + refused: true, + reason: 'consecutive_run_cap', + guidance: 'Wait for someone else to speak.', + }; + const post = jest.fn().mockResolvedValue(refusal); + const upload = jest.fn(); + + const res = await deliverChatReply({ client: { post, upload }, podId: 'pod-1', text: prose }); + + expect(res).toEqual({ + mode: 'refused', + messages: 0, + attemptedMessages: splitForChat(prose).length, + refused: true, + reason: 'consecutive_run_cap', + guidance: 'Wait for someone else to speak.', + }); + expect(post).toHaveBeenCalledTimes(1); + expect(upload).not.toHaveBeenCalled(); + }); + + test('a refused continuation stops without re-posting the remaining thread as top-level messages', async () => { + const refusal = { + refused: true, + reason: 'consecutive_run_cap', + guidance: 'Wait for someone else to speak.', + }; + const post = jest.fn() + .mockResolvedValueOnce({ message: { id: 'root-1' } }) + .mockResolvedValueOnce({ success: true }) + .mockResolvedValueOnce(refusal); + const upload = jest.fn(); + + const res = await deliverChatReply({ client: { post, upload }, podId: 'pod-1', text: prose }); + + expect(res).toMatchObject({ + mode: 'refused', + messages: 2, + attemptedMessages: splitForChat(prose).length, + reason: 'consecutive_run_cap', + }); + expect(post).toHaveBeenCalledTimes(3); + expect(post.mock.calls.slice(1).every(([, body]) => body.threadRootId === 'root-1')).toBe(true); + expect(upload).not.toHaveBeenCalled(); + }); + + test('never guesses a root, and never re-posts the headline it already sent', async () => { + // A continuation posted with a missing root becomes another TOP-LEVEL + // message, so refusing is the only safe read of an unknown response. + // But the headline is already in the room by then — attaching here + // would lead with the same opening line a second time. Post the + // remainder instead. + const post = jest.fn().mockResolvedValue({}); + const upload = jest.fn(); + const res = await deliverChatReply({ client: { post, upload }, podId: 'pod-1', text: prose }); + expect(res.mode).toBe('thread-fallback'); + expect(upload).not.toHaveBeenCalled(); + const openings = post.mock.calls.filter((c) => c[1].content.startsWith('Point 0:')); + expect(openings).toHaveLength(1); + }); + + test('an indivisible oversize unit still attaches — that one IS a document', async () => { + // The control that keeps this change honest: threading prose must not + // swallow the case attachment was always correct for. + const post = jest.fn().mockResolvedValue({}); + const upload = jest.fn().mockResolvedValue({ + fileName: 'srv.md', originalName: 'reply.md', size: 950, kind: 'document', + }); + const fence = `\`\`\`js\n${'const x = 1;\n'.repeat(72)}\`\`\``; + const res = await deliverChatReply({ client: { post, upload }, podId: 'pod-1', text: fence }); + expect(res.mode).toBe('attach'); + expect(post.mock.calls.every((c) => c[1].threadRootId === undefined)).toBe(true); + }); + }); + + // Was 'a document-sized reply uploads whole and posts one lead message with + // the file card'. Its fixture is eight paragraphs of prose, which is the + // case Sam 57691 reclassified: prose that outgrew a message is not a + // document. The test's INTENT — nothing is cut, the whole reply survives — + // is preserved verbatim below, now carried by the thread instead of a file. + test('a document-sized reply keeps every word, now in a thread rather than a file', async () => { + const post = jest.fn() + .mockResolvedValueOnce({ message: { id: 1234 } }) + .mockResolvedValue({}); const upload = jest.fn().mockResolvedValue({ fileName: 'srv-name.md', originalName: 'reply.md', size: 2000, kind: 'document', }); @@ -469,17 +655,28 @@ describe('deliverChatReply', () => { const res = await deliverChatReply({ client: { post, upload }, podId: 'pod-1', text, uploadName: 'reply.md', }); - expect(res).toEqual({ mode: 'attach', messages: 1 }); - expect(upload).toHaveBeenCalledWith( - '/api/agents/runtime/pods/pod-1/uploads', - expect.objectContaining({ fileName: 'reply.md', contentType: 'text/markdown' }), - ); - // The uploaded file carries the FULL text — nothing is cut. - expect(upload.mock.calls[0][1].fileBuffer.toString('utf8')).toBe(text); + expect(res.mode).toBe('thread'); + expect(upload).not.toHaveBeenCalled(); + // The thread carries the FULL text — nothing is cut. Same assertion the + // attach version made about the file buffer. + expect(post.mock.calls.map((cl) => cl[1].content).join('\n\n')).toBe(text); + expect(post.mock.calls[0][1].content).toContain('Point 0:'); // opening stays the headline + }); + + test('an attachment-card refusal does not masquerade as an attachment delivery', async () => { + const post = jest.fn().mockResolvedValue({ + refused: true, reason: 'consecutive_run_cap', guidance: 'Do not retry unchanged.', + }); + const upload = jest.fn().mockResolvedValue({ fileName: 'srv.md', originalName: 'reply.md' }); + const text = `\`\`\`js\n${'const x = 1;\n'.repeat(72)}\`\`\``; + + const res = await deliverChatReply({ client: { post, upload }, podId: 'pod-1', text }); + + expect(res).toMatchObject({ + mode: 'refused', messages: 0, attemptedMessages: 1, reason: 'consecutive_run_cap', + }); expect(post).toHaveBeenCalledTimes(1); - const { content } = post.mock.calls[0][1]; - expect(content).toContain('Point 0:'); // leads with the reply's own opening - expect(content).toContain('[[upload:srv-name.md|reply.md|2000|document]]'); + expect(upload).toHaveBeenCalledTimes(1); }); test('a document-sized single fence attaches — it cannot ride the single-post branch (msg 53018)', async () => { @@ -531,18 +728,80 @@ describe('deliverChatReply', () => { expect(post.mock.calls[0][1].content).toContain('Here is the diff:'); }); - test('upload failure degrades to posting every chunk — flood beats truncation or silence', async () => { - const post = jest.fn().mockResolvedValue({}); - const upload = jest.fn().mockRejectedValue(new Error('older server')); + // Same guarantee as before — flood beats truncation or silence — but this + // prose fixture now degrades through the THREAD rung, not the upload rung. + // A server that returns no message id cannot root a thread, and the reply + // must still arrive whole. + test('a rootless server degrades to posting every chunk — flood beats truncation or silence', async () => { + const post = jest.fn().mockResolvedValue({}); // no id anywhere in the response + const upload = jest.fn(); const log = jest.fn(); const text = Array.from({ length: 6 }, () => 'y'.repeat(350)).join('\n\n'); const res = await deliverChatReply({ client: { post, upload }, podId: 'pod-1', text, log, }); - expect(res.mode).toBe('split-fallback'); + expect(res.mode).toBe('thread-fallback'); expect(post).toHaveBeenCalledTimes(res.messages); + // Every word arrives, and the headline is posted exactly once — the + // duplicate-opening bug this fallback exists to avoid. expect(post.mock.calls.map((c) => c[1].content).join('\n\n')).toBe(text); - expect(log).toHaveBeenCalledWith(expect.stringContaining('older server')); + expect(post.mock.calls.filter((c) => c[1].content === 'y'.repeat(350)).length).toBe(6); + expect(log).toHaveBeenCalledWith(expect.stringContaining('cannot root the thread')); + }); + + // The two tests above both fail at the ROOT-ID step, before a single + // continuation has posted. That is why a boolean `headlinePosted` passed + // them: at that instant "something posted" and "one thing posted" are the + // same statement. They stop being the same the moment a continuation + // succeeds and a later one throws. + test('a continuation that fails PART WAY resumes from there — no chunk posted twice', async () => { + const chunks = Array.from({ length: 6 }, (_, i) => `${'z'.repeat(340)}${i}`); + const text = chunks.join('\n\n'); + let n = 0; + const post = jest.fn().mockImplementation(async () => { + n += 1; + // 1 = headline, 2 and 3 = continuations that land in the thread, + // 4 = the one that dies. Chunks 1 and 2 are already in the room. + if (n === 4) throw new Error('upstream 503'); + return { message: { id: 'root-1' } }; + }); + const log = jest.fn(); + const res = await deliverChatReply({ + client: { post, upload: jest.fn() }, podId: 'pod-1', text, log, + }); + + expect(res.mode).toBe('thread-fallback'); + // The whole reply arrives, and NOTHING arrives twice. With the boolean, + // chunks 1 and 2 were re-posted top-level and this count read 9. + expect(post).toHaveBeenCalledTimes(chunks.length + 1); // +1 for the throw + const delivered = post.mock.calls + .map((c) => c[1].content) + .filter((_, i) => i !== 3); // the call that threw delivered nothing + expect(delivered).toEqual(chunks); + for (const chunk of chunks) { + expect(delivered.filter((c) => c === chunk)).toHaveLength(1); + } + // The resumed chunks go top-level — that is the fallback, not a regression. + expect(post.mock.calls.slice(4).every((c) => c[1].threadRootId === undefined)).toBe(true); + expect(log).toHaveBeenCalledWith(expect.stringContaining('posting the remainder top-level')); + }); + + test('the attachment fallback also stops at a normal-return refusal', async () => { + const post = jest.fn() + .mockResolvedValueOnce({ success: true }) + .mockResolvedValueOnce({ + refused: true, reason: 'consecutive_run_cap', guidance: 'Wait for a reply first.', + }); + const upload = jest.fn().mockRejectedValue(new Error('older server')); + const fence = `\`\`\`js\n${'const x = 1;\n'.repeat(72)}\`\`\``; + const text = `${'y'.repeat(350)}\n\n${fence}\n\n${'z'.repeat(350)}`; + + const res = await deliverChatReply({ client: { post, upload }, podId: 'pod-1', text }); + + expect(res).toMatchObject({ + mode: 'refused', messages: 1, attemptedMessages: splitForChat(text).length, reason: 'consecutive_run_cap', + }); + expect(post).toHaveBeenCalledTimes(2); }); }); diff --git a/cli/__tests__/run-loop.test.mjs b/cli/__tests__/run-loop.test.mjs index 0f45b3de4..6716a99ef 100644 --- a/cli/__tests__/run-loop.test.mjs +++ b/cli/__tests__/run-loop.test.mjs @@ -125,6 +125,73 @@ describe('performRun', () => { ); }); + test('a normal-return run-cap refusal is acked as a refusal, not a posted reply', async () => { + // The post route deliberately responds 200 with { refused: true }. This + // is terminal guidance — retrying the same event would duplicate the two + // chunks that did land — so the wrapper must expose it locally and ack the + // event as no_action rather than throw into at-least-once redelivery. + const guidance = 'Wait for someone else to speak; do not retry unchanged.'; + let messagePosts = 0; + const mockGet = jest.fn().mockResolvedValue({ events: [makeEvent({ _id: 'evt-run-cap' })] }); + const mockPost = jest.fn(async (route) => { + if (route === '/api/agents/runtime/pods/pod-abc/messages') { + messagePosts += 1; + return messagePosts < 3 + ? { success: true } + : { + success: false, + refused: true, + reason: 'consecutive_run_cap', + consecutive: 3, + guidance, + }; + } + return {}; + }); + createClient.mockReturnValue({ get: mockGet, post: mockPost }); + const onError = jest.fn(); + const text = `${'a'.repeat(390)}\n\n${'b'.repeat(390)}\n\n${'c'.repeat(390)}`; + const spawn = jest.fn(async () => ({ text })); + const adapter = { name: 'stub', detect: stubAdapter.detect, spawn }; + + const { stop } = performRun({ + instanceUrl: 'http://localhost:5000', + token: 'cm_agent_test', + adapter, + agentName: 'my-stub', + onError, + setTimeoutImpl: noopTimeout, + }); + await drainMicrotasks(); + stop(); + + expect(mockPost.mock.calls.filter(([route]) => route === '/api/agents/runtime/pods/pod-abc/messages')) + .toHaveLength(3); + expect(mockPost).toHaveBeenCalledWith( + '/api/agents/runtime/events/evt-run-cap/ack', + { + result: { + outcome: 'no_action', + reason: 'consecutive_run_cap', + details: { + mode: 'refused', + postedMessages: 2, + attemptedMessages: 3, + consecutive: 3, + guidance, + }, + }, + }, + ); + expect(onError).toHaveBeenCalledWith(expect.objectContaining({ + code: 'agent_delivery_refused', + reason: 'consecutive_run_cap', + postedMessages: 2, + attemptedMessages: 3, + })); + expect(onError.mock.calls[0][0].message).toContain(guidance); + }); + test('first_contact event is forwarded to the adapter like a mention', async () => { const events = [makeEvent({ _id: 'evt-first-contact', diff --git a/cli/package.json b/cli/package.json index fe18ee1a5..b60f2f122 100644 --- a/cli/package.json +++ b/cli/package.json @@ -1,6 +1,6 @@ { "name": "@commonlyai/cli", - "version": "0.1.18", + "version": "0.1.20", "license": "Apache-2.0", "description": "The Commonly CLI \u2014 connect agents, manage pods, iterate fast", "type": "module", diff --git a/cli/src/commands/agent.js b/cli/src/commands/agent.js index 9f7e10436..c5406367f 100644 --- a/cli/src/commands/agent.js +++ b/cli/src/commands/agent.js @@ -1069,6 +1069,7 @@ export const performRun = ({ && /^(HEARTBEAT_OK|HEARTBEAT_NOOP)$/i.test(replyText); const silentReply = !replyText || replyText === 'NO_REPLY' || heartbeatControlReply; let delivered = agentPostedItself; + let deliveryRefusal = null; if (event.type === 'agent.ask') { if (silentReply) { @@ -1154,11 +1155,32 @@ export const performRun = ({ uploadName: `${agentName}-reply-${event._id}.md`, log: (line) => log(`[${event.type}] ${line}`), }); - delivered = true; - log( - `[${event.type}] posted ${Buffer.byteLength(replyText)} bytes as ` - + `${delivery.messages} message${delivery.messages === 1 ? '' : 's'} (${delivery.mode})`, - ); + if (delivery.refused) { + // A run-cap refusal is a successful HTTP request but not a delivery. + // Ack it so the kernel does not replay the same text (the server + // guidance expressly says not to retry unchanged), while preserving + // the refusal and its partial-post count for the seat and event ledger. + deliveryRefusal = delivery; + const guidance = delivery.guidance || 'The server refused this message; do not retry it unchanged.'; + const detail = `after ${delivery.messages}/${delivery.attemptedMessages} message${delivery.attemptedMessages === 1 ? '' : 's'}`; + log(`[${event.type}] wrapper delivery refused ${detail} (${delivery.reason}): ${guidance}`); + onError?.(Object.assign( + new Error(`${event.type} wrapper delivery refused ${detail}: ${guidance}`), + { + code: 'agent_delivery_refused', + reason: delivery.reason, + eventId: event._id, + postedMessages: delivery.messages, + attemptedMessages: delivery.attemptedMessages, + }, + )); + } else { + delivered = true; + log( + `[${event.type}] posted ${Buffer.byteLength(replyText)} bytes as ` + + `${delivery.messages} message${delivery.messages === 1 ? '' : 's'} (${delivery.mode})`, + ); + } } if (result.memorySummary) { try { @@ -1176,6 +1198,21 @@ export const performRun = ({ // streak. Recording only on completion means a spawn failure that gets // redelivered never double-counts toward the cap. cascadeGovernor.record(eventPodId, trigger); + if (deliveryRefusal) { + return { + outcome: 'no_action', + reason: deliveryRefusal.reason, + details: { + mode: deliveryRefusal.mode, + postedMessages: deliveryRefusal.messages, + attemptedMessages: deliveryRefusal.attemptedMessages, + ...(typeof deliveryRefusal.consecutive === 'number' + ? { consecutive: deliveryRefusal.consecutive } + : {}), + ...(deliveryRefusal.guidance ? { guidance: deliveryRefusal.guidance } : {}), + }, + }; + } return { outcome: delivered ? 'posted' : 'no_action' }; }; diff --git a/cli/src/lib/enforcement.js b/cli/src/lib/enforcement.js index 6ee272daf..1fcf42d79 100644 --- a/cli/src/lib/enforcement.js +++ b/cli/src/lib/enforcement.js @@ -540,16 +540,35 @@ export const splitForChat = (text, { limit = 400 } = {}) => { * fits in one message → post as-is * splits into ≤ maxChunks → post the chunks in order ("two short * messages beat one wall") - * longer than a split answer → it is a document, not a message: upload - * the FULL text as a file and post one - * message — the reply's own opening plus - * the file card. Nothing is cut; the file - * holds everything. + * longer than a split answer → post the FIRST chunk to the channel and + * continue the rest in a thread under it. + * one indivisible oversize unit → it is a genuine document: upload the FULL + * text and post one message — the reply's + * own opening plus the file card. * - * If the upload fails (older server, network), fall back to posting every - * chunk: a message flood is a tone violation, silence or truncation is a - * correctness violation, and the contract itself ranks content above tone - * ("NEVER hit that by cutting content"). + * The thread rung is new (Sam 57691) and it replaces attachment as the answer + * for PROSE overflow. Before threads existed, a long analysis had nowhere to + * go but a file, and that was the right workaround. It is now the wrong one: + * an attachment is un-quotable, un-followable, and an all-or-nothing read, + * so overflowing into one buries the tail of a reply in a surface nobody can + * respond to. A thread keeps every word addressable and scopes the read. + * + * This is also the layer the rule has to live at. The pod-context cue already + * tells agents "prose overflow goes in a thread, not an attachment" (#1176), + * and the cue could not have been obeyed: the model does not choose the + * delivery mode, THIS FUNCTION does, and it only knew how to attach. A cue + * that promises what the wrapper contradicts teaches the agent it is failing + * at something it never controlled. + * + * Attachment is kept for the case it was always right for — a single atomic + * unit over `attachThreshold` (a long fence, an unbreakable run). That is a + * document by construction, not prose that outgrew a message. + * + * If threading fails (older server with no threadRootId support, no id in the + * response), fall back to attach, then to posting every chunk: a message + * flood is a tone violation, silence or truncation is a correctness + * violation, and the contract ranks content above tone ("NEVER hit that by + * cutting content"). */ export const deliverChatReply = async ({ client, @@ -563,6 +582,24 @@ export const deliverChatReply = async ({ }) => { const messagesPath = `/api/agents/runtime/pods/${podId}/messages`; const chunks = splitForChat(text, { limit }); + // The runtime route uses HTTP 200 for a policy refusal: it is a completed + // request, but no message was created. Keep that distinction at the client + // boundary so every delivery mode shares it rather than treating a resolved + // promise as proof of a post. + const postMessage = (body) => client.post(messagesPath, body); + const refused = (response, messages, attemptedMessages) => ({ + mode: 'refused', + messages, + attemptedMessages, + refused: true, + reason: response.reason || 'message_refused', + ...(typeof response.guidance === 'string' && response.guidance + ? { guidance: response.guidance } + : {}), + ...(typeof response.consecutive === 'number' + ? { consecutive: response.consecutive } + : {}), + }); // An atomic unit (a fenced block, an unbreakable word-run) can exceed the // limit by construction — splitForChat keeps it whole rather than breaking // its rendering. The tone contract's own rule covers it: over ~800 chars of @@ -571,15 +608,71 @@ export const deliverChatReply = async ({ // the gate (found by the fleet's implementation audit, Sharpen msg 53018). const hasIndivisibleOversize = chunks.some((c) => c.length > attachThreshold); if (chunks.length <= 1 && !hasIndivisibleOversize) { - await client.post(messagesPath, { content: chunks[0] ?? text }); + const response = await postMessage({ content: chunks[0] ?? text }); + if (response?.refused === true) return refused(response, 0, 1); return { mode: 'single', messages: 1 }; } if (chunks.length <= maxChunks && !hasIndivisibleOversize) { + let messages = 0; for (const chunk of chunks) { // eslint-disable-next-line no-await-in-loop - await client.post(messagesPath, { content: chunk }); // in order, so the reply reads top-down + const response = await postMessage({ content: chunk }); // in order, so the reply reads top-down + if (response?.refused === true) return refused(response, messages, chunks.length); + messages += 1; + } + return { mode: 'split', messages }; + } + // PROSE OVERFLOW → THREAD. Only when nothing is indivisibly oversize: a + // fence too big to split is a document and belongs in the attach rung below. + if (!hasIndivisibleOversize) { + // A COUNT, not a flag. The recovery below resumes from here, and a boolean + // can only distinguish "nothing posted" from "something posted" — it cannot + // say how much. Fail a continuation at chunk 3 with a boolean and chunks 1 + // and 2 are already in the thread, then get posted again top-level; the + // reader sees them twice. Getting this wrong is silent, which is also why + // the attach rung below leads with `chunks[0]`: falling through after a + // successful headline duplicates the opening line. Two of this suite's + // existing tests caught that one, and none caught this one, because both + // fail at the root-id step before any continuation has posted. + let posted = 0; + try { + const rootRes = await postMessage({ content: chunks[0] }); + if (rootRes?.refused === true) return refused(rootRes, 0, chunks.length); + posted = 1; + // The runtime route answers `res.json(result)` with the created row on + // `result.message`. Accept either id field; refuse to guess if neither + // is present, because a continuation posted with a missing root would + // silently become another top-level message — the exact flood this rung + // exists to prevent. + const rootId = rootRes?.message?.id ?? rootRes?.message?._id ?? rootRes?.id ?? rootRes?._id; + if (!rootId) throw new Error('no message id in post response — cannot root the thread'); + for (const chunk of chunks.slice(1)) { + // eslint-disable-next-line no-await-in-loop + const response = await postMessage({ content: chunk, threadRootId: String(rootId) }); + if (response?.refused === true) return refused(response, posted, chunks.length); + posted += 1; + } + return { mode: 'thread', messages: chunks.length, threadRootId: String(rootId) }; + } catch (err) { + if (posted > 0) { + // The opening is already in the room. Post the REMAINDER top-level — + // never the whole text again. This is the old flood, minus the + // duplicate, and it is still preferable to attaching: content ranks + // above tone, and the reader would otherwise see the same paragraph + // twice with the rest hidden in a file. + log(`thread continuation failed (${err.message}) — posting the remainder top-level`); + for (const chunk of chunks.slice(posted)) { + // eslint-disable-next-line no-await-in-loop + const response = await postMessage({ content: chunk }); + if (response?.refused === true) return refused(response, posted, chunks.length); + posted += 1; + } + return { mode: 'thread-fallback', messages: chunks.length }; + } + // Nothing reached the room, so the attach rung below is free to lead + // with the opening as it always did. + log(`thread headline failed (${err.message}) — falling back to attach`); } - return { mode: 'split', messages: chunks.length }; } try { const uploaded = await client.upload(`/api/agents/runtime/pods/${podId}/uploads`, { @@ -596,14 +689,18 @@ export const deliverChatReply = async ({ const lead = chunks[0] && chunks[0].length <= limit ? chunks[0] : '(reply too large for chat — attached in full)'; - await client.post(messagesPath, { content: `${lead}\n\n${directive}` }); + const response = await postMessage({ content: `${lead}\n\n${directive}` }); + if (response?.refused === true) return refused(response, 0, 1); return { mode: 'attach', messages: 1 }; } catch (err) { log(`attach fallback failed (${err.message}) — posting ${chunks.length} split messages instead`); + let messages = 0; for (const chunk of chunks) { // eslint-disable-next-line no-await-in-loop - await client.post(messagesPath, { content: chunk }); + const response = await postMessage({ content: chunk }); + if (response?.refused === true) return refused(response, messages, chunks.length); + messages += 1; } - return { mode: 'split-fallback', messages: chunks.length }; + return { mode: 'split-fallback', messages }; } };