diff --git a/docs/site/content/docs/cli/reference.mdx b/docs/site/content/docs/cli/reference.mdx index 131c8474c90..e4bf76f1f94 100644 --- a/docs/site/content/docs/cli/reference.mdx +++ b/docs/site/content/docs/cli/reference.mdx @@ -246,12 +246,12 @@ For long output, use cursor reads. Save `nextCursor` from one stream read, then orca agent list --json orca agent show orca_session_id:claude_1234 --json orca agent read orca_session_id:claude_1234 --limit 10 -orca agent read orca_session_id:claude_1234 --cursor --json +orca agent read orca_session_id:claude_1234 --json ``` Every agent, chat or terminal, has an Orca session ID (`orca_session_id:_`), what Copy Orca Session ID copies. `agent list` lists the agents open in this host's workspaces with their IDs; a terminal agent shows none until it first reports to Orca. A chat keeps its ID across `/clear`; a terminal agent gets a new one. IDs belong to the host that runs the agent: agents on other hosts are not listed, and the output names those hosts with the `--environment` that lists them. -`agent read` reads the conversation, not a screen: a chat's history, or a terminal agent's own transcript. The agent need not be running and reading never starts it. A terminal agent can be read while Orca knows where its transcript is (it is open, or asleep in a workspace; on Windows, a WSL agent only once it wakes). Pass the printed `cursor` back with `--cursor` to read only what came after. To message an agent, use `orca terminal send` or `orca orchestration send`. +`agent read` reads the conversation, not a screen: a chat's history, or a terminal agent's own transcript. The agent need not be running and reading never starts it. A terminal agent can be read while Orca knows where its transcript is (it is open, or asleep in a workspace; on Windows, a WSL agent only once it wakes). To message an agent, use `orca terminal send` or `orca orchestration send`. ## Chats diff --git a/skill-guides/orca-cli.md b/skill-guides/orca-cli.md index b41b2ca94d6..416f564d196 100644 --- a/skill-guides/orca-cli.md +++ b/skill-guides/orca-cli.md @@ -74,7 +74,7 @@ Every agent, chat or terminal, has an Orca session ID (`orca_session_id:_ - `ORCA agent list --json` lists the agents open in this host's workspaces, with `orcaSessionId`, `kind` (chat or terminal), state, workspace and last reply. A terminal agent has no ID until it first reports to Orca. Agents on other hosts are not listed; the output names those hosts and the `--environment` to use. - `ORCA agent show --json` shows where one agent is open (none if closed, two if resumed in two panes). -- `ORCA agent read --json` reads what the agent said: a chat's history or a terminal agent's transcript, never its screen. It needs no running agent and never starts one. Pass the returned `cursor` with `--cursor` to read only newer entries. A terminal agent is readable while Orca knows its transcript (open, or asleep in a workspace; on Windows, a WSL agent only once it wakes). +- `ORCA agent read --json` reads what the agent said: a chat's history or a terminal agent's transcript, never its screen. It needs no running agent and never starts one. A terminal agent is readable while Orca knows its transcript (open, or asleep in a workspace; on Windows, a WSL agent only once it wakes). A chat keeps its ID across `/clear`; a terminal agent gets a new one. A provider's own conversation id or a `term_…` handle is refused with the ID to use. To message an agent, use `terminal send` or `orchestration send` as below. diff --git a/src/cli/agents/agent-conversations.ts b/src/cli/agents/agent-conversations.ts index 4223d1d751f..149116861c6 100644 --- a/src/cli/agents/agent-conversations.ts +++ b/src/cli/agents/agent-conversations.ts @@ -70,11 +70,9 @@ export const AGENT_CONVERSATION_HANDLERS: Record = { `--limit can be at most ${AGENT_READ_MAX_LIMIT}.` ) } - const cursor = getOptionalStringFlag(flags, 'cursor') const result = await callAgentMethod(client, 'agent.read', { id: getRequiredStringFlag(flags, 'id'), - ...(limit !== undefined ? { limit } : {}), - ...(cursor ? { cursor } : {}) + ...(limit !== undefined ? { limit } : {}) }) printResult(result, json, formatAgentRead) } @@ -158,9 +156,6 @@ function formatAgentRead(result: AgentReadResult): string { if (lines.length === 0) { lines.push('Nothing new.') } - if (result.cursor) { - lines.push('', `More later: --cursor ${result.cursor}`) - } return lines.join('\n') } diff --git a/src/cli/discovery/command-scoped-flag-help.ts b/src/cli/discovery/command-scoped-flag-help.ts index 686bb0bf9f0..cf265c0c264 100644 --- a/src/cli/discovery/command-scoped-flag-help.ts +++ b/src/cli/discovery/command-scoped-flag-help.ts @@ -14,8 +14,7 @@ const COMMAND_SCOPED_FLAG_HELP: Record> = { 'agent show': { id: AGENT_ID_FLAG_HELP }, 'agent read': { id: AGENT_ID_FLAG_HELP, - limit: '--limit Entries to return (default 20, at most 100)', - cursor: '--cursor Cursor from a previous agent read; returns only what came after' + limit: '--limit Newest entries to return (default 20, at most 100)' }, 'extension reload': { 'extension-id': diff --git a/src/cli/skills/bundled-skill-guides.ts b/src/cli/skills/bundled-skill-guides.ts index 619aa750a55..3ba8bf6d02a 100644 --- a/src/cli/skills/bundled-skill-guides.ts +++ b/src/cli/skills/bundled-skill-guides.ts @@ -302,7 +302,7 @@ const ORCA_CLI_MARKDOWN = [ "", "- `ORCA agent list --json` lists the agents open in this host's workspaces, with `orcaSessionId`, `kind` (chat or terminal), state, workspace and last reply. A terminal agent has no ID until it first reports to Orca. Agents on other hosts are not listed; the output names those hosts and the `--environment` to use.", "- `ORCA agent show --json` shows where one agent is open (none if closed, two if resumed in two panes).", - "- `ORCA agent read --json` reads what the agent said: a chat's history or a terminal agent's transcript, never its screen. It needs no running agent and never starts one. Pass the returned `cursor` with `--cursor` to read only newer entries. A terminal agent is readable while Orca knows its transcript (open, or asleep in a workspace; on Windows, a WSL agent only once it wakes).", + "- `ORCA agent read --json` reads what the agent said: a chat's history or a terminal agent's transcript, never its screen. It needs no running agent and never starts one. A terminal agent is readable while Orca knows its transcript (open, or asleep in a workspace; on Windows, a WSL agent only once it wakes).", "", "A chat keeps its ID across `/clear`; a terminal agent gets a new one. A provider's own conversation id or a `term_…` handle is refused with the ID to use. To message an agent, use `terminal send` or `orchestration send` as below.", "", diff --git a/src/cli/specs/agent-conversations.ts b/src/cli/specs/agent-conversations.ts index d0f79d8f452..b7d36fd098a 100644 --- a/src/cli/specs/agent-conversations.ts +++ b/src/cli/specs/agent-conversations.ts @@ -30,17 +30,17 @@ export const AGENT_CONVERSATION_COMMAND_SPECS: CommandSpec[] = [ { path: ['agent', 'read'], summary: 'Read what an agent said, by its Orca session ID', - usage: 'orca agent read [--limit ] [--cursor ] [--json]', - allowedFlags: [...GLOBAL_FLAGS, 'id', 'limit', 'cursor'], + usage: 'orca agent read [--limit ] [--json]', + allowedFlags: [...GLOBAL_FLAGS, 'id', 'limit'], positionalArgs: ['id'], notes: [ "Reads the conversation, not a terminal screen: a chat's history, or a terminal agent's own transcript. The agent need not be running, and reading never starts it.", - '--limit is the number of entries (default 20, at most 100). Pass the printed --cursor to read only what came after.', + '--limit is the number of newest entries (default 20, at most 100).', 'A terminal agent can be read while Orca knows where its transcript is: while it is open, or asleep in a workspace. On Windows, an asleep agent that ran in WSL can be read once its workspace wakes.' ], examples: [ 'orca agent read orca_session_id:claude_1234 --limit 10', - 'orca agent read orca_session_id:codex_1234 --cursor --json' + 'orca agent read orca_session_id:codex_1234 --json' ] } ] diff --git a/src/main/agent-hooks/server/server-agent-identities.ts b/src/main/agent-hooks/server/server-agent-identities.ts index 8df857413b4..7458afcb8ba 100644 --- a/src/main/agent-hooks/server/server-agent-identities.ts +++ b/src/main/agent-hooks/server/server-agent-identities.ts @@ -39,8 +39,6 @@ type SleepingAgentSighting = Pick< export type TerminalAgentTranscriptSource = { key: AgentProviderKey providerSession: AgentProviderSessionMetadata - /** The identity store generation that vouched for `key`. */ - authority: number } /** Terminal agent ids are named when a row is admitted, so walks and publications only copy them. */ @@ -206,23 +204,17 @@ export abstract class AgentHookServerAgentIdentities extends AgentHookServerStat /** Where terminal agent `id`'s transcript is: a current row carrying the id, in any state (a read * needs no running agent), else a sleeping record naming its provider key. The session id and * path come from one payload, never from a pane, so a row attributed to the wrong pane still - * names its own conversation. null: nothing names it now; undefined: the store cannot answer. - * `known` is an earlier answer: an id is never relinked, so while its store still stands a - * recheck needs no second store question. */ + * names its own conversation. null: nothing names it now; undefined: the store cannot answer. */ async terminalAgentTranscriptSource( id: string, - readSleeping: () => Readonly>, - known?: TerminalAgentTranscriptSource + readSleeping: () => Readonly> ): Promise { const authority = this.agentIdentities.authority() - const live = this.liveTranscriptSource(id, authority) + const live = this.liveTranscriptSource(id) if (live) { return live } - const linked = - known?.authority === authority - ? { key: known.key } - : await this.agentIdentities.providerKeyOf(id) + const linked = await this.agentIdentities.providerKeyOf(id) if (!linked || authority !== this.agentIdentities.generation) { return undefined } @@ -238,15 +230,12 @@ export abstract class AgentHookServerAgentIdentities extends AgentHookServerStat } const record = sleepingRecordOfKey(readSleeping(), key, windows) if (record) { - return { key, providerSession: record.providerSession, authority } + return { key, providerSession: record.providerSession } } return null } - private liveTranscriptSource( - id: string, - authority: number - ): TerminalAgentTranscriptSource | null { + private liveTranscriptSource(id: string): TerminalAgentTranscriptSource | null { let best: { key: AgentProviderKey; row: EnrichedAgentHookEventPayload } | null = null for (const place of this.agentIdentities.placesOf(id)) { const subject = place.canonical @@ -277,7 +266,7 @@ export abstract class AgentHookServerAgentIdentities extends AgentHookServerStat } } return best?.row.providerSession - ? { key: best.key, providerSession: best.row.providerSession, authority } + ? { key: best.key, providerSession: best.row.providerSession } : null } diff --git a/src/main/agent-hooks/server/testing/identity/server-agent-transcript-source.test.ts b/src/main/agent-hooks/server/testing/identity/server-agent-transcript-source.test.ts index 3572517f9a5..cbd1b7803c8 100644 --- a/src/main/agent-hooks/server/testing/identity/server-agent-transcript-source.test.ts +++ b/src/main/agent-hooks/server/testing/identity/server-agent-transcript-source.test.ts @@ -8,7 +8,6 @@ import { makePaneKey } from '../../../../../shared/workspace-layout/stable-pane- import { InProcessAgentIdentityClient } from '../../../identity/agent-identity-client.test-fixtures' import type { AgentIdentityClient } from '../../../identity/agent-identity-worker-client' import { AgentHookServer } from '../../agent-hook-server' -import type { TerminalAgentTranscriptSource } from '../../server-agent-identities' import { buildBody } from '../server.test-fixtures' const WSL = 'wsl:Ubuntu' @@ -62,15 +61,6 @@ async function started(): Promise { return server } -/** The located source, for passing back as a recheck's earlier answer. */ -function sourceOf(found: unknown): TerminalAgentTranscriptSource { - if (!found || typeof found !== 'object' || !('key' in found)) { - throw new Error('expected a transcript source') - } - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: only a located source carries `key`. - return found as TerminalAgentTranscriptSource -} - const answered = () => new Promise((resolve) => setImmediate(resolve)) it('reads a terminal agent by its own conversation, never by the pane it was attributed to', async () => { @@ -127,10 +117,8 @@ it.skipIf(process.platform === 'win32')( key: { wslDistro: 'Ubuntu', providerSessionId: 'session-a' }, providerSession: { transcriptPath: '/home/u/.claude/projects/p/session-a.jsonl' } }) - // The post-read recheck reuses the answer it was given: one partition read, no store question. - expect(await server.terminalAgentTranscriptSource(id!, records, sourceOf(found))).toEqual(found) expect(server.identityClient.providerKeyRequests).toBe(1) - expect(partitionReads).toBe(2) + expect(partitionReads).toBe(1) } ) @@ -188,44 +176,19 @@ it.skipIf(process.platform === 'win32')( }, connectionId: WSL }) - for (const population of [300, 3_000]) { - let visits = 0 - const records: Record> = {} - for (let index = 0; index < population; index++) { - const record = sleeping(`other-${index}`) - records[`pane-${index}`] = { - ...record, - get providerSession() { - visits++ - return record.providerSession - } - } - } - records.a = sleeping('session-a') - const found = await server.terminalAgentTranscriptSource(id!, () => records) - visits = 0 - for (let read = 0; read < 10; read++) { - expect( - await server.terminalAgentTranscriptSource(id!, () => records, sourceOf(found)) - ).toEqual(found) - } - // Ten rechecks touch only the asked conversation's records, however many others sleep. - expect(visits).toBe(0) - } + expect( + await server.terminalAgentTranscriptSource(id!, () => ({ + other: sleeping('other'), + a: sleeping('session-a') + })) + ).toMatchObject({ key: { providerSessionId: 'session-a' } }) - const found = await server.terminalAgentTranscriptSource(id!, () => ({ - a: sleeping('session-a') - })) // Another store takes over (a different status path): it never linked this id. const other = mkdtempSync(join(tmpdir(), 'agent-transcript-store-b-')) cleanups.push(() => rmSync(other, { recursive: true, force: true })) server.moveStatusFile(join(other, 'status.json')) expect( - await server.terminalAgentTranscriptSource( - id!, - () => ({ a: sleeping('session-a') }), - sourceOf(found) - ) + await server.terminalAgentTranscriptSource(id!, () => ({ a: sleeping('session-a') })) ).toBeNull() } ) diff --git a/src/main/native-chat/agent-session-journal/projection/journal-agent-read-index.ts b/src/main/native-chat/agent-session-journal/projection/journal-agent-read-index.ts index d31098c18c1..82fc1ec3894 100644 --- a/src/main/native-chat/agent-session-journal/projection/journal-agent-read-index.ts +++ b/src/main/native-chat/agent-session-journal/projection/journal-agent-read-index.ts @@ -75,12 +75,6 @@ export class JournalAgentReadIndex { constructor(private readonly journal: AgentSessionJournal) {} - /** Where the current context starts: the latest `/clear` divider's sequence, or 0. */ - contextFloor(): number { - this.ensureBuilt() - return this.floor - } - /** Current-context entries, newest first, rendered as the public timeline presents them; the * walk stops at the latest `/clear` divider, as the chat does. */ *newestFirst(): Generator { diff --git a/src/main/native-chat/agent-session-journal/replay/journal-row-item-ids.ts b/src/main/native-chat/agent-session-journal/replay/journal-row-item-ids.ts index 53d683549b1..84e5278d351 100644 --- a/src/main/native-chat/agent-session-journal/replay/journal-row-item-ids.ts +++ b/src/main/native-chat/agent-session-journal/replay/journal-row-item-ids.ts @@ -45,30 +45,3 @@ export function journalRowItemIds( } return ids } - -/** The ids one row records, as written: before canonicalization, existence checks or - * visibility, repeats kept. Stable while the row exists, so a reader can address its slots. - * A queued card's row records the send it recovers, then its own hand-off. */ -export function journalRowRawSlots(row: JournalReadRow): string[] { - if (row.kind === 'outcome') { - return [...row.itemIds, ...row.submissionIds.map((id) => agentJournalSubmissionKey(id))] - } - if (row.kind === 'lifecycle-batch') { - return row.mutations.map((mutation) => mutation.itemId) - } - if (isJournalStopOrResumeRow(row)) { - return [] - } - if (row.kind === 'item' || row.kind === 'tombstone') { - return [row.itemId] - } - if (row.kind === 'submission' || row.kind === 'dispatch') { - return [agentJournalSubmissionKey(row.clientMessageId)] - } - if (row.kind === 'queue' && row.change.op === 'queued') { - return [row.change.recovers, row.change.attempt] - .filter((id) => typeof id === 'string') - .map((id) => agentJournalSubmissionKey(id)) - } - return [] -} diff --git a/src/main/native-chat/agent-session-wire/host/structured-agent-session-conversation-reads.ts b/src/main/native-chat/agent-session-wire/host/structured-agent-session-conversation-reads.ts index e0df370b5a7..e620e253167 100644 --- a/src/main/native-chat/agent-session-wire/host/structured-agent-session-conversation-reads.ts +++ b/src/main/native-chat/agent-session-wire/host/structured-agent-session-conversation-reads.ts @@ -59,8 +59,6 @@ export function createStructuredAgentSessionConversationReads(context: () => Rea return { journalCurrentRead: async (sessionId: string, inputIds?: readonly string[]) => (await context().conversation(sessionId)).journal.currentRead(inputIds), - /** The journal's head, opened at rest: a reader skips an unchanged history. */ - journalCursor: async (sessionId: string) => (await journalOf(sessionId)).cursor(), /** One `orca agent read` page, opened at rest; a pending prompt waits only while the held * session can still answer it, and every pending one does with no session held. */ agentReadPage: async ( diff --git a/src/main/native-chat/agent-session-wire/publication/history/agent-read-journal-page.ts b/src/main/native-chat/agent-session-wire/publication/history/agent-read-journal-page.ts index 52132c6f4ca..92ab9853716 100644 --- a/src/main/native-chat/agent-session-wire/publication/history/agent-read-journal-page.ts +++ b/src/main/native-chat/agent-session-wire/publication/history/agent-read-journal-page.ts @@ -1,11 +1,6 @@ -import type { AgentJournalCursor } from '../../../../../shared/agent-session/journal/agent-session-journal-identity' import type { AgentReadEntry } from '../../../../../shared/agent-conversations/agent-conversation-contracts' -import { compareAgentJournalItems } from '../../../../../shared/agent-session/journal/agent-session-journal-position' import { agentChatEntryOf } from '../../../../../shared/agent-conversations/agent-chat-entries' import type { AgentSessionJournal } from '../../../agent-session-journal/journal-store' -import { findSequenceGap } from '../../../agent-session-journal/replay/journal-cursor' -import { journalRowRawSlots } from '../../../agent-session-journal/replay/journal-row-item-ids' -import type { JournalReadRow } from '../../../agent-session-journal/replay/journal-read-row' import { journalAgentReadIndex } from '../../../agent-session-journal/projection/journal-agent-read-index' import type { AgentJournalRenderItem } from '../../../../../shared/agent-session/journal/agent-session-journal-types' import { agentJournalItemForWire } from './agent-session-item-wire-form' @@ -13,205 +8,53 @@ import { HISTORY_PAGE_CONTENT_BUDGET_BYTES, oversizedHistoryItem } from './agent-session-history-page-bounds' -import { renderJournalStatusItem } from '../../../agent-session-journal/projection/journal-public-read' -import { - agentReadPartialRowBytes, - agentReadRowFingerprint, - EmittedSlots, - type AgentReadPartialRow -} from './agent-read-row-progress' -/** Where an agent read continues: rows through `sequence` are read, and `partial` names the next - * row when part of it was (one row can name several items). `floor`: the `/clear` divider the - * read's current context starts at, absent before any. */ -export type AgentReadJournalPosition = { - cursor: AgentJournalCursor - partial?: AgentReadPartialRow - floor?: number -} +export type AgentReadJournalPage = { entries: AgentReadEntry[]; limited: boolean } -/** Cursor bytes the history envelope reserve carries; a row needing more takes the rest from its - * page's content budget. */ -const CURSOR_ALLOWANCE_BYTES = 32 * 1024 - -export type AgentReadJournalPage = - | { ok: true; entries: AgentReadEntry[]; next: AgentReadJournalPosition; limited: boolean } - | { ok: false } - -/** One `orca agent read` page from a chat's journal, with work bounded by the page: a tail walks - * back over the journal's live agent-read entries, and a cursor read takes at most `limit + 1` - * rows and renders only the items they name. Neither renders the timeline, builds a snapshot or - * scans for turns. Entries use the history wire form (visibly shortened when too large, or the - * history's too-large marker when even that can't fit a page), and a page stops at `limit` - * entries or the history page byte budget, whichever comes first. - * `waiting`: the live answer for a pending prompt. */ +/** One `orca agent read` page from a chat's journal: its newest entries in the current context, + * walked back over the journal's live agent-read entries, so work is bounded by the page. It + * never renders the timeline, builds a snapshot or scans for turns. Entries use the history wire + * form (visibly shortened when too large, or the history's too-large marker when even that can't + * fit a page), and a page stops at `limit` entries or the history page byte budget, whichever + * comes first. `waiting`: the live answer for a pending prompt. */ export function readAgentJournalPage( journal: AgentSessionJournal, - request: { limit: number; after?: AgentReadJournalPosition }, + request: { limit: number }, waiting: (itemId: string) => boolean ): AgentReadJournalPage { - const { limit, after } = request - const page = new PageBudget(limit) - const index = journalAgentReadIndex(journal) - const floor = index.contextFloor() - const bound = floor > 0 ? { floor } : {} - if (!after) { - const newestFirst: AgentReadEntry[] = [] - let limited = false - for (const item of index.newestFirst()) { - const entry = entryOf(item, waiting) - if (!entry) { - continue - } - if (!page.admit(entry)) { - limited = true - break - } - newestFirst.push(entry) - } - return { - ok: true, - entries: newestFirst.toReversed(), - next: { cursor: journal.cursor(), ...bound }, - limited - } - } - const since = journal.readSince(after.cursor, limit + 1) - if ( - !since.ok || - (since.rows.every((row) => row.kind !== 'outcome') && - findSequenceGap( - since.rows.map((row) => row.seq), - after.cursor.sequence + 1 - )) || - (after.partial && since.rows[0]?.seq !== after.partial.row) || - // A `/clear` since this cursor was cut starts a context it never read. - (after.floor ?? 0) !== floor - ) { - return { ok: false } - } - const rows = since.rows.slice(0, limit) - const display = journal.queuedMessages.displayState() - const entries: AgentReadEntry[] = [] - let consumed = after.cursor.sequence - for (const row of rows) { - const read = readRow( - journal, - display, - row, - row.seq === after.partial?.row ? after.partial : null, - floor - ) - if (!read) { - return { ok: false } - } - for (const group of read.groups) { - const entry = entryOf(group.item, waiting, read.reserve) - if (!entry) { - continue - } - if (!page.admit(entry, read.reserve)) { - // Stop before this item, with what this row has already represented. - return { - ok: true, - entries, - next: { - cursor: { epoch: after.cursor.epoch, sequence: consumed }, - partial: read.partial(), - ...bound - }, - limited: true - } - } - read.represent(group) - entries.push(entry) - } - consumed = row.seq - } - const more = since.rows.length > rows.length - return { - ok: true, - entries, - next: { - cursor: { epoch: after.cursor.epoch, sequence: more ? consumed : since.through }, - ...bound - }, - limited: more - } -} - -type RowGroup = { item: AgentJournalRenderItem; slots: number[] } - -/** One row's current unrepresented items in presented order, over its raw slots: a canonical - * item with any represented slot is represented, wherever it now sorts, and hidden or missing - * items, and items placed before the current context's `floor`, are left unrepresented. Null - * when `partial` was cut from a different row. */ -function readRow( - journal: AgentSessionJournal, - display: Parameters[0], - row: JournalReadRow, - partial: AgentReadPartialRow | null, - floor: number -) { - const slots = journalRowRawSlots(row) - const fingerprint = agentReadRowFingerprint(row.kind, slots) - const emitted = partial - ? partial.count === slots.length && partial.fingerprint === fingerprint - ? EmittedSlots.decode(partial.emitted, slots.length) - : null - : EmittedSlots.none(slots.length) - if (!emitted) { - return null - } - const bySlot = new Map() - slots.forEach((slot, index) => { - const id = journal.canonicalItemId(slot) - bySlot.set(id, [...(bySlot.get(id) ?? []), index]) - }) - const groups: RowGroup[] = [] - for (const [id, indexes] of bySlot) { - if (indexes.some((index) => emitted.has(index))) { - indexes.forEach((index) => emitted.add(index)) + const page = new PageBudget(request.limit) + const newestFirst: AgentReadEntry[] = [] + let limited = false + for (const item of journalAgentReadIndex(journal).newestFirst()) { + const entry = entryOf(item, waiting) + if (!entry) { continue } - const item = renderJournalStatusItem(display, id) - if (item && item.sequence >= floor) { - groups.push({ item, slots: indexes }) + if (!page.admit(entry)) { + limited = true + break } + newestFirst.push(entry) } - groups.sort( - (a, b) => - compareAgentJournalItems(a.item, b.item) || - (a.item.itemId < b.item.itemId ? -1 : a.item.itemId > b.item.itemId ? 1 : 0) - ) - return { - groups, - reserve: Math.max(0, agentReadPartialRowBytes(slots.length) - CURSOR_ALLOWANCE_BYTES), - represent: (group: RowGroup) => group.slots.forEach((index) => emitted.add(index)), - partial: (): AgentReadPartialRow => ({ - row: row.seq, - count: slots.length, - fingerprint, - emitted: emitted.encode() - }) - } + return { entries: newestFirst.toReversed(), limited } } /** The entry an item reads as, from its history wire form. One no page could carry reads as the * history's own too-large marker, said by Orca rather than the agent. */ function entryOf( item: AgentJournalRenderItem, - waiting: (itemId: string) => boolean, - reserve = 0 + waiting: (itemId: string) => boolean ): AgentReadEntry | null { - const budget = HISTORY_PAGE_CONTENT_BUDGET_BYTES - reserve const sent = agentJournalItemForWire(item) const entry = agentChatEntryOf(sent.item, waiting) if (!entry) { return null } const shown: AgentReadEntry = sent.item.shortened ? { ...entry, shortened: true } : entry - if (sent.bytes <= budget && entryBytes(shown) <= budget) { + if ( + sent.bytes <= HISTORY_PAGE_CONTENT_BUDGET_BYTES && + entryBytes(shown) <= HISTORY_PAGE_CONTENT_BUDGET_BYTES + ) { return shown } const marker = oversizedHistoryItem(item).body @@ -225,19 +68,16 @@ function entryOf( /** An entry's serialized size, with its separator in the page's entries array. */ const entryBytes = (entry: AgentReadEntry) => Buffer.byteLength(JSON.stringify(entry)) + 1 -/** `limit` entries within the history page's byte budget, less any cursor `reserve`. */ +/** `limit` entries within the history page's byte budget. */ class PageBudget { private count = 0 private bytes = 0 constructor(private readonly limit: number) {} - admit(entry: AgentReadEntry, reserve = 0): boolean { + admit(entry: AgentReadEntry): boolean { const size = entryBytes(entry) - if ( - this.count >= this.limit || - this.bytes + size + reserve > HISTORY_PAGE_CONTENT_BUDGET_BYTES - ) { + if (this.count >= this.limit || this.bytes + size > HISTORY_PAGE_CONTENT_BUDGET_BYTES) { return false } this.count++ diff --git a/src/main/native-chat/agent-session-wire/publication/history/agent-read-row-progress.ts b/src/main/native-chat/agent-session-wire/publication/history/agent-read-row-progress.ts deleted file mode 100644 index c5b60ed6353..00000000000 --- a/src/main/native-chat/agent-session-wire/publication/history/agent-read-row-progress.ts +++ /dev/null @@ -1,85 +0,0 @@ -import { createHash } from 'node:crypto' -import { isRecord } from '../../../../../shared/agent-status/child-work/agent-status-child-work-value-guards' - -/** An `orca agent read` stopped inside one journal row: which of that row's raw slots it has - * represented, pinned to the row's sequence, slot count and slot fingerprint. */ -export type AgentReadPartialRow = { - row: number - count: number - fingerprint: string - /** base64url bitset, one bit per raw slot, padding bits zero. */ - emitted: string -} - -const FINGERPRINT = /^[\w-]{22}$/ - -/** The row's kind and raw slots in order, each length-delimited, as 128 bits of SHA-256. */ -export function agentReadRowFingerprint(kind: string, slots: readonly string[]): string { - const hash = createHash('sha256') - for (const part of [kind, ...slots]) { - hash.update(`${Buffer.byteLength(part)}:${part}`) - } - return hash.digest('base64url').slice(0, 22) -} - -/** Which raw slots of one row have been represented. */ -export class EmittedSlots { - private constructor(private readonly bits: Uint8Array) {} - - static none(count: number): EmittedSlots { - return new EmittedSlots(new Uint8Array(Math.ceil(count / 8))) - } - - /** Null unless `emitted` is the canonical encoding of exactly `count` bits. */ - static decode(emitted: string, count: number): EmittedSlots | null { - const bits = new Uint8Array(Buffer.from(emitted, 'base64url')) - const slots = new EmittedSlots(bits) - const padding = bits.length * 8 - count - const last = bits.at(-1) ?? 0 - return bits.length === Math.ceil(count / 8) && - slots.encode() === emitted && - (last & ((1 << padding) - 1)) === 0 - ? slots - : null - } - - has(slot: number): boolean { - return ((this.bits[slot >> 3] ?? 0) & (0x80 >> (slot & 7))) !== 0 - } - - add(slot: number): void { - this.bits[slot >> 3] = (this.bits[slot >> 3] ?? 0) | (0x80 >> (slot & 7)) - } - - encode(): string { - return Buffer.from(this.bits).toString('base64url') - } -} - -/** A well-formed partial row strictly after the consumed `frontier`. */ -export function isAgentReadPartialRow( - value: unknown, - frontier: number -): value is AgentReadPartialRow { - if (!isRecord(value)) { - return false - } - const { row, count, fingerprint, emitted } = value - return ( - Number.isSafeInteger(row) && - typeof row === 'number' && - row > frontier && - Number.isSafeInteger(count) && - typeof count === 'number' && - count >= 0 && - typeof fingerprint === 'string' && - FINGERPRINT.test(fingerprint) && - typeof emitted === 'string' && - EmittedSlots.decode(emitted, count) !== null - ) -} - -/** Bytes a cursor carrying a partial row of `count` slots can add to a reply. */ -export function agentReadPartialRowBytes(count: number): number { - return 1024 + Math.ceil(count / 8) * 2 -} diff --git a/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-clear.test.ts b/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-clear.test.ts new file mode 100644 index 00000000000..c5e17378ca6 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-clear.test.ts @@ -0,0 +1,140 @@ +import { afterEach, expect, it } from 'vitest' +import { agentSessionJournalProviderHandle } from '../../../../../../shared/agent-session/lifecycle/agent-session-provider-handle-encoding' +import type { AgentSessionJournal } from '../../../../agent-session-journal/journal-store' +import { rewindRefusalBehindClear } from '../../../rewind/structured-rewind-refusal' +import { + closeAgentReadJournals, + identity, + journals, + message, + resetWork, + roots, + seed, + tail, + texts, + write +} from './agent-read-journal-fixture' + +afterEach(closeAgentReadJournals) + +const clear = ( + journal: AgentSessionJournal, + operationId: string, + receipt = { write: () => {}, committed: () => {} } +) => journal.context.clear({ operationId, afterFence: 1, clearedAt: 1_000 }, receipt) +const CLEARED = { kind: 'message', role: 'system', text: 'Context cleared' } as const + +/** APPLE and BANANA, then `/clear`. */ +async function clearedChat() { + const journal = await seed(0, () => null) + await write(journal, 'apple', [['apple', 'APPLE']]) + await write(journal, 'banana', [['banana', 'BANANA']]) + tail(journal, 100) + await clear(journal, 'clear-1') + return journal +} + +it('reads only the current context after /clear, from its divider', async () => { + const journal = await clearedChat() + expect(tail(journal, 100)).toMatchObject({ entries: [CLEARED], limited: false }) + + await write(journal, 'cherry', [['cherry', 'CHERRY']]) + // An old item revised after the clear stays out of the current context. + await write(journal, 'apple again', [['apple', 'APPLE revised']]) + + expect(tail(journal, 100)).toMatchObject({ + entries: [CLEARED, { text: 'CHERRY' }], + limited: false + }) + expect(tail(journal, 1)).toMatchObject({ entries: [{ text: 'CHERRY' }], limited: true }) + expect(tail(journal, 2)).toMatchObject({ entries: [CLEARED, { text: 'CHERRY' }], limited: false }) + // The archive keeps the old context. + expect(journal.statusRead().items.some((item) => item.itemId === 'apple')).toBe(true) +}) + +it('starts the current context at the newest of repeated clears, and after reopening', async () => { + const journal = await clearedChat() + await write(journal, 'cherry', [['cherry', 'CHERRY']]) + await clear(journal, 'clear-2') + await write(journal, 'date', [['date', 'DATE']]) + expect(texts(tail(journal, 100))).toEqual(['Context cleared', 'DATE']) + + await journal.close() + const reopened = await journals.open({ identity, stateDirectory: roots.at(-1) ?? '' }) + + expect(texts(tail(reopened, 100))).toEqual(['Context cleared', 'DATE']) +}) + +it.each([1_000, 10_000])( + 'reads the current context after %i cleared replies without walking them', + async (count) => { + const journal = await seed(count, (index) => message(`reply ${index}`)) + tail(journal, 1) + const work = resetWork(journal) + await clear(journal, 'clear') + await write(journal, 'cherry', [['cherry', 'CHERRY']]) + + expect(texts(tail(journal, 100))).toEqual(['Context cleared', 'CHERRY']) + // Two placements and a walk of the two current entries: tree paths, never the old context. + expect({ upkeep: work.upkeep, builds: work.builds, reads: work.reads }).toEqual({ + upkeep: 2, + builds: 0, + reads: 2 + }) + expect(work.nodes).toBeLessThan(80) + } +) + +it('leaves the read untouched when a clear commits no divider', async () => { + const journal = await clearedChat() + await write(journal, 'cherry', [['cherry', 'CHERRY']]) + const before = tail(journal, 100) + const head = journal.cursor() + + // The receipt's write failing rolls the clear back: no divider is committed. + await expect( + clear(journal, 'refused', { + write: () => { + throw new Error('refused-clear') + }, + committed: () => {} + }) + ).rejects.toThrow('refused-clear') + + expect(journal.cursor()).toEqual(head) + expect(tail(journal, 100)).toEqual(before) +}) + +it.each(['plain', 'forked'] as const)( + 'rewinds within the current context (%s), keeping its divider, and refuses a target before it', + async (kind) => { + const journal = await clearedChat() + await write(journal, 'cherry', [['cherry', 'CHERRY']]) + const before = tail(journal, 100) + const floor = journal.context.floor() + const work = resetWork(journal) + + // A rewind to an item before the divider is refused before any agent runs. + const epoch = journal.cursor().epoch + expect( + rewindRefusalBehindClear(journal, { itemId: 'banana', expectedEpoch: epoch }) + ).not.toBeNull() + expect(rewindRefusalBehindClear(journal, { itemId: 'cherry', expectedEpoch: epoch })).toBeNull() + expect(tail(journal, 100)).toEqual(before) + + await journal.context.rewind( + floor ?? { epoch, sequence: 0 }, + 1, + [], + () => ({ write: () => {}, committed: () => {} }), + // A provider that forked its history starts a new epoch, as structured rewind recovery does. + kind === 'forked' + ? { reason: 'handle_forked', providerHandle: agentSessionJournalProviderHandle(identity) } + : undefined + ) + + expect(tail(journal, 100)).toMatchObject({ entries: [CLEARED], limited: false }) + // A new epoch rebuilds the read owner from what survived; the same epoch keeps it. + expect(work.builds).toBe(kind === 'forked' ? 1 : 0) + } +) diff --git a/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-continuation.test.ts b/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-continuation.test.ts deleted file mode 100644 index 434db5ee02f..00000000000 --- a/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-continuation.test.ts +++ /dev/null @@ -1,656 +0,0 @@ -import type * as JournalPublicRead from '../../../../agent-session-journal/projection/journal-public-read' -import { afterEach, expect, it, vi } from 'vitest' -import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../../../../shared/agent-session/journal/agent-session-journal-types' -import { agentJournalItemKey } from '../../../../../../shared/agent-session/journal/agent-session-journal-item-key' -import type { AgentSessionJournal } from '../../../../agent-session-journal/journal-store' -import { rewindRefusalBehindClear } from '../../../rewind/structured-rewind-refusal' -import { agentSessionJournalProviderHandle } from '../../../../../../shared/agent-session/lifecycle/agent-session-provider-handle-encoding' -import { isMobileE2EETextPayloadWithinLimit } from '../../../../../runtime/rpc/mobile/e2ee/mobile-e2ee-outbound-admission' -import { - readAgentJournalPage, - type AgentReadJournalPage, - type AgentReadJournalPosition -} from '../agent-read-journal-page' -import { - abc, - at, - base, - closeAgentReadJournals, - HISTORY_CONTENT_BYTES, - identity, - journals, - message, - publicEntries, - readThrough, - remove, - resetWork, - reply, - roots, - rowAfterReply, - seed, - tail, - textOf, - texts, - waiting, - write -} from './agent-read-journal-fixture' - -/** Every item render, so a test can pin a read's work. */ -const renders = vi.hoisted(() => ({ count: 0 })) -vi.mock('../../../../agent-session-journal/projection/journal-public-read', async (load) => { - const actual = await load() - return { - ...actual, - renderJournalStatusItem: (...args: Parameters) => { - renders.count++ - return actual.renderJournalStatusItem(...args) - } - } -}) - -afterEach(closeAgentReadJournals) - -it.each([ - ['[a, b, c] at limit 1', 1, abc], - ['[a, b, c] at limit 2', 2, abc], - ['[a, b] at limit 1', 1, abc.slice(0, 2)] -])( - 'reads the rest of a row whose delivered first item is then deleted: %s', - async (_, limit, items) => { - const { journal, start, stillThere } = await rowAfterReply(items) - - const { read, after } = await readThrough(journal, limit, start, [ - async () => { - await remove(journal, 'a') - stillThere() - expect(texts(tail(journal, 3))).toContain('b') - } - ]) - - expect(read).toEqual(items.map(([, text]) => text)) - expect(after).toEqual({ cursor: journal.cursor() }) - } -) - -it('reads the rest of a row split by bytes after its delivered first item is deleted', async () => { - const long = (name: string) => `${name}:${'y'.repeat(3 * 1024 * 1024)}` - const { journal, start, stillThere } = await rowAfterReply([ - ['a', long('a')], - ['b', long('b')], - ['c', 'c'] - ]) - - const { pages, read } = await readThrough(journal, 50, start, [ - async () => { - await remove(journal, 'a') - stillThere() - } - ]) - - expect(texts(pages[0] ?? { entries: [] }).map((text) => text.split(':')[0])).toEqual(['a']) - expect(read.map((text) => text.split(':')[0])).toEqual(['a', 'b', 'c']) -}) - -it.each([ - [ - 'hides a delivered item', - abc, - (j: AgentSessionJournal) => write(j, 'hide', [['a', '']]), - ['a', 'b', 'c'] - ], - [ - 'hides an unread item', - abc, - (j: AgentSessionJournal) => write(j, 'hide', [['b', '']]), - ['a', 'c'] - ], - ['deletes an unread item', abc, (j: AgentSessionJournal) => remove(j, 'b'), ['a', 'c']], - ['removes every remaining item', abc, (j: AgentSessionJournal) => remove(j, 'b', 'c'), ['a']], - [ - // The row shows it in place; its revising write is a later entry of its own. - 'reveals a hidden unread item', - [ - ['a', 'a'], - ['h', ''], - ['c', 'c'] - ] satisfies [string, string][], - (j: AgentSessionJournal) => write(j, 'reveal', [['h', 'h']]), - ['a', 'h', 'c', 'h'] - ], - [ - 'revises an unread item', - abc, - (j: AgentSessionJournal) => write(j, 'revise', [['b', 'b revised']]), - ['a', 'b revised', 'c', 'b revised'] - ] -] as const)('continues a row when another write %s', async (_, items, change, expected) => { - const { journal, start, stillThere } = await rowAfterReply([...items]) - - const { read, after } = await readThrough(journal, 1, start, [ - async () => { - await change(journal) - stillThere() - } - ]) - - expect(read).toEqual(expected) - expect(after).toEqual({ cursor: journal.cursor() }) -}) - -it.each([ - // Deleted and written again, a delivered item moves after its siblings and isn't read twice - // in this row. - ['a delivered item', 'a', ['b', 'c', 'a again'], ['a', 'b', 'c']], - // An unread item moved past its sibling reads after it. - ['an unread item', 'b', ['a', 'c', 'b again'], ['a', 'c', 'b again']] -] as const)('continues a row when %s moves', async (_, moved, presented, row) => { - const { journal, start, stillThere } = await rowAfterReply(abc) - - const { read } = await readThrough(journal, 1, start, [ - async () => { - await remove(journal, moved) - await write(journal, 'again', [[moved, `${moved} again`]]) - stillThere() - expect(texts(tail(journal, 3))).toEqual(presented) - } - ]) - - // The deleting and rewriting writes each name the item again: later entries, current content. - expect(read).toEqual([...row, `${moved} again`, `${moved} again`]) -}) - -it.each([ - // A hidden slot later named as a send's provider item: never delivered, so the send reads. - [ - 'an unread hidden item', - [ - ['a', 'a'], - ['provider', ''], - ['b', 'b'] - ], - ['a', 'b', 'hi'] - ], - // A delivered slot later named as that send's provider item: the send is already represented. - [ - 'a delivered item', - [ - ['provider', 'hello'], - ['a', 'a'], - ['b', 'b'] - ], - ['hello', 'a', 'b'] - ] -])('continues a row when %s becomes an alias of a send between reads', async (_, items, row) => { - const provider = at(50) - const journal = await seed( - 0, - () => null, - (seq) => [ - { - ...base, - kind: 'lifecycle-batch', - seq, - settlementId: 'row', - mutations: items.map(([name, text]) => ({ - kind: 'item' as const, - itemId: name === 'provider' ? agentJournalItemKey(provider) : name, - revision: 1, - body: - name === 'provider' - ? { - kind: 'message' as const, - role: 'user' as const, - blocks: [{ type: 'text' as const, text }] - } - : message(text) - })) - } - ] - ) - await journal.appendSubmission({ - clientMessageId: 'send-1', - payloadFingerprint: 'send-1', - fence: 1, - handoverRecorded: true, - body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] } - }) - const start: AgentReadJournalPosition = { cursor: { epoch: 'epoch-1', sequence: 1 } } - - const { pages } = await readThrough(journal, 1, start, [ - async () => { - await journal.resolveDispatch({ - clientMessageId: 'send-1', - state: 'pending', - fence: 1, - turnScope: AGENT_JOURNAL_THREAD_SCOPE - }) - await journal.resolveDispatch({ - clientMessageId: 'send-1', - state: 'accepted', - providerIdentity: provider, - fence: 1 - }) - expect(journal.canonicalItemId(agentJournalItemKey(provider))).toBe('orca:send-1') - } - ]) - - expect(pages.slice(0, row.length).map(texts)).toEqual(row.map((text) => [text])) -}) - -it.each([1, 2])( - 'reads a saved row of [z-question, a-answer] as presented at limit %i', - async (limit) => { - const journal = await seed( - 0, - () => null, - (seq) => [ - { - ...base, - kind: 'lifecycle-batch', - seq, - settlementId: 'exchange', - mutations: [ - { - kind: 'item', - itemId: 'z-question', - revision: 1, - body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'question' }] } - }, - { kind: 'item', itemId: 'a-answer', revision: 1, body: message('answer') } - ] - } - ] - ) - - const { read } = await readThrough( - journal, - limit, - { cursor: { epoch: 'epoch-1', sequence: 1 } }, - [] - ) - - expect(read).toEqual(['question', 'answer']) - expect(publicEntries(journal).map(textOf)).toEqual(['question', 'answer']) - } -) - -it.each([1_000, 10_000])( - 'reads a small row after %i replies with work set by the row alone', - async (count) => { - const journal = await seed(count, (index) => message(`reply ${index}`)) - const start: AgentReadJournalPosition = { cursor: journal.cursor() } - await write(journal, 'exchange', [ - ['z-question', 'question'], - ['a-answer', 'answer'], - ['m-more', 'more'] - ]) - // The owner's index is built once, by the first read of any kind. - tail(journal, 1) - renders.count = 0 - - const { read } = await readThrough(journal, 1, start, [() => remove(journal, 'z-question')]) - - expect(read).toEqual(['question', 'answer', 'more']) - // Seven renders read the row; one keeps the index current through the deletion. - expect(renders.count).toBe(8) - } -) - -// A write of 10,000 items is stored as groups of at most 200, the saved batch limit. -it.each(['saved', 'written'] as const)( - 'carries a cursor inside a 200-item %s batch within the reply bounds', - async (how) => { - const items = Array.from( - { length: 200 }, - (_, index) => `item ${String(index).padStart(5, '0')}` - ) - const journal = - how === 'saved' - ? await seed( - 0, - () => null, - (seq) => [ - { - ...base, - kind: 'lifecycle-batch', - seq, - settlementId: 'batch', - mutations: items.map((itemId) => ({ - kind: 'item' as const, - itemId, - revision: 1, - body: message(itemId) - })) - } - ] - ) - : await seed(1, () => message('before')) - const start: AgentReadJournalPosition = { - cursor: how === 'saved' ? { epoch: 'epoch-1', sequence: 1 } : journal.cursor() - } - if (how === 'written') { - await write( - journal, - 'batch', - items.map((itemId) => [itemId, itemId]) - ) - } - - const first = readAgentJournalPage(journal, { limit: 1, after: start }, waiting) - if (!first.ok) { - throw new Error('expected a page') - } - const second = readAgentJournalPage(journal, { limit: 2, after: first.next }, waiting) - - expect(first.next.partial?.count).toBe(200) - expect(isMobileE2EETextPayloadWithinLimit(reply(first).json)).toBe(true) - expect(second.ok && texts(second)).toEqual(['item 00001', 'item 00002']) - } -) - -it('keeps a cursor inside a 100,000-slot row within the reply bounds', () => { - const emitted = Buffer.alloc(100_000 / 8, 0xff).toString('base64url') - const page: Extract = { - ok: true, - entries: [{ kind: 'message', role: 'assistant', text: 'x'.repeat(HISTORY_CONTENT_BYTES - 64) }], - next: { - cursor: { epoch: 'epoch-1', sequence: 1 }, - partial: { row: 2, count: 100_000, fingerprint: 'A'.repeat(22), emitted } - }, - limited: true - } - - expect(Buffer.byteLength(reply(page).cursor)).toBeLessThan(32 * 1024) - expect(isMobileE2EETextPayloadWithinLimit(reply(page).json)).toBe(true) -}) - -it('resumes a row read in part after the journal reopens, and refuses a cursor cut elsewhere', async () => { - const { journal, start } = await rowAfterReply(abc) - const first = readAgentJournalPage(journal, { limit: 1, after: start }, waiting) - if (!first.ok || !first.next.partial) { - throw new Error('expected a partial page') - } - const partial = first.next.partial - await journal.close() - const reopened = await journals.open({ identity, stateDirectory: roots.at(-1) ?? '' }) - const resume = (next: AgentReadJournalPosition) => - readAgentJournalPage(reopened, { limit: 5, after: next }, waiting) - - expect(resume(first.next)).toMatchObject({ ok: true, entries: [{ text: 'b' }, { text: 'c' }] }) - // Nothing of the row emitted yet: a valid cursor that reads it whole. - expect(resume({ ...first.next, partial: { ...partial, emitted: 'AA' } })).toMatchObject({ - entries: [{ text: 'a' }, { text: 'b' }, { text: 'c' }] - }) - for (const changed of [ - { ...partial, fingerprint: 'A'.repeat(22) }, - { ...partial, count: 4 }, - { ...partial, row: partial.row + 1 } - ]) { - expect(resume({ ...first.next, partial: changed })).toEqual({ ok: false }) - } - await reopened.rollEpoch('corruption', 1) - expect(resume(first.next)).toEqual({ ok: false }) -}) - -const clear = ( - journal: AgentSessionJournal, - operationId: string, - receipt = { write: () => {}, committed: () => {} } -) => journal.context.clear({ operationId, afterFence: 1, clearedAt: 1_000 }, receipt) -const CLEARED = { kind: 'message', role: 'system', text: 'Context cleared' } as const - -/** APPLE and BANANA, then `/clear`, then CHERRY, with cursors cut before the clear. */ -async function clearedChat() { - const journal = await seed(0, () => null) - await write(journal, 'apple', [['apple', 'APPLE']]) - await write(journal, 'banana', [['banana', 'BANANA']]) - const full = tail(journal, 100) - const start: AgentReadJournalPosition = { cursor: { epoch: journal.cursor().epoch, sequence: 1 } } - const partial = readAgentJournalPage(journal, { limit: 1, after: start }, waiting) - await clear(journal, 'clear-1') - return { journal, full, partial } -} - -it('reads only the current context after /clear, from its divider', async () => { - const { journal } = await clearedChat() - expect(tail(journal, 100)).toMatchObject({ entries: [CLEARED], limited: false }) - - await write(journal, 'cherry', [['cherry', 'CHERRY']]) - - expect(tail(journal, 100)).toMatchObject({ - entries: [CLEARED, { text: 'CHERRY' }], - limited: false - }) - expect(tail(journal, 1)).toMatchObject({ entries: [{ text: 'CHERRY' }], limited: true }) - expect(tail(journal, 2)).toMatchObject({ entries: [CLEARED, { text: 'CHERRY' }], limited: false }) - // The archive keeps the old context. - expect(journal.statusRead().items.some((item) => item.itemId === 'apple')).toBe(true) -}) - -it('refuses a cursor cut before /clear, and follows one cut after it', async () => { - const { journal, full, partial } = await clearedChat() - expect(partial.ok && partial.limited).toBe(true) - - for (const before of [full.next, partial.ok ? partial.next : full.next]) { - expect(readAgentJournalPage(journal, { limit: 5, after: before }, waiting)).toEqual({ - ok: false - }) - } - const fresh = tail(journal, 100) - await write(journal, 'cherry', [['cherry', 'CHERRY']]) - // An old item revised after the clear stays out of the current context. - await write(journal, 'apple again', [['apple', 'APPLE revised']]) - - await expect(readThrough(journal, 1, fresh.next, [])).resolves.toMatchObject({ read: ['CHERRY'] }) - expect(texts(tail(journal, 100))).toEqual(['Context cleared', 'CHERRY']) -}) - -it('starts the current context at the newest of repeated clears, and after reopening', async () => { - const { journal } = await clearedChat() - await write(journal, 'cherry', [['cherry', 'CHERRY']]) - await clear(journal, 'clear-2') - await write(journal, 'date', [['date', 'DATE']]) - expect(texts(tail(journal, 100))).toEqual(['Context cleared', 'DATE']) - - await journal.close() - const reopened = await journals.open({ identity, stateDirectory: roots.at(-1) ?? '' }) - - expect(texts(tail(reopened, 100))).toEqual(['Context cleared', 'DATE']) -}) - -it.each([1_000, 10_000])( - 'reads the current context after %i cleared replies without walking them', - async (count) => { - const journal = await seed(count, (index) => message(`reply ${index}`)) - tail(journal, 1) - const work = resetWork(journal) - await clear(journal, 'clear') - await write(journal, 'cherry', [['cherry', 'CHERRY']]) - - expect(texts(tail(journal, 100))).toEqual(['Context cleared', 'CHERRY']) - // Two placements and a walk of the two current entries: tree paths, never the old context. - expect({ upkeep: work.upkeep, builds: work.builds, reads: work.reads }).toEqual({ - upkeep: 2, - builds: 0, - reads: 2 - }) - expect(work.nodes).toBeLessThan(80) - } -) - -/** One completed Claude turn, written in the order QA 5385's real chat committed it (recovered - * from that run's journal; see `orca-qa/pr-41815/qa5385-artifacts/banana-sequence.md`): the - * send, its hand-over, the previous turn's two late usage revisions, the new turn opening, the - * provider accepting the send, the reply written twice with the same text, and three turn - * revisions. Identities are shortened; their order and repetitions are the run's. */ -async function completedTurn( - journal: AgentSessionJournal, - word: string, - previous?: string, - answer = word -) { - const id = `send-${word}` - const user = { - kind: 'message' as const, - role: 'user' as const, - blocks: [{ type: 'text' as const, text: `Reply with the single word ${word}` }] - } - const turn = { - provider: 'codex' as const, - threadId: 'thread-1', - turnId: `turn-${word}`, - ordinal: 0 - } - const turnBody = (state: 'running' | 'completed', durationMs?: number) => ({ - kind: 'turn' as const, - turnId: word, - state, - ...(state === 'completed' ? { outcome: 'success' as const } : {}), - ...(durationMs === undefined ? {} : { durationMs }) - }) - await journal.appendSubmission({ - clientMessageId: id, - payloadFingerprint: id, - fence: 1, - handoverRecorded: true, - body: user - }) - await journal.resolveDispatch({ - clientMessageId: id, - state: 'pending', - fence: 1, - turnScope: AGENT_JOURNAL_THREAD_SCOPE - }) - if (previous) { - const prior = { - provider: 'codex' as const, - threadId: 'thread-1', - turnId: `turn-${previous}`, - ordinal: 0 - } - for (const durationMs of [5, 6]) { - await journal.appendItem(prior, turnBody('completed', durationMs), { - fence: 1, - turnScope: AGENT_JOURNAL_THREAD_SCOPE - }) - } - } - await journal.appendItem(turn, turnBody('running'), { - fence: 1, - turnScope: AGENT_JOURNAL_THREAD_SCOPE - }) - await journal.resolveDispatch({ - clientMessageId: id, - state: 'accepted', - providerIdentity: { ...turn, turnId: `user-${word}` }, - fence: 1 - }) - for (let revision = 0; revision < 2; revision++) { - await journal.appendItem({ ...turn, turnId: `reply-${word}` }, message(answer), { - fence: 1, - turnScope: AGENT_JOURNAL_THREAD_SCOPE - }) - } - for (const [state, durationMs] of [ - ['running', 1], - ['completed', 2], - ['completed', 3] - ] as const) { - await journal.appendItem(turn, turnBody(state, durationMs), { - fence: 1, - turnScope: AGENT_JOURNAL_THREAD_SCOPE - }) - } -} - -// QA 5385 Q-11: awaits the conversation-unit forward read (plan amendment "#41815 -// conversation-unit forward read"). Expected to fail until then. The byte-split case keeps the -// run's write order with a 3 MiB answer, so its pages split by bytes. -it.fails.each([ - ['limit 1', 1, 'BANANA'], - ['limit 2', 2, 'BANANA'], - ['the default limit', 100, 'BANANA'], - ['a byte split', 100, `BANANA:${'y'.repeat(3 * 1024 * 1024)}`] -])('reads a completed turn once each, user then answer, at %s', async (_, limit, answer) => { - const journal = await seed(0, () => null) - await completedTurn(journal, 'APPLE') - const after = tail(journal, 100).next - await completedTurn(journal, 'BANANA', 'APPLE', answer) - - const { pages, read, after: end } = await readThrough(journal, limit, after, []) - - expect(read.map((text) => text.split(':')[0])).toEqual([ - 'Reply with the single word BANANA', - 'BANANA' - ]) - expect(pages.filter((page) => page.limited && page.entries.length === 0)).toEqual([]) - expect(readAgentJournalPage(journal, { limit, after: end }, waiting)).toMatchObject({ - entries: [], - limited: false - }) -}) - -it('leaves the read untouched when a clear commits no divider', async () => { - const { journal } = await clearedChat() - await write(journal, 'cherry', [['cherry', 'CHERRY']]) - const before = tail(journal, 100) - const head = journal.cursor() - - // The receipt's write failing rolls the clear back: no divider is committed. - await expect( - clear(journal, 'refused', { - write: () => { - throw new Error('refused-clear') - }, - committed: () => {} - }) - ).rejects.toThrow('refused-clear') - - expect(journal.cursor()).toEqual(head) - expect(tail(journal, 100)).toEqual(before) - expect(readAgentJournalPage(journal, { limit: 5, after: before.next }, waiting)).toMatchObject({ - ok: true, - entries: [] - }) -}) - -it.each(['plain', 'forked'] as const)( - 'rewinds within the current context (%s), keeping its divider, and refuses a target before it', - async (kind) => { - const { journal } = await clearedChat() - await write(journal, 'cherry', [['cherry', 'CHERRY']]) - const before = tail(journal, 100) - const floor = before.next.floor ?? 0 - const work = resetWork(journal) - - // A rewind to an item before the divider is refused before any agent runs. - const epoch = journal.cursor().epoch - expect( - rewindRefusalBehindClear(journal, { itemId: 'banana', expectedEpoch: epoch }) - ).not.toBeNull() - expect(rewindRefusalBehindClear(journal, { itemId: 'cherry', expectedEpoch: epoch })).toBeNull() - expect(tail(journal, 100)).toEqual(before) - - await journal.context.rewind( - { epoch, sequence: floor }, - 1, - [], - () => ({ write: () => {}, committed: () => {} }), - // A provider that forked its history starts a new epoch, as structured rewind recovery does. - kind === 'forked' - ? { reason: 'handle_forked', providerHandle: agentSessionJournalProviderHandle(identity) } - : undefined - ) - - expect(tail(journal, 100)).toMatchObject({ entries: [CLEARED], limited: false }) - const old = readAgentJournalPage(journal, { limit: 5, after: before.next }, waiting) - if (kind === 'forked') { - // A new epoch: pre-rewind cursors are refused and the owner rebuilds from what survived. - expect(old).toEqual({ ok: false }) - expect(work.builds).toBe(1) - } else { - // The same epoch: an old cursor reads the removal as nothing new, at the head. - expect(old).toMatchObject({ ok: true, entries: [], limited: false, next: { floor } }) - } - } -) diff --git a/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-fixture.ts b/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-fixture.ts index 68df0bf444c..70815d2f8de 100644 --- a/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-fixture.ts +++ b/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-fixture.ts @@ -2,7 +2,7 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' -import { expect, vi } from 'vitest' +import { vi } from 'vitest' import { AGENT_JOURNAL_THREAD_SCOPE, AGENT_SESSION_JOURNAL_SCHEMA_VERSION, @@ -28,13 +28,8 @@ import type { JournalRow } from '../../../../agent-session-journal/durable-rows/ import { journalAgentReadIndex } from '../../../../agent-session-journal/projection/journal-agent-read-index' import { agentChatEntryOf } from '../../../../../../shared/agent-conversations/agent-chat-entries' import type { AgentReadEntry } from '../../../../../../shared/agent-conversations/agent-conversation-contracts' -import { - readAgentJournalPage, - type AgentReadJournalPage, - type AgentReadJournalPosition -} from '../agent-read-journal-page' +import { readAgentJournalPage } from '../agent-read-journal-page' import { agentJournalItemForWire } from '../agent-session-item-wire-form' -import { encodeReadCursor } from '../../../../../runtime/agent-conversations/agent-read-cursor' export const identity: AgentSessionJournalIdentity = { sessionId: 'agent-read-page', @@ -124,13 +119,8 @@ export async function seed( export const append = (journal: AgentSessionJournal, index: number, body: AgentJournalItemBody) => journal.appendItem(at(index), body, { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE }) -export const tail = (journal: AgentSessionJournal, limit: number) => { - const page = readAgentJournalPage(journal, { limit }, waiting) - if (!page.ok) { - throw new Error('expected a page') - } - return page -} +export const tail = (journal: AgentSessionJournal, limit: number) => + readAgentJournalPage(journal, { limit }, waiting) export const textOf = (entry: AgentReadEntry) => ('text' in entry ? String(entry.text) : '') export const texts = (page: { entries: AgentReadEntry[] }) => page.entries.map(textOf) @@ -146,28 +136,6 @@ export function resetWork(journal: AgentSessionJournal) { return work } -/** Pages forward from `after` until a page isn't limited. */ -export function readForward( - journal: AgentSessionJournal, - limit: number, - from: AgentReadJournalPosition -) { - let after = from - const pages: Extract[] = [] - for (let page = 0; page < 20; page++) { - const next = readAgentJournalPage(journal, { limit, after }, waiting) - if (!next.ok) { - throw new Error('expected a page') - } - pages.push(next) - after = next.next - if (!next.limited) { - break - } - } - return { entries: pages.flatMap((page) => page.entries), pages, after } -} - export const write = ( journal: AgentSessionJournal, settlementId: string, @@ -183,71 +151,5 @@ export const write = ( turnScope: AGENT_JOURNAL_THREAD_SCOPE })) }) -export const remove = (journal: AgentSessionJournal, ...itemIds: string[]) => - journal.appendLifecycleBatch({ - settlementId: `remove ${itemIds.join(' ')}`, - fence: 1, - mutations: itemIds.map((itemId) => ({ kind: 'tombstone' as const, itemId })) - }) -export const abc: [string, string][] = [ - ['a', 'a'], - ['b', 'b'], - ['c', 'c'] -] - -/** Reads forward a page at a time, running `between[n]` after page n, through to the head. */ -export async function readThrough( - journal: AgentSessionJournal, - limit: number, - from: AgentReadJournalPosition, - between: (() => Promise)[] -) { - let after = from - const pages: Extract[] = [] - for (let page = 0; page < 40; page++) { - const next = readAgentJournalPage(journal, { limit, after }, waiting) - if (!next.ok) { - throw new Error('expected a page') - } - pages.push(next) - after = next.next - await between[page]?.() - if (!next.limited && page >= between.length) { - break - } - } - return { pages, read: pages.flatMap(texts), after } -} - -/** A row written after one reply, the position before it, and its sequence. */ -export async function rowAfterReply(items: [string, string][]) { - const journal = await seed(1, () => message('before')) - const start: AgentReadJournalPosition = { cursor: journal.cursor() } - await write(journal, 'row', items) - const row = journal.cursor().sequence - /** The row is still there to continue, so a reset can't pass for a read. */ - const stillThere = () => { - const source = journal.readSince(start.cursor) - expect(source.ok && source.rows.some((stored) => stored.seq === row)).toBe(true) - } - return { journal, start, stillThere } -} - -/** An RPC reply as the read method sends it, with the encoded cursor. */ -export function reply(page: Extract) { - const cursor = encodeReadCursor({ - v: 2, - id: 'codex_0b8f2a4e_6c1d_4f7a_9e3b_5a2c8d1f0e47', - s: 'chat', - session: identity.sessionId, - journal: page.next.cursor, - ...(page.next.partial ? { partial: page.next.partial } : {}) - }) - return { - json: JSON.stringify({ id: 'r', ok: true, result: { entries: page.entries, cursor } }), - cursor - } -} - // The history page content budget: half the 4 MiB remote reply, less a 64 KiB envelope reserve. export const HISTORY_CONTENT_BYTES = 2 * 1024 * 1024 - 64 * 1024 diff --git a/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-page.test.ts b/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-page.test.ts index 4e6f93a7111..51cece820ff 100644 --- a/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-page.test.ts +++ b/src/main/native-chat/agent-session-wire/publication/history/testing/agent-read-journal-page.test.ts @@ -10,17 +10,10 @@ import { } from '../../../../../../shared/agent-session/journal/agent-session-journal-item-key' import type { AgentSessionJournal } from '../../../../agent-session-journal/journal-store' import { journalAgentReadIndex } from '../../../../agent-session-journal/projection/journal-agent-read-index' -import { agentChatEntryOf } from '../../../../../../shared/agent-conversations/agent-chat-entries' import type { AgentReadEntry } from '../../../../../../shared/agent-conversations/agent-conversation-contracts' import { isMobileE2EETextPayloadWithinLimit } from '../../../../../runtime/rpc/mobile/e2ee/mobile-e2ee-outbound-admission' import { DISPATCH_REJECTED_HOST_RESTARTED } from '../../../../../../shared/agent-session/requests/structured-agent-session-dispatch-rejection' -import { - readAgentJournalPage, - type AgentReadJournalPage, - type AgentReadJournalPosition -} from '../agent-read-journal-page' -import { readAgentSessionHistory } from '../agent-session-history-page' -import { agentSessionSendBodyFingerprint } from '../../../../../../shared/agent-session/requests/structured-agent-session-send-mutation' +import { readAgentJournalPage, type AgentReadJournalPage } from '../agent-read-journal-page' import { buildJournalQueueRow } from '../../../../agent-session-journal/queued-messages/journal-queue-events' import { append, @@ -28,11 +21,8 @@ import { base, closeAgentReadJournals, HISTORY_CONTENT_BYTES, - identity, - key, message, publicEntries, - readForward, resetWork, seed, status, @@ -94,13 +84,10 @@ it.each([1_000, 10_000])( expect(work.reads).toBe(15 * 4) expect(work.nodes).toBeLessThanOrEqual(15 * NODES_PER_READ) - const before = tail(journal, 1) resetWork(journal) await append(journal, count, message('newest reply')) expect(work.upkeep).toBe(1) expect(work.nodes).toBeLessThanOrEqual(NODES_PER_READ) - const next = readAgentJournalPage(journal, { limit: 3, after: before.next }, waiting) - expect(next).toMatchObject({ ok: true, entries: [{ text: 'newest reply' }], limited: false }) expect(texts(tail(journal, 1))).toEqual(['newest reply']) expect(work.builds).toBe(0) expect(publicRead).not.toHaveBeenCalled() @@ -153,7 +140,7 @@ it.each([1_000, 10_000])( expect(work.nodes).toBeLessThanOrEqual(3 * 2 * NODES_PER_READ) expect(texts(tail(journal, 2))).toEqual([`reply ${count - 2}`, 'revised newest']) const page = readAgentJournalPage(journal, { limit: 100 }, waiting) - expect(page.ok && page.entries.length).toBe(100) + expect(page.entries.length).toBe(100) await append(journal, 5, status('early note')) await append(journal, 7, message('reply 7 again')) @@ -190,7 +177,6 @@ it('matches the public timeline through Stop notes, rejected sends, empty heads ) const sameAsPublic = () => expect(readAgentJournalPage(journal, { limit: 100 }, waiting)).toMatchObject({ - ok: true, entries: publicEntries(journal) }) sameAsPublic() @@ -214,22 +200,17 @@ it('matches the public timeline through Stop notes, rejected sends, empty heads await journal.rejectPendingSubmissions(1, HOST_RESTARTED) sameAsPublic() - // A head that shows nothing new advances the cursor and returns no entry. - const before = tail(journal, 1) + // A head that shows nothing new adds no entry. + const before = tail(journal, 3) await append(journal, 200, status('working')) - const after = readAgentJournalPage(journal, { limit: 3, after: before.next }, waiting) - expect(after).toMatchObject({ ok: true, entries: [] }) - expect(after.ok && after.next.cursor.sequence).toBeGreaterThan(before.next.cursor.sequence) + expect(tail(journal, 3)).toEqual(before) - // A new epoch starts a new history; the old cursor no longer reads. + // A new epoch starts a new history. const work = resetWork(journal) await journal.rollEpoch('corruption', 1) sameAsPublic() tail(journal, 1) expect(work.builds).toBe(1) - expect(readAgentJournalPage(journal, { limit: 3, after: before.next }, waiting)).toEqual({ - ok: false - }) }) it('answers a pending prompt from the live owner, not from the index', async () => { @@ -249,53 +230,6 @@ it('answers a pending prompt from the live owner, not from the index', async () expect(work.upkeep).toBe(0) }) -it.each([1, 2])( - 'continues a row naming [a, b, a, c] at limit %i: each once, current content', - async (limit) => { - const mutation = (id: string, revision: number, text: string) => ({ - kind: 'item' as const, - itemId: id, - revision, - body: message(text) - }) - const journal = await seed( - 1, - () => message('before'), - (seq) => [ - { - ...base, - kind: 'lifecycle-batch', - seq, - settlementId: 'batch', - mutations: [ - mutation('a', 1, 'a first'), - mutation('b', 1, 'b'), - mutation('a', 2, 'a revised'), - mutation('c', 1, 'c') - ] - } - ] - ) - let after: AgentReadJournalPosition = { - cursor: { epoch: journal.cursor().epoch, sequence: 2 } - } - const read: string[] = [] - for (let page = 0; page < 5; page++) { - const next = readAgentJournalPage(journal, { limit, after }, waiting) - if (!next.ok) { - throw new Error('expected a page') - } - read.push(...texts(next)) - after = next.next - if (!next.limited) { - break - } - } - expect(read).toEqual(['a revised', 'b', 'c']) - expect(after).toEqual({ cursor: { epoch: journal.cursor().epoch, sequence: 3 } }) - } -) - it.each([10_000, 100_000])( 'delivers fifty %i-character replies in pages a remote reply can carry', async (size) => { @@ -335,90 +269,6 @@ it.each([10_000, 100_000])( expect(textOf(latest.entries.at(-1) ?? { kind: 'error', text: '' }).startsWith('huge:')).toBe( true ) - - let after: AgentReadJournalPosition = { - cursor: { epoch: journal.cursor().epoch, sequence: 1 } - } - const seen: string[] = [] - for (let page = 0; page < 200; page++) { - const next = readAgentJournalPage(journal, { limit: 50, after }, waiting) - if (!next.ok) { - throw new Error('expected a page') - } - expect(fits(next)).toBe(true) - seen.push(...next.entries.map((entry) => textOf(entry).split(':')[0] ?? '')) - after = next.next - if (!next.limited) { - break - } - } - expect(seen).toEqual([ - ...Array.from({ length: 50 }, (_, index) => String(index)), - 'big 0', - 'big 1', - 'big 2', - 'huge' - ]) - } -) - -it.each([1, 2])( - 'continues a row naming two ids for one send at limit %i: once, in place', - async (limit) => { - const user = (text: string): AgentJournalMessageItem => ({ - kind: 'message', - role: 'user', - blocks: [{ type: 'text', text }] - }) - const accepted = key(1) - const echo = key(0) - const journal = await seed( - 0, - () => null, - (seq) => [ - { - ...base, - kind: 'submission', - seq, - clientMessageId: 'send-1', - payloadFingerprint: agentSessionSendBodyFingerprint(identity.sessionId, user('hi')), - providerHandle: { kind: 'codex', threadId: 'thread-1' }, - body: user('hi') - }, - { - ...base, - kind: 'dispatch', - seq: seq + 1, - clientMessageId: 'send-1', - state: 'accepted', - providerItemId: accepted, - reason: null - }, - { - ...base, - kind: 'lifecycle-batch', - seq: seq + 2, - settlementId: 'echoes', - mutations: [ - { kind: 'item', itemId: echo, revision: 1, body: user('hi') }, - { kind: 'item', itemId: 'b', revision: 1, body: message('b') }, - { kind: 'item', itemId: accepted, revision: 1, body: user('hi') }, - { kind: 'item', itemId: 'c', revision: 1, body: message('c') } - ] - } - ] - ) - expect([journal.canonicalItemId(echo), journal.canonicalItemId(accepted)]).toEqual([ - 'orca:send-1', - 'orca:send-1' - ]) - - const read = readForward(journal, limit, { - cursor: { epoch: 'epoch-1', sequence: 3 } - }) - - expect(texts(read)).toEqual(['hi', 'b', 'c']) - expect(read.after).toEqual({ cursor: { epoch: 'epoch-1', sequence: 4 } }) } ) @@ -432,7 +282,7 @@ it.each([1, 2, 50])( 'reads a reply no page can carry as the history too-large marker, at limit %i', async (limit) => { const journal = await seed(1, () => message('before')) - const start = { cursor: journal.cursor() } + // 70,000 blocks of 64 characters: past any page, and no block can be shortened further. const blocks: AgentJournalItemBody = { kind: 'message', @@ -464,12 +314,10 @@ it.each([1, 2, 50])( expect( isMobileE2EETextPayloadWithinLimit(JSON.stringify({ id: 'r', ok: true, result: page })) ).toBe(true) - expect(Buffer.byteLength(JSON.stringify(page.ok ? page.entries : []))).toBeLessThanOrEqual( + expect(Buffer.byteLength(JSON.stringify(page.entries))).toBeLessThanOrEqual( HISTORY_CONTENT_BYTES + 1 ) } - const label = (entry: AgentReadEntry) => - JSON.stringify(entry) === JSON.stringify(TOO_LARGE) ? 'marker' : textOf(entry).split(':')[0] const oversizedLatest = tail(journal, limit) carried(oversizedLatest) @@ -482,12 +330,6 @@ it.each([1, 2, 50])( if (limit > 1) { expect(latest.entries.at(-2)).toEqual(TOO_LARGE) } - - const forward = readForward(journal, limit, start) - forward.pages.forEach(carried) - expect(forward.entries.map(label)).toEqual(['long 1', 'long 2', 'marker', 'after']) - expect(forward.entries.filter((entry) => 'shortened' in entry)).toHaveLength(2) - expect(forward.after).toEqual({ cursor: journal.cursor() }) } ) @@ -545,7 +387,6 @@ it.each([1_000, 10_000])( expect(publicEntries(journal).slice(-2).map(textOf)).toEqual(['old', 'new']) expect(texts(tail(journal, 3))).toEqual([`reply ${count - 1}`, 'old', 'new']) expect(texts(tail(journal, 3))).toEqual([`reply ${count - 1}`, 'old', 'new']) - expect(texts(readForward(journal, 1, before.next))).toEqual(['old', 'new']) expect(work.builds).toBe(0) expect(work.upkeep).toBe(2) } @@ -592,45 +433,3 @@ it.each([1_000, 10_000])( expect(index.noteDependencies()).toEqual({ turns: 3, notes: 3 }) } ) - -it.each([1, 2])( - 'reads one write forward in presented order, not stored-id order, at limit %i', - async (limit) => { - const journal = await seed(1, () => message('before')) - const start = { cursor: journal.cursor() } - await journal.appendLifecycleBatch({ - settlementId: 'exchange', - fence: 1, - mutations: [ - { - kind: 'item', - itemId: 'z-question', - body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'question' }] }, - turnScope: AGENT_JOURNAL_THREAD_SCOPE - }, - { - kind: 'item', - itemId: 'a-answer', - body: message('answer'), - turnScope: AGENT_JOURNAL_THREAD_SCOPE - } - ] - }) - const history = readAgentSessionHistory(journal, { - sessionId: identity.sessionId, - direction: 'after', - cursor: start.cursor, - limit: 50 - }) - - const forward = readForward(journal, limit, start) - - expect(texts(forward)).toEqual(['question', 'answer']) - expect(publicEntries(journal).slice(-2).map(textOf)).toEqual(['question', 'answer']) - expect( - history.ok && - history.page.items.flatMap((item) => agentChatEntryOf(item, waiting) ?? []).map(textOf) - ).toEqual(['question', 'answer']) - expect(forward.after).toEqual({ cursor: journal.cursor() }) - } -) diff --git a/src/main/runtime/agent-conversations/agent-conversation-read.test.ts b/src/main/runtime/agent-conversations/agent-conversation-read.test.ts index 338725d0c30..0e136452e3e 100644 --- a/src/main/runtime/agent-conversations/agent-conversation-read.test.ts +++ b/src/main/runtime/agent-conversations/agent-conversation-read.test.ts @@ -1,19 +1,8 @@ -import { beforeEach, expect, it, vi } from 'vitest' +import { expect, it, vi } from 'vitest' import type { RpcContext } from '../rpc/core' import { readAgentConversation } from './agent-conversation-read' import type { ResolvedAgentConversation } from './agent-conversation-address' -import { decodeReadCursor, encodeReadCursor } from './agent-read-cursor' -import type { AgentReadPartialRow } from '../../native-chat/agent-session-wire/publication/history/agent-read-row-progress' -const SESSION = 'codex_0b8f2a4e_6c1d_4f7a_9e3b_5a2c8d1f0e47' -const host = vi.hoisted(() => ({ - reads: { journalCursor: vi.fn(), agentReadPage: vi.fn() } -})) - -vi.mock('../rpc/methods/structured-sessions/creation/structured-agent-session-gate', () => ({ - requireStructuredAgentAudience: () => undefined, - requireInstalledStructuredHost: async () => host -})) const terminal = vi.hoisted(() => ({ source: vi.fn(), read: vi.fn() @@ -25,143 +14,9 @@ vi.mock('../orchestration/workers/transcripts/worker-transcript-read', () => ({ canReadWorkerTranscript: () => true, readWorkerTranscript: terminal.read })) -vi.mock('../orchestration/addressing/session-lineage/structured-session-lineage', () => ({ - resolveLineageRunningSession: () => ({ kind: 'here', sessionId: SESSION }) -})) -// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the chat read reads only the record's id and agent. -const chat = { - kind: 'chat', - orcaSessionId: SESSION, - record: { sessionId: SESSION, provider: 'codex' }, - store: {} -} as unknown as ResolvedAgentConversation -// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the chat branch passes the context only to the mocked gates. -const ctx = {} as RpcContext - -function page(entries: string[], sequence: number, partial?: AgentReadPartialRow) { - return { - ok: true, - entries: entries.map((text) => ({ kind: 'message', role: 'assistant', text })), - next: { cursor: { epoch: 'e', sequence }, ...(partial ? { partial } : {}) }, - limited: partial !== undefined - } -} - -// Row 5 has three raw slots; the first is read. -const PARTIAL: AgentReadPartialRow = { - row: 5, - count: 3, - fingerprint: 'AAAAAAAAAAAAAAAAAAAAAA', - emitted: 'gA' -} -const legacy = (fields: Record) => - `oar1_${Buffer.from(JSON.stringify({ v: 1, id: SESSION, s: 'chat', session: SESSION, journal: { epoch: 'e', sequence: 4 }, ...fields })).toString('base64url')}` -const current = (fields: Record) => - `oar2_${Buffer.from(JSON.stringify({ v: 2, id: SESSION, s: 'chat', session: SESSION, journal: { epoch: 'e', sequence: 4 }, ...fields })).toString('base64url')}` - -beforeEach(() => { - host.reads.agentReadPage.mockReset() - host.reads.journalCursor.mockReset() -}) - -it.each([20_000, 200_000])( - 'follows an unchanged chat of %i rows without reading its journal', - async (rows) => { - host.reads.journalCursor.mockResolvedValue({ epoch: 'e', sequence: rows }) - host.reads.agentReadPage.mockResolvedValue(page([`reply ${rows}`], rows)) - const first = await readAgentConversation(ctx, chat, { limit: 1 }) - host.reads.agentReadPage.mockClear() - - for (let read = 0; read < 10; read++) { - const again = await readAgentConversation(ctx, chat, { cursor: first.cursor! }) - expect(again).toMatchObject({ entries: [], cursor: first.cursor }) - } - - expect(host.reads.agentReadPage).not.toHaveBeenCalled() - } -) - -it.each([ - ['some', PARTIAL], - ['none', { ...PARTIAL, emitted: 'AA' }] -])('carries a row read in part (%s of it emitted) to the next read', async (_, partial) => { - host.reads.journalCursor.mockResolvedValue({ epoch: 'e', sequence: 4 }) - host.reads.agentReadPage.mockResolvedValueOnce(page(['first of three'], 4, partial)) - const first = await readAgentConversation(ctx, chat, { limit: 1 }) - host.reads.agentReadPage.mockResolvedValueOnce(page(['second of three'], 5)) - - expect(first.cursor?.startsWith('oar2_')).toBe(true) - await readAgentConversation(ctx, chat, { cursor: first.cursor!, limit: 2 }) - - expect(host.reads.journalCursor).not.toHaveBeenCalled() - expect(host.reads.agentReadPage).toHaveBeenLastCalledWith(SESSION, { - limit: 2, - after: { cursor: { epoch: 'e', sequence: 4 }, partial } - }) -}) - -it('reads a version 1 chat cursor at a row boundary, and refuses one inside a row', async () => { - host.reads.journalCursor.mockResolvedValue({ epoch: 'e', sequence: 9 }) - host.reads.agentReadPage.mockResolvedValue(page(['next'], 9)) - - await readAgentConversation(ctx, chat, { cursor: legacy({}) }) - expect(host.reads.agentReadPage).toHaveBeenLastCalledWith(SESSION, { - limit: expect.any(Number), - after: { cursor: { epoch: 'e', sequence: 4 } } - }) - await expect( - readAgentConversation(ctx, chat, { cursor: legacy({ w: 1 }) }) - ).rejects.toMatchObject({ - code: 'agent_read_cursor_invalid' - }) -}) - -it.each([ - ['a bitset longer than its slots', { ...PARTIAL, emitted: 'gAA' }], - ['a set padding bit', { ...PARTIAL, emitted: 'gQ' }], - ['a row at the frontier', { ...PARTIAL, row: 4 }], - ['a negative slot count', { ...PARTIAL, count: -1 }], - ['an unknown fingerprint form', { ...PARTIAL, fingerprint: 'short' }] -])('refuses a chat cursor with %s', async (_, partial) => { - await expect( - readAgentConversation(ctx, chat, { cursor: current({ partial }) }) - ).rejects.toMatchObject({ - code: 'agent_read_cursor_invalid' - }) - expect(host.reads.agentReadPage).not.toHaveBeenCalled() -}) - -it("refuses another agent's cursor and a chat payload under the old prefix", async () => { - const other = current({ id: 'claude_other' }) - await expect(readAgentConversation(ctx, chat, { cursor: other })).rejects.toMatchObject({ - code: 'agent_read_cursor_invalid' - }) - const mislabeled = `oar1_${current({}).slice('oar2_'.length)}` - await expect(readAgentConversation(ctx, chat, { cursor: mislabeled })).rejects.toMatchObject({ - code: 'agent_read_cursor_invalid' - }) -}) - -it('keeps transcript cursors on version 1', () => { - const cursor = { v: 1, id: SESSION, s: 'transcript', i: 'source', p: 42, c: 'check' } as const - - const encoded = encodeReadCursor(cursor) - - expect(encoded.startsWith('oar1_')).toBe(true) - expect(decodeReadCursor(encoded, SESSION)).toEqual(cursor) -}) - -it('takes a cursor ahead of the journal to the reader, not as an empty read', async () => { - host.reads.journalCursor.mockResolvedValue({ epoch: 'e', sequence: 100 }) - host.reads.agentReadPage.mockResolvedValueOnce(page([], 1_100)) - const ahead = await readAgentConversation(ctx, chat, { limit: 1 }) - host.reads.agentReadPage.mockResolvedValueOnce({ ok: false }) - - await expect(readAgentConversation(ctx, chat, { cursor: ahead.cursor! })).rejects.toMatchObject({ - code: 'agent_read_source_changed' - }) -}) +// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the terminal read passes the context only to the mocked hook server. +const ctx = { runtime: { readLocalSleepingAgentSessions: () => ({}) } } as unknown as RpcContext it.each([ [ @@ -187,8 +42,7 @@ it.each([ providerSession: { id: 'p', transcriptPath: '/home/qa/.claude-custom/projects/repo/p.jsonl' - }, - authority: 1 + } }) terminal.read.mockResolvedValue({ ok: false, reason: 'transcript_missing', warnings: [] }) // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the terminal read uses only the id and the mocked hook server and reader. diff --git a/src/main/runtime/agent-conversations/agent-conversation-read.ts b/src/main/runtime/agent-conversations/agent-conversation-read.ts index 52097b85b88..169b92d3e99 100644 --- a/src/main/runtime/agent-conversations/agent-conversation-read.ts +++ b/src/main/runtime/agent-conversations/agent-conversation-read.ts @@ -4,7 +4,6 @@ import { type AgentReadEntry, type AgentReadResult } from '../../../shared/agent-conversations/agent-conversation-contracts' -import type { AgentReadJournalPosition } from '../../native-chat/agent-session-wire/publication/history/agent-read-journal-page' import type { NativeChatMessage } from '../../../shared/native-chat/native-chat-types' import type { RpcContext } from '../rpc/core' import { @@ -16,10 +15,8 @@ import { canReadWorkerTranscript, readWorkerTranscript } from '../orchestration/workers/transcripts/worker-transcript-read' -import { createWorkerOutputSourceIdentity } from '../orchestration/workers/transcripts/worker-output-cursor' import { agentHookServer } from '../../agent-hooks/server/agent-hook-server' import type { TerminalAgentTranscriptSource } from '../../agent-hooks/server/server-agent-identities' -import { sameProviderKey } from '../../agent-hooks/identity/agent-provider-key' import { tuiAgentDisplayName } from '../../../shared/agent-launch/catalog/tui-agent-display-names' import { needsWslHostTranslation } from '../../native-chat/transcript-discovery/host-readable-transcript-path' import { @@ -28,14 +25,8 @@ import { } from '../../agent-hooks/identity/agent-identity-worker-protocol' import type { ResolvedAgentConversation } from './agent-conversation-address' import { AgentConversationRefusal } from './agent-conversation-refusal' -import { - decodeReadCursor, - encodeReadCursor, - invalidCursor, - type ReadCursor -} from './agent-read-cursor' -type ReadRequest = { limit?: number; cursor?: string } +type ReadRequest = { limit?: number } /** One page of what an agent said, by its conversation, never by a pane: a chat's own history, * or a terminal agent's transcript. Reading needs no running agent and never starts one. @@ -50,22 +41,17 @@ export async function readAgentConversation( request: ReadRequest ): Promise { const limit = request.limit ?? AGENT_READ_DEFAULT_LIMIT - const cursor = request.cursor ? decodeReadCursor(request.cursor, resolved.orcaSessionId) : null return resolved.kind === 'chat' - ? readChat(ctx, resolved, limit, cursor) - : readTerminalTranscript(ctx, resolved.orcaSessionId, limit, cursor) + ? readChat(ctx, resolved, limit) + : readTerminalTranscript(ctx, resolved.orcaSessionId, limit) } async function readChat( ctx: RpcContext, resolved: Extract, - limit: number, - cursor: ReadCursor | null + limit: number ): Promise { const { record, store, orcaSessionId } = resolved - if (cursor && cursor.s !== 'chat') { - throw invalidCursor() - } requireStructuredAgentAudience(ctx, record.provider) // The session that continues the conversation now: a /clear root reads its current chat. const running = resolveLineageRunningSession(store, record.sessionId) @@ -76,58 +62,14 @@ async function readChat( ) } const sessionId = running.kind === 'here' ? running.sessionId : record.sessionId - if (cursor && cursor.session !== sessionId) { - throw sourceChanged(orcaSessionId) - } const host = await requireInstalledStructuredHost(ctx, sessionId) - const encode = (next: AgentReadJournalPosition): string => - encodeReadCursor({ - v: 2, - id: orcaSessionId, - s: 'chat', - session: sessionId, - journal: next.cursor, - ...(next.partial ? { partial: next.partial } : {}), - ...(next.floor ? { floor: next.floor } : {}) - }) - const after = cursor - ? { - cursor: cursor.journal, - ...(cursor.partial ? { partial: cursor.partial } : {}), - ...(cursor.floor ? { floor: cursor.floor } : {}) - } - : undefined - // A follower re-reading with its cursor does no journal work until the journal moves; a cursor - // inside a row always reads it. - const head = after && !after.partial ? await host.reads.journalCursor(sessionId) : null - if ( - after && - head && - head.epoch === after.cursor.epoch && - // Only exact equality: a cursor ahead of the journal takes the reader's own reset path. - head.sequence === after.cursor.sequence - ) { - return { - orcaSessionId, - kind: 'chat', - agent: record.provider, - source: 'chat', - entries: [], - cursor: encode(after), - limited: false - } - } - const page = await host.reads.agentReadPage(sessionId, { limit, ...(after ? { after } : {}) }) - if (!page.ok) { - throw sourceChanged(orcaSessionId) - } + const page = await host.reads.agentReadPage(sessionId, { limit }) return { orcaSessionId, kind: 'chat', agent: record.provider, source: 'chat', entries: page.entries, - cursor: encode(page.next), limited: page.limited } } @@ -135,21 +77,12 @@ async function readChat( async function readTerminalTranscript( ctx: RpcContext, orcaSessionId: string, - limit: number, - cursor: ReadCursor | null + limit: number ): Promise { - if (cursor && cursor.s !== 'transcript') { - throw invalidCursor() - } - const locate = ( - known?: TerminalAgentTranscriptSource - ): Promise => - agentHookServer.terminalAgentTranscriptSource( - orcaSessionId, - () => ctx.runtime.readLocalSleepingAgentSessions(), - known + const source: TerminalAgentTranscriptSource | ReverseLookupUnavailable | null | undefined = + await agentHookServer.terminalAgentTranscriptSource(orcaSessionId, () => + ctx.runtime.readLocalSleepingAgentSessions() ) - const source = await locate() if (source === undefined) { throw new AgentConversationRefusal( CODES.unavailable, @@ -188,62 +121,26 @@ async function readTerminalTranscript( transcriptPath: providerSession.transcriptPath, // The key's own distro keeps a WSL agent's guest path on that guest. ...(key.wslDistro ? { wslDistro: key.wslDistro } : {}), - limit, - ...(cursor?.s === 'transcript' - ? { offset: cursor.p, ...(cursor.c ? { expectedBoundaryCheckpoint: cursor.c } : {}) } - : {}) + limit }) if (!transcript.ok) { - if (cursor) { - throw sourceChanged(orcaSessionId) - } throw new AgentConversationRefusal( CODES.noTranscript, `Orca couldn't read agent ${orcaSessionId}'s transcript (${transcript.reason.replaceAll('_', ' ')}).` ) } - const sourceIdentity = createWorkerOutputSourceIdentity([ - 'agent-transcript', - key.executionHost, - key.wslDistro, - key.agent, - key.providerSessionId, - key.locator, - transcript.filePath, - transcript.sourceFingerprint - ]) - // The conversation must still name the file just read: its row may have moved on meanwhile. - const after = await locate(source) - if ( - isReverseLookupUnavailable(after) || - (cursor?.s === 'transcript' && cursor.i !== sourceIdentity) || - !after || - !sameProviderKey(after.key, key) || - after.providerSession.id !== providerSession.id || - after.providerSession.transcriptPath !== providerSession.transcriptPath - ) { - throw sourceChanged(orcaSessionId) - } return { orcaSessionId, kind: 'terminal', agent: key.agent, source: 'transcript', entries: transcriptEntries(transcript.messages), - cursor: encodeReadCursor({ - v: 1, - id: orcaSessionId, - s: 'transcript', - i: sourceIdentity, - p: transcript.nextOffset, - ...(transcript.boundaryCheckpoint ? { c: transcript.boundaryCheckpoint } : {}) - }), limited: transcript.limited } } /** One entry per transcript message, its tool calls folded in, so `limit` messages are at most - * `limit` entries and the cursor after them skips nothing. */ + * `limit` entries. */ function transcriptEntries(messages: readonly NativeChatMessage[]): AgentReadEntry[] { const entries: AgentReadEntry[] = [] for (const message of messages) { @@ -271,10 +168,3 @@ function transcriptEntries(messages: readonly NativeChatMessage[]): AgentReadEnt } return entries } - -function sourceChanged(orcaSessionId: string): AgentConversationRefusal { - return new AgentConversationRefusal( - CODES.sourceChanged, - `What agent ${orcaSessionId}'s cursor pointed into has changed (the chat was cleared or its transcript moved). Read again without --cursor.` - ) -} diff --git a/src/main/runtime/agent-conversations/agent-read-cursor.ts b/src/main/runtime/agent-conversations/agent-read-cursor.ts deleted file mode 100644 index 396a28dd462..00000000000 --- a/src/main/runtime/agent-conversations/agent-read-cursor.ts +++ /dev/null @@ -1,100 +0,0 @@ -import { AGENT_CONVERSATION_REFUSAL_CODES as CODES } from '../../../shared/agent-conversations/agent-conversation-contracts' -import type { AgentJournalCursor } from '../../../shared/agent-session/journal/agent-session-journal-identity' -import { AgentConversationRefusal } from './agent-conversation-refusal' -import { - isAgentReadPartialRow, - type AgentReadPartialRow -} from '../../native-chat/agent-session-wire/publication/history/agent-read-row-progress' -import { isRecord } from '../../../shared/agent-status/child-work/agent-status-child-work-value-guards' - -/** Chat cursors are version 2; transcript cursors stay version 1. */ -const PREFIXES = { 1: 'oar1_', 2: 'oar2_' } as const - -/** Where the next `orca agent read` continues, pinned to one agent and one source. */ -export type ReadCursor = - /** `partial`: the next journal row, when part of it was read. `floor`: the `/clear` divider - * the read's context starts at. */ - | { - v: 2 - id: string - s: 'chat' - session: string - journal: AgentJournalCursor - partial?: AgentReadPartialRow - floor?: number - } - | { v: 1; id: string; s: 'transcript'; i: string; p: number; c?: string } - -export function encodeReadCursor(cursor: ReadCursor): string { - return `${PREFIXES[cursor.v]}${Buffer.from(JSON.stringify(cursor)).toString('base64url')}` -} - -export function decodeReadCursor(value: string, orcaSessionId: string): ReadCursor { - const version = value.startsWith(PREFIXES[2]) ? 2 : value.startsWith(PREFIXES[1]) ? 1 : null - let parsed: unknown - try { - parsed = - version === null - ? null - : JSON.parse( - Buffer.from(value.slice(PREFIXES[version].length), 'base64url').toString('utf8') - ) - } catch { - throw invalidCursor() - } - if (isRecord(parsed) && parsed.v === 1 && version === 1 && parsed.s === 'chat') { - // A version 1 chat cursor at a row boundary still reads; one inside a row (`w`) counted - // positions no current reader can map back. - parsed = parsed.w === undefined ? { ...parsed, v: 2 } : null - } else if (!isRecord(parsed) || parsed.v !== version) { - parsed = null - } - if (!isReadCursor(parsed)) { - throw invalidCursor() - } - if (parsed.id !== orcaSessionId) { - throw new AgentConversationRefusal( - CODES.invalidCursor, - `That cursor is from another agent's read; read agent ${orcaSessionId} without it first.` - ) - } - return parsed -} - -function isReadCursor(cursor: unknown): cursor is ReadCursor { - if (!isRecord(cursor) || typeof cursor.id !== 'string') { - return false - } - if (cursor.v === 2 && cursor.s === 'chat') { - return ( - typeof cursor.session === 'string' && - isRecord(cursor.journal) && - typeof cursor.journal.epoch === 'string' && - typeof cursor.journal.sequence === 'number' && - Number.isSafeInteger(cursor.journal.sequence) && - cursor.journal.sequence >= 0 && - (cursor.partial === undefined || - isAgentReadPartialRow(cursor.partial, cursor.journal.sequence)) && - (cursor.floor === undefined || - (typeof cursor.floor === 'number' && - Number.isSafeInteger(cursor.floor) && - cursor.floor > 0)) - ) - } - return ( - cursor.v === 1 && - cursor.s === 'transcript' && - typeof cursor.i === 'string' && - typeof cursor.p === 'number' && - Number.isSafeInteger(cursor.p) && - cursor.p >= 0 && - (cursor.c === undefined || typeof cursor.c === 'string') - ) -} - -export function invalidCursor(): AgentConversationRefusal { - return new AgentConversationRefusal( - CODES.invalidCursor, - 'That read cursor is not valid. Read again without --cursor.' - ) -} diff --git a/src/shared/agent-conversations/agent-conversation-contracts.ts b/src/shared/agent-conversations/agent-conversation-contracts.ts index aaaac73ec4a..0132b1a118d 100644 --- a/src/shared/agent-conversations/agent-conversation-contracts.ts +++ b/src/shared/agent-conversations/agent-conversation-contracts.ts @@ -56,8 +56,6 @@ export type AgentReadResult = { /** `chat`: the chat's own history; `transcript`: the terminal agent's transcript file. */ source: 'chat' | 'transcript' entries: AgentReadEntry[] - /** Pass back as `cursor` to read only what came after this page; tied to this source. */ - cursor: string | null /** Older entries exist that this page did not include. */ limited: boolean } @@ -72,7 +70,5 @@ export const AGENT_CONVERSATION_REFUSAL_CODES = { terminalHandle: 'agent_terminal_handle', invalidId: 'agent_invalid_id', unavailable: 'agent_unavailable', - noTranscript: 'agent_no_transcript', - sourceChanged: 'agent_read_source_changed', - invalidCursor: 'agent_read_cursor_invalid' + noTranscript: 'agent_no_transcript' } as const diff --git a/src/shared/rpc-contract/agent-conversation-params.ts b/src/shared/rpc-contract/agent-conversation-params.ts index 2b8564bd9d2..fa4f5d20f08 100644 --- a/src/shared/rpc-contract/agent-conversation-params.ts +++ b/src/shared/rpc-contract/agent-conversation-params.ts @@ -15,6 +15,5 @@ export const AgentShowParams = z.object({ export const AgentReadParams = z.object({ id: requiredString('Missing agent ID'), - limit: z.number().int().positive().max(AGENT_READ_MAX_LIMIT).optional(), - cursor: z.string().min(1).max(4096).optional() + limit: z.number().int().positive().max(AGENT_READ_MAX_LIMIT).optional() })