orca/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts

// 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' }
      })
    })
  })
})