Merge remote-tracking branch 'origin/brennanb2025/native-chat-1d-wire' into brennanb2025/native-chat-structured-base

This commit is contained in:
Brennan Benson
2026-08-09 14:16:37 -07:00
32 changed files with 4860 additions and 4 deletions
+23 -4
View File
@@ -95,7 +95,26 @@ negotiated capabilities differ from the contract. Adding an optional field keeps
green (Rule 1); making a client depend on that field turns the new-client/old-host
pairing red.
The harness covers the terminal stream only. It does **not** cover the session-tab
sync channel, agent-session publications, file or Git RPCs, mobile/E2EE framing, or
the relay transport. A change on those paths still needs its own reasoning against
the three rules above.
`tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts` pairs the
same two builds over the structured `agentSession.*` surface. Because a released build
cannot name a capability string its own source never contains, the old side's advertised
list and registered method names are read from the extracted checkout rather than
hand-written. It covers the three skews that surface can fail on:
- an old client — advertising only what the baseline build defines — is told the whole
surface does not exist and reaches no host method;
- a new client against the old dispatcher gets `method_not_found` on every method, and
can see the absence during negotiation instead of by calling;
- a cursor survives a host restart: the client's fence is refused as stale with the live
one attached, and resuming from the held cursor replays only what it missed.
Run it with:
```bash
pnpm exec vitest run --config config/vitest.config.ts tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts
```
The harness covers the terminal stream and the structured agent-session surface. It does
**not** cover the session-tab sync channel, legacy agent-session publications, file or Git
RPCs, mobile/E2EE framing, or the relay transport. A change on those paths still needs its
own reasoning against the three rules above.
@@ -0,0 +1,246 @@
import { 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 type {
AgentJournalItemBody,
AgentJournalItemIdentity,
AgentSessionJournalIdentity
} from '../../../shared/agent-session-journal-types'
import { AGENT_SESSION_HISTORY_MAX_LIMIT } from '../../../shared/agent-session-wire'
import {
openAgentSessionJournal,
type AgentSessionJournal
} from '../agent-session-journal/journal-store'
import { projectJournalBatch } from './agent-session-journal-batch'
import { readAgentSessionHistory, resolveHistoryLimit } from './agent-session-history-page'
const IDENTITY: AgentSessionJournalIdentity = {
sessionId: 'session-1',
workspaceId: 'ws-1',
hostId: 'host-1',
agent: 'codex',
providerHandle: { kind: 'codex', threadId: 'thread-1' }
}
let root: string
let clock = 1_000
let epochs = 0
let journal: AgentSessionJournal
function tick(): number {
clock += 1
return clock
}
function item(ordinal: number): AgentJournalItemIdentity {
return { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal }
}
function body(text: string): AgentJournalItemBody {
return { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text }] }
}
async function appendItems(count: number): Promise<void> {
for (let ordinal = 1; ordinal <= count; ordinal += 1) {
await journal.appendItem(item(ordinal), body(`item-${ordinal}`), { fence: 1 })
}
}
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'orca-wire-history-'))
clock = 1_000
epochs = 0
journal = await openAgentSessionJournal({
identity: IDENTITY,
journalDir: root,
now: tick,
mintEpoch: () => {
epochs += 1
return `epoch-${epochs}`
}
})
})
afterEach(async () => {
await rm(root, { recursive: true, force: true })
})
describe('resolveHistoryLimit', () => {
it('clamps rather than rejecting so a mid-scroll client keeps paging', () => {
expect(resolveHistoryLimit(undefined)).toBe(40)
expect(resolveHistoryLimit(0)).toBe(1)
expect(resolveHistoryLimit(-5)).toBe(1)
expect(resolveHistoryLimit(10_000)).toBe(AGENT_SESSION_HISTORY_MAX_LIMIT)
expect(resolveHistoryLimit(Number.NaN)).toBe(40)
})
})
describe('readAgentSessionHistory', () => {
it('serves the newest page on tail and pages backward from it', async () => {
await appendItems(5)
const tail = readAgentSessionHistory(journal, {
sessionId: 'session-1',
direction: 'tail',
limit: 2
})
if (!tail.ok) {
throw new Error(`expected a page, got reset ${tail.reset}`)
}
expect(tail.page.items.map((entry) => entry.body)).toEqual([body('item-4'), body('item-5')])
expect(tail.page.hasOlder).toBe(true)
expect(tail.page.hasNewer).toBe(false)
const older = readAgentSessionHistory(journal, {
sessionId: 'session-1',
direction: 'before',
cursor: tail.page.window.nextCursor,
limit: 2
})
if (!older.ok) {
throw new Error(`expected a page, got reset ${older.reset}`)
}
expect(older.page.items.map((entry) => entry.body)).toEqual([body('item-2'), body('item-3')])
expect(older.page.hasNewer).toBe(true)
})
it('catches a live reader up from its cursor and stops at the limit', async () => {
await appendItems(2)
const cursor = journal.cursor()
await appendItems(5)
const page = readAgentSessionHistory(journal, {
sessionId: 'session-1',
direction: 'after',
cursor,
limit: 2
})
if (!page.ok) {
throw new Error(`expected a page, got reset ${page.reset}`)
}
expect(page.page.items).toHaveLength(2)
expect(page.page.hasNewer).toBe(true)
expect(page.page.window.nextCursor.sequence).toBeGreaterThan(cursor.sequence)
})
it('carries tombstones in a forward catch-up page', async () => {
await appendItems(1)
const cursor = journal.cursor()
await journal.appendTombstone(item(1), { fence: 1 })
const page = readAgentSessionHistory(journal, {
sessionId: 'session-1',
direction: 'after',
cursor
})
if (!page.ok) {
throw new Error(`expected a page, got reset ${page.reset}`)
}
expect(page.page.items).toHaveLength(0)
expect(page.page.removedItemIds).toEqual(['codex:thread-1:turn-1:1'])
})
it('reports a forward read with no cursor as cursor_ahead rather than serving the tail', async () => {
await appendItems(1)
expect(
readAgentSessionHistory(journal, { sessionId: 'session-1', direction: 'after' })
).toMatchObject({ ok: false, reset: 'cursor_ahead' })
})
it('resets a cursor from a previous epoch', async () => {
await appendItems(1)
const stale = journal.cursor()
await journal.rollEpoch('legacy_import', 2)
for (const direction of ['before', 'after'] as const) {
expect(
readAgentSessionHistory(journal, { sessionId: 'session-1', direction, cursor: stale })
).toMatchObject({ ok: false, reset: 'epoch_changed' })
}
})
it('resets a cursor ahead of the journal', async () => {
await appendItems(1)
const ahead = { epoch: journal.epoch, sequence: journal.cursor().sequence + 10 }
expect(
readAgentSessionHistory(journal, {
sessionId: 'session-1',
direction: 'before',
cursor: ahead
})
).toMatchObject({ ok: false, reset: 'cursor_ahead' })
})
it('carries the submission for a message on the page', async () => {
await journal.appendSubmission({
clientMessageId: 'msg-1',
payloadFingerprint: 'a'.repeat(64),
body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] },
fence: 1
})
const page = readAgentSessionHistory(journal, { sessionId: 'session-1', direction: 'tail' })
if (!page.ok) {
throw new Error(`expected a page, got reset ${page.reset}`)
}
expect(page.page.items[0]?.itemId).toBe(agentJournalSubmissionKey('msg-1'))
expect(page.page.submissions).toHaveLength(1)
expect(page.page.submissions[0]).toMatchObject({
clientMessageId: 'msg-1',
dispatchState: 'pending'
})
})
})
describe('projectJournalBatch', () => {
it('reports a hole in the row sequence as journal_gap', async () => {
await appendItems(3)
const since = journal.readSince({ epoch: journal.epoch, sequence: 0 })
if (!since.ok) {
throw new Error(`expected rows, got reset ${since.reset}`)
}
const withHole = since.rows.filter((row) => row.seq !== since.rows[1]?.seq)
expect(
projectJournalBatch({ rows: withHole, snapshot: journal.snapshot(), afterSequence: 0 })
).toEqual({ ok: false, reset: 'journal_gap' })
})
it('publishes touched items at their current reduced state, not as a delta', async () => {
await appendItems(1)
const cursor = journal.cursor()
await journal.appendItem(item(1), body('revised'), { fence: 1 })
const since = journal.readSince(cursor)
if (!since.ok) {
throw new Error(`expected rows, got reset ${since.reset}`)
}
const projected = projectJournalBatch({
rows: since.rows,
snapshot: journal.snapshot(),
afterSequence: cursor.sequence
})
if (!projected.ok) {
throw new Error(`expected a batch, got reset ${projected.reset}`)
}
expect(projected.batch.items).toHaveLength(1)
expect(projected.batch.items[0]).toMatchObject({ body: body('revised'), revision: 2 })
expect(projected.batch.cursor).toEqual(journal.cursor())
})
it('lists a tombstoned item as removed', async () => {
await appendItems(1)
const cursor = journal.cursor()
await journal.appendTombstone(item(1), { fence: 1 })
const since = journal.readSince(cursor)
if (!since.ok) {
throw new Error(`expected rows, got reset ${since.reset}`)
}
const projected = projectJournalBatch({
rows: since.rows,
snapshot: journal.snapshot(),
afterSequence: cursor.sequence
})
if (!projected.ok) {
throw new Error(`expected a batch, got reset ${projected.reset}`)
}
expect(projected.batch.removedItemIds).toHaveLength(1)
expect(projected.batch.items).toHaveLength(0)
})
})
@@ -0,0 +1,146 @@
// Paged history over one journal.
//
// `tail` and `before` read the REDUCED timeline, so backward paging keeps
// working after compaction — the folded snapshot still holds every live item.
// `after` is the catch-up direction and must read rows instead: an item created
// early and revised late orders by its creation sequence, so an item-window
// 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 type {
AgentJournalCursor,
AgentJournalRenderItem,
AgentJournalSnapshot
} from '../../../shared/agent-session-journal-types'
import {
AGENT_SESSION_HISTORY_DEFAULT_LIMIT,
AGENT_SESSION_HISTORY_MAX_LIMIT,
type AgentSessionHistoryDirection,
type AgentSessionHistoryPage,
type AgentSessionHistoryRequest,
type AgentSessionHistoryResult
} from '../../../shared/agent-session-wire'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import { projectJournalBatch } from './agent-session-journal-batch'
/** Clamped, never rejected: a client asking for more than the host will serve
* should get a smaller page and keep paging, not an error mid-scroll. */
export function resolveHistoryLimit(limit: number | undefined): number {
if (limit === undefined || !Number.isFinite(limit)) {
return AGENT_SESSION_HISTORY_DEFAULT_LIMIT
}
return Math.min(AGENT_SESSION_HISTORY_MAX_LIMIT, Math.max(1, Math.floor(limit)))
}
export function readAgentSessionHistory(
journal: AgentSessionJournal,
request: AgentSessionHistoryRequest
): AgentSessionHistoryResult {
const snapshot = journal.snapshot()
if (journal.isReadOnly) {
return { ok: false, reset: 'schema_unreadable', snapshot }
}
const limit = resolveHistoryLimit(request.limit)
if (request.direction === 'after') {
return readForward(journal, snapshot, request.cursor, limit)
}
const cursor = request.direction === 'before' ? request.cursor : undefined
if (cursor) {
if (cursor.epoch !== snapshot.cursor.epoch) {
return { ok: false, reset: 'epoch_changed', snapshot }
}
if (cursor.sequence > snapshot.cursor.sequence) {
return { ok: false, reset: 'cursor_ahead', snapshot }
}
}
const older = cursor
? snapshot.items.filter((item) => item.sequence < cursor.sequence)
: snapshot.items
const items = older.slice(Math.max(0, older.length - limit))
return {
ok: true,
page: buildPage({
snapshot,
direction: request.direction,
items,
hasOlder: older.length > items.length,
hasNewer: older.length < snapshot.items.length,
fallbackCursor: cursor ?? { epoch: snapshot.cursor.epoch, sequence: 0 },
nextCursor: items[0]
? { epoch: snapshot.cursor.epoch, sequence: items[0].sequence }
: undefined
})
}
}
function readForward(
journal: AgentSessionJournal,
snapshot: AgentJournalSnapshot,
cursor: AgentJournalCursor | undefined,
limit: number
): AgentSessionHistoryResult {
if (!cursor) {
// Why: forward paging replays rows after a position; without one there is
// nothing to be after, and silently serving the tail would hand the client
// a page it cannot place.
return { ok: false, reset: 'cursor_ahead', snapshot }
}
const since = journal.readSince(cursor)
if (!since.ok) {
return { ok: false, reset: since.reset, snapshot }
}
const rows = since.rows.slice(0, limit)
const projected = projectJournalBatch({ rows, snapshot, afterSequence: cursor.sequence })
if (!projected.ok) {
return { ok: false, reset: projected.reset, snapshot }
}
const lastSequence = rows.at(-1)?.seq ?? cursor.sequence
return {
ok: true,
page: buildPage({
snapshot,
direction: 'after',
items: projected.batch.items,
removedItemIds: projected.batch.removedItemIds,
// Reading after a position means there is something before it.
hasOlder: cursor.sequence > 0,
hasNewer: since.rows.length > rows.length,
fallbackCursor: cursor,
nextCursor: { epoch: cursor.epoch, sequence: lastSequence }
})
}
}
function buildPage(input: {
snapshot: AgentJournalSnapshot
direction: AgentSessionHistoryDirection
items: AgentJournalRenderItem[]
removedItemIds?: string[]
hasOlder: boolean
hasNewer: boolean
fallbackCursor: AgentJournalCursor
nextCursor: AgentJournalCursor | undefined
}): AgentSessionHistoryPage {
const epoch = input.snapshot.cursor.epoch
const pageItemIds = new Set(input.items.map((item) => item.itemId))
const oldest = input.items[0]
const newest = input.items.at(-1)
return {
sessionId: input.snapshot.sessionId,
epoch,
direction: input.direction,
items: input.items,
removedItemIds: input.removedItemIds ?? [],
submissions: input.snapshot.submissions.filter((submission) =>
pageItemIds.has(agentJournalSubmissionKey(submission.clientMessageId))
),
window: {
oldest: oldest ? { epoch, sequence: oldest.sequence } : null,
newest: newest ? { epoch, sequence: newest.sequence } : null,
nextCursor: input.nextCursor ?? input.fallbackCursor
},
hasOlder: input.hasOlder,
hasNewer: input.hasNewer
}
}
@@ -0,0 +1,79 @@
// Rows appended since a cursor, projected into what a subscriber must apply.
//
// The batch carries each touched item at its CURRENT reduced state rather than
// the raw rows: a client that applies the same batch twice converges, and a
// provider echo that was adopted into a submission slot arrives under the slot
// key instead of appearing as a second copy of the user's own message.
import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
import type { AgentJournalSnapshot } from '../../../shared/agent-session-journal-types'
import type { AgentSessionJournalBatch } from '../../../shared/agent-session-wire'
import { findSequenceGap } from '../agent-session-journal/journal-cursor'
import type { JournalRow } from '../agent-session-journal/journal-row-schema'
export type JournalBatchProjection =
| { ok: true; batch: AgentSessionJournalBatch }
/** A missing sequence means the tail lost a row; the subscriber reloads
* rather than rendering a timeline with a hole in it. */
| { ok: false; reset: 'journal_gap' }
export function projectJournalBatch(input: {
rows: readonly JournalRow[]
snapshot: AgentJournalSnapshot
/** Sequence the subscriber has already applied. */
afterSequence: number
}): JournalBatchProjection {
const gap = findSequenceGap(
input.rows.map((row) => row.seq),
input.afterSequence + 1
)
if (gap) {
return { ok: false, reset: 'journal_gap' }
}
const aliases = submissionAliases(input.snapshot)
const touchedItemIds = new Set<string>()
const touchedClientMessageIds = new Set<string>()
for (const row of input.rows) {
if (row.kind === 'item' || row.kind === 'tombstone') {
touchedItemIds.add(aliases.get(row.itemId) ?? row.itemId)
continue
}
if (row.kind === 'submission' || row.kind === 'dispatch') {
touchedClientMessageIds.add(row.clientMessageId)
touchedItemIds.add(agentJournalSubmissionKey(row.clientMessageId))
}
}
const live = new Map(input.snapshot.items.map((item) => [item.itemId, item]))
const items = [...touchedItemIds]
.map((itemId) => live.get(itemId))
.filter((item) => item !== undefined)
.sort((a, b) => a.sequence - b.sequence)
return {
ok: true,
batch: {
cursor: input.snapshot.cursor,
items,
removedItemIds: [...touchedItemIds].filter((itemId) => !live.has(itemId)),
submissions: input.snapshot.submissions.filter((submission) =>
touchedClientMessageIds.has(submission.clientMessageId)
)
}
}
}
/**
* Provider item id → the submission slot that adopted it, rebuilt from the
* snapshot's own accepted submissions. This mirrors the alias the reducer
* writes on an accepted dispatch; deriving it here keeps the projection a pure
* function of published state instead of reaching into reducer internals.
*/
function submissionAliases(snapshot: AgentJournalSnapshot): Map<string, string> {
const aliases = new Map<string, string>()
for (const submission of snapshot.submissions) {
if (submission.dispatchState === 'accepted' && submission.providerItemId) {
aliases.set(submission.providerItemId, agentJournalSubmissionKey(submission.clientMessageId))
}
}
return aliases
}
@@ -0,0 +1,175 @@
// Recovery drives the real journal loader against real on-disk damage: a hole
// punched in the log, and a row stamped with a schema this host cannot read.
import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import type {
AgentJournalItemIdentity,
AgentSessionJournalIdentity
} from '../../../shared/agent-session-journal-types'
import { openAgentSessionJournal } from '../agent-session-journal/journal-store'
import {
openAgentSessionJournalWithRecovery,
providerHistoryId,
recoveryJournalDir
} from './agent-session-journal-recovery'
const CODEX_SESSION = '019fd532-7c11-7a90-b6de-4e1a2c3d5f60'
const IDENTITY: AgentSessionJournalIdentity = {
sessionId: CODEX_SESSION,
workspaceId: 'ws-1',
hostId: 'host-1',
agent: 'codex',
providerHandle: { kind: 'codex', threadId: CODEX_SESSION }
}
const CODEX_LINES = [
{
type: 'session_meta',
timestamp: '2026-08-05T10:00:00.000Z',
payload: {
id: CODEX_SESSION,
session_id: CODEX_SESSION,
cwd: '/Users/dev/project',
originator: 'codex_cli_rs',
cli_version: '0.146.1'
}
},
{
type: 'event_msg',
timestamp: '2026-08-05T10:00:02.000Z',
payload: { type: 'user_message', message: 'add a retry', kind: 'plain' }
},
{
type: 'event_msg',
timestamp: '2026-08-05T10:00:05.000Z',
payload: { type: 'agent_message', message: 'On it.' }
}
]
let root: string
let journalDir: string
let historyFilePath: string
function item(ordinal: number): AgentJournalItemIdentity {
return { provider: 'codex', threadId: CODEX_SESSION, turnId: 'turn-1', ordinal }
}
/** Fills a journal with `count` items and hands back the raw log lines. */
async function seedJournal(count: number): Promise<string[]> {
const journal = await openAgentSessionJournal({ identity: IDENTITY, journalDir })
for (let ordinal = 1; ordinal <= count; ordinal += 1) {
await journal.appendItem(
item(ordinal),
{ kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: `item-${ordinal}` }] },
{ fence: 1 }
)
}
const raw = await readFile(join(journalDir, 'log.jsonl'), 'utf-8')
return raw.split('\n').filter((line) => line.trim().length > 0)
}
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'orca-wire-recovery-'))
journalDir = join(root, 'journal')
historyFilePath = join(root, 'rollout.jsonl')
await writeFile(
historyFilePath,
`${CODEX_LINES.map((line) => JSON.stringify(line)).join('\n')}\n`,
'utf-8'
)
})
afterEach(async () => {
await rm(root, { recursive: true, force: true })
})
describe('providerHistoryId', () => {
it('uses the provider handle, never the Orca session id', () => {
expect(providerHistoryId({ kind: 'codex', threadId: 'thread-9' })).toBe('thread-9')
expect(providerHistoryId({ kind: 'claude', sessionId: 'sess-9', leafUuid: null })).toBe(
'sess-9'
)
})
})
describe('openAgentSessionJournalWithRecovery', () => {
it('opens a healthy journal untouched', async () => {
await seedJournal(2)
const opened = await openAgentSessionJournalWithRecovery({
identity: IDENTITY,
journalDir,
fence: 1,
historyFilePath
})
expect(opened.recovery).toBeNull()
expect(opened.journal.snapshot().items).toHaveLength(2)
})
it('rebuilds a holed journal in place on a fresh epoch', async () => {
const lines = await seedJournal(3)
const holed = lines.filter((_line, index) => index !== 1)
await writeFile(join(journalDir, 'log.jsonl'), `${holed.join('\n')}\n`, 'utf-8')
const opened = await openAgentSessionJournalWithRecovery({
identity: IDENTITY,
journalDir,
fence: 1,
historyFilePath
})
expect(opened.recovery).toMatchObject({ trigger: 'journal_corrupt', reset: 'epoch_changed' })
expect(opened.recovery?.imported).toBeGreaterThan(0)
expect(opened.journal.isReadOnly).toBe(false)
// The rebuilt timeline is the only content of its epoch — nothing from the
// damaged prefix survives into it.
const texts = opened.journal.snapshot().items.map((entry) => JSON.stringify(entry.body))
expect(texts.some((text) => text.includes('item-1'))).toBe(false)
expect(texts.some((text) => text.includes('add a retry'))).toBe(true)
})
it('reconstructs a future-schema journal into a schema-scoped sibling, never in place', async () => {
const lines = await seedJournal(1)
await writeFile(
join(journalDir, 'log.jsonl'),
`${lines.join('\n')}\n${JSON.stringify({ v: 99, seq: 2, epoch: 'e', kind: 'item' })}\n`,
'utf-8'
)
const opened = await openAgentSessionJournalWithRecovery({
identity: IDENTITY,
journalDir,
fence: 1,
historyFilePath
})
expect(opened.recovery).toMatchObject({
trigger: 'schema_unreadable',
reset: 'schema_unreadable'
})
expect(opened.recovery?.imported).toBeGreaterThan(0)
// The unreadable journal is left exactly as found; a newer host still owns it.
const untouched = await readFile(join(journalDir, 'log.jsonl'), 'utf-8')
expect(untouched).toContain('"v":99')
const sibling = await readFile(join(recoveryJournalDir(journalDir), 'log.jsonl'), 'utf-8')
expect(sibling).toContain('add a retry')
})
it('still opens the session when provider history cannot be read', async () => {
const lines = await seedJournal(3)
const holed = lines.filter((_line, index) => index !== 1)
await writeFile(join(journalDir, 'log.jsonl'), `${holed.join('\n')}\n`, 'utf-8')
const opened = await openAgentSessionJournalWithRecovery({
identity: IDENTITY,
journalDir,
fence: 1,
historyFilePath: join(root, 'missing.jsonl')
})
expect(opened.recovery).toMatchObject({ trigger: 'journal_corrupt', imported: 0 })
expect(opened.recovery?.error).toBeTruthy()
expect(opened.journal.snapshot().items).toHaveLength(0)
})
})
@@ -0,0 +1,116 @@
// Journal recovery: rehydrate the timeline from provider history.
//
// Two triggers, and they need different destinations. A journal whose prefix is
// unusable is writable, so it is rebuilt in place on a fresh epoch. A journal
// written by a NEWER schema is not writable by this host at all — rebuilding it
// in place would fork the sequence space a newer host still owns — so the
// reconstruction goes to a schema-scoped sibling directory that is only ever
// written by hosts at this version and is never merged back.
import type { AgentType } from '../../../shared/agent-status-types'
import {
AGENT_SESSION_JOURNAL_SCHEMA_VERSION,
type AgentJournalResetReason,
type AgentSessionJournalIdentity,
type AgentSessionProviderHandle
} from '../../../shared/agent-session-journal-types'
import { importLegacyTranscriptIntoJournal } from '../agent-session-journal/journal-legacy-import'
import { loadJournal } from '../agent-session-journal/journal-open'
import {
openAgentSessionJournal,
type AgentSessionJournal
} from '../agent-session-journal/journal-store'
export type AgentSessionJournalRecovery = {
trigger: 'journal_corrupt' | 'schema_unreadable'
/** What subscribers are told; both force a clean snapshot reload. */
reset: AgentJournalResetReason
epoch: string
imported: number
/** Set when provider history could not be read. The session still opens, on
* an empty fresh epoch — an unreadable transcript is not a reason to refuse
* the user their session. */
error?: string
}
export type AgentSessionJournalOpened = {
journal: AgentSessionJournal
recovery: AgentSessionJournalRecovery | null
}
/** Where a reconstruction lands when the real journal cannot be written. */
export function recoveryJournalDir(journalDir: string): string {
return `${journalDir}-recovered-v${AGENT_SESSION_JOURNAL_SCHEMA_VERSION}`
}
/** The provider's own session id, which is what the transcript readers index
* by — never the Orca session id. */
export function providerHistoryId(handle: AgentSessionProviderHandle): string {
if (handle.kind === 'codex') {
return handle.threadId
}
return handle.kind === 'claude' ? handle.sessionId : handle.value
}
export async function openAgentSessionJournalWithRecovery(input: {
identity: AgentSessionJournalIdentity
journalDir: string
fence: number
/** Resolve directly to a transcript instead of discovering it by session id. */
historyFilePath?: string | null
}): Promise<AgentSessionJournalOpened> {
const probe = await loadJournal(input.journalDir, input.identity.sessionId)
if (probe?.readOnly) {
const journal = await openAgentSessionJournal({
identity: input.identity,
journalDir: recoveryJournalDir(input.journalDir)
})
return {
journal,
recovery: await rehydrate({ ...input, journal, trigger: 'schema_unreadable' })
}
}
const journal = await openAgentSessionJournal({
identity: input.identity,
journalDir: input.journalDir
})
if (!probe?.corrupt) {
return { journal, recovery: null }
}
// `open()` already rolled the epoch off the unusable prefix; the import rolls
// once more so the rebuilt timeline is the only content of its epoch.
return { journal, recovery: await rehydrate({ ...input, journal, trigger: 'journal_corrupt' }) }
}
async function rehydrate(input: {
identity: AgentSessionJournalIdentity
journal: AgentSessionJournal
fence: number
historyFilePath?: string | null
trigger: AgentSessionJournalRecovery['trigger']
}): Promise<AgentSessionJournalRecovery> {
const reset: AgentJournalResetReason =
input.trigger === 'schema_unreadable' ? 'schema_unreadable' : 'epoch_changed'
const result = await importLegacyTranscriptIntoJournal({
journal: input.journal,
agent: input.identity.agent satisfies AgentType,
sessionId: providerHistoryId(input.identity.providerHandle),
fence: input.fence,
...(input.historyFilePath ? { options: { filePath: input.historyFilePath } } : {})
})
if (!result.ok) {
return {
trigger: input.trigger,
reset,
epoch: input.journal.epoch,
imported: 0,
error: result.error
}
}
return {
trigger: input.trigger,
reset,
epoch: result.epoch,
imported: result.imported
}
}
@@ -0,0 +1,66 @@
// What the wire needs from a provider adapter.
//
// Phase 2 implements this over the Codex app-server and the Claude Agent SDK;
// nothing here starts, resumes, or talks to a process. The wire owns the
// journal and the lease, so an adapter only has to answer "did the provider
// take this?" — and it answers `unknown` rather than guessing, because the
// journal renders that as delivery unconfirmed instead of as failure.
import type {
AgentJournalItemIdentity,
AgentJournalMessageItem,
AgentSessionJournalIdentity
} from '../../../shared/agent-session-journal-types'
import type { AgentSessionProviderHandleLink } from '../../../shared/agent-session-provider-handle'
import type { AgentSessionProcessIdentity } from '../../../shared/agent-session-record'
/** What a reservation turns into once something is actually running under it:
* the process the host can probe, and the provider handle it was minted with. */
export type AgentSessionAcquisition = {
process: AgentSessionProcessIdentity
link: AgentSessionProviderHandleLink
}
export type AgentSessionDispatchOutcome =
/** The provider owns the turn now, under this identity. */
| { state: 'accepted'; providerIdentity: AgentJournalItemIdentity }
| { state: 'rejected'; reason: string }
/** The call did not settle. Never re-send on the user's behalf. */
| { state: 'unknown'; reason: string }
export type StructuredAgentSessionAdapter = {
/** Makes the reservation real. Called once per reservation, with the spawn
* token the lease was reserved under and the fence the handle must be minted
* at — the store rejects a link minted at any other fence. */
acquire(input: {
identity: AgentSessionJournalIdentity
fence: number
spawnToken: string
}): Promise<AgentSessionAcquisition>
dispatch(input: {
sessionId: string
clientMessageId: string
body: AgentJournalMessageItem
fence: number
}): Promise<AgentSessionDispatchOutcome>
/** Cancels one turn, not the session: a session-wide interrupt would also kill
* a turn the client never asked to stop. */
cancelTurn(input: {
sessionId: string
turnId: string
fence: number
}): Promise<{ cancelled: boolean }>
/** Fires the provider callback for an approval or a question. The wire calls
* this only after the durable compare-and-set won, so it runs exactly once. */
answerPrompt(input: {
sessionId: string
itemId: string
kind: 'approval' | 'question'
optionId: string
fence: number
}): Promise<void>
setOption(input: { sessionId: string; key: string; value: string; fence: number }): Promise<void>
/** Transcript path for journal recovery. Omit to let the existing session-file
* resolver discover it from the provider session id. */
historyFilePath?(input: { identity: AgentSessionJournalIdentity }): Promise<string | null>
}
@@ -0,0 +1,138 @@
// The attach transition end to end: reserve the lease, make the reservation
// real, open the journal.
//
// Split out of the host so the sequence reads in one place. The host still owns
// the decisions that must not be client-supplied — the spawn token, the claim
// key, the owner probe — and passes them in.
import { isDeepStrictEqual } from 'node:util'
import type {
AgentSessionAttachResult,
AgentSessionMutationResult
} from '../../../shared/agent-session-wire'
import { agentSessionLeaseAdmitsWriter } from '../../../shared/agent-session-lease-adjudication'
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import {
admitAttachOrRefuse,
attachJournal,
classifyStoreFailure,
journalIdentityFor,
reserveRequestFor,
type AgentSessionAttachAuthority,
type AgentSessionAttachParams,
type AttachedJournal
} from './structured-agent-session-attach'
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
export type AttachFlowInput = {
store: AgentSessionRecordStore
adapter: StructuredAgentSessionAdapter
journalRoot: string
authority: AgentSessionAttachAuthority
callerKey: string
params: AgentSessionAttachParams
now: () => number
/** Registers the opened journal and fans out to subscribers before the caller
* sees the result, so no client can send against a session the host has not
* finished publishing. */
onAttached: (attached: AttachedJournal) => void
}
export async function performAttach(
input: AttachFlowInput
): Promise<AgentSessionMutationResult<AgentSessionAttachResult>> {
const { params, store } = input
const sessionId = params.envelope.sessionId
const admitted = admitAttachOrRefuse(params)
if (!admitted.ok) {
return admitted
}
let record: AgentSessionRecord
let replayed = false
try {
const reserved = await store.reserveOwner(
reserveRequestFor({
sessionId,
params,
authority: input.authority,
callerKey: input.callerKey,
fingerprint: admitted.fingerprint,
now: input.now()
})
)
record = reserved.record
replayed = reserved.disposition === 'replayed'
if (!agentSessionLeaseAdmitsWriter(record.lease)) {
record = await acquireOwner(input, record)
}
} catch (error) {
return {
ok: false,
refusal: classifyStoreFailure(error, store.getRecord(sessionId)?.lease.runtimeFence ?? null)
}
}
const attached = await attachJournal({
record,
params,
journalRoot: input.journalRoot,
adapter: input.adapter
})
input.onAttached(attached)
await store.recordOperationOutcome({
callerKey: input.callerKey,
operationId: params.envelope.clientOperationId,
outcome: { status: 'succeeded', sessionId }
})
const fence = record.lease.runtimeFence
return {
ok: true,
replayed,
fence,
cursor: attached.journal.cursor(),
value: {
sessionId,
fence,
snapshot: attached.journal.snapshot(),
unconfirmedClientMessageIds: attached.unconfirmedClientMessageIds
}
}
}
/** A reservation with no process behind it is only a promise to spawn; the
* adapter makes it real and the store then grants the writer. */
async function acquireOwner(
input: AttachFlowInput,
record: AgentSessionRecord
): Promise<AgentSessionRecord> {
const fence = record.lease.runtimeFence
const spawnToken = record.lease.reservedSpawnToken
if (!spawnToken) {
throw new Error('agent_session_ownership_unknown')
}
const acquired = await input.adapter.acquire({
identity: journalIdentityFor(record.sessionId, input.params),
fence,
// Retries must recover the original reservation, not mint a second child.
spawnToken
})
if (record.lease.ownerProcess === null) {
await input.store.commitProcessIdentity({
sessionId: record.sessionId,
fence,
process: acquired.process,
now: input.now()
})
} else if (!isDeepStrictEqual(record.lease.ownerProcess, acquired.process)) {
throw new Error('agent_session_ownership_unknown')
}
return input.store.proveOwner({
sessionId: record.sessionId,
fence,
link: acquired.link,
now: input.now()
})
}
@@ -0,0 +1,197 @@
// Attach: reserve the session record, then open its journal.
//
// `create` and `ensure` are the same transition with a different starting
// point — a null expected fence means "no session exists yet". Both go through
// the record store's compare-and-swap, which also owns the idempotency row, so
// a retried attach replays instead of reserving a second owner.
import type { AgentType } from '../../../shared/agent-status-types'
import type {
AgentSessionJournalIdentity,
AgentSessionProviderHandle
} from '../../../shared/agent-session-journal-types'
import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication'
import type { AgentSessionHandleProvider } from '../../../shared/agent-session-provider-handle'
import type {
AgentSessionAccountHome,
AgentSessionExecutionLocation,
AgentSessionOwnerRuntimeKind,
AgentSessionRecord
} from '../../../shared/agent-session-record'
import {
AGENT_SESSION_WIRE_REFUSAL_CODES,
type AgentSessionMutationEnvelope,
type AgentSessionWireRefusal,
type AgentSessionWireRefusalCode
} from '../../../shared/agent-session-wire'
import {
agentSessionFingerprintConflict,
computeAgentSessionPayloadFingerprint
} from '../../../shared/agent-session-mutation-envelope'
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import { journalDirectoryFor } from '../agent-session-journal/journal-paths'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import {
openAgentSessionJournalWithRecovery,
type AgentSessionJournalRecovery
} from './agent-session-journal-recovery'
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
/**
* Everything a client may declare about the session it wants. Deliberately no
* spawn token, claim key, or owner probe: those are host observations, and a
* client that could assert "the previous owner is dead" could steal a live
* session. The host fills them in.
*/
export type AgentSessionAttachParams = {
envelope: AgentSessionMutationEnvelope
location: AgentSessionExecutionLocation
provider: AgentSessionHandleProvider
agent: AgentType
accountHome: AgentSessionAccountHome
runtimeKind: AgentSessionOwnerRuntimeKind
providerHandle: Exclude<AgentSessionProviderHandle, { kind: 'opaque' }>
}
/** Host-supplied half of the reservation. */
export type AgentSessionAttachAuthority = {
spawnToken: string
claimKeyId: string
handoffOperationId: string | null
probe: AgentSessionOwnerProbe
}
/** The fields that define WHICH session this call would attach to. Deliberately
* excludes the spawn token and the probe: those differ between a first attempt
* and its retry, and a retry must replay rather than conflict. */
export function attachFingerprintFields(params: AgentSessionAttachParams): Record<string, unknown> {
return {
location: params.location,
provider: params.provider,
agent: params.agent,
accountHome: params.accountHome,
runtimeKind: params.runtimeKind,
providerHandle: params.providerHandle,
expectedRuntimeFence: params.envelope.expectedRuntimeFence
}
}
/** Recomputes the fingerprint the client declared and refuses a mismatch before
* anything reaches the store. */
export function admitAttachOrRefuse(
params: AgentSessionAttachParams
): { ok: true; fingerprint: string } | { ok: false; refusal: AgentSessionWireRefusal } {
if (params.providerHandle.kind !== params.provider) {
return {
ok: false,
refusal: {
code: 'agent_session_operation_invalid',
message: `A ${params.provider} session requires a ${params.provider} provider handle.`
}
}
}
const fingerprint = computeAgentSessionPayloadFingerprint({
method: 'agentSession.attach',
sessionId: params.envelope.sessionId,
fields: attachFingerprintFields(params)
})
const conflict = agentSessionFingerprintConflict(params.envelope, fingerprint)
return conflict ? { ok: false, refusal: conflict } : { ok: true, fingerprint }
}
export function journalIdentityFor(
sessionId: string,
params: AgentSessionAttachParams
): AgentSessionJournalIdentity {
return {
sessionId,
workspaceId: params.location.workspaceId,
hostId: params.location.executionHostId,
agent: params.agent,
providerHandle: params.providerHandle
}
}
export type AttachedJournal = {
journal: AgentSessionJournal
recovery: AgentSessionJournalRecovery | null
/** Submissions the crash boundary settled as `unknown` on this open. */
unconfirmedClientMessageIds: string[]
}
/**
* Open the session's journal, recovering it when the stored one is unusable,
* and settle every submission left in flight by a previous process. Orca never
* re-sends those; they surface as delivery unconfirmed.
*/
export async function attachJournal(input: {
record: AgentSessionRecord
params: AgentSessionAttachParams
journalRoot: string
adapter: StructuredAgentSessionAdapter
}): Promise<AttachedJournal> {
const identity = journalIdentityFor(input.record.sessionId, input.params)
const fence = input.record.lease.runtimeFence
const historyFilePath = input.adapter.historyFilePath
? await input.adapter.historyFilePath({ identity })
: null
const opened = await openAgentSessionJournalWithRecovery({
identity,
journalDir: journalDirectoryFor(input.journalRoot, {
workspaceId: identity.workspaceId,
sessionId: identity.sessionId
}),
fence,
historyFilePath
})
return {
...opened,
unconfirmedClientMessageIds: await opened.journal.markPendingSubmissionsUnknown(fence)
}
}
export function reserveRequestFor(input: {
sessionId: string
params: AgentSessionAttachParams
authority: AgentSessionAttachAuthority
callerKey: string
fingerprint: string
now: number
}): Parameters<AgentSessionRecordStore['reserveOwner']>[0] {
const { params, authority } = input
return {
sessionId: input.sessionId,
location: params.location,
provider: params.provider,
accountHome: params.accountHome,
runtimeKind: params.runtimeKind,
expectedFence: params.envelope.expectedRuntimeFence,
spawnToken: authority.spawnToken,
claimKeyId: authority.claimKeyId,
handoffOperationId: authority.handoffOperationId,
probe: authority.probe,
operation: {
callerKey: input.callerKey,
operationId: params.envelope.clientOperationId,
fingerprint: input.fingerprint
},
now: input.now
}
}
/** The store signals refusals by throwing the refusal code. Anything not in the
* known set is a defect, not a client error, and is rethrown. */
export function classifyStoreFailure(
error: unknown,
currentFence: number | null
): AgentSessionWireRefusal {
const code = error instanceof Error ? error.message : String(error)
if (!(AGENT_SESSION_WIRE_REFUSAL_CODES as readonly string[]).includes(code)) {
throw error
}
return {
code: code as AgentSessionWireRefusalCode,
message: `The session store refused this call: ${code}.`,
...(code === 'agent_session_checkpoint_stale' && currentFence !== null ? { currentFence } : {})
}
}
@@ -0,0 +1,805 @@
// The host against a real record store and a real journal, with only the
// provider adapter stubbed — admission, idempotency, and the journal writes are
// exactly the ones that ship.
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest'
import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication'
import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope'
import type {
AgentSessionExecutionLocation,
AgentSessionRecord
} from '../../../shared/agent-session-record'
import type {
AgentSessionMutationEnvelope,
AgentSessionSubscribeEvent
} from '../../../shared/agent-session-wire'
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import { journalDirectoryFor } from '../agent-session-journal/journal-paths'
import { openAgentSessionJournal } from '../agent-session-journal/journal-store'
import type {
AgentSessionDispatchOutcome,
StructuredAgentSessionAdapter
} from './structured-agent-session-adapter'
import { attachFingerprintFields } from './structured-agent-session-attach'
import type { AgentSessionAttachParams } from './structured-agent-session-attach'
import { StructuredAgentSessionHost } from './structured-agent-session-host'
const NOW = 1_800_000_000_000
const SESSION = 'session-alpha'
const THREAD = '019fd532-7c11-7a90-b6de-4e1a2c3d5f60'
const LOCATION: AgentSessionExecutionLocation = {
executionHostId: 'local',
wslDistro: null,
workspaceId: 'workspace-1',
workspaceKind: 'git-worktree'
}
const CALLER = { callerKey: 'client-1' }
let operations = 0
/** `<13-digit ms>-<32 hex>`, the only shape the durable ledger accepts. */
function operationId(): string {
operations += 1
return `${NOW}-${operations.toString(16).padStart(32, '0')}`
}
function message(text: string): AgentJournalMessageItem {
return { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] }
}
function envelope(
method: string,
fields: Record<string, unknown>,
overrides: Partial<AgentSessionMutationEnvelope> = {}
): AgentSessionMutationEnvelope {
return {
sessionId: SESSION,
clientOperationId: operationId(),
// A mutation names the fence it believes it is writing under; only attach
// may leave it null.
expectedRuntimeFence: store.getRecord(SESSION)?.lease.runtimeFence ?? 1,
payloadFingerprint: computeAgentSessionPayloadFingerprint({
method,
sessionId: SESSION,
fields
}),
...overrides
}
}
function attachParams(overrides: Partial<AgentSessionAttachParams> = {}): AgentSessionAttachParams {
const params: AgentSessionAttachParams = {
envelope: {
sessionId: SESSION,
clientOperationId: operationId(),
expectedRuntimeFence: null,
payloadFingerprint: '0'.repeat(64)
},
location: LOCATION,
provider: 'codex',
agent: 'codex',
accountHome: { variable: 'CODEX_HOME', path: '/home/dev/.codex' },
runtimeKind: 'native',
providerHandle: { kind: 'codex', threadId: THREAD },
...overrides
}
return {
...params,
envelope: {
...params.envelope,
payloadFingerprint: computeAgentSessionPayloadFingerprint({
method: 'agentSession.attach',
sessionId: params.envelope.sessionId,
fields: attachFingerprintFields(params)
})
}
}
}
/** `ensure`: the same transition as create, from a fence the client already holds. */
function ensureParams(fence: number): AgentSessionAttachParams {
return attachParams({
envelope: {
sessionId: SESSION,
clientOperationId: operationId(),
expectedRuntimeFence: fence,
payloadFingerprint: '0'.repeat(64)
}
})
}
let root: string
let store: AgentSessionRecordStore
let host: StructuredAgentSessionHost
let dispatch: Mock<StructuredAgentSessionAdapter['dispatch']>
let cancelTurn: Mock<StructuredAgentSessionAdapter['cancelTurn']>
let answerPrompt: Mock<StructuredAgentSessionAdapter['answerPrompt']>
let setOption: Mock<StructuredAgentSessionAdapter['setOption']>
let ordinal = 0
function accepted(): AgentSessionDispatchOutcome {
ordinal += 1
return {
state: 'accepted',
providerIdentity: { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal }
}
}
function adapter(): StructuredAgentSessionAdapter {
return {
acquire: async ({ fence }) => ({
process: {
hostId: 'local',
pid: 4242,
processStartTimeMs: 1_700_000_000_000,
spawnToken: store.getRecord(SESSION)?.lease.reservedSpawnToken ?? 'spawn-a'
},
link: {
linkId: `link-${fence}`,
handle: { provider: 'codex', threadId: THREAD },
// A restarted host re-proves the thread it inherited; only the first
// owner of a session may claim to have created it.
origin: store.getRecord(SESSION)?.providerHandleChain.length ? 'resumed' : 'created',
mintedAtFence: fence,
observedAt: NOW
}
}),
dispatch,
cancelTurn,
answerPrompt,
setOption
}
}
async function attach(): Promise<AgentSessionRecord | null> {
const result = await host.attach(CALLER, attachParams())
expect(result.ok).toBe(true)
return store.getRecord(SESSION)
}
/** Puts a pending approval in the journal BEFORE attach, which is the only way
* 1d can stage one: the adapter that would emit it is phase 2's. */
async function seedApproval(optionId = 'allow'): Promise<{ itemId: string; revision: number }> {
const identity = { provider: 'codex' as const, threadId: THREAD, turnId: 'turn-1', ordinal: 99 }
const journalDir = journalDirectoryFor(root, { workspaceId: 'workspace-1', sessionId: SESSION })
const journal = await openAgentSessionJournal({
identity: {
sessionId: SESSION,
workspaceId: 'workspace-1',
hostId: 'local',
agent: 'codex',
providerHandle: { kind: 'codex', threadId: THREAD }
},
journalDir
})
const appended = await journal.appendItem(
identity,
{
kind: 'approval',
title: 'Run the command?',
detail: null,
options: [{ id: optionId, label: 'Allow' }],
resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null }
},
{ fence: 1 }
)
return { itemId: appended.itemId, revision: appended.revision }
}
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'orca-wire-host-'))
operations = 0
ordinal = 0
dispatch = vi.fn(async () => accepted())
cancelTurn = vi.fn(async () => ({ cancelled: true }))
answerPrompt = vi.fn(async () => undefined)
setOption = vi.fn(async () => undefined)
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
host = new StructuredAgentSessionHost({
store,
adapter: adapter(),
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken: () => 'spawn-a',
now: () => NOW
})
})
afterEach(async () => {
await rm(root, { recursive: true, force: true })
})
describe('attach', () => {
it('reserves the lease, spawns through the adapter, and opens the journal', async () => {
const result = await host.attach(CALLER, attachParams())
expect(result).toMatchObject({ ok: true, replayed: false })
expect(host.hasSession(SESSION)).toBe(true)
const record = store.getRecord(SESSION)
expect(record?.lease.ownerProcess?.pid).toBe(4242)
expect(record?.lease.handoffStage).toBeNull()
})
it('refuses a payload the client fingerprinted wrong', async () => {
const params = attachParams()
const result = await host.attach(CALLER, {
...params,
envelope: { ...params.envelope, payloadFingerprint: 'a'.repeat(64) }
})
expect(result).toMatchObject({
ok: false,
refusal: { code: 'agent_session_operation_conflict' }
})
})
it('refuses a provider handle that belongs to a different provider', async () => {
const params = attachParams({
providerHandle: { kind: 'claude', sessionId: 'claude-session', leafUuid: null }
})
expect(await host.attach(CALLER, params)).toMatchObject({
ok: false,
refusal: { code: 'agent_session_operation_invalid' }
})
expect(store.getRecord(SESSION)).toBeNull()
})
it('refuses a second create against a live session', async () => {
await attach()
expect(await host.attach(CALLER, attachParams())).toMatchObject({ ok: false })
})
it('replays a retried attach instead of reserving a second owner', async () => {
const params = attachParams()
await host.attach(CALLER, params)
const retry = await host.attach(CALLER, params)
expect(retry).toMatchObject({ ok: true, replayed: true })
})
it('reuses the persisted spawn token when acquisition is retried', async () => {
const spawnTokens: string[] = []
const acquire = vi
.fn<StructuredAgentSessionAdapter['acquire']>()
.mockImplementationOnce(async ({ spawnToken }) => {
spawnTokens.push(spawnToken)
throw new Error('reply lost')
})
.mockImplementation(async ({ fence, spawnToken }) => {
spawnTokens.push(spawnToken)
return {
process: {
hostId: 'local',
pid: 4242,
processStartTimeMs: 1_700_000_000_000,
spawnToken
},
link: {
linkId: `link-${fence}`,
handle: { provider: 'codex', threadId: THREAD },
origin: 'created',
mintedAtFence: fence,
observedAt: NOW
}
}
})
let token = 0
host = new StructuredAgentSessionHost({
store,
adapter: { ...adapter(), acquire },
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken: () => `spawn-${++token}`,
now: () => NOW
})
const params = attachParams()
await expect(host.attach(CALLER, params)).rejects.toThrow('reply lost')
expect(await host.attach(CALLER, params)).toMatchObject({ ok: true, replayed: true })
expect(spawnTokens).toEqual(['spawn-1', 'spawn-1'])
})
it('finishes proof when a retry finds the process identity already committed', async () => {
const acquire = vi
.fn<StructuredAgentSessionAdapter['acquire']>()
.mockImplementationOnce(async ({ fence, spawnToken }) => ({
process: {
hostId: 'local',
pid: 4242,
processStartTimeMs: 1_700_000_000_000,
spawnToken
},
link: {
linkId: 'stale-link',
handle: { provider: 'codex', threadId: THREAD },
origin: 'created',
mintedAtFence: fence + 1,
observedAt: NOW
}
}))
.mockImplementation(async ({ fence, spawnToken }) => ({
process: {
hostId: 'local',
pid: 4242,
processStartTimeMs: 1_700_000_000_000,
spawnToken
},
link: {
linkId: `link-${fence}`,
handle: { provider: 'codex', threadId: THREAD },
origin: 'created',
mintedAtFence: fence,
observedAt: NOW
}
}))
host = new StructuredAgentSessionHost({
store,
adapter: { ...adapter(), acquire },
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken: () => 'spawn-a',
now: () => NOW
})
const params = attachParams()
await expect(host.attach(CALLER, params)).rejects.toThrow(
'agent_session_provider_handle_stale_fence'
)
expect(await host.attach(CALLER, params)).toMatchObject({ ok: true, replayed: true })
expect(acquire).toHaveBeenCalledTimes(2)
expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('live')
})
})
describe('send', () => {
it('writes the submission before dispatching and resolves it accepted', async () => {
await attach()
const body = message('add a retry')
const result = await host.send(CALLER, {
envelope: envelope('agentSession.send', { body }),
body
})
if (!result.ok) {
throw new Error(`expected a send, got ${result.refusal.code}`)
}
expect(result.value.submission.dispatchState).toBe('accepted')
expect(dispatch).toHaveBeenCalledTimes(1)
const page = host.history({ sessionId: SESSION, direction: 'tail' })
expect(page.ok && page.page.items).toHaveLength(1)
})
it('settles a thrown dispatch as unknown, never as a rejection', async () => {
await attach()
dispatch.mockRejectedValueOnce(new Error('socket closed'))
const body = message('add a retry')
const result = await host.send(CALLER, {
envelope: envelope('agentSession.send', { body }),
body
})
expect(result).toMatchObject({ ok: true, value: { submission: { dispatchState: 'unknown' } } })
})
it('replays a retried send from the journal without dispatching twice', async () => {
await attach()
const body = message('add a retry')
const params = { envelope: envelope('agentSession.send', { body }), body }
await host.send(CALLER, params)
const retry = await host.send(CALLER, params)
expect(retry).toMatchObject({ ok: true, replayed: true })
expect(dispatch).toHaveBeenCalledTimes(1)
})
it('refuses a stale fence and hands back the current one', async () => {
const record = await attach()
const body = message('add a retry')
const result = await host.send(CALLER, {
envelope: envelope(
'agentSession.send',
{ body },
{ expectedRuntimeFence: (record?.lease.runtimeFence ?? 1) + 5 }
),
body
})
expect(result).toMatchObject({
ok: false,
refusal: { code: 'agent_session_checkpoint_stale', currentFence: record?.lease.runtimeFence }
})
})
it('does not let a refused call leave a ledger row that replays past the fence', async () => {
const record = await attach()
const body = message('add a retry')
const params = {
envelope: envelope(
'agentSession.send',
{ body },
{ expectedRuntimeFence: (record?.lease.runtimeFence ?? 1) + 5 }
),
body
}
await host.send(CALLER, params)
expect(await host.send(CALLER, params)).toMatchObject({
ok: false,
refusal: { code: 'agent_session_checkpoint_stale' }
})
expect(dispatch).not.toHaveBeenCalled()
})
it('refuses any mutation against a session this host has not attached', async () => {
const body = message('add a retry')
expect(
await host.send(CALLER, { envelope: envelope('agentSession.send', { body }), body })
).toMatchObject({ ok: false, refusal: { code: 'agent_session_ownership_unknown' } })
})
})
describe('cancel', () => {
it('records the outcome as a status item keyed by the operation id', async () => {
await attach()
const result = await host.cancel(CALLER, {
envelope: envelope('agentSession.cancel', { turnId: 'turn-1' }),
turnId: 'turn-1'
})
expect(result).toMatchObject({ ok: true, value: { cancelled: true } })
const page = host.history({ sessionId: SESSION, direction: 'tail' })
expect(page.ok && page.page.items[0]?.body).toMatchObject({ kind: 'status' })
})
it('reports an unconfirmed cancellation rather than failing the call', async () => {
await attach()
cancelTurn.mockRejectedValueOnce(new Error('no answer'))
const result = await host.cancel(CALLER, {
envelope: envelope('agentSession.cancel', { turnId: 'turn-1' }),
turnId: 'turn-1'
})
expect(result).toMatchObject({ ok: true, value: { cancelled: false } })
})
it('never interrupts twice on a replay', async () => {
await attach()
const params = {
envelope: envelope('agentSession.cancel', { turnId: 'turn-1' }),
turnId: 'turn-1'
}
await host.cancel(CALLER, params)
expect(await host.cancel(CALLER, params)).toMatchObject({
ok: true,
replayed: true,
value: { cancelled: false }
})
expect(cancelTurn).toHaveBeenCalledTimes(1)
})
})
describe('respondToPrompt', () => {
it('commits the answer before the provider callback', async () => {
const prompt = await seedApproval()
await attach()
const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' }
const result = await host.respondToPrompt(CALLER, {
envelope: envelope('agentSession.respondTo:approval', fields),
kind: 'approval',
...fields
})
expect(result).toMatchObject({
ok: true,
value: { resolution: { state: 'resolved', selectedOptionId: 'allow' } }
})
expect(answerPrompt).toHaveBeenCalledTimes(1)
})
it('refuses a second answer to one prompt and says which answer won', async () => {
const prompt = await seedApproval()
await attach()
const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' }
await host.respondToPrompt(CALLER, {
envelope: envelope('agentSession.respondTo:approval', fields),
kind: 'approval',
...fields
})
const loser = await host.respondToPrompt(
{ callerKey: 'client-2' },
{
envelope: envelope('agentSession.respondTo:approval', fields),
kind: 'approval',
...fields
}
)
expect(loser).toMatchObject({
ok: false,
refusal: {
code: 'agent_session_item_revision_stale',
resolution: { selectedOptionId: 'allow' }
}
})
expect(answerPrompt).toHaveBeenCalledTimes(1)
})
it('refuses an option the prompt does not offer', async () => {
const prompt = await seedApproval()
await attach()
const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'deny' }
expect(
await host.respondToPrompt(CALLER, {
envelope: envelope('agentSession.respondTo:approval', fields),
kind: 'approval',
...fields
})
).toMatchObject({ ok: false, refusal: { code: 'agent_session_operation_invalid' } })
expect(answerPrompt).not.toHaveBeenCalled()
})
it("does not turn a recorded refusal into another client's successful answer", async () => {
const prompt = await seedApproval()
await attach()
const rejectedFields = {
itemId: prompt.itemId,
expectedRevision: prompt.revision,
optionId: 'deny'
}
const rejected = {
envelope: envelope('agentSession.respondTo:approval', rejectedFields),
kind: 'approval' as const,
...rejectedFields
}
await host.respondToPrompt(CALLER, rejected)
const acceptedFields = { ...rejectedFields, optionId: 'allow' }
await host.respondToPrompt(
{ callerKey: 'client-2' },
{
envelope: envelope('agentSession.respondTo:approval', acceptedFields),
kind: 'approval',
...acceptedFields
}
)
expect(await host.respondToPrompt(CALLER, rejected)).toMatchObject({
ok: false,
refusal: { code: 'agent_session_operation_invalid' }
})
})
it('keeps the answer and reports it undelivered when the provider callback throws', async () => {
const prompt = await seedApproval()
await attach()
answerPrompt.mockRejectedValueOnce(new Error('pipe closed'))
const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' }
const result = await host.respondToPrompt(CALLER, {
envelope: envelope('agentSession.respondTo:approval', fields),
kind: 'approval',
...fields
})
expect(result.ok).toBe(true)
const page = host.history({ sessionId: SESSION, direction: 'tail' })
const statusId = agentJournalItemKey({
provider: 'orca',
clientMessageId: `${prompt.itemId}#delivery`
})
expect(page.ok && page.page.items.some((entry) => entry.itemId === statusId)).toBe(true)
})
})
describe('setOption', () => {
it('goes to the provider and writes nothing to the journal', async () => {
await attach()
const fields = { key: 'model', value: 'gpt-5' }
const result = await host.setOption(CALLER, {
envelope: envelope('agentSession.setOption', fields),
...fields
})
expect(result).toMatchObject({ ok: true, value: fields })
expect(setOption).toHaveBeenCalledTimes(1)
const page = host.history({ sessionId: SESSION, direction: 'tail' })
expect(page.ok && page.page.items).toHaveLength(0)
})
it('does not turn an unknown provider outcome into a successful replay', async () => {
await attach()
setOption.mockRejectedValueOnce(new Error('reply lost'))
const fields = { key: 'model', value: 'gpt-5' }
const params = {
envelope: envelope('agentSession.setOption', fields),
...fields
}
await expect(host.setOption(CALLER, params)).rejects.toThrow('reply lost')
expect(await host.setOption(CALLER, params)).toMatchObject({
ok: false,
refusal: { code: 'agent_session_operation_unknown' }
})
expect(setOption).toHaveBeenCalledTimes(1)
})
})
describe('restart', () => {
/** A restarted process: the same directories, a new store and a new host over
* them. Every lease loads unreconciled, so this is the state that decides
* whether a persisted session is reachable at all. */
async function reboot(
probeOwner: (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>
) {
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
host = new StructuredAgentSessionHost({
store,
adapter: adapter(),
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken: () => 'spawn-b',
probeOwner,
now: () => NOW
})
}
/** The refusal a restarted host owes a client holding the dead generation's
* fence: stale, with the live fence attached so the retry can succeed. */
async function staleFenceFrom(held: number): Promise<number> {
const refused = await host.attach(CALLER, ensureParams(held))
if (refused.ok) {
throw new Error('a fence from the previous host generation was accepted')
}
expect(refused.refusal.code).toBe('agent_session_checkpoint_stale')
const current = refused.refusal.currentFence
expect(current).toBeGreaterThan(held)
return current ?? 0
}
it('adjudicates the leases it loaded before deciding who may write', async () => {
const before = await attach()
const held = before?.lease.runtimeFence ?? 0
await reboot(async () => ({ outcome: 'pid-absent' }))
const reattached = await host.attach(CALLER, ensureParams(await staleFenceFrom(held)))
expect(reattached).toMatchObject({ ok: true })
expect(store.getRecord(SESSION)?.lease.unreconciled).toBe(false)
expect(store.getRecord(SESSION)?.lease.ownerProcess?.pid).toBe(4242)
})
it("keeps a session whose owner cannot be probed out of a live writer's hands", async () => {
await attach()
const held = store.getRecord(SESSION)?.lease.runtimeFence ?? 0
await reboot(async () => ({ outcome: 'indeterminate', reason: 'no probe on this host' }))
// Nothing proved the previous process dead, so the lease routes to manual
// recovery rather than letting a second writer in behind it.
expect(await host.attach(CALLER, ensureParams(held))).toMatchObject({
ok: false,
refusal: { code: 'agent_session_ownership_unknown' }
})
})
it('does not remember a failed adjudication as done', async () => {
const before = await attach()
const held = before?.lease.runtimeFence ?? 0
const probe = vi
.fn<(record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>>()
.mockRejectedValueOnce(new Error('probe exploded'))
.mockResolvedValue({ outcome: 'pid-absent' })
await reboot(probe)
await expect(host.attach(CALLER, ensureParams(held))).rejects.toThrow('probe exploded')
// One unlucky startup must not strand the session for this host's lifetime.
const reattached = await host.attach(CALLER, ensureParams(await staleFenceFrom(held)))
expect(reattached).toMatchObject({ ok: true })
expect(probe).toHaveBeenCalledTimes(2)
})
})
describe('subscribe', () => {
it('opens with a snapshot and then streams cursor-qualified batches', async () => {
await attach()
const events: AgentSessionSubscribeEvent[] = []
const dispose = host.subscribe({
id: 'sub-1',
sessionId: SESSION,
emit: (event) => events.push(event)
})
const body = message('add a retry')
await host.send(CALLER, { envelope: envelope('agentSession.send', { body }), body })
expect(events[0]?.type).toBe('snapshot')
const batches = events.filter((event) => event.type === 'batch')
expect(batches.length).toBeGreaterThan(0)
const last = batches.at(-1)
expect(last?.type === 'batch' && last.batch.cursor.sequence).toBeGreaterThan(0)
dispose()
expect(events.at(-1)?.type).toBe('end')
})
it('resumes from a client cursor with only the rows it missed', async () => {
await attach()
const body = message('add a retry')
const first = await host.send(CALLER, {
envelope: envelope('agentSession.send', { body }),
body
})
if (!first.ok) {
throw new Error(`expected a send, got ${first.refusal.code}`)
}
const events: AgentSessionSubscribeEvent[] = []
host.subscribe({
id: 'sub-2',
sessionId: SESSION,
emit: (event) => events.push(event),
cursor: first.cursor
})
// Nothing new since that cursor: a resume is silent, not a redundant replay.
expect(events).toHaveLength(0)
const second = message('and a timeout')
await host.send(CALLER, {
envelope: envelope('agentSession.send', { body: second }),
body: second
})
expect(events.some((event) => event.type === 'batch')).toBe(true)
expect(events.some((event) => event.type === 'snapshot')).toBe(false)
})
it('drops a failed transport without aborting the mutation or other subscribers', async () => {
await attach()
const events: AgentSessionSubscribeEvent[] = []
host.subscribe({
id: 'dead-sub',
sessionId: SESSION,
emit: () => {
throw new Error('socket closed')
}
})
host.subscribe({
id: 'live-sub',
sessionId: SESSION,
emit: (event) => events.push(event)
})
const body = message('survive subscriber failure')
const result = await host.send(CALLER, {
envelope: envelope('agentSession.send', { body }),
body
})
expect(result).toMatchObject({ ok: true, value: { submission: { dispatchState: 'accepted' } } })
expect(dispatch).toHaveBeenCalledTimes(1)
expect(events.some((event) => event.type === 'batch')).toBe(true)
})
it('resets a subscriber whose epoch is gone', async () => {
await attach()
const events: AgentSessionSubscribeEvent[] = []
host.subscribe({
id: 'sub-3',
sessionId: SESSION,
emit: (event) => events.push(event),
cursor: { epoch: 'epoch-from-a-previous-life', sequence: 3 }
})
expect(events[0]).toMatchObject({ type: 'reset', reset: 'epoch_changed', fence: 1 })
})
it('publishes the replacement fence when the owner generation changes', async () => {
const record = await attach()
const events: AgentSessionSubscribeEvent[] = []
host.subscribe({
id: 'sub-4',
sessionId: SESSION,
emit: (event) => events.push(event)
})
const released = await store.evictProvenDeadOwner({
sessionId: SESSION,
expectedFence: record?.lease.runtimeFence ?? 1,
probe: { outcome: 'pid-absent' },
now: NOW
})
const replacement = await host.attach(CALLER, ensureParams(released.lease.runtimeFence))
if (!replacement.ok) {
throw new Error(`expected replacement owner, got ${replacement.refusal.code}`)
}
expect(events.at(-1)).toMatchObject({ type: 'snapshot', fence: replacement.fence })
})
})
@@ -0,0 +1,357 @@
// The structured agent-session host: one place where the lease, the journal,
// and the provider adapter meet.
//
// Every mutating call takes the same route — recompute the fingerprint, admit
// through the durable operation ledger, check the lease, then act — so no
// method can grow its own admission rules. Calls against one session are
// serialized; the journal's own queue orders writes, but admission has to be
// single-file too or two clients could both read one prompt as pending.
import { randomUUID } from 'node:crypto'
import type {
AgentJournalCursor,
AgentJournalMessageItem
} from '../../../shared/agent-session-journal-types'
import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication'
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import {
admitAgentSessionMutation,
agentSessionFingerprintConflict,
computeAgentSessionPayloadFingerprint
} from '../../../shared/agent-session-mutation-envelope'
import type {
AgentSessionAttachResult,
AgentSessionCancelResult,
AgentSessionHistoryRequest,
AgentSessionHistoryResult,
AgentSessionMutationEnvelope,
AgentSessionMutationResult,
AgentSessionOptionResult,
AgentSessionPromptResult,
AgentSessionSendResult,
AgentSessionWireRefusal
} from '../../../shared/agent-session-wire'
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import { readAgentSessionHistory } from './agent-session-history-page'
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
import type { AgentSessionAttachParams } from './structured-agent-session-attach'
import { performAttach } from './structured-agent-session-attach-flow'
import { createRestartReconciler } from './structured-agent-session-restart-reconcile'
import {
cancelPlan,
promptPlan,
sendPlan,
setOptionPlan,
type MutationPlan
} from './structured-agent-session-mutation-plans'
import { runSettledAgentSessionMutation } from './structured-agent-session-operation-settlement'
import { resolveAgentSessionReplayOutcome } from './structured-agent-session-replay-outcome'
import {
AgentSessionSubscribers,
type AgentSessionSubscriberEmit
} from './structured-agent-session-subscribers'
import type { AgentSessionTurnContext } from './structured-agent-session-turns'
export type StructuredAgentSessionCaller = {
/** Stable per-client identity; scopes the operation ledger and records who
* answered a prompt. */
callerKey: string
}
export type StructuredAgentSessionHostDeps = {
store: AgentSessionRecordStore
adapter: StructuredAgentSessionAdapter
journalRoot: string
/** Key id the host's claims are minted under. */
claimKeyId: string
probeOwner?: (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>
mintSpawnToken?: () => string
now?: () => number
}
type SessionState = {
journal: AgentSessionJournal
params: AgentSessionAttachParams
fence: number
}
function refuse(refusal: AgentSessionWireRefusal): { ok: false; refusal: AgentSessionWireRefusal } {
return { ok: false, refusal }
}
const NO_SESSION: AgentSessionWireRefusal = {
code: 'agent_session_ownership_unknown',
message: 'This host holds no attached session by that id.'
}
export class StructuredAgentSessionHost {
private readonly sessions = new Map<string, SessionState>()
private readonly subscribers = new AgentSessionSubscribers()
private readonly chains = new Map<string, Promise<unknown>>()
private readonly reconcileLeases: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
constructor(private readonly deps: StructuredAgentSessionHostDeps) {
this.reconcileLeases = createRestartReconciler({
store: deps.store,
probe: (record) => this.probeRecord(record),
now: () => this.now()
})
}
private now = (): number => this.deps.now?.() ?? Date.now()
hasSession(sessionId: string): boolean {
return this.sessions.has(sessionId)
}
/** Per-session single file. A rejected task must not poison the chain. */
private serialize<T>(sessionId: string, task: () => Promise<T>): Promise<T> {
const prior = this.chains.get(sessionId) ?? Promise.resolve()
const next = prior.then(task, task)
this.chains.set(
sessionId,
next.catch(() => undefined)
)
return next
}
/**
* `create` and `ensure` in one: a null expected fence means the session must
* not exist yet, any other value is a compare-and-swap against the lease.
*/
attach(
caller: StructuredAgentSessionCaller,
params: AgentSessionAttachParams
): Promise<AgentSessionMutationResult<AgentSessionAttachResult>> {
return this.serialize(params.envelope.sessionId, async () => {
const sessionId = params.envelope.sessionId
const unreconciled = await this.reconcileLeases(sessionId)
if (unreconciled) {
return refuse(unreconciled)
}
return performAttach({
store: this.deps.store,
adapter: this.deps.adapter,
journalRoot: this.deps.journalRoot,
authority: {
spawnToken: this.deps.mintSpawnToken?.() ?? randomUUID(),
claimKeyId: this.deps.claimKeyId,
handoffOperationId: params.envelope.clientOperationId,
probe: await this.probeOwner(sessionId)
},
callerKey: caller.callerKey,
params,
now: () => this.now(),
onAttached: (attached) => {
const fence = this.deps.store.getRecord(sessionId)?.lease.runtimeFence ?? 0
const previousFence = this.sessions.get(sessionId)?.fence
this.sessions.set(sessionId, { journal: attached.journal, params, fence })
if (attached.recovery) {
this.subscribers.reset(sessionId, attached.journal, attached.recovery.reset, fence)
} else if (previousFence !== undefined && previousFence !== fence) {
this.subscribers.snapshot(sessionId, attached.journal, fence)
} else {
this.subscribers.publish(sessionId, attached.journal)
}
}
})
})
}
/**
* Liveness of whatever currently owns the session. Never client-asserted: a
* client that could claim the owner is gone could evict a live one. Without a
* host prober the answer is `indeterminate`, which routes to manual recovery
* rather than a silent takeover.
*/
private probeOwner(sessionId: string): Promise<AgentSessionOwnerProbe> {
const record = this.deps.store.getRecord(sessionId)
if (!record || record.lease.ownerProcess === null) {
return Promise.resolve({ outcome: 'reservation-unused' })
}
return this.probeRecord(record)
}
private probeRecord(record: AgentSessionRecord): Promise<AgentSessionOwnerProbe> {
return (
this.deps.probeOwner?.(record) ??
Promise.resolve({
outcome: 'indeterminate',
reason: 'This host cannot probe structured session owners.'
})
)
}
send(
caller: StructuredAgentSessionCaller,
params: { envelope: AgentSessionMutationEnvelope; body: AgentJournalMessageItem }
): Promise<AgentSessionMutationResult<AgentSessionSendResult>> {
return this.mutate(caller, params.envelope, sendPlan(params))
}
cancel(
caller: StructuredAgentSessionCaller,
params: { envelope: AgentSessionMutationEnvelope; turnId: string }
): Promise<AgentSessionMutationResult<AgentSessionCancelResult>> {
return this.mutate(caller, params.envelope, cancelPlan(params))
}
respondToPrompt(
caller: StructuredAgentSessionCaller,
params: {
envelope: AgentSessionMutationEnvelope
kind: 'approval' | 'question'
itemId: string
expectedRevision: number
optionId: string
}
): Promise<AgentSessionMutationResult<AgentSessionPromptResult>> {
return this.mutate(caller, params.envelope, promptPlan(params))
}
setOption(
caller: StructuredAgentSessionCaller,
params: { envelope: AgentSessionMutationEnvelope; key: string; value: string }
): Promise<AgentSessionMutationResult<AgentSessionOptionResult>> {
return this.mutate(caller, params.envelope, setOptionPlan(params))
}
private mutate<TValue>(
caller: StructuredAgentSessionCaller,
envelope: AgentSessionMutationEnvelope,
plan: MutationPlan<TValue>
): Promise<AgentSessionMutationResult<TValue>> {
return this.serialize(envelope.sessionId, () => this.runMutation(caller, envelope, plan))
}
private async runMutation<TValue>(
caller: StructuredAgentSessionCaller,
envelope: AgentSessionMutationEnvelope,
plan: MutationPlan<TValue>
): Promise<AgentSessionMutationResult<TValue>> {
const session = this.sessions.get(envelope.sessionId)
const record = this.deps.store.getRecord(envelope.sessionId)
if (!session || !record) {
return refuse(NO_SESSION)
}
const hostFingerprint = computeAgentSessionPayloadFingerprint({
method: plan.method,
sessionId: envelope.sessionId,
fields: plan.fields
})
const conflict = agentSessionFingerprintConflict(envelope, hostFingerprint)
if (conflict) {
return refuse(conflict)
}
const admission = admitAgentSessionMutation({
envelope,
hostFingerprint,
ledger: await this.deps.store.admitOperation({
callerKey: caller.callerKey,
operationId: envelope.clientOperationId,
fingerprint: hostFingerprint,
now: this.now()
}),
lease: record.lease
})
if (admission.decision === 'refused') {
return refuse(admission.refusal)
}
const fence = record.lease.runtimeFence
const ctx = this.contextFor(caller, session, envelope.sessionId, fence)
if (admission.decision === 'replay') {
const replay = resolveAgentSessionReplayOutcome({
operationId: envelope.clientOperationId,
outcome: admission.row.outcome,
reconstruct: () => plan.replay(ctx, admission.row.outcome)
})
if (replay.decision === 'refuse') {
return refuse(replay.refusal)
}
if (replay.decision === 'replay') {
return {
ok: true,
replayed: true,
fence,
cursor: ctx.journal.cursor(),
value: replay.value
}
}
// Nothing durable landed, so this id is about to run for the first time.
// A refused call leaves its ledger row behind, and replaying past the
// lease and the fence would let a resend act under an owner that has since
// changed — so a first run pays the full admission price either way.
const rerun = admitAgentSessionMutation({
envelope,
hostFingerprint,
ledger: { decision: 'admit', row: admission.row },
lease: record.lease
})
if (rerun.decision === 'refused') {
return refuse(rerun.refusal)
}
}
const outcome = await runSettledAgentSessionMutation({
store: this.deps.store,
callerKey: caller.callerKey,
envelope,
plan,
context: ctx
})
return outcome.ok
? { ok: true, replayed: false, fence, cursor: ctx.journal.cursor(), value: outcome.value }
: refuse(outcome.refusal)
}
private contextFor(
caller: StructuredAgentSessionCaller,
session: SessionState,
sessionId: string,
fence: number
): AgentSessionTurnContext {
return {
sessionId,
journal: session.journal,
fence,
adapter: this.deps.adapter,
resolvedBy: caller.callerKey,
publish: () => this.subscribers.publish(sessionId, session.journal),
now: () => this.now()
}
}
/**
* Paged read. Throws `agent_session_ownership_unknown` for a session this host
* has not attached — a reset would tell the client to reload something that
* does not exist here.
*/
history(request: AgentSessionHistoryRequest): AgentSessionHistoryResult {
return readAgentSessionHistory(this.requireSession(request.sessionId).journal, request)
}
subscribe(input: {
id: string
sessionId: string
emit: AgentSessionSubscriberEmit
cursor?: AgentJournalCursor
}): () => void {
const session = this.requireSession(input.sessionId)
const fence = this.deps.store.getRecord(input.sessionId)?.lease.runtimeFence ?? 0
return this.subscribers.open({ ...input, journal: session.journal, fence })
}
unsubscribe(sessionId: string, id: string): void {
this.subscribers.close(sessionId, id)
}
private requireSession(sessionId: string): SessionState {
const session = this.sessions.get(sessionId)
if (!session) {
throw new Error(NO_SESSION.code)
}
return session
}
}
@@ -0,0 +1,117 @@
// One plan per mutating method: what it fingerprints, what it does, and how its
// answer is rebuilt on a replay.
//
// The replay half matters more than it looks. The ledger records only that an
// operation happened, so the durable answer has to come back out of the journal.
// A plan that cannot find its effect returns null, and the call runs for real —
// which is exactly right when the crash landed before the journal write.
import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
import type { AgentSessionOperationOutcome } from '../../../shared/agent-session-operation-ledger'
import type {
AgentSessionCancelResult,
AgentSessionMutationEnvelope,
AgentSessionOptionResult,
AgentSessionPromptResult,
AgentSessionSendResult
} from '../../../shared/agent-session-wire'
import {
performCancel,
performPrompt,
performSend,
performSetOption,
type AgentSessionTurnContext,
type TurnOutcome
} from './structured-agent-session-turns'
export type MutationPlan<TValue> = {
method: string
fields: Record<string, unknown>
run: (ctx: AgentSessionTurnContext) => Promise<TurnOutcome<TValue>>
replay: (ctx: AgentSessionTurnContext, outcome: AgentSessionOperationOutcome) => TValue | null
}
export function sendPlan(params: {
envelope: AgentSessionMutationEnvelope
body: AgentJournalMessageItem
}): MutationPlan<AgentSessionSendResult> {
// The operation id IS the client message id: one send, one durable row, one
// key the client reconciles its optimistic bubble against.
const clientMessageId = params.envelope.clientOperationId
return {
method: 'agentSession.send',
fields: { body: params.body },
run: (ctx) =>
performSend(ctx, {
clientMessageId,
payloadFingerprint: params.envelope.payloadFingerprint,
body: params.body
}),
replay: (ctx) => {
const submission = ctx.journal
.submissions()
.find((entry) => entry.clientMessageId === clientMessageId)
return submission ? { clientMessageId, submission } : null
}
}
}
export function cancelPlan(params: {
envelope: AgentSessionMutationEnvelope
turnId: string
}): MutationPlan<AgentSessionCancelResult> {
return {
method: 'agentSession.cancel',
fields: { turnId: params.turnId },
run: (ctx) =>
performCancel(ctx, {
clientOperationId: params.envelope.clientOperationId,
turnId: params.turnId
}),
// Interrupting twice would kill a turn the client never asked to stop, so a
// replay reports the turn as already handled instead.
replay: () => ({ turnId: params.turnId, cancelled: false })
}
}
export function promptPlan(params: {
kind: 'approval' | 'question'
itemId: string
expectedRevision: number
optionId: string
}): MutationPlan<AgentSessionPromptResult> {
return {
method: `agentSession.respondTo:${params.kind}`,
fields: {
itemId: params.itemId,
expectedRevision: params.expectedRevision,
optionId: params.optionId
},
run: (ctx) => performPrompt(ctx, params),
replay: (ctx) => {
const item = ctx.journal.snapshot().items.find((entry) => entry.itemId === params.itemId)
const body = item?.body
if (!item || !body || (body.kind !== 'approval' && body.kind !== 'question')) {
return null
}
return body.resolution.state === 'pending'
? null
: { itemId: item.itemId, revision: item.revision, resolution: body.resolution }
}
}
}
export function setOptionPlan(params: {
key: string
value: string
}): MutationPlan<AgentSessionOptionResult> {
return {
method: 'agentSession.setOption',
fields: { key: params.key, value: params.value },
run: (ctx) => performSetOption(ctx, params),
// A pending row may have crashed before the adapter call. Reapplying the
// same assignment is safe; only a settled success can be answered directly.
replay: (_ctx, outcome) =>
outcome.status === 'succeeded' ? { key: params.key, value: params.value } : null
}
}
@@ -0,0 +1,33 @@
import type { AgentSessionMutationEnvelope } from '../../../shared/agent-session-wire'
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import type { MutationPlan } from './structured-agent-session-mutation-plans'
import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns'
export async function runSettledAgentSessionMutation<TValue>(input: {
store: AgentSessionRecordStore
callerKey: string
envelope: AgentSessionMutationEnvelope
plan: MutationPlan<TValue>
context: AgentSessionTurnContext
}): Promise<TurnOutcome<TValue>> {
const settle = (
outcome: Parameters<AgentSessionRecordStore['recordOperationOutcome']>[0]['outcome']
) =>
input.store.recordOperationOutcome({
callerKey: input.callerKey,
operationId: input.envelope.clientOperationId,
outcome
})
try {
const outcome = await input.plan.run(input.context)
await settle(
outcome.ok
? { status: 'succeeded', sessionId: input.envelope.sessionId }
: { status: 'failed', code: outcome.refusal.code }
)
return outcome
} catch (error) {
await settle({ status: 'unknown' })
throw error
}
}
@@ -0,0 +1,18 @@
// Where the RPC layer finds the host.
//
// The runtime service is already far past its size budget, so structured
// sessions hang off a module-level slot instead of another field on it — the
// same shape the native-chat RPC methods use to reach their own collaborators.
// Tests install a host with a stub adapter and clear it on teardown.
import type { StructuredAgentSessionHost } from './structured-agent-session-host'
let host: StructuredAgentSessionHost | null = null
export function setStructuredAgentSessionHost(next: StructuredAgentSessionHost | null): void {
host = next
}
export function getStructuredAgentSessionHost(): StructuredAgentSessionHost | null {
return host
}
@@ -0,0 +1,53 @@
import type { AgentSessionOperationOutcome } from '../../../shared/agent-session-operation-ledger'
import {
AGENT_SESSION_WIRE_REFUSAL_CODES,
type AgentSessionWireRefusal,
type AgentSessionWireRefusalCode
} from '../../../shared/agent-session-wire'
export type AgentSessionReplayOutcomeDecision<TValue> =
| { decision: 'replay'; value: TValue }
| { decision: 'rerun' }
| { decision: 'refuse'; refusal: AgentSessionWireRefusal }
export function resolveAgentSessionReplayOutcome<TValue>(input: {
operationId: string
outcome: AgentSessionOperationOutcome
reconstruct: () => TValue | null
}): AgentSessionReplayOutcomeDecision<TValue> {
const { operationId, outcome } = input
if (outcome.status === 'failed') {
const code = (AGENT_SESSION_WIRE_REFUSAL_CODES as readonly string[]).includes(outcome.code)
? (outcome.code as AgentSessionWireRefusalCode)
: 'agent_session_operation_invalid'
return {
decision: 'refuse',
refusal: {
code,
message: `Operation ${operationId} was already refused: ${outcome.code}.`
}
}
}
if (outcome.status === 'unknown') {
return {
decision: 'refuse',
refusal: {
code: 'agent_session_operation_unknown',
message: `The outcome of operation ${operationId} is unknown; it was not run again.`
}
}
}
const recorded = input.reconstruct()
if (recorded) {
return { decision: 'replay', value: recorded }
}
return outcome.status === 'succeeded'
? {
decision: 'refuse',
refusal: {
code: 'agent_session_operation_unknown',
message: `Operation ${operationId} succeeded, but its result is no longer reconstructable.`
}
}
: { decision: 'rerun' }
}
@@ -0,0 +1,29 @@
import { describe, expect, it, vi } from 'vitest'
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import { createRestartReconciler } from './structured-agent-session-restart-reconcile'
describe('createRestartReconciler', () => {
it('reruns after an external store refresh introduces unreconciled leases', async () => {
let record = { sessionId: 'session-1', lease: { unreconciled: true } } as AgentSessionRecord
const reconcileOnRestart = vi.fn(async () => {
record = { ...record, lease: { ...record.lease, unreconciled: false } }
return new Map()
})
const store = {
listRecords: () => [record],
getRecord: () => record,
reconcileOnRestart
} as unknown as AgentSessionRecordStore
const reconcile = createRestartReconciler({
store,
probe: async () => ({ outcome: 'pid-absent' }),
now: () => 1
})
expect(await reconcile('session-1')).toBeNull()
record = { ...record, lease: { ...record.lease, unreconciled: true } }
expect(await reconcile('session-1')).toBeNull()
expect(reconcileOnRestart).toHaveBeenCalledTimes(2)
})
})
@@ -0,0 +1,42 @@
// Every lease loads from disk unreconciled: the process that wrote it may still
// be alive, so nothing the store persisted grants a writer until this host has
// adjudicated it. Without this an attach after a restart is refused forever with
// `execution_owner_reconciling`, and the session becomes unreachable.
import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication'
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire'
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import { classifyStoreFailure } from './structured-agent-session-attach'
/** Adjudicates leases loaded by this process or refreshed from another writer.
* Answers with the refusal attach owes its caller, or null once settled. */
export function createRestartReconciler(deps: {
store: AgentSessionRecordStore
probe: (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>
now: () => number
}): (sessionId: string) => Promise<AgentSessionWireRefusal | null> {
let pending: Promise<void> | null = null
return async (sessionId) => {
if (!deps.store.listRecords().some((record) => record.lease.unreconciled)) {
return null
}
if (!pending) {
const run = deps.store
.reconcileOnRestart({ probe: deps.probe, now: deps.now() })
.then(() => undefined)
pending = run.finally(() => {
pending = null
})
}
try {
await pending
return null
} catch (error) {
return classifyStoreFailure(
error,
deps.store.getRecord(sessionId)?.lease.runtimeFence ?? null
)
}
}
}
@@ -0,0 +1,172 @@
// Per-subscriber cursors over one session's journal.
//
// Each subscriber advances independently: a client that connected two epochs
// ago gets a reset while a caught-up one gets a batch from the same publish.
// Nothing raw reaches a subscriber — every event carries reducer output.
import type {
AgentJournalCursor,
AgentJournalResetReason
} from '../../../shared/agent-session-journal-types'
import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import { projectJournalBatch } from './agent-session-journal-batch'
export type AgentSessionSubscriberEmit = (event: AgentSessionSubscribeEvent) => void
type Subscriber = {
id: string
sessionId: string
emit: AgentSessionSubscriberEmit
cursor: AgentJournalCursor
fence: number
}
export class AgentSessionSubscribers {
private readonly bySession = new Map<string, Map<string, Subscriber>>()
/** Opens the stream with a snapshot or, when the client's cursor still
* resolves, with the rows it missed. Returns the disposer. */
open(input: {
id: string
sessionId: string
journal: AgentSessionJournal
fence: number
emit: AgentSessionSubscriberEmit
cursor?: AgentJournalCursor
}): () => void {
const snapshot = input.journal.snapshot()
const subscriber: Subscriber = {
id: input.id,
sessionId: input.sessionId,
emit: input.emit,
cursor: input.cursor ?? { epoch: snapshot.cursor.epoch, sequence: 0 },
fence: input.fence
}
const session = this.bySession.get(input.sessionId) ?? new Map<string, Subscriber>()
session.set(input.id, subscriber)
this.bySession.set(input.sessionId, session)
if (input.cursor) {
this.deliver(subscriber, input.journal)
} else {
this.emit(subscriber, {
type: 'snapshot',
sessionId: input.sessionId,
snapshot,
fence: input.fence
})
subscriber.cursor = snapshot.cursor
}
return () => this.close(input.sessionId, input.id)
}
close(sessionId: string, id: string): void {
const session = this.bySession.get(sessionId)
const subscriber = session?.get(id)
if (!session || !subscriber) {
return
}
this.drop(subscriber)
try {
subscriber.emit({ type: 'end' })
} catch {
// The transport is already gone; teardown must remain idempotent.
}
}
/** Fan out whatever each subscriber has not yet seen. */
publish(sessionId: string, journal: AgentSessionJournal): void {
for (const subscriber of this.subscribers(sessionId)) {
this.deliver(subscriber, journal)
}
}
/** Force every subscriber back to a clean snapshot — recovery, epoch
* rollover, an unreadable schema. */
reset(
sessionId: string,
journal: AgentSessionJournal,
reason: AgentJournalResetReason,
fence: number
): void {
const snapshot = journal.snapshot()
for (const subscriber of this.subscribers(sessionId)) {
this.emit(subscriber, { type: 'reset', sessionId, reset: reason, snapshot, fence })
subscriber.cursor = snapshot.cursor
subscriber.fence = fence
}
}
snapshot(sessionId: string, journal: AgentSessionJournal, fence: number): void {
const snapshot = journal.snapshot()
for (const subscriber of this.subscribers(sessionId)) {
this.emit(subscriber, { type: 'snapshot', sessionId, snapshot, fence })
subscriber.cursor = snapshot.cursor
subscriber.fence = fence
}
}
private subscribers(sessionId: string): Subscriber[] {
return [...(this.bySession.get(sessionId)?.values() ?? [])]
}
private deliver(subscriber: Subscriber, journal: AgentSessionJournal): void {
const since = journal.readSince(subscriber.cursor)
const snapshot = journal.snapshot()
if (!since.ok) {
this.emit(subscriber, {
type: 'reset',
sessionId: subscriber.sessionId,
reset: since.reset,
snapshot,
fence: subscriber.fence
})
subscriber.cursor = snapshot.cursor
return
}
if (since.rows.length === 0) {
return
}
const projected = projectJournalBatch({
rows: since.rows,
snapshot,
afterSequence: subscriber.cursor.sequence
})
if (!projected.ok) {
this.emit(subscriber, {
type: 'reset',
sessionId: subscriber.sessionId,
reset: projected.reset,
snapshot,
fence: subscriber.fence
})
subscriber.cursor = snapshot.cursor
return
}
this.emit(subscriber, {
type: 'batch',
sessionId: subscriber.sessionId,
batch: projected.batch
})
subscriber.cursor = projected.batch.cursor
}
/** A dead transport cannot be allowed to turn a durable mutation into an
* unknown outcome or poison every later publication. */
private emit(subscriber: Subscriber, event: AgentSessionSubscribeEvent): void {
try {
subscriber.emit(event)
} catch {
this.drop(subscriber)
}
}
private drop(subscriber: Subscriber): void {
const session = this.bySession.get(subscriber.sessionId)
session?.delete(subscriber.id)
if (session?.size === 0) {
this.bySession.delete(subscriber.sessionId)
}
}
}
@@ -0,0 +1,246 @@
// The effects behind send / cancel / respond / setOption.
//
// Admission (lease, fence, idempotency) has already passed by the time anything
// here runs; these functions own only the journal writes and the adapter call,
// in that order. Journal first is deliberate: a crash between the two leaves a
// row the next attach settles as `unknown`, whereas the reverse would lose a
// turn the provider already accepted.
import type {
AgentJournalItemBody,
AgentJournalMessageItem,
AgentJournalResolution
} from '../../../shared/agent-session-journal-types'
import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
import type {
AgentSessionCancelResult,
AgentSessionOptionResult,
AgentSessionPromptResult,
AgentSessionSendResult,
AgentSessionWireRefusal
} from '../../../shared/agent-session-wire'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import type {
AgentSessionDispatchOutcome,
StructuredAgentSessionAdapter
} from './structured-agent-session-adapter'
export type AgentSessionTurnContext = {
sessionId: string
journal: AgentSessionJournal
fence: number
adapter: StructuredAgentSessionAdapter
/** Opaque client identity recorded as the resolver of a prompt. */
resolvedBy: string
publish: () => void
now: () => number
}
export type TurnOutcome<TValue> =
| { ok: true; value: TValue }
| { ok: false; refusal: AgentSessionWireRefusal }
function invalid(message: string): { ok: false; refusal: AgentSessionWireRefusal } {
return { ok: false, refusal: { code: 'agent_session_operation_invalid', message } }
}
/** A thrown adapter error is indistinguishable from a lost reply, so it settles
* as `unknown` rather than as a rejection. */
async function dispatchSafely(
ctx: AgentSessionTurnContext,
clientMessageId: string,
body: AgentJournalMessageItem
): Promise<AgentSessionDispatchOutcome> {
try {
return await ctx.adapter.dispatch({
sessionId: ctx.sessionId,
clientMessageId,
body,
fence: ctx.fence
})
} catch (error) {
return { state: 'unknown', reason: error instanceof Error ? error.message : String(error) }
}
}
async function appendStatus(
ctx: AgentSessionTurnContext,
clientMessageId: string,
text: string
): Promise<void> {
await ctx.journal.appendItem(
{ provider: 'orca', clientMessageId },
{ kind: 'status', text },
{ fence: ctx.fence }
)
ctx.publish()
}
export async function performSend(
ctx: AgentSessionTurnContext,
input: { clientMessageId: string; payloadFingerprint: string; body: AgentJournalMessageItem }
): Promise<TurnOutcome<AgentSessionSendResult>> {
await ctx.journal.appendSubmission({ ...input, fence: ctx.fence })
ctx.publish()
const outcome = await dispatchSafely(ctx, input.clientMessageId, input.body)
await ctx.journal.resolveDispatch(
outcome.state === 'accepted'
? {
clientMessageId: input.clientMessageId,
state: 'accepted',
providerIdentity: outcome.providerIdentity,
fence: ctx.fence
}
: {
clientMessageId: input.clientMessageId,
state: outcome.state,
reason: outcome.reason,
fence: ctx.fence
}
)
ctx.publish()
const submission = ctx.journal
.submissions()
.find((entry) => entry.clientMessageId === input.clientMessageId)
if (!submission) {
throw new Error('agent_session_submission_lost')
}
return { ok: true, value: { clientMessageId: input.clientMessageId, submission } }
}
export async function performCancel(
ctx: AgentSessionTurnContext,
input: { clientOperationId: string; turnId: string }
): Promise<TurnOutcome<AgentSessionCancelResult>> {
let cancelled = false
let note = `Cancelled turn ${input.turnId}.`
try {
cancelled = (
await ctx.adapter.cancelTurn({
sessionId: ctx.sessionId,
turnId: input.turnId,
fence: ctx.fence
})
).cancelled
if (!cancelled) {
note = `The provider had already finished turn ${input.turnId}.`
}
} catch (error) {
note = `Cancellation of turn ${input.turnId} was not confirmed: ${
error instanceof Error ? error.message : String(error)
}`
}
// Keyed by the operation id so a replayed cancel upserts one item, not two.
await appendStatus(ctx, input.clientOperationId, note)
return { ok: true, value: { turnId: input.turnId, cancelled } }
}
function promptBodyOf(
body: AgentJournalItemBody
): { options: readonly { id: string }[]; resolution: AgentJournalResolution } | null {
return body.kind === 'approval' || body.kind === 'question' ? body : null
}
/**
* Durable compare-and-set on (itemId, revision) plus the pending state. The
* journal write commits before the provider callback fires, so two clients
* answering one prompt produce exactly one callback and the loser is told which
* answer won.
*/
export async function performPrompt(
ctx: AgentSessionTurnContext,
input: {
itemId: string
expectedRevision: number
optionId: string
kind: 'approval' | 'question'
}
): Promise<TurnOutcome<AgentSessionPromptResult>> {
const item = ctx.journal.snapshot().items.find((entry) => entry.itemId === input.itemId)
if (!item) {
return invalid(`No item ${input.itemId} in session ${ctx.sessionId}.`)
}
const prompt = promptBodyOf(item.body)
if (!prompt || item.body.kind !== input.kind) {
return invalid(`Item ${input.itemId} is not a pending ${input.kind}.`)
}
if (item.revision !== input.expectedRevision) {
return {
ok: false,
refusal: {
code: 'agent_session_item_revision_stale',
message: `Item ${input.itemId} has moved on.`,
currentRevision: item.revision,
resolution: prompt.resolution
}
}
}
if (prompt.resolution.state !== 'pending') {
return {
ok: false,
refusal: {
code: 'agent_session_already_resolved',
message: `Item ${input.itemId} was already ${prompt.resolution.state}.`,
currentRevision: item.revision,
resolution: prompt.resolution
}
}
}
if (!prompt.options.some((option) => option.id === input.optionId)) {
return invalid(`Option ${input.optionId} is not offered by item ${input.itemId}.`)
}
const identity = parseAgentJournalItemKey(input.itemId)
if (!identity) {
return invalid(`Item id ${input.itemId} is not a well-formed item key.`)
}
const resolution: AgentJournalResolution = {
state: 'resolved',
selectedOptionId: input.optionId,
resolvedBy: ctx.resolvedBy,
resolvedAt: ctx.now()
}
const appended = await ctx.journal.appendItem(
identity,
{ ...item.body, resolution },
{
fence: ctx.fence
}
)
ctx.publish()
try {
await ctx.adapter.answerPrompt({
sessionId: ctx.sessionId,
itemId: input.itemId,
kind: input.kind,
optionId: input.optionId,
fence: ctx.fence
})
} catch (error) {
// The answer is committed and will not be offered again; say so rather than
// reopening the prompt and risking a second callback.
await appendStatus(
ctx,
`${input.itemId}#delivery`,
`Your answer was recorded but the agent did not confirm it: ${
error instanceof Error ? error.message : String(error)
}`
)
}
return {
ok: true,
value: { itemId: appended.itemId, revision: appended.revision, resolution }
}
}
/** Options live on the provider, not in the journal, so this writes nothing. */
export async function performSetOption(
ctx: AgentSessionTurnContext,
input: { key: string; value: string }
): Promise<TurnOutcome<AgentSessionOptionResult>> {
await ctx.adapter.setOption({ sessionId: ctx.sessionId, ...input, fence: ctx.fence })
return { ok: true, value: input }
}
@@ -0,0 +1,33 @@
// Ledger admission for mutations that are not reservations — send, cancel, an
// approval answer. Split from the store so the store keeps only the transaction.
import {
agentSessionOperationKey,
evaluateAgentSessionOperation,
pruneAgentSessionOperationRows,
type AgentSessionOperationDecision,
type AgentSessionOperationRow
} from '../../shared/agent-session-operation-ledger'
export type AgentSessionOperationAdmission = {
callerKey: string
operationId: string
fingerprint: string
now: number
}
type OperationRows = Map<string, AgentSessionOperationRow>
/** Prune, evaluate, and (on admit) place the row. The caller runs this inside one
* transaction, so two concurrent copies of an operation id cannot both admit. */
export function admitAgentSessionOperationRow(
rows: OperationRows,
args: AgentSessionOperationAdmission
): { rows: OperationRows; decision: AgentSessionOperationDecision } {
const pruned = pruneAgentSessionOperationRows(rows, args.now)
const decision = evaluateAgentSessionOperation({ rows: pruned, ...args })
if (decision.decision === 'admit') {
pruned.set(agentSessionOperationKey(args.callerKey, args.operationId), decision.row)
}
return { rows: pruned, decision }
}
@@ -12,9 +12,14 @@
import {
pruneAgentSessionOperationRows,
agentSessionOperationKey,
type AgentSessionOperationDecision,
type AgentSessionOperationOutcome,
type AgentSessionOperationRow
} from '../../shared/agent-session-operation-ledger'
import {
admitAgentSessionOperationRow,
type AgentSessionOperationAdmission
} from './agent-session-operation-admission'
import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication'
import { classifyObservedAgentSessionSpawnToken } from '../../shared/agent-session-lease-adjudication'
import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle'
@@ -276,6 +281,17 @@ export class AgentSessionRecordStore {
})
}
/** Admits one non-reservation mutation through the durable ledger. */
async admitOperation(
args: AgentSessionOperationAdmission
): Promise<AgentSessionOperationDecision> {
return this.transact(() => {
const admitted = admitAgentSessionOperationRow(this.state.operations, args)
this.state.operations = admitted.rows
return admitted.decision
})
}
async recordOperationOutcome(args: {
callerKey: string
operationId: string
+2
View File
@@ -39,6 +39,7 @@ import { EMULATOR_METHODS } from './emulator'
import { PAIRING_METHODS } from './pairing'
import { UPDATER_METHODS } from './updater'
import { AGENT_SESSION_METHODS } from './agent-session'
import { STRUCTURED_AGENT_SESSION_METHODS } from './structured-agent-session'
import { ARTIFACT_METHODS } from './artifacts'
// Why: a flat manifest keeps registration order explicit and provides one
@@ -52,6 +53,7 @@ export const ALL_RPC_METHODS: readonly RpcAnyMethod[] = [
...REPO_METHODS,
...WORKTREE_METHODS,
...AGENT_SESSION_METHODS,
...STRUCTURED_AGENT_SESSION_METHODS,
...TERMINAL_METHODS,
...TERMINAL_ORPHAN_METHODS,
...BROWSER_CORE_METHODS,
@@ -0,0 +1,173 @@
// Wire validation for `agentSession.*`.
//
// Strict objects throughout: zod drops unknown keys, and a silently dropped key
// is how a newer client's field becomes a different effect on an older host.
import { z } from 'zod'
import { isAgentSessionId } from '../../../../shared/agent-session-record'
import {
AGENT_SESSION_HISTORY_DIRECTIONS,
AGENT_SESSION_HISTORY_MAX_LIMIT
} from '../../../../shared/agent-session-wire'
import { normalizeExecutionHostId } from '../../../../shared/execution-host'
const MAX_ID_LENGTH = 512
const MAX_PROMPT_BYTES = 256 * 1024
const MAX_BLOCKS = 64
const MAX_OPTION_LABEL = 512
export const SessionId = z
.string()
.max(MAX_ID_LENGTH)
.refine(isAgentSessionId, 'Invalid agent session id')
const Identifier = (message: string) =>
z
.string()
.min(1, message)
.max(MAX_ID_LENGTH, message)
.refine((value) => value === value.trim(), message)
export const JournalCursor = z
.object({
epoch: Identifier('Invalid journal epoch'),
sequence: z.number().int().nonnegative()
})
.strict()
export const MutationEnvelope = z
.object({
sessionId: SessionId,
clientOperationId: Identifier('Invalid client operation id'),
/** Null is the "must not exist yet" case; every other call fences. */
expectedRuntimeFence: z.number().int().positive().nullable(),
payloadFingerprint: z
.string()
.regex(/^[0-9a-f]{64}$/, 'Payload fingerprint must be a sha256 hex digest')
})
.strict()
const ProviderHandle = z.discriminatedUnion('kind', [
z.object({ kind: z.literal('codex'), threadId: Identifier('Invalid thread id') }).strict(),
z
.object({
kind: z.literal('claude'),
sessionId: Identifier('Invalid provider session id'),
leafUuid: Identifier('Invalid leaf uuid').nullable()
})
.strict()
])
const ExecutionHostId = z
.string()
.max(MAX_ID_LENGTH)
.transform((value) => normalizeExecutionHostId(value))
.refine((value): value is NonNullable<typeof value> => value !== null, {
message: 'Invalid execution host id'
})
const ExecutionLocation = z
.object({
executionHostId: ExecutionHostId,
wslDistro: Identifier('Invalid WSL distro').nullable(),
workspaceId: Identifier('Invalid workspace id'),
workspaceKind: z.enum(['git-worktree', 'folder'])
})
.strict()
const AccountHome = z
.object({
variable: z.enum(['CLAUDE_CONFIG_DIR', 'CODEX_HOME']),
path: z.string().min(1).max(4096)
})
.strict()
export const AttachParams = z
.object({
envelope: MutationEnvelope,
location: ExecutionLocation,
provider: z.enum(['codex', 'claude']),
agent: Identifier('Invalid agent'),
accountHome: AccountHome,
runtimeKind: z.enum(['native', 'tui']),
providerHandle: ProviderHandle
})
.strict()
/** Clients may only author user turns. Accepting an assistant or tool role here
* would let one client write words into the agent's mouth in another's
* timeline, and the provider — not the client — owns those. */
const SendBlock = z.discriminatedUnion('type', [
z.object({ type: z.literal('text'), text: z.string() }).strict(),
z
.object({
type: z.literal('image-ref'),
path: z.string().min(1).max(4096).optional(),
url: z.string().min(1).max(4096).optional(),
alt: z.string().max(MAX_OPTION_LABEL).optional()
})
.strict()
.refine(
(value) => Boolean(value.path) !== Boolean(value.url),
'Provide exactly one of path/url'
)
])
export const SendParams = z
.object({
envelope: MutationEnvelope,
body: z
.object({
kind: z.literal('message'),
role: z.literal('user'),
blocks: z.array(SendBlock).min(1).max(MAX_BLOCKS)
})
.strict()
.refine(
(value) => Buffer.byteLength(JSON.stringify(value.blocks), 'utf8') <= MAX_PROMPT_BYTES,
'Message is too large'
)
})
.strict()
export const CancelParams = z
.object({ envelope: MutationEnvelope, turnId: Identifier('Invalid turn id') })
.strict()
export const RespondParams = z
.object({
envelope: MutationEnvelope,
itemId: Identifier('Invalid item id'),
/** Compare-and-set: the revision the client had on screen. */
expectedRevision: z.number().int().positive(),
optionId: Identifier('Invalid option id')
})
.strict()
export const SetOptionParams = z
.object({
envelope: MutationEnvelope,
key: Identifier('Invalid option key'),
value: z.string().max(MAX_OPTION_LABEL)
})
.strict()
export const HistoryParams = z
.object({
sessionId: SessionId,
direction: z.enum(AGENT_SESSION_HISTORY_DIRECTIONS),
cursor: JournalCursor.optional(),
limit: z.number().int().positive().max(AGENT_SESSION_HISTORY_MAX_LIMIT).optional()
})
.strict()
export const SubscribeParams = z
.object({ sessionId: SessionId, cursor: JournalCursor.optional() })
.strict()
export const UnsubscribeParams = z
.object({
sessionId: SessionId,
subscriptionId: Identifier('Invalid subscription id').optional()
})
.strict()
@@ -0,0 +1,292 @@
// The wire boundary: who may see `agentSession.*` at all, and what shapes it
// accepts once they can.
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import type { StructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-host'
import { setStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry'
import {
RUNTIME_CAPABILITIES,
RUNTIME_PROTOCOL_VERSION,
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY
} from '../../../../shared/protocol-version'
import type { OrcaRuntimeService } from '../../orca-runtime'
import type { RpcRequest, RpcResponse } from '../core'
import { RpcDispatcher } from '../dispatcher'
import { ALL_RPC_METHODS } from './index'
import { STRUCTURED_AGENT_SESSION_METHODS } from './structured-agent-session'
const SESSION = 'session-alpha'
const FINGERPRINT = 'f'.repeat(64)
const OPERATION = '1800000000000-00000000000000000000000000000001'
function envelope(overrides: Record<string, unknown> = {}) {
return {
sessionId: SESSION,
clientOperationId: OPERATION,
expectedRuntimeFence: 1,
payloadFingerprint: FINGERPRINT,
...overrides
}
}
function sendParams(overrides: Record<string, unknown> = {}) {
return {
envelope: envelope(),
body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] },
...overrides
}
}
function attachParams(overrides: Record<string, unknown> = {}) {
return {
envelope: envelope({ expectedRuntimeFence: null }),
location: {
executionHostId: 'local',
wslDistro: null,
workspaceId: 'workspace-1',
workspaceKind: 'git-worktree'
},
provider: 'codex',
agent: 'codex',
accountHome: { variable: 'CODEX_HOME', path: '/home/dev/.codex' },
runtimeKind: 'native',
providerHandle: { kind: 'codex', threadId: 'thread-1' },
...overrides
}
}
function request(method: string, params: unknown): RpcRequest {
return { id: 'request-1', authToken: 'token', method, params }
}
let hostCalls: Record<string, ReturnType<typeof vi.fn>>
function hostStub(): StructuredAgentSessionHost {
hostCalls = {
attach: vi.fn(async () => ({ ok: true, replayed: false })),
send: vi.fn(async () => ({ ok: true, replayed: false })),
cancel: vi.fn(async () => ({ ok: true, replayed: false })),
respondToPrompt: vi.fn(async () => ({ ok: true, replayed: false })),
setOption: vi.fn(async () => ({ ok: true, replayed: false })),
history: vi.fn(() => ({ ok: true, page: { items: [] } })),
subscribe: vi.fn(() => () => undefined),
unsubscribe: vi.fn()
}
return hostCalls as unknown as StructuredAgentSessionHost
}
function dispatcher(): RpcDispatcher {
const runtime = {
getRuntimeId: () => 'runtime-1',
registerSubscriptionCleanup: vi.fn(),
cleanupSubscription: vi.fn(),
cleanupSubscriptionsByPrefix: vi.fn()
}
return new RpcDispatcher({
runtime: runtime as unknown as OrcaRuntimeService,
methods: STRUCTURED_AGENT_SESSION_METHODS
})
}
/** The reply path is the only one that carries a client's negotiated identity,
* which is exactly what the capability gate reads. */
async function call(
method: string,
params: unknown,
client?: { clientKind?: 'mobile' | 'runtime'; clientCapabilities?: string[] }
): Promise<RpcResponse> {
const replies: RpcResponse[] = []
await dispatcher().dispatchStreaming(
request(method, params),
(raw) => replies.push(JSON.parse(raw) as RpcResponse),
client
)
const first = replies[0]
if (!first) {
throw new Error(`no reply for ${method}`)
}
return first
}
const STRUCTURED_CLIENT = {
clientKind: 'mobile' as const,
clientCapabilities: [STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY]
}
beforeEach(() => {
setStructuredAgentSessionHost(hostStub())
})
afterEach(() => {
setStructuredAgentSessionHost(null)
})
describe('capability gating', () => {
it('advertises the capability without bumping the protocol version', () => {
expect(RUNTIME_CAPABILITIES).toContain(STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY)
// Additive methods do not break an old client; bumping would strand every
// paired device that has not updated.
expect(RUNTIME_PROTOCOL_VERSION).toBe(3)
})
it('registers every structured method on the runtime manifest', () => {
const names = new Set(ALL_RPC_METHODS.map((method) => method.name))
for (const method of STRUCTURED_AGENT_SESSION_METHODS) {
expect(names).toContain(method.name)
}
expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(10)
})
it('hides the surface from a declared client that did not advertise it', async () => {
const response = await call('agentSession.send', sendParams(), {
clientKind: 'mobile',
clientCapabilities: ['terminal.stream.v1']
})
expect(response).toMatchObject({
ok: false,
error: { message: expect.stringContaining('structured_agent_session_unsupported') }
})
expect(hostCalls.send).not.toHaveBeenCalled()
})
it('serves a client that advertised it', async () => {
const response = await call('agentSession.send', sendParams(), STRUCTURED_CLIENT)
expect(response).toMatchObject({ ok: true })
expect(hostCalls.send).toHaveBeenCalledTimes(1)
})
it('serves an in-process caller, which negotiates no capabilities at all', async () => {
const response = await call('agentSession.send', sendParams())
expect(response).toMatchObject({ ok: true })
})
it('reports the surface as absent when no host is installed', async () => {
setStructuredAgentSessionHost(null)
const response = await call('agentSession.send', sendParams(), STRUCTURED_CLIENT)
expect(response).toMatchObject({ ok: false })
})
})
describe('method routing', () => {
it('separates create from ensure by the fence the client may declare', async () => {
const created = await call('agentSession.create', attachParams(), STRUCTURED_CLIENT)
expect(created).toMatchObject({ ok: true })
const fenced = await call(
'agentSession.create',
attachParams({ envelope: envelope() }),
STRUCTURED_CLIENT
)
expect(fenced).toMatchObject({ ok: false })
const ensured = await call(
'agentSession.ensure',
attachParams({ envelope: envelope() }),
STRUCTURED_CLIENT
)
expect(ensured).toMatchObject({ ok: true })
})
it('tags the prompt kind from the method name, not from the client', async () => {
const params = {
envelope: envelope(),
itemId: 'item-1',
expectedRevision: 1,
optionId: 'allow'
}
await call('agentSession.respondToApproval', params, STRUCTURED_CLIENT)
await call('agentSession.respondToQuestion', params, STRUCTURED_CLIENT)
expect(hostCalls.respondToPrompt.mock.calls.map((invocation) => invocation[1].kind)).toEqual([
'approval',
'question'
])
})
})
describe('parameter validation', () => {
const rejects = async (method: string, params: unknown): Promise<void> => {
const response = await call(method, params, STRUCTURED_CLIENT)
expect(response).toMatchObject({ ok: false, error: { code: 'invalid_argument' } })
}
it('rejects an unknown key rather than dropping it', async () => {
await rejects('agentSession.send', { ...sendParams(), replyToItemId: 'item-1' })
await rejects('agentSession.send', {
...sendParams(),
envelope: { ...envelope(), priority: 'high' }
})
})
it('refuses to let a client author anything but a user turn', async () => {
await rejects(
'agentSession.send',
sendParams({
body: { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'hi' }] }
})
)
await rejects(
'agentSession.send',
sendParams({
body: { kind: 'message', role: 'user', blocks: [{ type: 'tool-call', name: 'Bash' }] }
})
)
})
it('rejects a journal-only opaque provider handle', async () => {
await rejects(
'agentSession.create',
attachParams({ providerHandle: { kind: 'opaque', agent: 'codex', value: 'thread-1' } })
)
})
it('requires a sha256 fingerprint and a positive fence', async () => {
await rejects(
'agentSession.send',
sendParams({ envelope: envelope({ payloadFingerprint: 'f' }) })
)
await rejects(
'agentSession.send',
sendParams({ envelope: envelope({ payloadFingerprint: 'F'.repeat(64) }) })
)
await rejects(
'agentSession.send',
sendParams({ envelope: envelope({ expectedRuntimeFence: 0 }) })
)
})
it('requires the item revision on a prompt answer', async () => {
await rejects('agentSession.respondToApproval', {
envelope: envelope(),
itemId: 'item-1',
optionId: 'allow'
})
})
it('bounds a history page and validates its cursor', async () => {
await rejects('agentSession.history', {
sessionId: SESSION,
direction: 'tail',
limit: 100_000
})
await rejects('agentSession.history', { sessionId: SESSION, direction: 'sideways' })
await rejects('agentSession.history', {
sessionId: SESSION,
direction: 'after',
cursor: { epoch: 'epoch-1', sequence: -1 }
})
})
it('accepts a well-formed history request', async () => {
const response = await call(
'agentSession.history',
{
sessionId: SESSION,
direction: 'after',
cursor: { epoch: 'epoch-1', sequence: 4 },
limit: 40
},
STRUCTURED_CLIENT
)
expect(response).toMatchObject({ ok: true })
})
})
@@ -0,0 +1,159 @@
// `agentSession.*` — the structured session RPC surface.
//
// Every method here is gated on the client advertising
// `agent-session.structured.v1`. A client that does not is told the surface does
// not exist rather than being handed a session it cannot render or drive; that
// is the whole visibility rule, because nothing else on the runtime publishes a
// structured session.
import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version'
import { getStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry'
import type {
StructuredAgentSessionCaller,
StructuredAgentSessionHost
} from '../../../native-chat/agent-session-wire/structured-agent-session-host'
import { defineMethod, defineStreamingMethod, type RpcAnyMethod, type RpcContext } from '../core'
import {
AttachParams,
CancelParams,
HistoryParams,
RespondParams,
SendParams,
SetOptionParams,
SubscribeParams,
UnsubscribeParams
} from './structured-agent-session-schemas'
const SUBSCRIPTION_PREFIX = 'agentSession'
/**
* In-process callers are the same build as the host, so they carry no negotiated
* capability list; every remote client must say it can read structured sessions.
*/
function supportsStructuredSessions(ctx: RpcContext): boolean {
return (
ctx.clientKind === undefined ||
(ctx.clientCapabilities?.includes(STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY) ?? false)
)
}
function requireHost(ctx: RpcContext): StructuredAgentSessionHost {
const host = supportsStructuredSessions(ctx) ? getStructuredAgentSessionHost() : null
if (!host) {
throw new Error('structured_agent_session_unsupported')
}
return host
}
/** Mirrors the existing agent-session host-authority derivation so one client
* gets one operation namespace across both surfaces. */
function callerFor(ctx: RpcContext): StructuredAgentSessionCaller {
return {
callerKey: ctx.clientId?.trim() || `trusted-local:${ctx.clientKind ?? 'runtime'}`
}
}
function subscriptionIdFor(ctx: RpcContext, sessionId: string): string {
const base = `${SUBSCRIPTION_PREFIX}:${ctx.connectionId ?? 'local'}:${sessionId}`
// Shared control multiplexes several streams over one socket; the frame id
// keeps one subscriber from evicting another on the same session.
return ctx.requestId ? `${base}:${ctx.requestId}` : base
}
export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [
defineMethod({
name: 'agentSession.create',
params: AttachParams,
handler: async (params, ctx) => {
if (params.envelope.expectedRuntimeFence !== null) {
throw new Error('agent_session_operation_invalid')
}
return requireHost(ctx).attach(callerFor(ctx), params)
}
}),
defineMethod({
name: 'agentSession.ensure',
params: AttachParams,
handler: async (params, ctx) => requireHost(ctx).attach(callerFor(ctx), params)
}),
defineMethod({
name: 'agentSession.send',
params: SendParams,
handler: async (params, ctx) => requireHost(ctx).send(callerFor(ctx), params)
}),
defineMethod({
name: 'agentSession.cancel',
params: CancelParams,
handler: async (params, ctx) => requireHost(ctx).cancel(callerFor(ctx), params)
}),
defineMethod({
name: 'agentSession.respondToApproval',
params: RespondParams,
handler: async (params, ctx) =>
requireHost(ctx).respondToPrompt(callerFor(ctx), { ...params, kind: 'approval' })
}),
defineMethod({
name: 'agentSession.respondToQuestion',
params: RespondParams,
handler: async (params, ctx) =>
requireHost(ctx).respondToPrompt(callerFor(ctx), { ...params, kind: 'question' })
}),
defineMethod({
name: 'agentSession.setOption',
params: SetOptionParams,
handler: async (params, ctx) => requireHost(ctx).setOption(callerFor(ctx), params)
}),
defineMethod({
name: 'agentSession.history',
params: HistoryParams,
handler: async (params, ctx) => requireHost(ctx).history(params)
}),
defineStreamingMethod({
name: 'agentSession.subscribe',
params: SubscribeParams,
handler: async (params, ctx, emit) => {
const host = requireHost(ctx)
const subscriptionId = subscriptionIdFor(ctx, params.sessionId)
let closed = false
let dispose = (): void => {}
ctx.runtime.registerSubscriptionCleanup(
subscriptionId,
() => {
closed = true
dispose()
},
ctx.connectionId
)
if (closed) {
return
}
// The host emits the opening snapshot (or the missed batch) synchronously
// inside open(), so nothing between here and there can interleave.
dispose = host.subscribe({
id: subscriptionId,
sessionId: params.sessionId,
emit,
...(params.cursor ? { cursor: params.cursor } : {})
})
if (closed) {
dispose()
}
}
}),
defineMethod({
name: 'agentSession.unsubscribe',
params: UnsubscribeParams,
handler: async (params, ctx) => {
requireHost(ctx)
const connection = ctx.connectionId ?? 'local'
const base = `${SUBSCRIPTION_PREFIX}:${connection}:${params.sessionId}`
if (params.subscriptionId) {
ctx.runtime.cleanupSubscription(`${base}:${params.subscriptionId}`)
return { unsubscribed: true }
}
ctx.runtime.cleanupSubscription(base)
ctx.runtime.cleanupSubscriptionsByPrefix(`${base}:`)
return { unsubscribed: true }
}
})
]
@@ -49,3 +49,35 @@ export function agentJournalItemKey(identity: AgentJournalItemIdentity): string
export function agentJournalSubmissionKey(clientMessageId: string): string {
return agentJournalItemKey({ provider: 'orca', clientMessageId })
}
/**
* Inverse of {@link agentJournalItemKey}. Clients hold item KEYS, but an upsert
* needs the identity behind one — answering an approval re-appends the same
* item at the next revision. Every component is percent-encoded, and
* `encodeURIComponent` escapes the delimiter, so the split is unambiguous.
*/
export function parseAgentJournalItemKey(key: string): AgentJournalItemIdentity | null {
const parts = key.split(KEY_DELIMITER).map((part) => decodeURIComponent(part))
const [provider, ...rest] = parts
if (provider === 'codex' && rest.length === 3) {
const ordinal = Number(rest[2])
return Number.isSafeInteger(ordinal) && ordinal >= 0
? { provider, threadId: rest[0] as string, turnId: rest[1] as string, ordinal }
: null
}
if (provider === 'claude' && rest.length === 2) {
return { provider, sessionId: rest[0] as string, uuid: rest[1] as string }
}
if (provider === 'orca' && rest.length === 1) {
return { provider, clientMessageId: rest[0] as string }
}
if (provider === 'legacy' && rest.length === 3) {
return {
provider,
agent: rest[0] as string,
sessionId: rest[1] as string,
recordId: rest[2] as string
}
}
return null
}
@@ -0,0 +1,158 @@
import { describe, expect, it } from 'vitest'
import type { AgentSessionOperationDecision } from './agent-session-operation-ledger'
import { agentSessionLeaseFixture } from './agent-session-record.test-fixture'
import {
admitAgentSessionMutation,
agentSessionFingerprintConflict,
computeAgentSessionPayloadFingerprint
} from './agent-session-mutation-envelope'
import type { AgentSessionMutationEnvelope } from './agent-session-wire'
const LEASE = agentSessionLeaseFixture({
sessionId: 'session-1',
runtimeKind: 'native',
runtimeFence: 4
})
function envelope(overrides: Partial<AgentSessionMutationEnvelope> = {}) {
return {
sessionId: 'session-1',
clientOperationId: 'op-1',
expectedRuntimeFence: 4,
payloadFingerprint: 'f'.repeat(64),
...overrides
}
}
function row(fingerprint: string) {
return {
callerKey: 'caller-1',
operationId: 'op-1',
fingerprint,
operationTimestamp: 1_000,
recordedAt: 1_000,
expiresAt: 100_000,
outcome: { status: 'pending' as const }
}
}
const ADMIT = (fingerprint: string): AgentSessionOperationDecision => ({
decision: 'admit',
row: row(fingerprint)
})
describe('computeAgentSessionPayloadFingerprint', () => {
it('is stable across key order at every depth', () => {
const a = computeAgentSessionPayloadFingerprint({
method: 'agentSession.send',
sessionId: 'session-1',
fields: { body: { kind: 'message', blocks: [{ type: 'text', text: 'hi' }] }, extra: 1 }
})
const b = computeAgentSessionPayloadFingerprint({
method: 'agentSession.send',
sessionId: 'session-1',
fields: { extra: 1, body: { blocks: [{ text: 'hi', type: 'text' }], kind: 'message' } }
})
expect(a).toBe(b)
expect(a).toMatch(/^[0-9a-f]{64}$/)
})
it('separates one payload from another and one method from another', () => {
const send = computeAgentSessionPayloadFingerprint({
method: 'agentSession.send',
sessionId: 'session-1',
fields: { text: 'hi' }
})
expect(
computeAgentSessionPayloadFingerprint({
method: 'agentSession.send',
sessionId: 'session-1',
fields: { text: 'hi there' }
})
).not.toBe(send)
expect(
computeAgentSessionPayloadFingerprint({
method: 'agentSession.cancel',
sessionId: 'session-1',
fields: { text: 'hi' }
})
).not.toBe(send)
})
})
describe('agentSessionFingerprintConflict', () => {
it('refuses a payload that does not match what the client declared', () => {
const conflict = agentSessionFingerprintConflict(envelope(), 'a'.repeat(64))
expect(conflict?.code).toBe('agent_session_operation_conflict')
})
it('passes a matching declaration', () => {
expect(agentSessionFingerprintConflict(envelope(), 'f'.repeat(64))).toBeNull()
})
})
describe('admitAgentSessionMutation', () => {
const base = { envelope: envelope(), hostFingerprint: 'f'.repeat(64), lease: LEASE }
it('admits a first-time operation under a live lease at the expected fence', () => {
expect(admitAgentSessionMutation({ ...base, ledger: ADMIT('f'.repeat(64)) }).decision).toBe(
'admit'
)
})
it('replays a recorded operation without re-checking the fence', () => {
const admission = admitAgentSessionMutation({
...base,
envelope: envelope({ expectedRuntimeFence: 1 }),
ledger: { decision: 'replay', row: row('f'.repeat(64)) }
})
expect(admission.decision).toBe('replay')
})
it('refuses a stale fence and hands back the current one', () => {
const admission = admitAgentSessionMutation({
...base,
envelope: envelope({ expectedRuntimeFence: 3 }),
ledger: ADMIT('f'.repeat(64))
})
expect(admission).toMatchObject({
decision: 'refused',
refusal: { code: 'agent_session_checkpoint_stale', currentFence: 4 }
})
})
it('refuses a writer while the lease is unreconciled', () => {
const admission = admitAgentSessionMutation({
...base,
lease: { ...LEASE, unreconciled: true },
ledger: ADMIT('f'.repeat(64))
})
expect(admission).toMatchObject({
decision: 'refused',
refusal: { code: 'execution_owner_reconciling' }
})
})
it('refuses a writer mid-handoff', () => {
const admission = admitAgentSessionMutation({
...base,
lease: { ...LEASE, handoffStage: 'new-owner-proving' },
ledger: ADMIT('f'.repeat(64))
})
expect(admission).toMatchObject({
decision: 'refused',
refusal: { code: 'agent_session_conflict' }
})
})
it('surfaces a ledger refusal verbatim', () => {
const admission = admitAgentSessionMutation({
...base,
ledger: { decision: 'refused', code: 'agent_session_operation_expired' }
})
expect(admission).toMatchObject({
decision: 'refused',
refusal: { code: 'agent_session_operation_expired' }
})
})
})
@@ -0,0 +1,151 @@
// Admission for one mutating `agentSession.*` call.
//
// The rules themselves live in the durable ledger and the lease adjudicator;
// this is only the fixed order they are applied in, plus the payload
// fingerprint both peers derive from the same request fields. Nothing here
// re-derives who may write — that answer comes from
// `agentSessionLeaseAdmitsWriter` alone.
import { createHash } from 'node:crypto'
import type {
AgentSessionOperationDecision,
AgentSessionOperationRow
} from './agent-session-operation-ledger'
import {
agentSessionLeaseAdmitsWriter,
isAgentSessionFenceCurrent
} from './agent-session-lease-adjudication'
import type { AgentSessionLease } from './agent-session-record'
import type { AgentSessionMutationEnvelope, AgentSessionWireRefusal } from './agent-session-wire'
/**
* Stable digest over the fields that define what this call DOES. Keys are
* emitted in sorted order at every depth so two peers serializing the same
* request in different property order agree, and an undefined field is dropped
* rather than hashed as present-but-empty.
*/
export function computeAgentSessionPayloadFingerprint(input: {
method: string
sessionId: string
fields: Record<string, unknown>
}): string {
const canonical = canonicalize({
method: input.method,
sessionId: input.sessionId,
fields: input.fields
})
return createHash('sha256').update(canonical).digest('hex')
}
function canonicalize(value: unknown): string {
if (value === null || typeof value !== 'object') {
return JSON.stringify(value ?? null)
}
if (Array.isArray(value)) {
return `[${value.map(canonicalize).join(',')}]`
}
const entries = Object.entries(value as Record<string, unknown>)
.filter(([, entry]) => entry !== undefined)
.sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))
return `{${entries.map(([key, entry]) => `${JSON.stringify(key)}:${canonicalize(entry)}`).join(',')}}`
}
/**
* A retry whose payload changed is a different call wearing the same id.
* Checked BEFORE the ledger is consulted, so a refused call never leaves an
* admitted row that a later honest retry would replay as already-done.
*/
export function agentSessionFingerprintConflict(
envelope: AgentSessionMutationEnvelope,
hostFingerprint: string
): AgentSessionWireRefusal | null {
return envelope.payloadFingerprint === hostFingerprint
? null
: {
code: 'agent_session_operation_conflict',
message:
'The payload does not match the fingerprint the client declared for this operation.'
}
}
export type AgentSessionMutationAdmission =
| { decision: 'admit'; row: AgentSessionOperationRow }
/** The recorded outcome answers this call; do not run the effect again. */
| { decision: 'replay'; row: AgentSessionOperationRow }
| { decision: 'refused'; refusal: AgentSessionWireRefusal }
/**
* Fixed order: fingerprint agreement, then the ledger (so a retry replays
* before anything else can refuse it), then the lease, then the fence. Putting
* the ledger ahead of the fence is deliberate — a retry that crossed an owner
* change must still return its recorded answer instead of a stale-checkpoint
* refusal the client would then resend as a second effect.
*/
export function admitAgentSessionMutation(input: {
envelope: AgentSessionMutationEnvelope
/** Fingerprint the host computed from the request it actually received. */
hostFingerprint: string
/** Decision from the durable ledger, evaluated under `hostFingerprint`. */
ledger: AgentSessionOperationDecision
lease: AgentSessionLease
}): AgentSessionMutationAdmission {
const { envelope, lease, ledger } = input
const mismatch = agentSessionFingerprintConflict(envelope, input.hostFingerprint)
if (mismatch) {
return { decision: 'refused', refusal: mismatch }
}
if (ledger.decision === 'refused') {
return {
decision: 'refused',
refusal: {
code: ledger.code,
message: `Operation ${envelope.clientOperationId} was refused: ${ledger.code}.`
}
}
}
if (ledger.decision === 'replay') {
return { decision: 'replay', row: ledger.row }
}
const leaseRefusal = refuseUnlessWriterAdmitted(lease)
if (leaseRefusal) {
return { decision: 'refused', refusal: leaseRefusal }
}
if (
envelope.expectedRuntimeFence === null ||
!isAgentSessionFenceCurrent(lease, envelope.expectedRuntimeFence)
) {
return {
decision: 'refused',
refusal: {
code: 'agent_session_checkpoint_stale',
message: `Expected runtime fence ${envelope.expectedRuntimeFence ?? 'none'}; the session is at ${lease.runtimeFence}.`,
currentFence: lease.runtimeFence
}
}
}
return { decision: 'admit', row: ledger.row }
}
/** Why the single admission oracle said no, mapped to what the client can do
* about it. The predicate itself is never re-implemented here. */
function refuseUnlessWriterAdmitted(lease: AgentSessionLease): AgentSessionWireRefusal | null {
if (agentSessionLeaseAdmitsWriter(lease)) {
return null
}
if (lease.unreconciled) {
return {
code: 'execution_owner_reconciling',
message: 'This host has not yet adjudicated the session lease.'
}
}
if (lease.handoffStage !== null) {
return {
code: 'agent_session_conflict',
message: `The session is mid-handoff (${lease.handoffStage}).`
}
}
return {
code: 'agent_session_ownership_unknown',
message: 'The session has no live owner to accept writes.'
}
}
+170
View File
@@ -0,0 +1,170 @@
// ─── Structured agent-session wire contract ─────────────────────────────────
// The shapes `agentSession.*` accepts and publishes. Phase 2 builds provider
// adapters and the mobile client against exactly these types, so everything
// here must be plain JSON and every new field must be optional to old readers
// (docs/reference/remote-wire-compatibility.md).
import type {
AgentJournalCursor,
AgentJournalRenderItem,
AgentJournalResetReason,
AgentJournalResolution,
AgentJournalSnapshot,
AgentJournalSubmission
} from './agent-session-journal-types'
/** Backward paging is the client's normal read; 40 matches the page size the
* mobile list renders without a visible fill-in. */
export const AGENT_SESSION_HISTORY_DEFAULT_LIMIT = 40
export const AGENT_SESSION_HISTORY_MAX_LIMIT = 200
export const AGENT_SESSION_HISTORY_DIRECTIONS = ['tail', 'before', 'after'] as const
/** `tail` is the newest page, `before` pages backward, `after` catches a live
* reader up. Only `after` needs replayable rows; the other two read the
* reduced timeline and so survive compaction. */
export type AgentSessionHistoryDirection = (typeof AGENT_SESSION_HISTORY_DIRECTIONS)[number]
export type AgentSessionHistoryRequest = {
sessionId: string
direction: AgentSessionHistoryDirection
/** Required for `before` and `after`; ignored for `tail`. */
cursor?: AgentJournalCursor
limit?: number
}
export type AgentSessionHistoryPage = {
sessionId: string
epoch: string
direction: AgentSessionHistoryDirection
items: AgentJournalRenderItem[]
/** Populated by `after` reads so a disconnected client can apply tombstones. */
removedItemIds: string[]
/** Submissions overlapping this page, so an unconfirmed bubble renders with
* its dispatch state instead of as a plain message. */
submissions: AgentJournalSubmission[]
/** Page edges. `nextCursor` is what the client sends back for the same
* direction; it equals the request cursor when the page is empty. */
window: {
oldest: AgentJournalCursor | null
newest: AgentJournalCursor | null
nextCursor: AgentJournalCursor
}
hasOlder: boolean
hasNewer: boolean
}
export type AgentSessionHistoryResult =
| { ok: true; page: AgentSessionHistoryPage }
/** Every reset forces a clean reload; the snapshot is inlined so the client
* never has to make a second call to recover. */
| { ok: false; reset: AgentJournalResetReason; snapshot: AgentJournalSnapshot }
/** Cursor-qualified incremental publication. Items and submissions carry their
* CURRENT reduced state rather than a delta, so applying a batch twice
* converges instead of double-appending. */
export type AgentSessionJournalBatch = {
cursor: AgentJournalCursor
items: AgentJournalRenderItem[]
removedItemIds: string[]
submissions: AgentJournalSubmission[]
}
export type AgentSessionSubscribeEvent =
| { type: 'snapshot'; sessionId: string; snapshot: AgentJournalSnapshot; fence: number }
| { type: 'batch'; sessionId: string; batch: AgentSessionJournalBatch }
| {
type: 'reset'
sessionId: string
reset: AgentJournalResetReason
snapshot: AgentJournalSnapshot
fence: number
}
| { type: 'end' }
// ─── Mutation envelope ──────────────────────────────────────────────────────
/**
* The four fields every mutating call carries. Same operation id and same
* fingerprint replays the recorded outcome; a different fingerprint under one
* operation id is a conflict, never a second effect.
*/
export type AgentSessionMutationEnvelope = {
sessionId: string
clientOperationId: string
/** Null only on a create for a session that does not exist yet. */
expectedRuntimeFence: number | null
/** Client-declared; the host recomputes it and compares. */
payloadFingerprint: string
}
export const AGENT_SESSION_WIRE_REFUSAL_CODES = [
'structured_agent_session_unsupported',
'agent_session_checkpoint_stale',
'agent_session_conflict',
'agent_session_ownership_unknown',
'agent_session_operation_conflict',
'agent_session_operation_expired',
'agent_session_operation_capacity',
'agent_session_operation_invalid',
'agent_session_operation_unknown',
'agent_session_item_revision_stale',
'agent_session_already_resolved',
'agent_session_identity_required',
'agent_session_journal_unreadable',
'execution_owner_reconciling'
] as const
export type AgentSessionWireRefusalCode = (typeof AGENT_SESSION_WIRE_REFUSAL_CODES)[number]
export type AgentSessionWireRefusal = {
code: AgentSessionWireRefusalCode
message: string
/** On a stale fence, so the client can retry without another round trip. */
currentFence?: number
/** On a lost compare-and-set: the winning answer and who gave it. */
resolution?: AgentJournalResolution
/** On a lost compare-and-set: the revision the host actually holds. */
currentRevision?: number
}
export type AgentSessionMutationResult<TValue> =
| {
ok: true
/** True when the recorded outcome was returned instead of a new effect. */
replayed: boolean
fence: number
cursor: AgentJournalCursor
value: TValue
}
| { ok: false; refusal: AgentSessionWireRefusal }
// ─── Per-method payloads ────────────────────────────────────────────────────
export type AgentSessionAttachResult = {
sessionId: string
fence: number
snapshot: AgentJournalSnapshot
/** Submissions the crash boundary settled as `unknown` while attaching. */
unconfirmedClientMessageIds: string[]
}
export type AgentSessionSendResult = {
clientMessageId: string
submission: AgentJournalSubmission
}
export type AgentSessionCancelResult = {
/** The turn the client named, echoed so a late reply can be matched. */
turnId: string
cancelled: boolean
}
export type AgentSessionPromptResult = {
itemId: string
revision: number
resolution: AgentJournalResolution
}
export type AgentSessionOptionResult = {
key: string
value: string
}
+6
View File
@@ -79,6 +79,11 @@ export const AGENT_SESSION_HOST_AUTHORITY_RUNTIME_CAPABILITY =
'agent-session.host-authority.v1' as const
export const AGENT_SESSION_OMP_RESUME_PATH_RUNTIME_CAPABILITY =
'agent-session.omp-resume-path.v1' as const
// Why: structured sessions are journal-backed, not PTY-backed, so a client that
// cannot read them must not see them at all — it would render an agent tab it
// can neither display nor drive. The host also refuses every agentSession.*
// method from a connection that does not advertise this.
export const STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY = 'agent-session.structured.v1' as const
// Why: older runtimes strip mutation owner fields, so clients must fence writes before RPC.
export const FILE_MUTATION_OWNERSHIP_RUNTIME_CAPABILITY = 'files.mutation-ownership.v1' as const
export const FILE_MUTATION_OWNERSHIP_UPDATE_REQUIRED_MESSAGE =
@@ -114,6 +119,7 @@ export const RUNTIME_CAPABILITIES = [
REMOTE_SERVER_UPDATE_CAPABILITY,
AGENT_SESSION_HOST_AUTHORITY_RUNTIME_CAPABILITY,
AGENT_SESSION_OMP_RESUME_PATH_RUNTIME_CAPABILITY,
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
FILE_MUTATION_OWNERSHIP_RUNTIME_CAPABILITY,
ACCOUNT_IMPORT_RUNTIME_CAPABILITY,
CODEX_RESET_CREDIT_RUNTIME_CAPABILITY
@@ -0,0 +1,461 @@
// Cross-version coverage for the structured agent-session surface, paired the same
// way the terminal wire harness is: current code against a real published release.
//
// Three skews matter here, and none can be checked from one build alone — an old
// client must not be shown a session it cannot render, a new client must find an
// old host's missing surface cleanly, and a client's cursor must survive the host
// process that minted it.
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
import type { StructuredAgentSessionAdapter } from '../../../src/main/native-chat/agent-session-wire/structured-agent-session-adapter'
import { attachFingerprintFields } from '../../../src/main/native-chat/agent-session-wire/structured-agent-session-attach'
import type { AgentSessionAttachParams } from '../../../src/main/native-chat/agent-session-wire/structured-agent-session-attach'
import { StructuredAgentSessionHost } from '../../../src/main/native-chat/agent-session-wire/structured-agent-session-host'
import { setStructuredAgentSessionHost } from '../../../src/main/native-chat/agent-session-wire/structured-agent-session-registry'
import { AgentSessionRecordStore } from '../../../src/main/runtime/agent-session-record-store'
import { computeAgentSessionPayloadFingerprint } from '../../../src/shared/agent-session-mutation-envelope'
import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire'
import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../../src/shared/protocol-version'
import { resolveBaselineReleaseRef } from './release-checkout'
import {
loadAgentSessionWireBuild,
WORKING_TREE,
type AgentSessionWireBuild,
type RpcClientIdentity,
type RpcReply
} from './versioned-agent-session-wire'
// Why: a cold CI run extracts the baseline checkout before the first pairing.
const SUITE_TIMEOUT_MS = 180_000
const SESSION = 'session-alpha'
const WORKSPACE = 'workspace-1'
const THREAD = '019fd532-7c11-7a90-b6de-4e1a2c3d5f60'
const NOW = 1_800_000_000_000
/** Every method the structured surface publishes, paired with the host method it
* must reach — a gate that hides one method and leaks another is the bug. */
const STRUCTURED_CALLS: { method: string; hostMethod: string | null }[] = [
{ method: 'agentSession.create', hostMethod: 'attach' },
{ method: 'agentSession.ensure', hostMethod: 'attach' },
{ method: 'agentSession.send', hostMethod: 'send' },
{ method: 'agentSession.cancel', hostMethod: 'cancel' },
{ method: 'agentSession.respondToApproval', hostMethod: 'respondToPrompt' },
{ method: 'agentSession.respondToQuestion', hostMethod: 'respondToPrompt' },
{ method: 'agentSession.setOption', hostMethod: 'setOption' },
{ method: 'agentSession.history', hostMethod: 'history' },
{ method: 'agentSession.subscribe', hostMethod: 'subscribe' },
// Teardown runs through the runtime's subscription registry rather than the
// host, so its reply is the only signal that the gate opened.
{ method: 'agentSession.unsubscribe', hostMethod: null }
]
let baselineRef: string
let current: AgentSessionWireBuild
let baseline: AgentSessionWireBuild
let operations = 0
beforeAll(async () => {
baselineRef = resolveBaselineReleaseRef()
current = await loadAgentSessionWireBuild(WORKING_TREE)
baseline = await loadAgentSessionWireBuild(baselineRef)
}, SUITE_TIMEOUT_MS)
/** `<13-digit ms>-<32 hex>`, the only shape the durable ledger accepts. */
function operationId(): string {
operations += 1
return `${NOW}-${operations.toString(16).padStart(32, '0')}`
}
function envelope(args: {
method: string
fields: Record<string, unknown>
fence: number | null
}): Record<string, unknown> {
return {
sessionId: SESSION,
clientOperationId: operationId(),
expectedRuntimeFence: args.fence,
payloadFingerprint: computeAgentSessionPayloadFingerprint({
method: args.method,
sessionId: SESSION,
fields: args.fields
})
}
}
function attachParams(fence: number | null): Record<string, unknown> {
const params = {
envelope: { sessionId: SESSION, clientOperationId: operationId(), expectedRuntimeFence: fence },
location: {
executionHostId: 'local',
wslDistro: null,
workspaceId: WORKSPACE,
workspaceKind: 'git-worktree'
},
provider: 'codex',
agent: 'codex',
accountHome: { variable: 'CODEX_HOME', path: '/home/dev/.codex' },
runtimeKind: 'native',
providerHandle: { kind: 'codex', threadId: THREAD }
}
return {
...params,
envelope: {
...params.envelope,
payloadFingerprint: computeAgentSessionPayloadFingerprint({
method: 'agentSession.attach',
sessionId: SESSION,
fields: attachFingerprintFields(params as unknown as AgentSessionAttachParams)
})
}
}
}
function sendParams(text: string, fence: number): Record<string, unknown> {
const body = { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] }
return { envelope: envelope({ method: 'agentSession.send', fields: { body }, fence }), body }
}
/** Schema-valid params per method; values only need to survive validation. */
function paramsFor(method: string): unknown {
const fence = 1
switch (method) {
case 'agentSession.create':
return attachParams(null)
case 'agentSession.ensure':
return attachParams(fence)
case 'agentSession.send':
return sendParams('hi', fence)
case 'agentSession.cancel':
return {
envelope: envelope({ method: 'agentSession.cancel', fields: { turnId: 'turn-1' }, fence }),
turnId: 'turn-1'
}
case 'agentSession.respondToApproval':
case 'agentSession.respondToQuestion': {
const fields = { itemId: 'item-1', expectedRevision: 1, optionId: 'allow' }
return { envelope: envelope({ method, fields, fence }), ...fields }
}
case 'agentSession.setOption': {
const fields = { key: 'model', value: 'gpt-5' }
return { envelope: envelope({ method, fields, fence }), ...fields }
}
case 'agentSession.history':
return { sessionId: SESSION, direction: 'tail' }
default:
return { sessionId: SESSION }
}
}
function runtimeStub(): unknown {
const cleanups = new Map<string, () => void>()
return {
getRuntimeId: () => 'runtime-1',
registerSubscriptionCleanup: (id: string, cleanup: () => void) => cleanups.set(id, cleanup),
cleanupSubscription: (id: string) => {
cleanups.get(id)?.()
cleanups.delete(id)
},
cleanupSubscriptionsByPrefix: (prefix: string) => {
for (const [id, cleanup] of cleanups) {
if (id.startsWith(prefix)) {
cleanup()
cleanups.delete(id)
}
}
}
}
}
/** Every reply one call produced. Streaming methods answer more than once, and a
* refusal has to arrive as a reply rather than as silence. */
async function callBuild(
build: AgentSessionWireBuild,
method: string,
params: unknown,
client: RpcClientIdentity,
runtime: unknown = runtimeStub()
): Promise<RpcReply[]> {
const replies: RpcReply[] = []
await build
.createDispatcher(runtime)
.dispatchStreaming(
{ id: `request-${method}`, authToken: 'cross-version-token', method, params },
(raw) => replies.push(JSON.parse(raw) as RpcReply),
client
)
return replies
}
describe('cross-version structured agent sessions', () => {
it(
'skews current code against a real published release',
() => {
expect(baselineRef).toMatch(/^v?\d/)
expect(baseline.revision).toMatch(/^[0-9a-f]{40}$/)
expect(baseline.revision).not.toBe(current.revision)
// The anti-vacuous oracle for the source scan: a scan that found nothing
// would make every "no structured method here" claim below meaningless.
expect(baseline.methodNames).toContain('terminal.create')
expect(current.methodNames).toContain('terminal.create')
},
SUITE_TIMEOUT_MS
)
describe('a client that never asked for structured sessions', () => {
let hostCalls: Record<string, ReturnType<typeof vi.fn>>
beforeEach(() => {
operations = 0
hostCalls = {
attach: vi.fn(async () => ({ ok: true, replayed: false })),
send: vi.fn(async () => ({ ok: true, replayed: false })),
cancel: vi.fn(async () => ({ ok: true, replayed: false })),
respondToPrompt: vi.fn(async () => ({ ok: true, replayed: false })),
setOption: vi.fn(async () => ({ ok: true, replayed: false })),
history: vi.fn(() => ({ ok: true, page: { items: [] } })),
subscribe: vi.fn(() => () => undefined),
unsubscribe: vi.fn()
}
setStructuredAgentSessionHost(hostCalls as unknown as StructuredAgentSessionHost)
})
afterEach(() => {
setStructuredAgentSessionHost(null)
})
it('is told the whole surface does not exist, and reaches no host method', async () => {
// The old build cannot name the capability, so its clients never send it.
expect(baseline.capabilities).not.toContain(STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY)
for (const { method } of STRUCTURED_CALLS) {
const replies = await callBuild(current, method, paramsFor(method), {
clientKind: 'mobile',
clientCapabilities: baseline.capabilities
})
expect(replies, `${method} must answer exactly once`).toHaveLength(1)
expect(replies[0]).toMatchObject({
ok: false,
error: { message: expect.stringContaining('structured_agent_session_unsupported') }
})
}
for (const [name, spy] of Object.entries(hostCalls)) {
expect(spy, `${name} ran for a client without the capability`).not.toHaveBeenCalled()
}
})
it('is served the same calls once it advertises the capability', async () => {
for (const { method, hostMethod } of STRUCTURED_CALLS) {
const replies = await callBuild(current, method, paramsFor(method), {
clientKind: 'mobile',
clientCapabilities: [
...baseline.capabilities,
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY
]
})
// A subscription that opens with nothing to say answers with no reply at
// all, so reaching the host is the signal that the gate opened.
if (hostMethod) {
expect(hostCalls[hostMethod], `${method} did not reach the host`).toHaveBeenCalled()
} else {
expect(replies[0], `${method} was refused`).toMatchObject({ ok: true })
}
for (const reply of replies) {
expect(reply, `${method} was refused`).toMatchObject({ ok: true })
}
}
})
})
describe('a new client against an old host', () => {
it('finds no structured method registered on the old build', () => {
expect(baseline.methodNames.filter((name) => name.startsWith('agentSession.'))).toEqual([])
expect(current.methodNames.filter((name) => name.startsWith('agentSession.'))).toHaveLength(
10
)
})
it('can detect the absence during negotiation instead of by calling', () => {
expect(current.capabilities).toContain(STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY)
expect(baseline.capabilities).not.toContain(STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY)
// Additive surface: bumping the protocol number would strand every paired
// device on this release rather than degrade one feature.
expect(current.protocolVersion).toBe(baseline.protocolVersion)
})
it('gets a clean method_not_found from the old dispatcher rather than silence', async () => {
for (const { method } of STRUCTURED_CALLS) {
const replies = await callBuild(baseline, method, paramsFor(method), {
clientKind: 'mobile',
clientCapabilities: current.capabilities
})
expect(replies, `${method} must answer exactly once`).toHaveLength(1)
expect(replies[0], `${method} on the old host`).toMatchObject({
ok: false,
error: { code: 'method_not_found' }
})
}
})
})
describe('a cursor across a host restart', () => {
let root: string
let store: AgentSessionRecordStore
let runtime: unknown
/** Phase 2 owns provider processes; the adapter is the only stub here. */
function adapter(): StructuredAgentSessionAdapter {
return {
acquire: async ({ fence }) => ({
process: {
hostId: 'local',
pid: 4242,
processStartTimeMs: 1_700_000_000_000,
spawnToken: store.getRecord(SESSION)?.lease.reservedSpawnToken ?? 'spawn-a'
},
link: {
linkId: `link-${fence}`,
handle: { provider: 'codex', threadId: THREAD },
// A restarted host re-proves the thread it inherited; only the first
// owner of a session may claim to have created it.
origin: store.getRecord(SESSION)?.providerHandleChain.length ? 'resumed' : 'created',
mintedAtFence: fence,
observedAt: NOW
}
}),
dispatch: async () => ({
state: 'accepted',
providerIdentity: { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 1 }
}),
cancelTurn: async () => ({ cancelled: true }),
answerPrompt: async () => undefined,
setOption: async () => undefined
}
}
/** Reopens the store from disk and installs a fresh host over the same journal
* root — what a process restart actually leaves behind. */
async function bootHost(generation: string): Promise<void> {
store = await AgentSessionRecordStore.open({
directory: join(root, 'store'),
hostId: 'local'
})
setStructuredAgentSessionHost(
new StructuredAgentSessionHost({
store,
adapter: adapter(),
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken: () => `spawn-${generation}`,
// The provider died with the host that spawned it, which is what makes
// the restarted host the legitimate next writer.
probeOwner: async () => ({ outcome: 'pid-absent' }),
now: () => NOW
})
)
}
type HostAnswer = {
ok: boolean
fence: number
cursor: { epoch: string; sequence: number }
refusal?: { code: string; currentFence?: number }
}
/** Reattaching after a restart: the client's fence died with the previous
* host, and the refusal that says so is what hands it the live one. */
async function reattach(staleFence: number): Promise<HostAnswer> {
const refused = await answer('agentSession.ensure', attachParams(staleFence))
expect(refused).toMatchObject({
ok: false,
refusal: { code: 'agent_session_checkpoint_stale' }
})
const currentFence = refused.refusal?.currentFence
expect(currentFence).toBeGreaterThan(staleFence)
const reattached = await answer('agentSession.ensure', attachParams(currentFence ?? 0))
expect(reattached).toMatchObject({ ok: true })
return reattached
}
async function call(method: string, params: unknown): Promise<RpcReply[]> {
return callBuild(
current,
method,
params,
{
clientKind: 'mobile',
clientCapabilities: [STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY],
clientId: 'paired-device-1',
connectionId: 'connection-1'
},
runtime
)
}
/** The host's own answer, which carries its refusals inside a successful RPC. */
async function answer(method: string, params: unknown): Promise<HostAnswer> {
const reply = (await call(method, params))[0]
if (!reply?.ok) {
throw new Error(`${method} failed at the wire: ${JSON.stringify(reply?.error ?? reply)}`)
}
return reply.result as HostAnswer
}
beforeEach(async () => {
operations = 0
root = await mkdtemp(join(tmpdir(), 'orca-cross-version-agent-session-'))
runtime = runtimeStub()
await bootHost('a')
})
afterEach(async () => {
setStructuredAgentSessionHost(null)
await rm(root, { recursive: true, force: true })
})
it('resumes from the cursor the client held, with no snapshot and no replay', async () => {
const created = await answer('agentSession.create', attachParams(null))
expect(created.ok).toBe(true)
const first = await answer('agentSession.send', sendParams('before restart', created.fence))
expect(first.ok).toBe(true)
const held = first.cursor
await bootHost('b')
// A restart is a new writer generation; the old fence must not survive it.
const reattached = await reattach(created.fence)
expect(reattached.fence).toBeGreaterThan(created.fence)
const second = await answer(
'agentSession.send',
sendParams('after restart', reattached.fence)
)
expect(second.ok).toBe(true)
const events = (
await call('agentSession.subscribe', { sessionId: SESSION, cursor: held })
).map((reply) => reply.result as AgentSessionSubscribeEvent)
expect(events.map((event) => event.type)).toEqual(['batch'])
const batch = events[0]?.type === 'batch' ? events[0].batch : null
const rendered = JSON.stringify(batch?.items ?? [])
expect(rendered).toContain('after restart')
// Everything the client already had stays out of the resume.
expect(rendered).not.toContain('before restart')
expect(batch?.cursor.epoch).toBe(held.epoch)
expect(batch?.cursor.sequence).toBeGreaterThan(held.sequence)
})
it('refuses a write still fenced to the host generation that died', async () => {
const created = await answer('agentSession.create', attachParams(null))
await bootHost('b')
const reattached = await reattach(created.fence)
expect(reattached.fence).toBeGreaterThan(created.fence)
expect(await answer('agentSession.send', sendParams('stale', created.fence))).toMatchObject({
ok: false,
refusal: { code: 'agent_session_checkpoint_stale' }
})
})
})
})
@@ -0,0 +1,149 @@
import { readFileSync, readdirSync } from 'node:fs'
import { join } from 'node:path'
import { materializeReleaseCheckout, REPO_ROOT, type ReleaseCheckout } from './release-checkout'
/**
* The two things that decide whether a structured agent session exists for a given
* pairing: the capability strings a build can name, and the RPC methods it
* registers. Both are read per build, so "the old side does not have it" is a fact
* about a real release rather than a hand-written list.
*/
export const WORKING_TREE = 'working-tree' as const
export type RpcReply = {
id: string
ok: boolean
streaming?: true
result?: unknown
error?: { code: string; message: string }
}
export type RpcClientIdentity = {
clientKind?: 'mobile' | 'runtime'
clientCapabilities?: readonly string[]
connectionId?: string
clientId?: string
}
export type AgentSessionDispatcher = {
dispatchStreaming: (
request: { id: string; authToken: string; method: string; params?: unknown },
reply: (message: string) => void,
options?: RpcClientIdentity
) => Promise<void>
}
export type AgentSessionWireBuild = {
/** Human label used in test names and failure messages. */
label: string
/** `working-tree` for current code, otherwise the resolved release commit. */
revision: string
/** Capability strings this build defines. A peer cannot advertise — nor a client
* ask for — a string its own source never names. */
capabilities: readonly string[]
protocolVersion: number
/** RPC method names the build registers, read from source. */
methodNames: readonly string[]
/** A dispatcher carrying a method set this build really ships, so an
* unknown-method answer is about the method and not an empty registry. */
createDispatcher: (runtime: unknown) => AgentSessionDispatcher
}
type DispatcherModule = {
RpcDispatcher: new (options: { runtime: unknown; methods: unknown[] }) => AgentSessionDispatcher
}
// A dotted literal in a `name:` position. Deliberately loose: over-matching only
// makes "this build registers no agentSession method" a stronger claim.
const METHOD_NAME = /\bname:\s*'([A-Za-z][A-Za-z0-9]*(?:\.[A-Za-z0-9]+)+)'/g
/**
* Method names declared under `runtime/rpc/methods`, scanned rather than imported:
* a released build's method manifest reaches Electron, which cannot load here.
*/
function scanMethodNames(root: string): string[] {
const names = new Set<string>()
const walk = (directory: string): void => {
for (const entry of readdirSync(directory, { withFileTypes: true })) {
const full = join(directory, entry.name)
if (entry.isDirectory()) {
walk(full)
} else if (entry.isFile() && entry.name.endsWith('.ts')) {
for (const match of readFileSync(full, 'utf8').matchAll(METHOD_NAME)) {
names.add(match[1]!)
}
}
}
}
walk(join(root, 'src', 'main', 'runtime', 'rpc', 'methods'))
return [...names].sort()
}
function capabilityStrings(module: Record<string, unknown>): readonly string[] {
const declared = module.RUNTIME_CAPABILITIES
if (!Array.isArray(declared) || declared.length === 0) {
throw new Error('Cross-version harness found no RUNTIME_CAPABILITIES to compare')
}
return declared as readonly string[]
}
async function loadWorkingTreeBuild(): Promise<AgentSessionWireBuild> {
const [protocol, dispatcher, structured] = await Promise.all([
import('../../../src/shared/protocol-version'),
import('../../../src/main/runtime/rpc/dispatcher'),
import('../../../src/main/runtime/rpc/methods/structured-agent-session')
])
const module = dispatcher as unknown as DispatcherModule
return {
label: WORKING_TREE,
revision: WORKING_TREE,
capabilities: capabilityStrings(protocol as unknown as Record<string, unknown>),
protocolVersion: protocol.RUNTIME_PROTOCOL_VERSION,
methodNames: scanMethodNames(REPO_ROOT),
createDispatcher: (runtime) =>
new module.RpcDispatcher({
runtime,
methods: structured.STRUCTURED_AGENT_SESSION_METHODS as unknown[]
})
}
}
// Why @vite-ignore: the checkout is created at run time, so Vite cannot glob it at
// transform time. Vite-node still resolves and transforms the target on demand.
function importFromCheckout(specifier: string): Promise<Record<string, unknown>> {
return import(/* @vite-ignore */ specifier) as Promise<Record<string, unknown>>
}
async function loadReleaseBuild(checkout: ReleaseCheckout): Promise<AgentSessionWireBuild> {
const base = `${checkout.root}/src`
const [protocol, dispatcher, terminalMethods] = await Promise.all([
importFromCheckout(`${base}/shared/protocol-version.ts`),
importFromCheckout(`${base}/main/runtime/rpc/dispatcher.ts`),
importFromCheckout(`${base}/main/runtime/rpc/methods/terminal.ts`)
])
const module = dispatcher as unknown as DispatcherModule
return {
label: checkout.ref,
revision: checkout.commit,
capabilities: capabilityStrings(protocol),
protocolVersion: protocol.RUNTIME_PROTOCOL_VERSION as number,
methodNames: scanMethodNames(checkout.root),
createDispatcher: (runtime) =>
new module.RpcDispatcher({
runtime,
methods: terminalMethods.TERMINAL_METHODS as unknown[]
})
}
}
/**
* Load the structured-session wire surface for one build. `WORKING_TREE` imports
* current source; any other value is a git ref extracted into a cached checkout.
*/
export async function loadAgentSessionWireBuild(ref: string): Promise<AgentSessionWireBuild> {
if (ref === WORKING_TREE) {
return loadWorkingTreeBuild()
}
return loadReleaseBuild(materializeReleaseCheckout(ref))
}