feat(cli): take agent read --cursor out of this PR

agent read returns the newest page only. Removes the cursor codec, the
forward and partial-row readers, the row-local emitted bitset and its raw
slot extractor, the post-read source re-check, and the cursor refusal
codes. --cursor now gets the normal unknown-flag error. Following a chat
comes in a later PR.
This commit is contained in:
Brennan Benson
2026-11-13 18:01:15 -08:00
parent 08fdd24d82
commit 8ffd9cdd3c
22 changed files with 221 additions and 1731 deletions
+2 -2
View File
@@ -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 <cursor> --json
orca agent read orca_session_id:claude_1234 --json
```
Every agent, chat or terminal, has an Orca session ID (`orca_session_id:<agent>_<uuid>`), 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
+1 -1
View File
@@ -74,7 +74,7 @@ Every agent, chat or terminal, has an Orca session ID (`orca_session_id:<agent>_
- `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 <id> --json` shows where one agent is open (none if closed, two if resumed in two panes).
- `ORCA agent read <id> --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 <id> --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.
+1 -6
View File
@@ -70,11 +70,9 @@ export const AGENT_CONVERSATION_HANDLERS: Record<string, CommandHandler> = {
`--limit can be at most ${AGENT_READ_MAX_LIMIT}.`
)
}
const cursor = getOptionalStringFlag(flags, 'cursor')
const result = await callAgentMethod<AgentReadResult>(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')
}
@@ -14,8 +14,7 @@ const COMMAND_SCOPED_FLAG_HELP: Record<string, Record<string, string>> = {
'agent show': { id: AGENT_ID_FLAG_HELP },
'agent read': {
id: AGENT_ID_FLAG_HELP,
limit: '--limit <n> Entries to return (default 20, at most 100)',
cursor: '--cursor <cursor> Cursor from a previous agent read; returns only what came after'
limit: '--limit <n> Newest entries to return (default 20, at most 100)'
},
'extension reload': {
'extension-id':
+1 -1
View File
@@ -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 <id> --json` shows where one agent is open (none if closed, two if resumed in two panes).",
"- `ORCA agent read <id> --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 <id> --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.",
"",
+4 -4
View File
@@ -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 <id> [--limit <n>] [--cursor <cursor>] [--json]',
allowedFlags: [...GLOBAL_FLAGS, 'id', 'limit', 'cursor'],
usage: 'orca agent read <id> [--limit <n>] [--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 <cursor> --json'
'orca agent read orca_session_id:codex_1234 --json'
]
}
]
@@ -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<Record<string, SleepingAgentSighting>>,
known?: TerminalAgentTranscriptSource
readSleeping: () => Readonly<Record<string, SleepingAgentSighting>>
): Promise<TerminalAgentTranscriptSource | ReverseLookupUnavailable | null | undefined> {
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
}
@@ -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<IdentityTestServer> {
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<string, ReturnType<typeof sleeping>> = {}
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()
}
)
@@ -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<AgentJournalRenderItem> {
@@ -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 []
}
@@ -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 (
@@ -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<typeof renderJournalStatusItem>[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<string, number[]>()
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++
@@ -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
}
@@ -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)
}
)
@@ -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<typeof JournalPublicRead>()
return {
...actual,
renderJournalStatusItem: (...args: Parameters<typeof actual.renderJournalStatusItem>) => {
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<AgentReadJournalPage, { ok: true }> = {
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 } })
}
}
)
@@ -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<AgentReadJournalPage, { ok: true }>[] = []
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<unknown>)[]
) {
let after = from
const pages: Extract<AgentReadJournalPage, { ok: true }>[] = []
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<AgentReadJournalPage, { ok: true }>) {
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
@@ -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() })
}
)
@@ -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<string, unknown>) =>
`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<string, unknown>) =>
`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.
@@ -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<AgentReadResult> {
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<ResolvedAgentConversation, { kind: 'chat' }>,
limit: number,
cursor: ReadCursor | null
limit: number
): Promise<AgentReadResult> {
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<AgentReadResult> {
if (cursor && cursor.s !== 'transcript') {
throw invalidCursor()
}
const locate = (
known?: TerminalAgentTranscriptSource
): Promise<TerminalAgentTranscriptSource | ReverseLookupUnavailable | null | undefined> =>
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.`
)
}
@@ -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.'
)
}
@@ -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
@@ -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()
})