fix(native-chat): harden round-two reliability

This commit is contained in:
Brennan Benson
2026-08-27 23:42:56 -07:00
parent ba0ff0fb29
commit 78fc4f091a
15 changed files with 843 additions and 44 deletions
+4
View File
@@ -93,6 +93,10 @@ const CROSS_VERSION_WIRE_PREFIXES = [
'src/shared/browser-client-host-protocol',
'src/shared/browser-network-tunnel-protocol',
'src/shared/browser-client-host-placement',
'src/shared/agent-session-wire',
'src/shared/agent-session-mutation-envelope',
'src/main/native-chat/agent-session-wire/',
'src/main/runtime/agent-session-record-store',
'src/main/runtime/rpc/dispatcher',
'src/main/runtime/rpc/methods/browser-tab-create-schema',
'src/main/runtime/rpc/methods/terminal',
@@ -181,6 +181,10 @@ describe('per-job path classification', () => {
for (const file of [
'src/shared/protocol-version.ts',
'src/shared/terminal-stream-protocol.ts',
'src/shared/agent-session-wire.ts',
'src/shared/agent-session-mutation-envelope.ts',
'src/main/native-chat/agent-session-wire/structured-agent-session-host.ts',
'src/main/runtime/agent-session-record-store.ts',
'src/main/runtime/rpc/dispatcher.ts',
'src/main/runtime/rpc/methods/browser-tab-create-schema.ts',
'src/main/runtime/rpc/methods/terminal.ts',
@@ -1,4 +1,4 @@
import type { ChildProcess } from 'node:child_process'
import type { ChildProcessHandle } from '../../shared/child-process/run-process'
import { captureDescendantSnapshot, type DescendantSnapshot } from '../pty-descendant-termination'
import { terminateDescendantSnapshotAndWait } from '../pty-descendant-exit-verification'
import { terminateWindowsProcessTree } from '../windows-process-tree-kill'
@@ -8,7 +8,7 @@ const TOKEN_PROCESS_EXIT_TIMEOUT_MS = 3_500
const TOKEN_PROCESS_POLL_MS = 25
const activeTeardowns = new WeakMap<object, Promise<boolean>>()
type TeardownChild = Pick<ChildProcess, 'pid' | 'kill'>
type TeardownChild = Pick<ChildProcessHandle, 'pid' | 'kill'>
export type CodexAppServerProcessTeardownDeps = {
platform?: NodeJS.Platform
@@ -0,0 +1,168 @@
import { mkdtemp, readdir, rm, writeFile } from 'node:fs/promises'
import type * as FsPromises from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { appendJournalRows, JOURNAL_SNAPSHOT_FILE, readJournalSnapshot } from './journal-log-file'
import type { JournalSnapshotFile } from './journal-log-file'
import type { JournalRow } from './journal-row-schema'
import { openAgentSessionJournal } from './journal-store'
type FakeDirectoryHandle = { sync: ReturnType<typeof vi.fn>; close: ReturnType<typeof vi.fn> }
let openDirectoryHook: ((path: unknown, flags: unknown) => FakeDirectoryHandle | undefined) | null =
null
vi.mock('node:fs/promises', async (importOriginal) => {
const actual = await importOriginal<typeof FsPromises>()
return {
...actual,
open: (async (...args: Parameters<typeof actual.open>) => {
const fake = openDirectoryHook?.(args[0], args[1])
return fake ?? actual.open(...args)
}) as typeof actual.open
}
})
let root: string
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'orca-journal-log-file-'))
openDirectoryHook = null
})
afterEach(async () => {
openDirectoryHook = null
await rm(root, { recursive: true, force: true })
})
function validSnapshot(): JournalSnapshotFile {
return {
v: 1,
epoch: 'epoch-A',
compactedThrough: 2,
highestFence: 1,
items: [
{
itemId: 'codex:thread-1:turn-1:1',
revision: 1,
body: { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'hi' }] },
sequence: 2,
observedAt: 1_000
}
],
submissions: [],
receipts: [],
aliases: [],
tombstones: [{ itemId: 'codex:thread-1:turn-1:2', revision: 3 }],
tail: []
}
}
async function writeSnapshot(snapshot: unknown): Promise<void> {
await writeFile(join(root, JOURNAL_SNAPSHOT_FILE), JSON.stringify(snapshot), 'utf-8')
}
describe('readJournalSnapshot validation', () => {
it('accepts a well-formed snapshot, with and without the tombstones collection', async () => {
await writeSnapshot(validSnapshot())
expect((await readJournalSnapshot(root)).status).toBe('valid')
const { tombstones: _tombstones, ...withoutTombstones } = validSnapshot()
await writeSnapshot(withoutTombstones)
expect((await readJournalSnapshot(root)).status).toBe('valid')
})
it('classifies a JSON-valid non-array tombstones collection as invalid instead of valid', async () => {
await writeSnapshot({ ...validSnapshot(), tombstones: {} })
expect((await readJournalSnapshot(root)).status).toBe('invalid')
})
it('rejects tombstone entries that would poison seeding', async () => {
for (const tombstones of [
[{ itemId: 42, revision: 1 }],
[{ itemId: 'codex:thread-1:turn-1:1', revision: 'one' }],
[{ itemId: 'codex:thread-1:turn-1:1', revision: Number.NaN }],
['codex:thread-1:turn-1:1']
]) {
await writeSnapshot({ ...validSnapshot(), tombstones })
expect((await readJournalSnapshot(root)).status).toBe('invalid')
}
})
it('rejects items and counters that only look shallowly plausible', async () => {
const missingSequence = validSnapshot()
missingSequence.items = [
{ itemId: 'codex:thread-1:turn-1:1', revision: 1, body: { kind: 'status', text: 'x' } }
] as unknown as JournalSnapshotFile['items']
await writeSnapshot(missingSequence)
expect((await readJournalSnapshot(root)).status).toBe('invalid')
await writeSnapshot({ ...validSnapshot(), compactedThrough: Number.NaN })
expect((await readJournalSnapshot(root)).status).toBe('invalid')
})
})
describe('journal startup isolation from a malformed snapshot', () => {
it('quarantines a JSON-valid malformed snapshot instead of throwing through open', async () => {
await writeSnapshot({ ...validSnapshot(), tombstones: {} })
const journal = await openAgentSessionJournal({
identity: {
sessionId: 'session-1',
workspaceId: 'ws-1',
hostId: 'host-1',
agent: 'codex',
providerHandle: { kind: 'codex', threadId: 'thread-1' }
},
journalDir: root
})
// Degraded exactly like other corrupt snapshots: quarantined on disk, never
// silently deleted, and the session does not adopt state it cannot trust.
const entries = await readdir(root)
expect(entries.some((entry) => entry.startsWith('quarantine-snapshot-'))).toBe(true)
expect(entries.includes(JOURNAL_SNAPSHOT_FILE)).toBe(false)
expect(journal.snapshot().items).toEqual([])
})
})
describe('appendJournalRows directory fsync', () => {
const ROW: JournalRow = {
kind: 'epoch',
reason: 'session_created',
providerHandle: { kind: 'codex', threadId: 'thread-1' },
v: 1,
epoch: 'epoch-A',
seq: 1,
fence: 0,
ts: 1_000
}
function hookDirectoryOpen(sync: ReturnType<typeof vi.fn>): FakeDirectoryHandle {
const fake: FakeDirectoryHandle = { sync, close: vi.fn(async () => undefined) }
openDirectoryHook = (path, flags) => (path === root && flags === 'r' ? fake : undefined)
return fake
}
it('closes the directory handle when directory fsync fails', async () => {
const fake = hookDirectoryOpen(
vi.fn(async () => {
throw new Error('EINVAL: sync')
})
)
// Tolerating unsupported directory fsync must not turn into a leak.
await expect(appendJournalRows(root, [ROW])).resolves.toBeUndefined()
expect(fake.sync).toHaveBeenCalledTimes(1)
expect(fake.close).toHaveBeenCalledTimes(1)
})
it('closes the directory handle when directory fsync succeeds', async () => {
const fake = hookDirectoryOpen(vi.fn(async () => undefined))
await expect(appendJournalRows(root, [ROW])).resolves.toBeUndefined()
expect(fake.sync).toHaveBeenCalledTimes(1)
expect(fake.close).toHaveBeenCalledTimes(1)
})
})
@@ -8,7 +8,7 @@
// between publishing the snapshot and truncating the log leaves the log a
// superset of the tail, and recovery unions the two by sequence — never a hole.
import { appendFile, mkdir, open, readFile } from 'node:fs/promises'
import { appendFile, mkdir, open, readFile, type FileHandle } from 'node:fs/promises'
import { randomUUID } from 'node:crypto'
import { join } from 'node:path'
import { durableWriteTempPath, renameDurable, writeFileDurable } from '../../durable-file-write'
@@ -156,12 +156,17 @@ function isJournalSnapshotFile(value: unknown): value is JournalSnapshotFile {
typeof snapshot.v === 'number' &&
typeof snapshot.epoch === 'string' &&
snapshot.epoch.length > 0 &&
typeof snapshot.compactedThrough === 'number' &&
typeof snapshot.highestFence === 'number' &&
Number.isInteger(snapshot.compactedThrough) &&
(snapshot.compactedThrough as number) >= 0 &&
Number.isInteger(snapshot.highestFence) &&
arrayOf(snapshot.items, isRenderItem) &&
arrayOf(snapshot.submissions, isSubmission) &&
arrayOf(snapshot.receipts, isReceipt) &&
arrayOf(snapshot.aliases, isAlias) &&
// Older snapshots predate tombstones; absence is fine, a non-array is not —
// seeding iterates this collection, so a JSON-valid wrong shape must land
// in quarantine rather than throw through startup restoration.
(snapshot.tombstones === undefined || arrayOf(snapshot.tombstones, isTombstone)) &&
arrayOf(snapshot.tail, (row) => parseJournalRow(JSON.stringify(row)).ok)
)
}
@@ -181,11 +186,19 @@ function isRenderItem(value: unknown): boolean {
return Boolean(
item &&
typeof item.itemId === 'string' &&
typeof item.revision === 'number' &&
Number.isInteger(item.revision) &&
Number.isInteger(item.sequence) &&
recordOf(item.body)
)
}
function isTombstone(value: unknown): boolean {
const tombstone = recordOf(value)
return Boolean(
tombstone && typeof tombstone.itemId === 'string' && Number.isInteger(tombstone.revision)
)
}
function isSubmission(value: unknown): boolean {
const submission = recordOf(value)
return Boolean(submission && typeof submission.clientMessageId === 'string')
@@ -242,12 +255,16 @@ export async function appendJournalRows(
} finally {
await handle.close()
}
let directory: FileHandle | undefined
try {
const directory = await open(journalDir, 'r')
directory = await open(journalDir, 'r')
await directory.sync()
await directory.close()
} catch {
// Directory fsync is unavailable on some platforms (notably Windows).
} finally {
// The tolerance above must not leak the descriptor when open succeeded
// but sync failed — one leaked handle per append adds up fast.
await directory?.close().catch(() => undefined)
}
}
@@ -1,17 +1,31 @@
import { mkdtemp, rm } from 'node:fs/promises'
import { appendFile, mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
import {
agentJournalSubmissionKey,
boundJournalKeyComponent
} from '../../../shared/agent-session-journal-item-key'
import type {
AgentJournalItemBody,
AgentJournalItemIdentity,
AgentJournalMessageItem,
AgentSessionJournalIdentity
} from '../../../shared/agent-session-journal-types'
import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types'
import { AGENT_SESSION_HISTORY_MAX_LIMIT } from '../../../shared/agent-session-wire'
import { serializeRemoteRuntimePayload } from '../../../shared/remote-runtime-memory-limits'
import {
REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES,
serializeRemoteRuntimePayload
} from '../../../shared/remote-runtime-memory-limits'
import { structuredAgentSessionPayloadFingerprint } from '../../../shared/structured-agent-session-mutation'
import { JOURNAL_LOG_FILE } from '../agent-session-journal/journal-log-file'
import {
serializeJournalRow,
type JournalItemRow,
type JournalRow,
type JournalTombstoneRow
} from '../agent-session-journal/journal-row-schema'
import {
openAgentSessionJournal,
type AgentSessionJournal
@@ -399,3 +413,220 @@ describe('projectJournalBatch', () => {
expect(page.page.removedItemIds).toEqual([])
})
})
/** Serialize through the actual channel gate and hand back the byte length. */
function serializedPageBytes(value: unknown): number {
return Buffer.byteLength(serializeRemoteRuntimePayload(value), 'utf8')
}
type RawSeedRow =
| Omit<JournalItemRow, 'v' | 'epoch' | 'fence' | 'ts'>
| Omit<JournalTombstoneRow, 'v' | 'epoch' | 'fence' | 'ts'>
/** Simulate rows admitted before identity bounding existed: written straight
* into the log, then loaded by a fresh journal instance. */
async function reopenWithRawRows(rows: readonly RawSeedRow[]): Promise<AgentSessionJournal> {
const full = rows.map(
(row) =>
({
...row,
v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION,
epoch: journal.epoch,
fence: 1,
ts: tick()
}) as JournalRow
)
await appendFile(
join(root, JOURNAL_LOG_FILE),
`${full.map(serializeJournalRow).join('\n')}\n`,
'utf-8'
)
return openAgentSessionJournal({ identity: IDENTITY, journalDir: root, now: tick })
}
describe('pre-existing oversized identities', () => {
// The exact escape from the round-two review: a legal 5 MiB Codex turnId
// admitted before bounding, then tombstoned. Its removal id alone exceeds
// the 4 MiB outbound cap, so no page can ever carry it.
const HUGE_ITEM_ID = `codex:thread-1:${'h'.repeat(5 * 1024 * 1024)}:1`
it('answers an unfittable pre-existing removal with a bounded reset instead of an unsendable page', async () => {
const seq = journal.cursor().sequence
const reopened = await reopenWithRawRows([
{ kind: 'item', itemId: HUGE_ITEM_ID, revision: 1, seq: seq + 1, body: body('big') },
{ kind: 'tombstone', itemId: HUGE_ITEM_ID, revision: 2, seq: seq + 2 }
])
for (const sequence of [seq, seq + 1]) {
const result = readAgentSessionHistory(reopened, {
sessionId: 'session-1',
direction: 'after',
cursor: { epoch: reopened.epoch, sequence },
limit: 40
})
expect(result.ok).toBe(false)
if (result.ok) {
throw new Error('expected a reset')
}
expect(result.reset).toBe('cursor_compacted')
expect(serializedPageBytes(result.page)).toBeLessThanOrEqual(
REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES
)
// The reset page replaces client state wholesale, so the removal is
// applied without carrying the id, and resuming from the live cursor
// starts past the unsendable row.
expect(result.page.items).toEqual([])
const liveCursor = result.page.liveCursor
if (!liveCursor) {
throw new Error('expected a live cursor on the reset page')
}
expect(liveCursor.sequence).toBeGreaterThanOrEqual(seq + 2)
const resumed = readAgentSessionHistory(reopened, {
sessionId: 'session-1',
direction: 'after',
cursor: liveCursor,
limit: 40
})
expect(resumed.ok).toBe(true)
}
})
it('charges removal ids into the page budget and splits catch-up instead of overflowing', async () => {
const seq = journal.cursor().sequence
const removalIds = Array.from(
{ length: 30 },
(_, index) => `codex:thread-1:${'r'.repeat(250 * 1024)}:${index}`
)
const reopened = await reopenWithRawRows(
removalIds.map((itemId, index) => ({
kind: 'tombstone' as const,
itemId,
revision: 1,
seq: seq + 1 + index
}))
)
const seen = new Set<string>()
let cursor = { epoch: reopened.epoch, sequence: seq }
let guard = 0
while (true) {
guard += 1
expect(guard).toBeLessThan(30)
const result = readAgentSessionHistory(reopened, {
sessionId: 'session-1',
direction: 'after',
cursor,
limit: 40
})
if (!result.ok) {
throw new Error(`expected a page, got reset ${result.reset}`)
}
expect(serializedPageBytes(result.page)).toBeLessThanOrEqual(
REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES
)
for (const removed of result.page.removedItemIds) {
seen.add(removed)
}
cursor = result.page.window.nextCursor
if (!result.page.hasNewer) {
break
}
}
expect(seen).toEqual(new Set(removalIds))
})
it('bounds the truncation marker id for a live oversized-id item', async () => {
const seq = journal.cursor().sequence
const reopened = await reopenWithRawRows([
{ kind: 'item', itemId: HUGE_ITEM_ID, revision: 1, seq: seq + 1, body: body('big') }
])
const tail = readAgentSessionHistory(reopened, {
sessionId: 'session-1',
direction: 'tail',
limit: 40
})
if (!tail.ok) {
throw new Error(`expected a page, got reset ${tail.reset}`)
}
expect(serializedPageBytes(tail.page)).toBeLessThanOrEqual(
REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES
)
expect(tail.page.items).toHaveLength(1)
const marker = tail.page.items[0]
expect(marker?.body.kind).toBe('status')
expect(marker?.itemId).toBe(boundJournalKeyComponent(HUGE_ITEM_ID))
expect(marker?.itemId.length).toBeLessThan(2048)
const forward = readAgentSessionHistory(reopened, {
sessionId: 'session-1',
direction: 'after',
cursor: { epoch: reopened.epoch, sequence: seq },
limit: 40
})
if (!forward.ok) {
throw new Error(`expected a page, got reset ${forward.reset}`)
}
expect(serializedPageBytes(forward.page)).toBeLessThanOrEqual(
REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES
)
expect(forward.page.items[0]?.itemId).toBe(boundJournalKeyComponent(HUGE_ITEM_ID))
})
})
describe('identity bounding at admission', () => {
it('bounds a new oversized provider identity so its item and removal share one sendable key', async () => {
const oversized: AgentJournalItemIdentity = {
provider: 'codex',
threadId: 'thread-1',
turnId: 'T'.repeat(5 * 1024 * 1024),
ordinal: 1
}
const start = { epoch: journal.epoch, sequence: journal.cursor().sequence }
const appended = await journal.appendItem(oversized, body('bounded'), { fence: 1 })
expect(appended.itemId.length).toBeLessThan(2048)
expect(appended.itemId).toContain('~orca-oversized~')
const beforeTombstone = journal.cursor()
await journal.appendTombstone(oversized, { fence: 1 })
const created = readAgentSessionHistory(journal, {
sessionId: 'session-1',
direction: 'after',
cursor: start,
limit: 40
})
if (!created.ok) {
throw new Error(`expected a page, got reset ${created.reset}`)
}
expect(serializedPageBytes(created.page)).toBeLessThanOrEqual(
REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES
)
expect(created.page.removedItemIds).toEqual([appended.itemId])
const removal = readAgentSessionHistory(journal, {
sessionId: 'session-1',
direction: 'after',
cursor: beforeTombstone,
limit: 40
})
if (!removal.ok) {
throw new Error(`expected a page, got reset ${removal.reset}`)
}
expect(serializedPageBytes(removal.page)).toBeLessThanOrEqual(
REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES
)
expect(removal.page.removedItemIds).toEqual([appended.itemId])
})
it('keeps bounded keys deterministic and a fixed point of re-derivation', () => {
const oversized = 'x'.repeat(2 * 1024 * 1024)
const bounded = boundJournalKeyComponent(oversized)
expect(bounded).toBe(boundJournalKeyComponent(oversized))
expect(boundJournalKeyComponent(bounded)).toBe(bounded)
expect(bounded.length).toBeLessThan(2048)
expect(boundJournalKeyComponent(`${oversized}y`)).not.toBe(bounded)
expect(boundJournalKeyComponent('short')).toBe('short')
})
})
@@ -7,7 +7,10 @@
// read would silently skip that revision. Rows carry the revision, which is why
// `after` is the only direction that can answer `cursor_compacted`.
import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
import {
agentJournalSubmissionKey,
boundJournalKeyComponent
} from '../../../shared/agent-session-journal-item-key'
import type {
AgentJournalCursor,
AgentJournalRenderItem,
@@ -34,6 +37,14 @@ import { projectJournalBatch } from './agent-session-journal-batch'
* client paging. */
export const AGENT_SESSION_HISTORY_MAX_PAGE_BYTES = REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES / 2
/** Reserved for everything the page carries beyond its items and removal ids:
* cursors, session/epoch ids, and the RPC envelope. Charged up front so the
* content budget bounds the COMPLETE serialized result, not just the rows. */
const HISTORY_PAGE_ENVELOPE_RESERVE_BYTES = 64 * 1024
const HISTORY_PAGE_CONTENT_BUDGET_BYTES =
AGENT_SESSION_HISTORY_MAX_PAGE_BYTES - HISTORY_PAGE_ENVELOPE_RESERVE_BYTES
/** Item bytes plus the submission the page would carry alongside it. */
function historyEntryBytes(
item: AgentJournalRenderItem,
@@ -62,6 +73,10 @@ function oversizedHistoryItem(
): AgentJournalRenderItem {
return {
...item,
// A pre-bounding id can exceed the budget by itself; the stand-in must not
// re-inflate the page it exists to bound. Bounding is deterministic, so
// re-reads keep deduplicating on the same key.
itemId: boundJournalKeyComponent(item.itemId),
body: {
kind: 'status',
text: `[Orca: item truncated — ${byteLength} bytes exceeds the history page budget]`
@@ -141,7 +156,7 @@ export function readAgentSessionHistory(
windowed,
'newest',
submissionBytesByItemId(snapshot.submissions),
AGENT_SESSION_HISTORY_MAX_PAGE_BYTES
HISTORY_PAGE_CONTENT_BUDGET_BYTES
)
return {
ok: true,
@@ -175,7 +190,7 @@ function buildHydrationPage(
items,
'newest',
submissionBytesByItemId(snapshot.submissions),
AGENT_SESSION_HISTORY_MAX_PAGE_BYTES
HISTORY_PAGE_CONTENT_BUDGET_BYTES
)
return buildPage({
snapshot,
@@ -219,8 +234,18 @@ function readForward(
return historyReset(snapshot, since.reset)
}
const submissionBytes = submissionBytesByItemId(snapshot.submissions)
const batchBytes = (items: readonly AgentJournalRenderItem[]): number =>
items.reduce((total, item) => total + historyEntryBytes(item, submissionBytes), 0)
// The page cost is EVERYTHING variable it carries: items with their
// submissions AND removal ids — a legal pre-bounding tombstone id can dwarf
// every item on the page.
const pageContentBytes = (
items: readonly AgentJournalRenderItem[],
removedItemIds: readonly string[]
): number =>
items.reduce((total, item) => total + historyEntryBytes(item, submissionBytes), 0) +
removedItemIds.reduce(
(total, itemId) => total + Buffer.byteLength(JSON.stringify(itemId), 'utf8') + 1,
0
)
// Rows replay forward, so the byte bound shrinks the ROW window rather than
// clipping projected items: dropping an item while advancing the cursor past
// the rows that touched it would lose that revision for good.
@@ -236,7 +261,8 @@ function readForward(
}
while (
rows.length > 1 &&
batchBytes(projected.batch.items) > AGENT_SESSION_HISTORY_MAX_PAGE_BYTES
pageContentBytes(projected.batch.items, projected.batch.removedItemIds) >
HISTORY_PAGE_CONTENT_BUDGET_BYTES
) {
rows = rows.slice(0, Math.ceil(rows.length / 2))
const shrunk = projectJournalBatch({
@@ -252,14 +278,23 @@ function readForward(
}
// One row can still touch an over-budget item; degrade it visibly.
const items =
batchBytes(projected.batch.items) > AGENT_SESSION_HISTORY_MAX_PAGE_BYTES
pageContentBytes(projected.batch.items, projected.batch.removedItemIds) >
HISTORY_PAGE_CONTENT_BUDGET_BYTES
? projected.batch.items.map((item) => {
const bytes = historyEntryBytes(item, submissionBytes)
return bytes > AGENT_SESSION_HISTORY_MAX_PAGE_BYTES
return bytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES
? oversizedHistoryItem(item, bytes)
: item
})
: projected.batch.items
if (pageContentBytes(items, projected.batch.removedItemIds) > HISTORY_PAGE_CONTENT_BUDGET_BYTES) {
// A single row's semantic payload — in practice a pre-bounding oversized
// removal id — can never fit any page, and truncating a removal id would
// break the client's keying. A bounded tail replaces the client's state
// wholesale, which applies the removal without carrying the id, and the
// client resumes from the live cursor past this row.
return historyReset(snapshot, 'cursor_compacted')
}
const lastSequence = rows.at(-1)?.seq ?? cursor.sequence
return {
ok: true,
@@ -0,0 +1,43 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
const { restoreOnRestart } = vi.hoisted(() => ({ restoreOnRestart: vi.fn() }))
vi.mock('./structured-agent-session-restart-restore', () => ({
restoreStructuredAgentSessionsOnRestart: restoreOnRestart
}))
import { StructuredAgentSessionReadableRestorer } from './structured-agent-session-readable-restorer'
describe('StructuredAgentSessionReadableRestorer', () => {
beforeEach(() => {
restoreOnRestart.mockReset().mockResolvedValue(undefined)
})
it('passes targeted records to the restore pool in visible-first order', async () => {
const records = ['background-a', 'visible-b', 'visible-a', 'background-b'].map(
(sessionId) => ({ sessionId }) as AgentSessionRecord
)
const restorer = new StructuredAgentSessionReadableRestorer({
store: { listRecords: () => records } as never,
journalRoot: '/tmp/journals',
supportsRecord: () => true,
reconcile: async () => null,
resolveRecovery: async () => undefined,
serialize: async (_sessionId, task) => task(),
hasSession: () => false,
onReadable: () => undefined,
restoreHandoff: async () => undefined
})
await restorer.restore(['visible-a', 'visible-b', 'background-a', 'background-b'])
expect(restoreOnRestart).toHaveBeenCalledOnce()
expect(restoreOnRestart.mock.calls[0][0].records.map((record) => record.sessionId)).toEqual([
'visible-a',
'visible-b',
'background-a',
'background-b'
])
})
})
@@ -30,15 +30,23 @@ export class StructuredAgentSessionReadableRestorer {
}
private async restoreReadableSessions(sessionIds?: readonly string[]): Promise<void> {
const targeted = sessionIds ? new Set(sessionIds) : null
const targetOrder = sessionIds
? new Map(sessionIds.map((sessionId, index) => [sessionId, index]))
: null
const records = this.input.store
.listRecords()
.filter(
(record) =>
this.input.supportsRecord(record) && (!targetOrder || targetOrder.has(record.sessionId))
)
if (targetOrder) {
records.sort(
(left, right) => targetOrder.get(left.sessionId)! - targetOrder.get(right.sessionId)!
)
}
await restoreStructuredAgentSessionsOnRestart({
...this.input,
records: this.input.store
.listRecords()
.filter(
(record) =>
this.input.supportsRecord(record) && (!targeted || targeted.has(record.sessionId))
)
records
})
}
}
@@ -1,11 +1,18 @@
import { mkdtemp, rm } from 'node:fs/promises'
import { appendFile, mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types'
import type {
AgentSessionHandoffStatus,
AgentSessionSubscribeEvent
} from '../../../shared/agent-session-wire'
import {
REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES,
serializeRemoteRuntimePayload
} from '../../../shared/remote-runtime-memory-limits'
import { JOURNAL_LOG_FILE } from '../agent-session-journal/journal-log-file'
import { serializeJournalRow, type JournalRow } from '../agent-session-journal/journal-row-schema'
import { openAgentSessionJournal } from '../agent-session-journal/journal-store'
import { AgentSessionSubscribers } from './structured-agent-session-subscribers'
@@ -102,4 +109,87 @@ describe('AgentSessionSubscribers', () => {
handoff
})
})
it('catches a subscriber up past a pre-existing unsendable removal with a bounded reset', async () => {
const journalDir = join(root, 'oversized-removal-journal')
const seeded = await openAgentSessionJournal({
identity: {
sessionId: SESSION,
workspaceId: 'workspace-1',
hostId: 'local',
agent: 'codex',
providerHandle: { kind: 'codex', threadId: 'thread-1' }
},
journalDir
})
// A row admitted before identity bounding: its removal id alone exceeds
// the outbound cap, so no catch-up batch can ever carry it.
const hugeItemId = `codex:thread-1:${'h'.repeat(5 * 1024 * 1024)}:1`
const resumeCursor = seeded.cursor()
const seq = resumeCursor.sequence
const rows: JournalRow[] = [
{
kind: 'item',
itemId: hugeItemId,
revision: 1,
body: { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'big' }] },
v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION,
epoch: seeded.epoch,
seq: seq + 1,
fence: 1,
ts: 2_000
},
{
kind: 'tombstone',
itemId: hugeItemId,
revision: 2,
v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION,
epoch: seeded.epoch,
seq: seq + 2,
fence: 1,
ts: 2_001
}
]
await appendFile(
join(journalDir, JOURNAL_LOG_FILE),
`${rows.map(serializeJournalRow).join('\n')}\n`,
'utf-8'
)
const journal = await openAgentSessionJournal({
identity: {
sessionId: SESSION,
workspaceId: 'workspace-1',
hostId: 'local',
agent: 'codex',
providerHandle: { kind: 'codex', threadId: 'thread-1' }
},
journalDir
})
const subscribers = new AgentSessionSubscribers()
const events: AgentSessionSubscribeEvent[] = []
subscribers.open({
id: 'subscriber-1',
sessionId: SESSION,
journal,
fence: 1,
cursor: resumeCursor,
emit: (event) => events.push(event)
})
// Every event a remote subscriber receives must fit the outbound channel.
for (const event of events) {
expect(Buffer.byteLength(serializeRemoteRuntimePayload(event), 'utf8')).toBeLessThanOrEqual(
REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES
)
}
const reset = events.find((event) => event.type === 'reset')
expect(reset).toBeDefined()
// Forward progress: the reset advanced the subscriber past the unsendable
// row, so the next publish has nothing stale to re-deliver.
const settled = events.length
subscribers.publish(SESSION, journal)
expect(events.slice(settled).filter((event) => event.type === 'reset')).toEqual([])
})
})
@@ -4,26 +4,26 @@ import {
type StructuredTuiRecoveryClaimCandidate
} from './structured-tui-recovery-claim-match'
// Captured from packaged run 31759132745 while the first restart claim failed.
// Modeled after a packaged restart whose first recovery claim failed.
const PACKAGED_CANDIDATE: StructuredTuiRecoveryClaimCandidate = {
expectedWorkspaceId:
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/brennanbenson/orca/workspaces/orca/unified-gate',
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/alice/orca/workspaces/orca/recovery-fixture',
claimMatches: true,
pty: {
connected: true,
ptyId:
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/brennanbenson/orca/workspaces/orca/unified-gate@@fda34510',
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/alice/orca/workspaces/orca/recovery-fixture@@fda34510',
incarnationId: '4cf23679-8987-487d-a24c-dba3bed1b442',
worktreeId:
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/brennanbenson/orca/workspaces/orca/unified-gate'
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/alice/orca/workspaces/orca/recovery-fixture'
},
owner: {
phase: 'live',
ptyId:
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/brennanbenson/orca/workspaces/orca/unified-gate@@fda34510',
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/alice/orca/workspaces/orca/recovery-fixture@@fda34510',
surface: {
worktreeId:
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/brennanbenson/orca/workspaces/orca/unified-gate',
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/alice/orca/workspaces/orca/recovery-fixture',
tabId: 'ced3bd39-262b-41f3-a446-92ceab4f938c',
leafId: 'd4e9d94d-8ec3-4d0d-8ca3-52730ba61c24'
}
@@ -32,7 +32,7 @@ const PACKAGED_CANDIDATE: StructuredTuiRecoveryClaimCandidate = {
sessionResolved: true,
tabPresent: true,
ptyId:
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/brennanbenson/orca/workspaces/orca/unified-gate@@fda34510',
'4d68c30a-b7eb-4078-a1ca-e44e9fa75024::/Users/alice/orca/workspaces/orca/recovery-fixture@@fda34510',
incarnationId: null
}
}
@@ -0,0 +1,134 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import type { AgentJournalCursor } from '../../../../shared/agent-session-journal-types'
import type {
AgentSessionHistoryPage,
AgentSessionSubscribeEvent
} from '../../../../shared/agent-session-wire'
const mocks = vi.hoisted(() => ({ subscribe: vi.fn() }))
vi.mock('@/runtime/structured-agent-session-client', () => ({
subscribeStructuredAgentSession: mocks.subscribe
}))
import { startStructuredAgentSessionReadTransport } from './structured-agent-session-read-transport'
type SubscribeAttempt = {
closed: PromiseWithResolvers<{ unsubscribe: () => void }>
onClose: () => void
onError: (error: unknown) => void
onEvent: (event: AgentSessionSubscribeEvent) => void
unsubscribe: ReturnType<typeof vi.fn<() => void>>
}
const target = { kind: 'local' } as const
function snapshot(sequence: number): AgentSessionSubscribeEvent {
const cursor: AgentJournalCursor = { epoch: 'epoch-a', sequence }
const page: AgentSessionHistoryPage = {
sessionId: 'session-a',
epoch: cursor.epoch,
direction: 'tail',
items: [],
removedItemIds: [],
submissions: [],
window: { oldest: null, newest: null, nextCursor: cursor },
liveCursor: cursor,
hasOlder: false,
hasNewer: false
}
return { type: 'snapshot', sessionId: 'session-a', page, fence: sequence }
}
async function flushPromises(): Promise<void> {
await Promise.resolve()
await Promise.resolve()
}
describe('structured agent-session read transport generations', () => {
const attempts: SubscribeAttempt[] = []
beforeEach(() => {
attempts.length = 0
vi.clearAllMocks()
mocks.subscribe.mockImplementation((_target, _params, onEvent, onError, onClose) => {
const attempt: SubscribeAttempt = {
closed: Promise.withResolvers<{ unsubscribe: () => void }>(),
onClose,
onError,
onEvent,
unsubscribe: vi.fn<() => void>()
}
attempts.push(attempt)
return attempt.closed.promise
})
})
function start(applyEvent: (event: AgentSessionSubscribeEvent) => void, applyError = vi.fn()) {
return startStructuredAgentSessionReadTransport({
applyEvent,
applyError,
getCursor: () => null,
refreshTail: async () => undefined,
sessionId: 'session-a',
target
})
}
it('ignores opening frames after disposal and a replacement transport starts', async () => {
const applyEvent = vi.fn()
const applyError = vi.fn()
const retired = start(applyEvent, applyError)
await flushPromises()
expect(attempts).toHaveLength(1)
retired.dispose()
const replacement = start(applyEvent, applyError)
await flushPromises()
expect(attempts).toHaveLength(2)
attempts[0].onEvent(snapshot(1))
attempts[0].onError(new Error('retired error'))
attempts[0].onClose()
expect(applyEvent).not.toHaveBeenCalled()
expect(applyError).not.toHaveBeenCalled()
attempts[0].closed.resolve({ unsubscribe: attempts[0].unsubscribe })
attempts[1].closed.resolve({ unsubscribe: attempts[1].unsubscribe })
await flushPromises()
expect(attempts[0].unsubscribe).toHaveBeenCalledOnce()
attempts[1].onEvent(snapshot(2))
expect(applyEvent).toHaveBeenCalledExactlyOnceWith(snapshot(2))
replacement.dispose()
})
it('ignores callbacks from a subscription superseded by reconnect', async () => {
vi.useFakeTimers()
try {
const applyEvent = vi.fn()
const applyError = vi.fn()
const transport = start(applyEvent, applyError)
await flushPromises()
attempts[0].closed.resolve({ unsubscribe: attempts[0].unsubscribe })
await flushPromises()
attempts[0].onClose()
await vi.advanceTimersByTimeAsync(750)
expect(attempts).toHaveLength(2)
attempts[0].onEvent(snapshot(1))
attempts[0].onError(new Error('stale error'))
expect(applyEvent).not.toHaveBeenCalled()
expect(applyError).not.toHaveBeenCalled()
attempts[1].onEvent(snapshot(2))
expect(applyEvent).toHaveBeenCalledExactlyOnceWith(snapshot(2))
attempts[1].closed.resolve({ unsubscribe: attempts[1].unsubscribe })
await flushPromises()
transport.dispose()
} finally {
vi.useRealTimers()
}
})
})
@@ -39,6 +39,7 @@ export function startStructuredAgentSessionReadTransport(args: {
let stopped = false
let connected = false
let opening = false
let generation = 0
let unsubscribe = (): void => {}
let resumeCursor = args.getCursor()
const coalescer = createStructuredAgentSessionEventCoalescer(args.applyEvent)
@@ -46,7 +47,11 @@ export function startStructuredAgentSessionReadTransport(args: {
shouldStop: () => stopped || connected,
reconnect: () => void open()
})
const handleEvent = (event: AgentSessionSubscribeEvent): void => {
const isCurrentGeneration = (candidate: number): boolean => !stopped && candidate === generation
const handleEvent = (event: AgentSessionSubscribeEvent, eventGeneration: number): void => {
if (!isCurrentGeneration(eventGeneration)) {
return
}
if (event.type === 'snapshot' || event.type === 'reset') {
resumeCursor = event.page.liveCursor ?? event.page.window.nextCursor
} else if (
@@ -69,6 +74,8 @@ export function startStructuredAgentSessionReadTransport(args: {
return
}
opening = true
coalescer.flush()
const openGeneration = ++generation
unsubscribe()
unsubscribe = (): void => {}
try {
@@ -76,22 +83,28 @@ export function startStructuredAgentSessionReadTransport(args: {
const handle = await subscribeStructuredAgentSession(
args.target,
{ sessionId: args.sessionId, ...(resumeCursor ? { cursor: resumeCursor } : {}) },
handleEvent,
(event) => handleEvent(event, openGeneration),
(error) => {
if (!isCurrentGeneration(openGeneration)) {
return
}
closedDuringOpen = true
connected = false
args.applyError(String(error))
reconnectScheduler.schedule()
},
() => {
if (!isCurrentGeneration(openGeneration)) {
return
}
closedDuringOpen = true
connected = false
reconnectScheduler.schedule()
}
)
if (stopped || closedDuringOpen) {
if (!isCurrentGeneration(openGeneration) || closedDuringOpen) {
handle.unsubscribe()
if (!stopped) {
if (isCurrentGeneration(openGeneration)) {
reconnectScheduler.schedule()
}
} else {
@@ -99,37 +112,57 @@ export function startStructuredAgentSessionReadTransport(args: {
unsubscribe = handle.unsubscribe
}
} catch (error) {
if (!isCurrentGeneration(openGeneration)) {
return
}
connected = false
args.applyError(String(error))
reconnectScheduler.schedule()
} finally {
opening = false
if (openGeneration === generation) {
opening = false
}
}
}
const refresh = (): void => {
const refreshGeneration = generation
void args
.refreshTail(() => stopped)
.then(() => {
if (!isCurrentGeneration(refreshGeneration)) {
return
}
resumeCursor = args.getCursor()
if (!connected) {
reconnectScheduler.schedule(0)
}
})
.catch((error) => args.applyError(String(error)))
.catch((error) => {
if (isCurrentGeneration(refreshGeneration)) {
args.applyError(String(error))
}
})
}
const initialGeneration = generation
void args
.refreshTail(() => stopped)
.then(() => {
if (!isCurrentGeneration(initialGeneration)) {
return
}
resumeCursor = args.getCursor()
return open()
})
.catch((error) => {
args.applyError(String(error))
reconnectScheduler.schedule()
if (isCurrentGeneration(initialGeneration)) {
args.applyError(String(error))
reconnectScheduler.schedule()
}
})
return {
dispose: () => {
stopped = true
generation += 1
reconnectScheduler.dispose()
coalescer.dispose()
unsubscribe()
+32 -1
View File
@@ -8,8 +8,39 @@ import type { AgentJournalItemIdentity } from './agent-session-journal-types'
const KEY_DELIMITER = ':'
/** Longest raw component a key may embed. Real provider ids are tens of bytes;
* anything larger would push the composed key past wire page budgets, so it
* travels as a stable digest instead of verbatim. */
export const MAX_JOURNAL_KEY_COMPONENT_CHARS = 1024
/**
* Deterministic stand-in for an oversized key component: same input, same
* output, so revisions and tombstones of one identity still share a key, and
* re-deriving from a parsed key is a fixed point (the bounded form is far
* below the cap). The head keeps keys debuggable; length plus two independent
* hashes makes an accidental collision practically impossible. Pure JS because
* clients derive keys too and cannot reach node:crypto.
*/
export function boundJournalKeyComponent(value: string): string {
if (value.length <= MAX_JOURNAL_KEY_COMPONENT_CHARS) {
return value
}
const h1 = fnv1a32(value, 0x811c9dc5).toString(16).padStart(8, '0')
const h2 = fnv1a32(value, 0x0100_0193).toString(16).padStart(8, '0')
return `${value.slice(0, 40)}~orca-oversized~${value.length}~${h1}${h2}`
}
function fnv1a32(value: string, seed: number): number {
let hash = seed >>> 0
for (let index = 0; index < value.length; index += 1) {
hash ^= value.charCodeAt(index)
hash = Math.imul(hash, 0x0100_0193) >>> 0
}
return hash >>> 0
}
function encodePart(value: string | number): string {
return encodeURIComponent(String(value))
return encodeURIComponent(boundJournalKeyComponent(String(value)))
}
/**
@@ -14,6 +14,8 @@
# `detached: true` for a process-group kill, a synchronous taskkill, and an stdio shape other
# than all-pipes (a long-lived `ssh -N` must not be handed a stdin pipe nobody closes).
# Migrating them means teaching run-process.ts those options first.
# codex-app-server-posix-supervisor.ts requires child_process only inside its
# dependency-free POSIX supervisor script, never in Orca's process.
#
# The src/main/git/command-runner/* entries are the seven spawn sites that used
# to live in the single line src/main/git/runner.ts; splitting that file moved
@@ -45,7 +47,6 @@ src/main/codex-accounts/runtime-home-service.ts
src/main/codex-accounts/service.ts
src/main/codex/codex-app-server-client.ts
src/main/codex/codex-app-server-posix-supervisor.ts
src/main/codex/codex-app-server-process-teardown.ts
src/main/codex/codex-app-server-session.ts
src/main/codex/codex-state-db-backfill-recovery.ts
src/main/codex/codex-wsl-hook-install-plan.ts