Files
orca/tests/e2e/cross-version-wire/versioned-agent-session-wire.ts
T
Brennan Benson 280ef3c6f3 fix(native-chat): accept sends through their own receipt (#26672)
* feat(native-chat): add a chat-owned command receipt store

Adds the agent_session_command_receipts table and an insert-if-absent
receipt write that runs inside a command's own journal transaction. A
receipt records the command's identity, canonical hash, status and a
pointer to the row holding its result; it dies with its chat record.
Nothing uses it yet: the existing operation ledger stays authoritative.

* fix(native-chat): key command receipts like today's ledger ids

Rows are keyed by caller and operation id; a send looks across callers and
other methods look up only their caller's row, matching the existing ledger.
Bad rows and newer table shapes read as unreadable instead of throwing.

* fix(native-chat): order receipt lookups by insertion and key by operation id

Global lookups now take rows in insertion order, so a read and an insert
agree on which receipt answers an id regardless of the host clock. The
primary key leads with the operation id, so its own index serves both
lookups.

* fix(native-chat): accept sends through their own receipt

Send (direct, queued, steer) and /compact no longer commit a separate
ledger reservation before the message is saved. Their command receipt is
inserted-if-absent in the same transaction as the submission or queued
draft; a repeated id replays from it, a changed request conflicts.
Operation ids are checked for syntax only, so age, clock skew, pruning
and the reservation's own storage can no longer refuse a send.

The ledger success row they already co-committed stays as a temporary
insert-if-absent write so families not yet switched keep seeing these
ids; it is never read on this path.

* fix(native-chat): answer a send's receipt one way, share /compact's id space with Send, wake delivery once

- After its commit, a send whose publication fails answers through the same receipt replay a retry takes, so one id gets one answer; the in-memory copy and journal reload are gone.
- /compact ids are global like Send's: a retry under a new connection replays instead of needing a second acknowledgment write.
- The receipt route calls its runner directly; the runner wakes delivery when a new message row commits.
- The compatibility ledger row expires from host time; each temporary site is labelled.
- Receipt read errors and post-commit failures are logged.

* test(native-chat): expect receipt answers where tests assumed the send ledger

Restart-continuation tests now model the case Send cannot vouch for (saved but
not loaded into memory) and keep it unconfirmed; a later throw after the message
is in the chat answers continued, as a resend would. Drop /compact's unreachable
legacy replay branch and fix stale comments.

* test: run the queued tests that open the journal database in the Node runtime project
2026-10-09 17:25:43 -07:00

256 lines
10 KiB
TypeScript

import type * as ClientReducer from '../../../src/shared/structured-agent-session-reducer'
import type * as ClientProjection from '../../../src/shared/structured-agent-session-projection'
import type * as ClientSchemas from '../../../src/shared/agent-session-journal-schemas'
import type { sendPlan } from '../../../src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans'
import {
importReleaseCheckoutModule,
materializeReleaseCheckout,
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 AgentSessionClientProjection = Pick<
typeof ClientReducer,
'EMPTY_STRUCTURED_AGENT_SESSION' | 'reduceStructuredAgentSession'
> &
Pick<typeof ClientProjection, 'projectStructuredItemsToNativeChat'> &
Pick<typeof ClientSchemas, 'AgentJournalRenderItemSchema'>
/** Each build owns its own copy of the module-level host slot, so a host installed
* in current source is invisible to a release checkout's dispatcher. */
const STRUCTURED_HOST_REGISTRY =
'/src/main/native-chat/agent-session-wire/structured-agent-session-registry.ts'
const MUTATION_PLANS =
'/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts'
const MUTATION_ADMISSION =
'/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts'
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[]
updateClientCapabilities?: (capabilities: readonly string[]) => void
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
/** Put a host in *this* build's slot. Loaded on call so a release that predates
* the surface stays loadable, and throws rather than no-opping so a build with
* no slot cannot read as a surface that answered. */
installStructuredHost: (host: unknown) => Promise<void>
/** This build's own admission of the `agentSession.send` params its host was handed. With no
* journal it stops after the fingerprint check: a fingerprint it derives differently refuses
* as `fingerprintMismatch`, one it agrees with as `sessionNotAttached`. */
admitSend: (sent: SentMessage) => Promise<unknown>
clientProjection: () => Promise<AgentSessionClientProjection>
}
async function loadClientProjection(
load: (path: string) => Promise<Record<string, unknown>>
): Promise<AgentSessionClientProjection> {
const modules = await Promise.all([
load('/src/shared/structured-agent-session-reducer.ts'),
load('/src/shared/structured-agent-session-projection.ts'),
load('/src/shared/agent-session-journal-schemas.ts')
])
const client = Object.assign({}, ...modules)
if (
typeof client.reduceStructuredAgentSession !== 'function' ||
typeof client.projectStructuredItemsToNativeChat !== 'function' ||
!client.EMPTY_STRUCTURED_AGENT_SESSION ||
typeof client.AgentJournalRenderItemSchema?.parse !== 'function'
) {
throw new Error('Release does not export the structured transcript reader')
}
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: checked release exports; skew tests exercise admission, reducer and transcript signatures against real host frames.
return client as AgentSessionClientProjection
}
type DispatcherModule = {
RpcDispatcher: new (options: { runtime: unknown; methods: unknown[] }) => AgentSessionDispatcher
}
function registeredMethodNames(methods: readonly unknown[]): string[] {
return methods
.flatMap((method) => {
if (!method || typeof method !== 'object' || !('name' in method)) {
return []
}
const { name } = method
return typeof name === 'string' ? [name] : []
})
.sort()
}
function applyStructuredHost(module: Record<string, unknown>, label: string, host: unknown): void {
const install = module.setStructuredAgentSessionHost
if (typeof install !== 'function') {
throw new Error(`Build ${label} publishes no structured agent-session host registry`)
}
;(install as (next: unknown) => void)(host)
}
/** The `agentSession.send` params a host is handed. */
export type SentMessage = Parameters<typeof sendPlan>[0]
type SendAdmissionModules = {
sendPlan: (sent: SentMessage) => unknown
admitAndRunAgentSessionMutation: (request: {
plan: unknown
envelope: SentMessage['envelope']
journal: () => undefined
store: { getRecord: () => null }
}) => Promise<unknown>
}
async function admitSend(
plans: Record<string, unknown>,
admission: Record<string, unknown>,
sent: SentMessage
): Promise<unknown> {
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the build's own send plan and admission; the absent session and journal stop both admission routes after fingerprint validation. A drifted export or shape fails this check.
const build = { ...plans, ...admission } as unknown as SendAdmissionModules
return build.admitAndRunAgentSessionMutation({
plan: build.sendPlan(sent),
envelope: sent.envelope,
journal: () => undefined,
store: { getRecord: () => null }
})
}
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, methodRegistry] = await Promise.all([
import('../../../src/shared/protocol-version'),
import('../../../src/main/runtime/rpc/dispatcher'),
import('../../../src/main/runtime/rpc/methods')
])
const module = dispatcher as unknown as DispatcherModule
const methods = methodRegistry.ALL_RPC_METHODS as unknown[]
return {
label: WORKING_TREE,
revision: WORKING_TREE,
capabilities: capabilityStrings(protocol as unknown as Record<string, unknown>),
protocolVersion: protocol.RUNTIME_PROTOCOL_VERSION,
methodNames: registeredMethodNames(methods),
clientProjection: async () => {
const [reducer, projection, schemas] = await Promise.all([
import('../../../src/shared/structured-agent-session-reducer'),
import('../../../src/shared/structured-agent-session-projection'),
import('../../../src/shared/agent-session-journal-schemas')
])
return { ...reducer, ...projection, ...schemas }
},
createDispatcher: (runtime) =>
new module.RpcDispatcher({
runtime,
methods
}),
installStructuredHost: async (host) => {
const registry =
await import('../../../src/main/native-chat/agent-session-wire/structured-agent-session-registry')
applyStructuredHost(registry as unknown as Record<string, unknown>, WORKING_TREE, host)
},
admitSend: async (sent) => {
const [plans, admission] = await Promise.all([
import('../../../src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans'),
import('../../../src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission')
])
return admitSend(plans, admission, sent)
}
}
}
async function loadReleaseBuild(checkout: ReleaseCheckout): Promise<AgentSessionWireBuild> {
const [protocol, dispatcher, methodRegistry] = await Promise.all([
importReleaseCheckoutModule(checkout, '/src/shared/protocol-version.ts'),
importReleaseCheckoutModule(checkout, '/src/main/runtime/rpc/dispatcher.ts'),
importReleaseCheckoutModule(checkout, '/src/main/runtime/rpc/methods/index.ts')
])
const module = dispatcher as unknown as DispatcherModule
const methods = methodRegistry.ALL_RPC_METHODS as unknown[]
return {
label: checkout.ref,
revision: checkout.commit,
capabilities: capabilityStrings(protocol),
protocolVersion: protocol.RUNTIME_PROTOCOL_VERSION as number,
methodNames: registeredMethodNames(methods),
clientProjection: () =>
loadClientProjection((path) => importReleaseCheckoutModule(checkout, path)),
createDispatcher: (runtime) =>
new module.RpcDispatcher({
runtime,
methods
}),
installStructuredHost: async (host) => {
applyStructuredHost(
await importReleaseCheckoutModule(checkout, STRUCTURED_HOST_REGISTRY),
checkout.ref,
host
)
},
admitSend: async (sent) => {
const [plans, admission] = await Promise.all([
importReleaseCheckoutModule(checkout, MUTATION_PLANS),
importReleaseCheckoutModule(checkout, MUTATION_ADMISSION)
])
return admitSend(plans, admission, sent)
}
}
}
/**
* 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(await materializeReleaseCheckout(ref))
}