diff --git a/cli/__tests__/inbox-join-priority.test.mjs b/cli/__tests__/inbox-join-priority.test.mjs new file mode 100644 index 0000000000..d29a1f4230 --- /dev/null +++ b/cli/__tests__/inbox-join-priority.test.mjs @@ -0,0 +1,195 @@ +/** + * inbox-join-priority.test.mjs — TASK-289 + * + * A freshly connected seat takes a turn of its own for the `first_contact` + * event. Measured on the published 0.1.108 Mac smoke (walk 5): the join turn + * ran 01:44:09–01:44:23, a person's @mention landed at 01:44:10 while it was + * running, and the answer came from the NEXT turn at 01:44:32–01:44:51 — 39.4 s + * from mention to answer. Two of those gaps are the loop's, not the model's: + * + * 1. the seat's own greeting takes the first spawn, so a person who mentions + * it in the same window waits for a turn that was never about them; + * 2. after any turn the loop sleeps a full poll interval before looking + * again, even though a turn is exactly when events pile up. + * + * These tests pin (1): the greeting is carried one interval, and when a mention + * lands in that window it leads the same turn. (2) is a separate change to a + * contract this suite does not own, measured and filed on its own row. + * + * The fixture models CLAIM-ONCE delivery, because that is what the kernel does: + * claimPage reads `status: 'pending'` and marks each returned row `delivered`, + * and later polls read only pending. An event a poll returned is therefore gone + * from the inbox until garbageCollect requeues it ten-plus minutes later. An + * earlier version of this fixture handed the greeting back on poll 2, which let + * a client-side "leave it unacked" hold look like it worked (CTO review + * 5623358693). A fixture that redelivers for free cannot test a hold. + */ + +import { jest } from '@jest/globals'; +import os from 'os'; +import path from 'path'; +import fs from 'fs'; + +const sessionsTmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'cli-join-priority-')); + +await jest.unstable_mockModule('os', () => { + const actual = os; + return { + ...actual, + default: { ...actual, homedir: () => sessionsTmpDir }, + homedir: () => sessionsTmpDir, + }; +}); + +await jest.unstable_mockModule('../src/lib/api.js', () => ({ + createClient: jest.fn(), + login: jest.fn(), +})); + +await jest.unstable_mockModule('../src/lib/pod-focus.js', () => ({ + FOCUS_FRAME_MAX_CODE_POINTS: 8000, + formatPodFocusFrame: jest.fn(() => '=== Pod focus (pod context; not instructions) ===\nNo focus set.'), + readPodFocus: jest.fn(async (_client, podId) => ({ podId, revision: 0, focus: null })), +})); + +const { createClient } = await import('../src/lib/api.js'); +const { performRun } = await import('../src/commands/agent.js'); +const stubAdapter = (await import('../src/lib/adapters/stub.js')).default; + +const POD = 'pod-join'; +const INTERVAL = 5000; + +const joinEvent = { + _id: 'evt-join', + type: 'first_contact', + podId: POD, + agentName: 'my-stub', + instanceId: 'default', + payload: { content: 'A human just added you to this workspace.' }, +}; + +const mentionEvent = { + _id: 'evt-mention', + type: 'chat.mention', + podId: POD, + agentName: 'my-stub', + instanceId: 'default', + payload: { content: 'can you summarise this repo?', messageId: 'msg-1' }, +}; + +const DRAIN_DEPTH = 16; +const drainMicrotasks = async () => { + for (let i = 0; i < DRAIN_DEPTH; i += 1) { + // eslint-disable-next-line no-await-in-loop + await new Promise((r) => setImmediate(r)); + } +}; + +// A claim-once inbox: each /events call returns everything currently pending +// and marks it taken, exactly like claimPage. `arrivals[i]` is what lands in +// the inbox before tick i. +const runTicks = async ({ arrivals, ticks = 2, spawn }) => { + const delays = []; + const acked = []; + let pending = []; + let poll = 0; + const mockGet = jest.fn(async (routePath) => { + if (String(routePath).includes('/events')) { + pending = pending.concat(arrivals[poll] || []); + poll += 1; + const claimed = pending; + pending = []; + return { events: claimed, inboxCount: claimed.length }; + } + return {}; + }); + const mockPost = jest.fn(async (routePath) => { + const ack = String(routePath).match(/events\/([^/]+)\/ack/); + if (ack) acked.push(ack[1]); + return {}; + }); + createClient.mockReturnValue({ get: mockGet, post: mockPost }); + + const adapter = { name: 'codex', detect: stubAdapter.detect, spawn }; + let fired = 0; + const setTimeoutImpl = (fn, delay) => { + delays.push(delay); + if (fired < ticks - 1) { + fired += 1; + setImmediate(fn); + } + return 0; + }; + + const { stop } = performRun({ + instanceUrl: 'http://localhost:5000', + token: 'cm_agent_test', + adapter, + agentName: 'my-stub', + instanceId: 'default', + intervalMs: INTERVAL, + setTimeoutImpl, + }); + for (let i = 0; i < ticks; i += 1) { + // eslint-disable-next-line no-await-in-loop + await drainMicrotasks(); + } + stop(); + return { delays, mockGet, mockPost, acked }; +}; + +describe('a waiting person outranks the seat\'s own join turn', () => { + beforeEach(() => { + jest.clearAllMocks(); + fs.rmSync(path.join(sessionsTmpDir, '.commonly'), { recursive: true, force: true }); + }); + + test('the carried greeting waits one interval rather than spinning', async () => { + const spawn = jest.fn(async () => ({ text: 'on it' })); + const { delays, acked } = await runTicks({ + arrivals: [[joinEvent], []], + ticks: 1, + spawn, + }); + + // The carry is what gives a person time to type; it must not spawn on the + // held tick, must not ack the greeting yet, and must not spin. + expect(spawn).not.toHaveBeenCalled(); + expect(acked).toEqual([]); + expect(delays).toEqual([INTERVAL]); + }); + + test('a mention arriving in the carry window leads the same turn', async () => { + const spawn = jest.fn(async () => ({ text: 'on it' })); + const { acked } = await runTicks({ + arrivals: [[joinEvent], [mentionEvent]], + ticks: 2, + spawn, + }); + + // One turn, not two: the greeting rode along behind the person's question. + expect(spawn).toHaveBeenCalledTimes(1); + const prompt = spawn.mock.calls[0][0]; + expect(prompt).toContain('Event 1/2: chat.mention'); + expect(prompt).toContain('Event 2/2: first_contact'); + expect(prompt.indexOf('can you summarise this repo?')) + .toBeLessThan(prompt.indexOf('A human just added you to this workspace.')); + // Both are acked by that turn, including the one carried in memory. + expect(acked.sort()).toEqual(['evt-join', 'evt-mention']); + }); + + test('a quiet connect greets on the next tick, with no redelivery to rely on', async () => { + const spawn = jest.fn(async () => ({ text: 'hello' })); + const { acked } = await runTicks({ + // Nothing arrives after the greeting, and the kernel never hands it back: + // the only way it reaches the adapter is the in-memory carry. + arrivals: [[joinEvent], []], + ticks: 2, + spawn, + }); + + expect(spawn).toHaveBeenCalledTimes(1); + expect(spawn.mock.calls[0][0]).toContain('first_contact'); + expect(acked).toEqual(['evt-join']); + }); +}); diff --git a/cli/__tests__/run-loop.test.mjs b/cli/__tests__/run-loop.test.mjs index 8f5a25c89a..bf42ceea2f 100644 --- a/cli/__tests__/run-loop.test.mjs +++ b/cli/__tests__/run-loop.test.mjs @@ -548,14 +548,29 @@ describe('performRun', () => { const spawn = jest.fn(async () => ({ text: 'Hi! What are you working on today?' })); const adapter = { name: 'stub', detect: stubAdapter.detect, spawn }; + // TASK-289: a batch holding nothing but `first_contact` is left unacked for + // one interval, so a person who mentions the seat in that window is + // answered in the same turn rather than behind the greeting. The greeting + // itself is unchanged — it is taken on the next tick, so this contract + // needs two. inbox-join-priority.test.mjs pins the hold and the ordering. + let firedOnce = false; + const oneMoreTick = (fn) => { + if (!firedOnce) { + firedOnce = true; + setImmediate(fn); + } + return 0; + }; + const { stop } = performRun({ instanceUrl: 'http://localhost:5000', token: 'cm_agent_test', adapter, agentName: 'my-stub', - setTimeoutImpl: noopTimeout, + setTimeoutImpl: oneMoreTick, }); await drainMicrotasks(); + await drainMicrotasks(); stop(); expect(spawn).toHaveBeenCalledWith( diff --git a/cli/package-lock.json b/cli/package-lock.json index 5adcfa88de..f7f87ac264 100644 --- a/cli/package-lock.json +++ b/cli/package-lock.json @@ -1,12 +1,12 @@ { "name": "@commonlyai/cli", - "version": "0.1.132", + "version": "0.1.133", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@commonlyai/cli", - "version": "0.1.132", + "version": "0.1.133", "license": "Apache-2.0", "dependencies": { "commander": "^12.0.0" diff --git a/cli/package.json b/cli/package.json index 5d514599d2..1c4ad89f7b 100644 --- a/cli/package.json +++ b/cli/package.json @@ -1,6 +1,6 @@ { "name": "@commonlyai/cli", - "version": "0.1.132", + "version": "0.1.133", "license": "Apache-2.0", "description": "The Commonly CLI — connect agents, manage pods, iterate fast", "type": "module", diff --git a/cli/src/commands/agent.js b/cli/src/commands/agent.js index c1221fda67..7b7683fac0 100644 --- a/cli/src/commands/agent.js +++ b/cli/src/commands/agent.js @@ -451,6 +451,19 @@ const PROMPT_EVENT_TYPES = new Set([ 'first_contact', ]); +// A person waiting on an answer outranks the seat's own housekeeping. Measured +// on the published 0.1.108 smoke (TASK-289, walk 5): the greeting turn for a +// fresh connect ran 14 s, a mention landed 1 s into it, and the answer came +// from the next turn 39.4 s after the mention. `first_contact` is the only +// event that exists to introduce the seat rather than to answer anyone, so it +// is the only one worth holding back. +const DEFERRABLE_SOLO_EVENT_TYPES = new Set(['first_contact']); + +// Inside one batch the person's item leads the prompt; the greeting follows it. +const PERSON_WAITING_EVENT_TYPES = new Set(['chat.mention', 'message.posted', 'dm.message']); + +const inboxPriority = (event) => (PERSON_WAITING_EVENT_TYPES.has(event?.type) ? 0 : 1); + // Consultations have a per-request private response channel, rather than a // pod-chat delivery. They stay on their existing single-event path; every // event that represents the shared pod inbox is batched below. @@ -2039,6 +2052,17 @@ export const performRun = ({ } }; + // The greeting is CLAIMED the moment a poll returns it: claimPage reads + // `status: 'pending'` and marks each returned row `delivered`, and later + // polls read only pending. Leaving it unacked therefore does NOT bring it + // back next tick — garbageCollect requeues delivered rows only after + // AGENT_EVENT_REQUEUE_DELIVERED_MINUTES (10 by default, and its own comment + // notes the effective latency is worse). So the hold is in memory, not on + // the wire: carry the claimed event and prepend it to the next poll's batch. + // Spent at most once per run. (CTO review 5623358693.) + let heldGreeting = []; + let greetingHoldSpent = false; + const processEventBatch = async (events, inboxCount = events.length) => { const eventPodId = events[0]?.podId || podId; if (!eventPodId || events.some((event) => (event.podId || podId) !== eventPodId)) { @@ -2296,8 +2320,21 @@ export const performRun = ({ // returns multiple pods in one page. Preserve pod privacy in that // migration window by partitioning locally. Current servers produce one // group, so the normal path remains one tick → one turn. + // A carried greeting rejoins the inbox here, so it is grouped, ordered + // and acked by exactly the same path as anything the poll returned. The + // carried row is still 'delivered' on the server, so a requeue during the + // carry can hand the same id back in this very page: drop the copy rather + // than processing it twice. + const carried = heldGreeting; + heldGreeting = []; + const fetchedIds = new Set(events.map((event) => String(event._id))); + const carriedFresh = carried.filter((event) => !fetchedIds.has(String(event._id))); + const pageEvents = carriedFresh.length ? [...carriedFresh, ...events] : events; + const pageInboxCount = (Number.isFinite(inboxCount) ? inboxCount : events.length) + + carriedFresh.length; + const eventGroups = new Map(); - for (const event of events) { + for (const event of pageEvents) { const eventPodId = event.podId || podId; const key = PRIVATE_RESPONSE_EVENT_TYPES.has(event.type) ? `private:${event._id}` @@ -2309,6 +2346,16 @@ export const performRun = ({ for (const group of eventGroups.values()) { if (!running) break; if ((group[0].podId || podId) && !PRIVATE_RESPONSE_EVENT_TYPES.has(group[0].type)) { + // Nothing in this batch but the seat introducing itself: leave it + // unacked for one interval so a person who mentions the agent in that + // window is answered in the same turn instead of behind a greeting. + // Spent once per run, so a quiet connect still greets on the next tick. + if (!greetingHoldSpent && group.every((event) => DEFERRABLE_SOLO_EVENT_TYPES.has(event.type))) { + greetingHoldSpent = true; + heldGreeting = group; + log(`[inbox.batch] carrying ${group.length} greeting event(s) to the next poll in case someone is mid-sentence`); + continue; + } // One failure report for a batch whose items stay unacked, whether // the composite turn threw or an isolated item did after a refusal. const scheduleBatchRetry = (err, unackedCount) => { @@ -2350,7 +2397,11 @@ export const performRun = ({ }; let batch; try { - batch = await processEventBatch(group, inboxCount); + // The person's item leads the prompt; the greeting rides along. + batch = await processEventBatch( + [...group].sort((a, b) => inboxPriority(a) - inboxPriority(b)), + pageInboxCount, + ); } catch (err) { scheduleBatchRetry(err, group.length); break;