diff --git a/docs/reference/remote-wire-compatibility.md b/docs/reference/remote-wire-compatibility.md index 6ca32f49c48..544a5d45842 100644 --- a/docs/reference/remote-wire-compatibility.md +++ b/docs/reference/remote-wire-compatibility.md @@ -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. diff --git a/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts b/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts new file mode 100644 index 00000000000..e9cc27678c0 --- /dev/null +++ b/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts @@ -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 + fence: number | null +}): Record { + return { + sessionId: SESSION, + clientOperationId: operationId(), + expectedRuntimeFence: args.fence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: args.method, + sessionId: SESSION, + fields: args.fields + }) + } +} + +function attachParams(fence: number | null): Record { + 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 { + 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 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 { + 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> + + 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 { + 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 { + 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 { + 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 { + 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' } + }) + }) + }) +}) diff --git a/tests/e2e/cross-version-wire/versioned-agent-session-wire.ts b/tests/e2e/cross-version-wire/versioned-agent-session-wire.ts new file mode 100644 index 00000000000..1b2eb3dae97 --- /dev/null +++ b/tests/e2e/cross-version-wire/versioned-agent-session-wire.ts @@ -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 +} + +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() + 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): 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 { + 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), + 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> { + return import(/* @vite-ignore */ specifier) as Promise> +} + +async function loadReleaseBuild(checkout: ReleaseCheckout): Promise { + 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 { + if (ref === WORKING_TREE) { + return loadWorkingTreeBuild() + } + return loadReleaseBuild(materializeReleaseCheckout(ref)) +}