mirror of
https://github.com/stablyai/orca.git
synced 2026-10-27 00:02:48 +00:00
Merge branch 'brennanb2025/native-chat-1a-session-store' into brennanb2025/native-chat-structured-base
This commit is contained in:
Brennan Benson
@@ -0,0 +1,314 @@
|
||||
/**
|
||||
* Pure lease state transitions. Every function returns the next record or throws a typed error;
|
||||
* the store applies them inside one durable transaction so a rejected transition never lands.
|
||||
*
|
||||
* The invariant they exist to enforce: a session admits a writer only after a reservation, an
|
||||
* observed process identity, and a proved provider handle — in that order, at one fence.
|
||||
*/
|
||||
|
||||
import { isDeepStrictEqual } from 'node:util'
|
||||
import {
|
||||
adjudicateAgentSessionRestart,
|
||||
evaluateAgentSessionAcquisition,
|
||||
type AgentSessionOwnerProbe
|
||||
} from '../../shared/agent-session-lease-adjudication'
|
||||
import {
|
||||
appendAgentSessionProviderHandleLink,
|
||||
type AgentSessionProviderHandleLink
|
||||
} from '../../shared/agent-session-provider-handle'
|
||||
import type {
|
||||
AgentSessionJournalCheckpoint,
|
||||
AgentSessionHandoffStage,
|
||||
AgentSessionLease,
|
||||
AgentSessionOwnerRuntimeKind,
|
||||
AgentSessionProcessIdentity,
|
||||
AgentSessionRecord
|
||||
} from '../../shared/agent-session-record'
|
||||
|
||||
export type AgentSessionReservation = {
|
||||
runtimeKind: AgentSessionOwnerRuntimeKind
|
||||
spawnToken: string
|
||||
claimKeyId: string
|
||||
handoffOperationId: string | null
|
||||
leaseTtlMs: number
|
||||
now: number
|
||||
}
|
||||
|
||||
function withLease(record: AgentSessionRecord, lease: AgentSessionLease): AgentSessionRecord {
|
||||
return { ...record, lease, updatedAt: lease.lastRenewedAt }
|
||||
}
|
||||
|
||||
function assertFence(lease: AgentSessionLease, fence: number): void {
|
||||
if (lease.runtimeFence !== fence) {
|
||||
throw new Error('agent_session_checkpoint_stale')
|
||||
}
|
||||
if (lease.unreconciled) {
|
||||
throw new Error('execution_owner_reconciling')
|
||||
}
|
||||
}
|
||||
|
||||
export function agentSessionReconciliationTargetMatches(
|
||||
current: AgentSessionRecord,
|
||||
probed: AgentSessionRecord
|
||||
): boolean {
|
||||
return isDeepStrictEqual(current, probed)
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare-and-swap reservation. Writes the intent at fence + 1 before any process exists, so the
|
||||
* loser of a concurrent swap is refused and never spawns.
|
||||
*/
|
||||
export function reserveAgentSessionOwner(args: {
|
||||
record: AgentSessionRecord
|
||||
expectedFence: number
|
||||
probe: AgentSessionOwnerProbe
|
||||
reservation: AgentSessionReservation
|
||||
}): { record: AgentSessionRecord; disposition: 'reserved' | 'retry-reservation' } {
|
||||
const { record, reservation } = args
|
||||
const decision = evaluateAgentSessionAcquisition({
|
||||
lease: record.lease,
|
||||
expectedFence: args.expectedFence,
|
||||
handoffOperationId: reservation.handoffOperationId,
|
||||
probe: args.probe
|
||||
})
|
||||
if (decision.decision === 'refused') {
|
||||
throw new Error(decision.code)
|
||||
}
|
||||
if (decision.decision === 'retry-reservation') {
|
||||
return { record, disposition: 'retry-reservation' }
|
||||
}
|
||||
return {
|
||||
disposition: 'reserved',
|
||||
record: withLease(record, {
|
||||
...record.lease,
|
||||
runtimeKind: reservation.runtimeKind,
|
||||
runtimeFence: decision.nextFence,
|
||||
// Why: a reserved owner is not yet a writer; it may only talk to the provider to prove resume.
|
||||
handoffStage: 'new-owner-proving',
|
||||
provenHandleLinkId: null,
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: reservation.spawnToken,
|
||||
leaseDeadlineAt: reservation.now + reservation.leaseTtlMs,
|
||||
lastRenewedAt: reservation.now,
|
||||
handoffOperationId: reservation.handoffOperationId,
|
||||
claimKeyId: reservation.claimKeyId,
|
||||
claimStatus: 'reserved',
|
||||
deathEvidence: null
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/** Step 4 of acquisition: write the observed identity back into the same lease row. */
|
||||
export function commitAgentSessionProcessIdentity(args: {
|
||||
record: AgentSessionRecord
|
||||
fence: number
|
||||
process: AgentSessionProcessIdentity
|
||||
now: number
|
||||
}): AgentSessionRecord {
|
||||
const { record } = args
|
||||
assertFence(record.lease, args.fence)
|
||||
if (record.lease.claimStatus !== 'reserved' || record.lease.ownerProcess !== null) {
|
||||
throw new Error('agent_session_ownership_unknown')
|
||||
}
|
||||
if (record.lease.reservedSpawnToken !== args.process.spawnToken) {
|
||||
// Why: a child that cannot echo the reserved token is not the process Orca started.
|
||||
throw new Error('agent_session_ownership_unknown')
|
||||
}
|
||||
return withLease(record, { ...record.lease, ownerProcess: args.process, lastRenewedAt: args.now })
|
||||
}
|
||||
|
||||
/**
|
||||
* The new runtime proved it resumed the expected provider handle. Only now does the session have
|
||||
* a writer.
|
||||
*/
|
||||
export function proveAgentSessionOwner(args: {
|
||||
record: AgentSessionRecord
|
||||
fence: number
|
||||
link: AgentSessionProviderHandleLink
|
||||
now: number
|
||||
leaseTtlMs: number
|
||||
}): AgentSessionRecord {
|
||||
const { record } = args
|
||||
assertFence(record.lease, args.fence)
|
||||
if (
|
||||
record.lease.claimStatus !== 'reserved' ||
|
||||
record.lease.handoffStage !== 'new-owner-proving' ||
|
||||
record.lease.ownerProcess === null
|
||||
) {
|
||||
throw new Error('agent_session_ownership_unknown')
|
||||
}
|
||||
if (args.link.handle.provider !== record.provider) {
|
||||
throw new Error('agent_session_provider_handle_provider_mismatch')
|
||||
}
|
||||
if (args.link.mintedAtFence !== args.fence) {
|
||||
throw new Error('agent_session_provider_handle_stale_fence')
|
||||
}
|
||||
const providerHandleChain = appendAgentSessionProviderHandleLink(
|
||||
record.providerHandleChain,
|
||||
args.link
|
||||
)
|
||||
const head = providerHandleChain.at(-1)
|
||||
if (!head) {
|
||||
throw new Error('agent_session_provider_handle_invalid')
|
||||
}
|
||||
return {
|
||||
...record,
|
||||
providerHandleChain,
|
||||
lease: {
|
||||
...record.lease,
|
||||
handoffStage: null,
|
||||
provenHandleLinkId: head.linkId,
|
||||
claimStatus: 'live',
|
||||
leaseDeadlineAt: args.now + args.leaseTtlMs,
|
||||
lastRenewedAt: args.now,
|
||||
handoffOperationId: null
|
||||
},
|
||||
updatedAt: args.now
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A renewal asserts two things at once: the host is running its loop, and the child still matches
|
||||
* the recorded identity. A host that cannot re-verify the child stops renewing rather than
|
||||
* extending a lease it can no longer vouch for.
|
||||
*/
|
||||
export function renewAgentSessionLease(args: {
|
||||
record: AgentSessionRecord
|
||||
fence: number
|
||||
childProbe: AgentSessionOwnerProbe
|
||||
now: number
|
||||
leaseTtlMs: number
|
||||
}): AgentSessionRecord {
|
||||
const { record } = args
|
||||
assertFence(record.lease, args.fence)
|
||||
if (record.lease.ownerProcess === null) {
|
||||
throw new Error('agent_session_ownership_unknown')
|
||||
}
|
||||
if (args.childProbe.outcome !== 'identity-matched' || args.childProbe.matchedOn.length === 0) {
|
||||
throw new Error('agent_session_ownership_unknown')
|
||||
}
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
leaseDeadlineAt: args.now + args.leaseTtlMs,
|
||||
lastRenewedAt: args.now
|
||||
})
|
||||
}
|
||||
|
||||
/** Proven eviction — the only other thing besides acquisition that may move the fence. */
|
||||
export function evictAgentSessionOwner(args: {
|
||||
record: AgentSessionRecord
|
||||
expectedFence: number
|
||||
probe: AgentSessionOwnerProbe
|
||||
now: number
|
||||
}): AgentSessionRecord {
|
||||
const { record } = args
|
||||
assertFence(record.lease, args.expectedFence)
|
||||
const adjudication = adjudicateAgentSessionRestart({
|
||||
lease: record.lease,
|
||||
probe: args.probe,
|
||||
observedAt: args.now
|
||||
})
|
||||
if (adjudication.disposition !== 'evicted') {
|
||||
throw new Error('agent_session_ownership_unknown')
|
||||
}
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
runtimeFence: adjudication.nextFence,
|
||||
handoffStage: null,
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: null,
|
||||
claimStatus: 'released',
|
||||
lastRenewedAt: args.now,
|
||||
handoffOperationId: null,
|
||||
deathEvidence: adjudication.evidence
|
||||
})
|
||||
}
|
||||
|
||||
/** Apply one restart adjudication. Never consults deadlines — only proof moves a lease. */
|
||||
export function applyAgentSessionRestartAdjudication(args: {
|
||||
record: AgentSessionRecord
|
||||
probe: AgentSessionOwnerProbe
|
||||
now: number
|
||||
}): AgentSessionRecord {
|
||||
const { record } = args
|
||||
const adjudication = adjudicateAgentSessionRestart({
|
||||
lease: record.lease,
|
||||
probe: args.probe,
|
||||
observedAt: args.now
|
||||
})
|
||||
if (adjudication.disposition === 'readopt') {
|
||||
// Why: re-adoption is not a new generation, so the fence does not move.
|
||||
return withLease(record, { ...record.lease, unreconciled: false, lastRenewedAt: args.now })
|
||||
}
|
||||
if (adjudication.disposition === 'evicted') {
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
runtimeFence: adjudication.nextFence,
|
||||
handoffStage: null,
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: null,
|
||||
claimStatus: 'released',
|
||||
unreconciled: false,
|
||||
lastRenewedAt: args.now,
|
||||
handoffOperationId: null,
|
||||
deathEvidence: adjudication.evidence
|
||||
})
|
||||
}
|
||||
const stage: AgentSessionHandoffStage =
|
||||
adjudication.disposition === 'conflicted' ? 'manual-recovery' : adjudication.stage
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
handoffStage: stage,
|
||||
claimStatus:
|
||||
adjudication.disposition === 'conflicted' ? 'conflicted' : record.lease.claimStatus,
|
||||
unreconciled: false,
|
||||
lastRenewedAt: args.now
|
||||
})
|
||||
}
|
||||
|
||||
export function setAgentSessionHandoffStage(args: {
|
||||
record: AgentSessionRecord
|
||||
fence: number
|
||||
stage: AgentSessionHandoffStage | null
|
||||
handoffOperationId: string | null
|
||||
now: number
|
||||
}): AgentSessionRecord {
|
||||
const { record } = args
|
||||
assertFence(record.lease, args.fence)
|
||||
if (
|
||||
record.lease.handoffOperationId !== null &&
|
||||
args.handoffOperationId !== null &&
|
||||
args.handoffOperationId !== record.lease.handoffOperationId
|
||||
) {
|
||||
throw new Error('agent_session_operation_conflict')
|
||||
}
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
handoffStage: args.stage,
|
||||
handoffOperationId: args.handoffOperationId,
|
||||
lastRenewedAt: args.now
|
||||
})
|
||||
}
|
||||
|
||||
export function setAgentSessionJournalCheckpoint(args: {
|
||||
record: AgentSessionRecord
|
||||
fence: number
|
||||
checkpoint: AgentSessionJournalCheckpoint
|
||||
now: number
|
||||
}): AgentSessionRecord {
|
||||
const { record } = args
|
||||
assertFence(record.lease, args.fence)
|
||||
const current = record.lease.journalCheckpoint
|
||||
if (
|
||||
current &&
|
||||
(current.epoch > args.checkpoint.epoch ||
|
||||
(current.epoch === args.checkpoint.epoch && current.sequence > args.checkpoint.sequence))
|
||||
) {
|
||||
throw new Error('agent_session_checkpoint_stale')
|
||||
}
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
journalCheckpoint: args.checkpoint,
|
||||
lastRenewedAt: args.now
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,209 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record'
|
||||
import {
|
||||
PROCESS_START_TIME_TOLERANCE_MS,
|
||||
probeAgentSessionProcessIdentity,
|
||||
probeAgentSessionReservation,
|
||||
readProcessStartTimeMs,
|
||||
type AgentSessionProcessProbeDeps
|
||||
} from './agent-session-process-identity-probe'
|
||||
|
||||
const START_TIME = 1_700_000_000_000
|
||||
|
||||
const IDENTITY: AgentSessionProcessIdentity = {
|
||||
hostId: 'local',
|
||||
pid: 4242,
|
||||
processStartTimeMs: START_TIME,
|
||||
spawnToken: 'spawn-a'
|
||||
}
|
||||
|
||||
function deps(overrides: AgentSessionProcessProbeDeps = {}): AgentSessionProcessProbeDeps {
|
||||
return {
|
||||
isPidPresent: () => true,
|
||||
readProcessStartTimeMs: async () => START_TIME,
|
||||
readEchoedSpawnToken: async () => 'spawn-a',
|
||||
platform: 'linux',
|
||||
...overrides
|
||||
}
|
||||
}
|
||||
|
||||
describe('owner identity probe', () => {
|
||||
it('reports an observed exit without touching the host', () => {
|
||||
return expect(
|
||||
probeAgentSessionProcessIdentity({
|
||||
identity: IDENTITY,
|
||||
observedExit: true,
|
||||
deps: deps({
|
||||
isPidPresent: () => {
|
||||
throw new Error('must not probe after an observed exit')
|
||||
}
|
||||
})
|
||||
})
|
||||
).resolves.toEqual({ outcome: 'exit-observed' })
|
||||
})
|
||||
|
||||
it('reports an absent pid as proven death', async () => {
|
||||
await expect(
|
||||
probeAgentSessionProcessIdentity({
|
||||
identity: IDENTITY,
|
||||
deps: deps({ isPidPresent: () => false })
|
||||
})
|
||||
).resolves.toEqual({ outcome: 'pid-absent' })
|
||||
})
|
||||
|
||||
it('does not call an unexpected host probe error proof of death', async () => {
|
||||
const kill = vi.spyOn(process, 'kill').mockImplementation(() => {
|
||||
throw Object.assign(new Error('host probe unavailable'), { code: 'EIO' })
|
||||
})
|
||||
try {
|
||||
const probe = await probeAgentSessionProcessIdentity({
|
||||
identity: { ...IDENTITY, processStartTimeMs: null },
|
||||
deps: { readEchoedSpawnToken: async () => null }
|
||||
})
|
||||
expect(probe.outcome).toBe('indeterminate')
|
||||
} finally {
|
||||
kill.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it('catches pid reuse through the spawn token', async () => {
|
||||
// The pid is live and started at the recorded time, but it is a different process.
|
||||
await expect(
|
||||
probeAgentSessionProcessIdentity({
|
||||
identity: IDENTITY,
|
||||
deps: deps({ readEchoedSpawnToken: async () => 'spawn-other' })
|
||||
})
|
||||
).resolves.toEqual({ outcome: 'identity-mismatch', field: 'spawn-token' })
|
||||
})
|
||||
|
||||
it('catches pid reuse through the start time when no token comes back', async () => {
|
||||
await expect(
|
||||
probeAgentSessionProcessIdentity({
|
||||
identity: IDENTITY,
|
||||
deps: deps({
|
||||
readEchoedSpawnToken: async () => null,
|
||||
readProcessStartTimeMs: async () => START_TIME + PROCESS_START_TIME_TOLERANCE_MS + 1
|
||||
})
|
||||
})
|
||||
).resolves.toEqual({ outcome: 'identity-mismatch', field: 'process-start-time' })
|
||||
})
|
||||
|
||||
it('fails closed when an exact token and the reconstructed start time disagree', async () => {
|
||||
await expect(
|
||||
probeAgentSessionProcessIdentity({
|
||||
identity: IDENTITY,
|
||||
deps: deps({
|
||||
readProcessStartTimeMs: async () => START_TIME + PROCESS_START_TIME_TOLERANCE_MS + 1
|
||||
})
|
||||
})
|
||||
).resolves.toEqual({
|
||||
outcome: 'indeterminate',
|
||||
reason: 'process identity evidence contradicted'
|
||||
})
|
||||
})
|
||||
|
||||
it('tolerates start-time jitter inside the tolerance', async () => {
|
||||
await expect(
|
||||
probeAgentSessionProcessIdentity({
|
||||
identity: IDENTITY,
|
||||
deps: deps({
|
||||
readEchoedSpawnToken: async () => null,
|
||||
readProcessStartTimeMs: async () => START_TIME + PROCESS_START_TIME_TOLERANCE_MS
|
||||
})
|
||||
})
|
||||
).resolves.toEqual({ outcome: 'identity-matched', matchedOn: ['process-start-time'] })
|
||||
})
|
||||
|
||||
it('reports every element it could actually verify', async () => {
|
||||
await expect(
|
||||
probeAgentSessionProcessIdentity({ identity: IDENTITY, deps: deps() })
|
||||
).resolves.toEqual({
|
||||
outcome: 'identity-matched',
|
||||
matchedOn: ['spawn-token', 'process-start-time']
|
||||
})
|
||||
})
|
||||
|
||||
it('fails closed when the pid is live but nothing PID-reuse-safe could be checked', async () => {
|
||||
// The Windows case: no start time recorded and no token echo, so a bare pid match is all the
|
||||
// host has — and a bare pid match is what mints a second writer.
|
||||
const probe = await probeAgentSessionProcessIdentity({
|
||||
identity: { ...IDENTITY, processStartTimeMs: null },
|
||||
deps: deps({ readEchoedSpawnToken: async () => null, platform: 'win32' })
|
||||
})
|
||||
expect(probe.outcome).toBe('indeterminate')
|
||||
})
|
||||
|
||||
it('fails closed when the host errors instead of answering', async () => {
|
||||
const probe = await probeAgentSessionProcessIdentity({
|
||||
identity: { ...IDENTITY, processStartTimeMs: null },
|
||||
deps: deps({
|
||||
readEchoedSpawnToken: async () => {
|
||||
throw new Error('handshake unavailable')
|
||||
}
|
||||
})
|
||||
})
|
||||
expect(probe.outcome).toBe('indeterminate')
|
||||
})
|
||||
|
||||
it('still proves life from the token when the start time is unreadable', async () => {
|
||||
await expect(
|
||||
probeAgentSessionProcessIdentity({
|
||||
identity: IDENTITY,
|
||||
deps: deps({ readProcessStartTimeMs: async () => null })
|
||||
})
|
||||
).resolves.toEqual({ outcome: 'identity-matched', matchedOn: ['spawn-token'] })
|
||||
})
|
||||
|
||||
it('returns no start time on Windows rather than guessing one', async () => {
|
||||
await expect(readProcessStartTimeMs(process.pid, 'win32')).resolves.toBeNull()
|
||||
})
|
||||
|
||||
it('reads a start time for the current process on this platform', async () => {
|
||||
const observed = await readProcessStartTimeMs(process.pid)
|
||||
if (process.platform === 'linux' || process.platform === 'darwin') {
|
||||
expect(observed).not.toBeNull()
|
||||
expect(Math.abs((observed as number) - (Date.now() - process.uptime() * 1000))).toBeLessThan(
|
||||
60_000
|
||||
)
|
||||
} else {
|
||||
expect(observed).toBeNull()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('reservation probe', () => {
|
||||
it('declares a reservation unused only with positive proof nothing started', async () => {
|
||||
await expect(
|
||||
probeAgentSessionReservation({
|
||||
spawnToken: 'spawn-a',
|
||||
findProcessesWithSpawnToken: async () => [],
|
||||
hasProviderActivitySinceReservation: async () => false
|
||||
})
|
||||
).resolves.toEqual({ outcome: 'reservation-unused' })
|
||||
})
|
||||
|
||||
it.each([
|
||||
['a process still carries the token', async () => [999], async () => false],
|
||||
['the host cannot enumerate', async () => null, async () => false],
|
||||
['provider activity is unknown', async () => [], async () => null],
|
||||
['the provider saw activity', async () => [], async () => true]
|
||||
] as const)('stays indeterminate when %s', async (_name, findProcesses, hasActivity) => {
|
||||
const probe = await probeAgentSessionReservation({
|
||||
spawnToken: 'spawn-a',
|
||||
findProcessesWithSpawnToken: findProcesses,
|
||||
hasProviderActivitySinceReservation: hasActivity
|
||||
})
|
||||
expect(probe.outcome).toBe('indeterminate')
|
||||
})
|
||||
|
||||
it('stays indeterminate when enumeration throws', async () => {
|
||||
const probe = await probeAgentSessionReservation({
|
||||
spawnToken: 'spawn-a',
|
||||
findProcessesWithSpawnToken: async () => {
|
||||
throw new Error('ps unavailable')
|
||||
},
|
||||
hasProviderActivitySinceReservation: async () => false
|
||||
})
|
||||
expect(probe.outcome).toBe('indeterminate')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,174 @@
|
||||
/**
|
||||
* PID-reuse-safe process identity probe for the single-writer lease.
|
||||
*
|
||||
* Pids are reused within minutes on a busy host, and reuse happens precisely in the recovery
|
||||
* case, so a bare pid match is never proof. Every element of the identity tuple is unavailable
|
||||
* somewhere — start time costs a CIM query on Windows and is missing in some containers, /proc
|
||||
* does not exist on macOS — so an unanswerable probe reports `indeterminate` and the lease fails
|
||||
* closed to manual recovery rather than minting a second writer.
|
||||
*/
|
||||
|
||||
import { execFile } from 'node:child_process'
|
||||
import { readFile } from 'node:fs/promises'
|
||||
import { promisify } from 'node:util'
|
||||
import type {
|
||||
AgentSessionIdentityMatchField,
|
||||
AgentSessionOwnerProbe
|
||||
} from '../../shared/agent-session-lease-adjudication'
|
||||
import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record'
|
||||
|
||||
const execFileAsync = promisify(execFile)
|
||||
|
||||
/** Start times drift by scheduler granularity and clock reads; compare with a tolerance. */
|
||||
export const PROCESS_START_TIME_TOLERANCE_MS = 2_000
|
||||
|
||||
const PROCESS_START_TIME_TIMEOUT_MS = 5_000
|
||||
|
||||
export type AgentSessionProcessProbeDeps = {
|
||||
/** ESRCH means gone; EPERM means present but owned by another user. */
|
||||
isPidPresent?: (pid: number) => boolean
|
||||
readProcessStartTimeMs?: (pid: number, platform?: NodeJS.Platform) => Promise<number | null>
|
||||
/** Token the running child echoed back through the adapter handshake or provider hook. */
|
||||
readEchoedSpawnToken?: (identity: AgentSessionProcessIdentity) => Promise<string | null>
|
||||
platform?: NodeJS.Platform
|
||||
}
|
||||
|
||||
function defaultIsPidPresent(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
return true
|
||||
} catch (error) {
|
||||
// Why: only ESRCH proves absence; permission and transient host failures must fail closed.
|
||||
return (error as NodeJS.ErrnoException)?.code !== 'ESRCH'
|
||||
}
|
||||
}
|
||||
|
||||
async function readLinuxProcessStartTimeMs(pid: number): Promise<number | null> {
|
||||
try {
|
||||
const [stat, systemStat] = await Promise.all([
|
||||
readFile(`/proc/${pid}/stat`, 'utf-8'),
|
||||
readFile('/proc/stat', 'utf-8')
|
||||
])
|
||||
// Field 22 is starttime in clock ticks; the comm field can contain spaces, so cut past ") ".
|
||||
const fields = stat.slice(stat.lastIndexOf(') ') + 2).split(' ')
|
||||
const ticks = Number(fields[19])
|
||||
const bootTimeSeconds = Number(/^btime\s+(\d+)$/m.exec(systemStat)?.[1])
|
||||
if (!Number.isFinite(ticks) || !Number.isFinite(bootTimeSeconds)) {
|
||||
return null
|
||||
}
|
||||
return Math.round(bootTimeSeconds * 1000 + (ticks / 100) * 1000)
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
async function readDarwinProcessStartTimeMs(pid: number): Promise<number | null> {
|
||||
try {
|
||||
const { stdout } = await execFileAsync('ps', ['-o', 'lstart=', '-p', String(pid)], {
|
||||
timeout: PROCESS_START_TIME_TIMEOUT_MS
|
||||
})
|
||||
const parsed = Date.parse(stdout.trim())
|
||||
return Number.isFinite(parsed) ? parsed : null
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Windows returns null: reading a start time needs a CIM query Orca does not make here, so a
|
||||
* Windows child without a token echo qualifies for manual recovery only.
|
||||
*/
|
||||
export async function readProcessStartTimeMs(
|
||||
pid: number,
|
||||
platform: NodeJS.Platform = process.platform
|
||||
): Promise<number | null> {
|
||||
if (platform === 'linux') {
|
||||
return readLinuxProcessStartTimeMs(pid)
|
||||
}
|
||||
if (platform === 'darwin') {
|
||||
return readDarwinProcessStartTimeMs(pid)
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Probe one recorded owner. `observedExit` short-circuits everything: Orca watching that exact
|
||||
* process exit is the strongest evidence available.
|
||||
*/
|
||||
export async function probeAgentSessionProcessIdentity(args: {
|
||||
identity: AgentSessionProcessIdentity
|
||||
observedExit?: boolean
|
||||
deps?: AgentSessionProcessProbeDeps
|
||||
}): Promise<AgentSessionOwnerProbe> {
|
||||
const { identity } = args
|
||||
const deps = args.deps ?? {}
|
||||
if (args.observedExit) {
|
||||
return { outcome: 'exit-observed' }
|
||||
}
|
||||
const isPidPresent = deps.isPidPresent ?? defaultIsPidPresent
|
||||
if (!isPidPresent(identity.pid)) {
|
||||
return { outcome: 'pid-absent' }
|
||||
}
|
||||
const matchedOn: AgentSessionIdentityMatchField[] = []
|
||||
const echoedToken = await deps.readEchoedSpawnToken?.(identity).catch(() => null)
|
||||
if (echoedToken !== null && echoedToken !== undefined) {
|
||||
if (echoedToken !== identity.spawnToken) {
|
||||
return { outcome: 'identity-mismatch', field: 'spawn-token' }
|
||||
}
|
||||
matchedOn.push('spawn-token')
|
||||
}
|
||||
if (identity.processStartTimeMs !== null) {
|
||||
const readStartTime = deps.readProcessStartTimeMs ?? readProcessStartTimeMs
|
||||
const observed = await readStartTime(identity.pid, deps.platform ?? process.platform).catch(
|
||||
() => null
|
||||
)
|
||||
if (observed !== null) {
|
||||
if (Math.abs(observed - identity.processStartTimeMs) > PROCESS_START_TIME_TOLERANCE_MS) {
|
||||
if (matchedOn.includes('spawn-token')) {
|
||||
// Why: contradictory evidence cannot prove that a token-authenticated child is dead.
|
||||
return { outcome: 'indeterminate', reason: 'process identity evidence contradicted' }
|
||||
}
|
||||
return { outcome: 'identity-mismatch', field: 'process-start-time' }
|
||||
}
|
||||
matchedOn.push('process-start-time')
|
||||
}
|
||||
}
|
||||
if (matchedOn.length === 0) {
|
||||
// Why: the pid exists and nothing PID-reuse-safe could be checked. Reporting a match here is
|
||||
// exactly the case that produces two writers on one provider session.
|
||||
return {
|
||||
outcome: 'indeterminate',
|
||||
reason: 'pid present but neither spawn token nor start time could be verified'
|
||||
}
|
||||
}
|
||||
return { outcome: 'identity-matched', matchedOn }
|
||||
}
|
||||
|
||||
/**
|
||||
* Probe a reservation that has no recorded process. `reservation-unused` requires positive proof
|
||||
* that nothing started — no process carrying the token and no provider-side activity after the
|
||||
* reservation — not an assumption that the crash beat the spawn.
|
||||
*/
|
||||
export async function probeAgentSessionReservation(args: {
|
||||
spawnToken: string
|
||||
findProcessesWithSpawnToken: (spawnToken: string) => Promise<number[] | null>
|
||||
hasProviderActivitySinceReservation: () => Promise<boolean | null>
|
||||
}): Promise<AgentSessionOwnerProbe> {
|
||||
const pids = await args.findProcessesWithSpawnToken(args.spawnToken).catch(() => null)
|
||||
if (pids === null) {
|
||||
return { outcome: 'indeterminate', reason: 'host could not enumerate spawn tokens' }
|
||||
}
|
||||
if (pids.length > 0) {
|
||||
return {
|
||||
outcome: 'indeterminate',
|
||||
reason: `reservation spawn token is live on ${pids.length} process(es)`
|
||||
}
|
||||
}
|
||||
const providerActivity = await args.hasProviderActivitySinceReservation().catch(() => null)
|
||||
if (providerActivity === null) {
|
||||
return { outcome: 'indeterminate', reason: 'provider activity since reservation is unknown' }
|
||||
}
|
||||
return providerActivity
|
||||
? { outcome: 'indeterminate', reason: 'provider saw activity after the reservation' }
|
||||
: { outcome: 'reservation-unused' }
|
||||
}
|
||||
@@ -0,0 +1,246 @@
|
||||
/**
|
||||
* On-disk layer for the durable agent-session store.
|
||||
*
|
||||
* Every mutation is a whole-file atomic transaction — temp write, fsync, rename — so a SIGKILL
|
||||
* at any point leaves either the previous committed state or the next one, never a torn lease.
|
||||
* That matters because this host restarts its runtime often; a half-written lease would be
|
||||
* indistinguishable from an owner whose identity cannot be verified.
|
||||
*/
|
||||
|
||||
import { createHash } from 'node:crypto'
|
||||
import { mkdir, readFile, rename, rm } from 'node:fs/promises'
|
||||
import { dirname, join } from 'node:path'
|
||||
import {
|
||||
agentSessionOperationKey,
|
||||
isAgentSessionOperationRow,
|
||||
type AgentSessionOperationRow
|
||||
} from '../../shared/agent-session-operation-ledger'
|
||||
import { isAgentSessionRecord, type AgentSessionRecord } from '../../shared/agent-session-record'
|
||||
import { durableWriteTempPath, writeFileDurable } from '../durable-file-write'
|
||||
|
||||
export const AGENT_SESSION_STORE_SCHEMA_VERSION = 1 as const
|
||||
|
||||
export const AGENT_SESSION_STORE_FILE_NAME = 'agent-sessions.json'
|
||||
|
||||
export type RetiredAgentSessionClaimKey = { keyId: string; retiredAt: number }
|
||||
|
||||
export type AgentSessionStoreState = {
|
||||
schemaVersion: number
|
||||
hostId: string
|
||||
records: Map<string, AgentSessionRecord>
|
||||
operations: Map<string, AgentSessionOperationRow>
|
||||
retiredClaimKeys: RetiredAgentSessionClaimKey[]
|
||||
/** Rows this build cannot validate, kept verbatim so a rollback cannot delete another host's work. */
|
||||
unreadableRecords: Map<string, unknown>
|
||||
}
|
||||
|
||||
export type LoadedAgentSessionStore = {
|
||||
state: AgentSessionStoreState
|
||||
storeFound: boolean
|
||||
/** True when the file was written by a newer schema; this host reads but never writes it. */
|
||||
readOnly: boolean
|
||||
/** True when the primary file was unusable and the previous committed copy was used. */
|
||||
recoveredFromBackup: boolean
|
||||
}
|
||||
|
||||
export function agentSessionStorePath(directory: string): string {
|
||||
return join(directory, AGENT_SESSION_STORE_FILE_NAME)
|
||||
}
|
||||
|
||||
function backupPath(filePath: string): string {
|
||||
return `${filePath}.bak`
|
||||
}
|
||||
|
||||
function emptyState(hostId: string): AgentSessionStoreState {
|
||||
return {
|
||||
schemaVersion: AGENT_SESSION_STORE_SCHEMA_VERSION,
|
||||
hostId,
|
||||
records: new Map(),
|
||||
operations: new Map(),
|
||||
retiredClaimKeys: [],
|
||||
unreadableRecords: new Map()
|
||||
}
|
||||
}
|
||||
|
||||
export function agentSessionStoreRevision(state: AgentSessionStoreState): string {
|
||||
return createHash('sha256')
|
||||
.update(String(state.schemaVersion))
|
||||
.update('\0')
|
||||
.update(serializeState(state))
|
||||
.digest('hex')
|
||||
}
|
||||
|
||||
function parseState(raw: string, hostId: string): { state: AgentSessionStoreState } | null {
|
||||
let parsed: unknown
|
||||
try {
|
||||
parsed = JSON.parse(raw)
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
if (typeof parsed !== 'object' || parsed === null) {
|
||||
return null
|
||||
}
|
||||
const file = parsed as {
|
||||
schemaVersion?: unknown
|
||||
hostId?: unknown
|
||||
records?: unknown
|
||||
operations?: unknown
|
||||
retiredClaimKeys?: unknown
|
||||
}
|
||||
if (
|
||||
!Number.isSafeInteger(file.schemaVersion) ||
|
||||
(file.schemaVersion as number) < 0 ||
|
||||
typeof file.hostId !== 'string'
|
||||
) {
|
||||
return null
|
||||
}
|
||||
const schemaVersion = file.schemaVersion as number
|
||||
if (
|
||||
schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION &&
|
||||
(typeof file.records !== 'object' ||
|
||||
file.records === null ||
|
||||
Array.isArray(file.records) ||
|
||||
typeof file.operations !== 'object' ||
|
||||
file.operations === null ||
|
||||
Array.isArray(file.operations) ||
|
||||
(schemaVersion === AGENT_SESSION_STORE_SCHEMA_VERSION &&
|
||||
!Array.isArray(file.retiredClaimKeys)))
|
||||
) {
|
||||
return null
|
||||
}
|
||||
const state = emptyState(hostId)
|
||||
state.schemaVersion = schemaVersion
|
||||
state.hostId = file.hostId
|
||||
if (typeof file.records === 'object' && file.records !== null) {
|
||||
for (const [sessionId, value] of Object.entries(file.records)) {
|
||||
if (isAgentSessionRecord(value) && value.sessionId === sessionId) {
|
||||
state.records.set(sessionId, value)
|
||||
} else {
|
||||
// Why: a record this build cannot read must not silently vanish, and must never be
|
||||
// granted a writer; keep it and refuse ownership for that session id.
|
||||
state.unreadableRecords.set(sessionId, value)
|
||||
}
|
||||
}
|
||||
}
|
||||
if (typeof file.operations === 'object' && file.operations !== null) {
|
||||
for (const [key, value] of Object.entries(file.operations)) {
|
||||
if (!isAgentSessionOperationRow(value)) {
|
||||
if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) {
|
||||
return null
|
||||
}
|
||||
continue
|
||||
}
|
||||
if (key !== agentSessionOperationKey(value.callerKey, value.operationId)) {
|
||||
if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) {
|
||||
return null
|
||||
}
|
||||
continue
|
||||
}
|
||||
state.operations.set(key, value)
|
||||
}
|
||||
}
|
||||
if (Array.isArray(file.retiredClaimKeys)) {
|
||||
for (const entry of file.retiredClaimKeys) {
|
||||
const key = entry as Partial<RetiredAgentSessionClaimKey>
|
||||
if (
|
||||
typeof key?.keyId !== 'string' ||
|
||||
key.keyId.length === 0 ||
|
||||
key.keyId.length > 512 ||
|
||||
!Number.isSafeInteger(key.retiredAt) ||
|
||||
(key.retiredAt as number) < 0
|
||||
) {
|
||||
if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) {
|
||||
return null
|
||||
}
|
||||
continue
|
||||
}
|
||||
state.retiredClaimKeys.push({ keyId: key.keyId, retiredAt: key.retiredAt as number })
|
||||
}
|
||||
}
|
||||
return { state }
|
||||
}
|
||||
|
||||
export async function loadAgentSessionStore(
|
||||
filePath: string,
|
||||
hostId: string
|
||||
): Promise<LoadedAgentSessionStore> {
|
||||
let unusableStoreFound = false
|
||||
for (const [candidate, recoveredFromBackup] of [
|
||||
[filePath, false],
|
||||
[backupPath(filePath), true]
|
||||
] as const) {
|
||||
let raw: string
|
||||
try {
|
||||
raw = await readFile(candidate, 'utf-8')
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') {
|
||||
unusableStoreFound = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
const parsed = parseState(raw, hostId)
|
||||
if (!parsed) {
|
||||
unusableStoreFound = true
|
||||
continue
|
||||
}
|
||||
return {
|
||||
state: parsed.state,
|
||||
storeFound: true,
|
||||
readOnly: parsed.state.schemaVersion > AGENT_SESSION_STORE_SCHEMA_VERSION,
|
||||
recoveredFromBackup
|
||||
}
|
||||
}
|
||||
if (unusableStoreFound) {
|
||||
throw new Error('agent_session_store_corrupt')
|
||||
}
|
||||
return {
|
||||
state: emptyState(hostId),
|
||||
storeFound: false,
|
||||
readOnly: false,
|
||||
recoveredFromBackup: false
|
||||
}
|
||||
}
|
||||
|
||||
function serializeState(state: AgentSessionStoreState): string {
|
||||
const records: Record<string, unknown> = Object.create(null)
|
||||
for (const [sessionId, value] of state.unreadableRecords) {
|
||||
records[sessionId] = value
|
||||
}
|
||||
for (const [sessionId, record] of state.records) {
|
||||
records[sessionId] = record
|
||||
}
|
||||
return JSON.stringify({
|
||||
schemaVersion: AGENT_SESSION_STORE_SCHEMA_VERSION,
|
||||
hostId: state.hostId,
|
||||
records,
|
||||
operations: Object.fromEntries(state.operations),
|
||||
retiredClaimKeys: state.retiredClaimKeys
|
||||
})
|
||||
}
|
||||
|
||||
/** Commit the whole state. The previous file becomes the backup before the new one lands. */
|
||||
export async function saveAgentSessionStore(
|
||||
filePath: string,
|
||||
state: AgentSessionStoreState,
|
||||
options: { recoveredFromBackup?: boolean } = {}
|
||||
): Promise<void> {
|
||||
await mkdir(dirname(filePath), { recursive: true })
|
||||
const tmpPath = durableWriteTempPath(filePath)
|
||||
if (!options.recoveredFromBackup) {
|
||||
try {
|
||||
await rm(backupPath(filePath), { force: true })
|
||||
await rename(filePath, backupPath(filePath))
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') {
|
||||
throw error
|
||||
}
|
||||
// First write, or recovery already found the primary missing.
|
||||
}
|
||||
}
|
||||
try {
|
||||
await writeFileDurable(tmpPath, filePath, serializeState(state))
|
||||
} catch (error) {
|
||||
await rm(tmpPath, { force: true }).catch(() => {})
|
||||
throw error
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,877 @@
|
||||
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 { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication'
|
||||
import type {
|
||||
AgentSessionExecutionLocation,
|
||||
AgentSessionProcessIdentity,
|
||||
AgentSessionRecord
|
||||
} from '../../shared/agent-session-record'
|
||||
import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle'
|
||||
import {
|
||||
AGENT_SESSION_CLAIM_KEY_RETENTION_MS,
|
||||
AgentSessionRecordStore
|
||||
} from './agent-session-record-store'
|
||||
import {
|
||||
agentSessionStorePath,
|
||||
AGENT_SESSION_STORE_FILE_NAME
|
||||
} from './agent-session-record-store-file'
|
||||
import type { AgentSessionReserveRequest } from './agent-session-reservation-admission'
|
||||
|
||||
const NOW = 1_800_000_000_000
|
||||
|
||||
const NATIVE: AgentSessionExecutionLocation = {
|
||||
executionHostId: 'local',
|
||||
wslDistro: null,
|
||||
workspaceId: 'workspace-1',
|
||||
workspaceKind: 'git-worktree'
|
||||
}
|
||||
const WSL: AgentSessionExecutionLocation = { ...NATIVE, wslDistro: 'Ubuntu-22.04' }
|
||||
const SSH: AgentSessionExecutionLocation = { ...NATIVE, executionHostId: 'ssh:build-box' }
|
||||
const FOLDER: AgentSessionExecutionLocation = {
|
||||
...NATIVE,
|
||||
workspaceId: 'workspace-2',
|
||||
workspaceKind: 'folder'
|
||||
}
|
||||
|
||||
const MATCHED: AgentSessionOwnerProbe = { outcome: 'identity-matched', matchedOn: ['spawn-token'] }
|
||||
const INDETERMINATE: AgentSessionOwnerProbe = { outcome: 'indeterminate', reason: 'no answer' }
|
||||
const UNUSED: AgentSessionOwnerProbe = { outcome: 'reservation-unused' }
|
||||
const BAD_OP_STORE = '{"schemaVersion":0,"hostId":"","records":{},"operations":{"x":0}}'
|
||||
const BAD_KEY_STORE =
|
||||
'{"schemaVersion":1,"hostId":"","records":{},"operations":{},"retiredClaimKeys":[0]}'
|
||||
|
||||
let counter = 0
|
||||
|
||||
function operationId(now = NOW): string {
|
||||
counter += 1
|
||||
return `${now}-${String(counter)
|
||||
.padStart(32, '0')
|
||||
.replaceAll(/[^0-9a-f]/g, '0')}`
|
||||
}
|
||||
|
||||
function reserveRequest(
|
||||
overrides: Partial<AgentSessionReserveRequest> = {}
|
||||
): AgentSessionReserveRequest {
|
||||
return {
|
||||
sessionId: 'session-alpha',
|
||||
location: NATIVE,
|
||||
provider: 'claude',
|
||||
accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/dev/.claude-work' },
|
||||
runtimeKind: 'native',
|
||||
expectedFence: null,
|
||||
spawnToken: 'spawn-a',
|
||||
claimKeyId: 'key-1',
|
||||
handoffOperationId: null,
|
||||
probe: INDETERMINATE,
|
||||
operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-1' },
|
||||
now: NOW,
|
||||
...overrides
|
||||
}
|
||||
}
|
||||
|
||||
function processIdentity(
|
||||
overrides: Partial<AgentSessionProcessIdentity> = {}
|
||||
): AgentSessionProcessIdentity {
|
||||
return {
|
||||
hostId: 'local',
|
||||
pid: 4242,
|
||||
processStartTimeMs: 1_700_000_000_000,
|
||||
spawnToken: 'spawn-a',
|
||||
...overrides
|
||||
}
|
||||
}
|
||||
|
||||
function handleLink(
|
||||
overrides: Partial<AgentSessionProviderHandleLink> = {}
|
||||
): AgentSessionProviderHandleLink {
|
||||
return {
|
||||
linkId: 'link-1',
|
||||
handle: { provider: 'claude', sessionId: 'provider-session-1', leafUuid: 'leaf-1' },
|
||||
origin: 'created',
|
||||
mintedAtFence: 1,
|
||||
observedAt: NOW,
|
||||
...overrides
|
||||
}
|
||||
}
|
||||
|
||||
let directory: string
|
||||
|
||||
async function open(hostId = 'local'): Promise<AgentSessionRecordStore> {
|
||||
return AgentSessionRecordStore.open({ directory, hostId })
|
||||
}
|
||||
|
||||
/** Reserve, observe the spawn, prove the handle — the full path to an admitted writer. */
|
||||
async function establishOwner(
|
||||
store: AgentSessionRecordStore,
|
||||
overrides: Partial<AgentSessionReserveRequest> = {}
|
||||
): Promise<AgentSessionRecord> {
|
||||
const reserved = await store.reserveOwner(reserveRequest(overrides))
|
||||
const fence = reserved.record.lease.runtimeFence
|
||||
const sessionId = reserved.record.sessionId
|
||||
await store.commitProcessIdentity({
|
||||
sessionId,
|
||||
fence,
|
||||
process: processIdentity({ spawnToken: reserved.record.lease.reservedSpawnToken ?? 'spawn-a' }),
|
||||
now: NOW
|
||||
})
|
||||
return store.proveOwner({
|
||||
sessionId,
|
||||
fence,
|
||||
link: handleLink({ mintedAtFence: fence }),
|
||||
now: NOW
|
||||
})
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
directory = await mkdtemp(join(tmpdir(), 'orca-agent-session-store-'))
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await rm(directory, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe('acquisition path', () => {
|
||||
it('admits a writer only after reservation, observed identity, and a proved handle', async () => {
|
||||
const store = await open()
|
||||
const reserved = await store.reserveOwner(reserveRequest())
|
||||
expect(reserved.disposition).toBe('created')
|
||||
expect(reserved.record.lease).toMatchObject({
|
||||
runtimeFence: 1,
|
||||
claimStatus: 'reserved',
|
||||
handoffStage: 'new-owner-proving',
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: 'spawn-a'
|
||||
})
|
||||
|
||||
// Proving before the spawn is observed is refused.
|
||||
await expect(
|
||||
store.proveOwner({ sessionId: 'session-alpha', fence: 1, link: handleLink(), now: NOW })
|
||||
).rejects.toThrow('agent_session_ownership_unknown')
|
||||
|
||||
await store.commitProcessIdentity({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
process: processIdentity(),
|
||||
now: NOW
|
||||
})
|
||||
const proved = await store.proveOwner({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
link: handleLink(),
|
||||
now: NOW
|
||||
})
|
||||
expect(proved.lease).toMatchObject({
|
||||
claimStatus: 'live',
|
||||
handoffStage: null,
|
||||
provenHandleLinkId: 'link-1',
|
||||
runtimeFence: 1
|
||||
})
|
||||
expect(proved.providerHandleChain).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('refuses a child that cannot echo the reserved spawn token', async () => {
|
||||
const store = await open()
|
||||
await store.reserveOwner(reserveRequest())
|
||||
await expect(
|
||||
store.commitProcessIdentity({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
process: processIdentity({ spawnToken: 'spawn-other' }),
|
||||
now: NOW
|
||||
})
|
||||
).rejects.toThrow('agent_session_ownership_unknown')
|
||||
})
|
||||
|
||||
it('accepts provider proof only for the reserved provider at the current fence', async () => {
|
||||
const store = await open()
|
||||
await store.reserveOwner(reserveRequest())
|
||||
await store.commitProcessIdentity({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
process: processIdentity(),
|
||||
now: NOW
|
||||
})
|
||||
|
||||
await expect(
|
||||
store.proveOwner({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
link: handleLink({
|
||||
handle: { provider: 'codex', threadId: 'thread-1' }
|
||||
}),
|
||||
now: NOW
|
||||
})
|
||||
).rejects.toThrow('agent_session_provider_handle_provider_mismatch')
|
||||
await expect(
|
||||
store.proveOwner({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
link: handleLink({ mintedAtFence: 2 }),
|
||||
now: NOW
|
||||
})
|
||||
).rejects.toThrow('agent_session_provider_handle_stale_fence')
|
||||
})
|
||||
|
||||
it('does not let an established owner re-enter the proof transition', async () => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
|
||||
await expect(
|
||||
store.proveOwner({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
link: handleLink({
|
||||
linkId: 'link-2',
|
||||
origin: 'resumed',
|
||||
observedAt: NOW + 1
|
||||
}),
|
||||
now: NOW + 1
|
||||
})
|
||||
).rejects.toThrow('agent_session_ownership_unknown')
|
||||
})
|
||||
|
||||
it('refuses a create that carries a fence and a re-create that does not', async () => {
|
||||
const store = await open()
|
||||
await expect(store.reserveOwner(reserveRequest({ expectedFence: 0 }))).rejects.toThrow(
|
||||
'agent_session_checkpoint_stale'
|
||||
)
|
||||
await establishOwner(store)
|
||||
await expect(store.reserveOwner(reserveRequest({ expectedFence: null }))).rejects.toThrow(
|
||||
'agent_session_conflict'
|
||||
)
|
||||
})
|
||||
|
||||
it.each([
|
||||
[
|
||||
'provider',
|
||||
{ provider: 'codex', accountHome: { variable: 'CODEX_HOME', path: '/home/dev/.codex' } }
|
||||
],
|
||||
['account', { accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/dev/.claude-other' } }]
|
||||
] as const)("refuses to change a session's pinned %s", async (_name, overrides) => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
await expect(
|
||||
store.reserveOwner(
|
||||
reserveRequest({
|
||||
...overrides,
|
||||
expectedFence: 1,
|
||||
probe: { outcome: 'pid-absent' },
|
||||
operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' }
|
||||
})
|
||||
)
|
||||
).rejects.toThrow('agent_session_conflict')
|
||||
})
|
||||
})
|
||||
|
||||
describe('concurrent claims', () => {
|
||||
it('lets only one store instance reserve a session from the same disk snapshot', async () => {
|
||||
const [first, second] = await Promise.all([open(), open()])
|
||||
const results = await Promise.allSettled([
|
||||
first.reserveOwner(reserveRequest()),
|
||||
second.reserveOwner(reserveRequest())
|
||||
])
|
||||
|
||||
expect(results.filter((result) => result.status === 'fulfilled')).toHaveLength(1)
|
||||
const refused = results.find((result) => result.status === 'rejected')
|
||||
expect((refused as PromiseRejectedResult).reason.message).toBe('agent_session_conflict')
|
||||
|
||||
const persisted = await open()
|
||||
expect(persisted.getRecord('session-alpha')?.lease.runtimeFence).toBe(1)
|
||||
expect(persisted.listOperationRows()).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('lets exactly one of two concurrent reservations win and never spawns the loser', async () => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
const request = () =>
|
||||
reserveRequest({
|
||||
expectedFence: 1,
|
||||
probe: { outcome: 'pid-absent' },
|
||||
spawnToken: 'spawn-b',
|
||||
operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' }
|
||||
})
|
||||
const results = await Promise.allSettled([
|
||||
store.reserveOwner(request()),
|
||||
store.reserveOwner(request())
|
||||
])
|
||||
const granted = results.filter((result) => result.status === 'fulfilled')
|
||||
expect(granted).toHaveLength(1)
|
||||
const refused = results.find((result) => result.status === 'rejected')
|
||||
expect((refused as PromiseRejectedResult).reason.message).toBe('agent_session_checkpoint_stale')
|
||||
expect(store.getRecord('session-alpha')?.lease.runtimeFence).toBe(2)
|
||||
})
|
||||
|
||||
it('serializes concurrent creates of the same session id', async () => {
|
||||
const store = await open()
|
||||
const results = await Promise.allSettled([
|
||||
store.reserveOwner(reserveRequest()),
|
||||
store.reserveOwner(reserveRequest())
|
||||
])
|
||||
expect(results.filter((result) => result.status === 'fulfilled')).toHaveLength(1)
|
||||
expect(store.getRecord('session-alpha')?.lease.runtimeFence).toBe(1)
|
||||
})
|
||||
|
||||
it('replays a retried operation id instead of reserving twice', async () => {
|
||||
const store = await open()
|
||||
const operation = { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-1' }
|
||||
const first = await store.reserveOwner(reserveRequest({ operation }))
|
||||
const second = await store.reserveOwner(reserveRequest({ operation }))
|
||||
expect(second.disposition).toBe('replayed')
|
||||
expect(second.record.lease.runtimeFence).toBe(first.record.lease.runtimeFence)
|
||||
expect(store.listOperationRows()).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('refuses the same operation id carrying different parameters', async () => {
|
||||
const store = await open()
|
||||
const operationId_ = operationId()
|
||||
await store.reserveOwner(
|
||||
reserveRequest({
|
||||
operation: { callerKey: 'client-1', operationId: operationId_, fingerprint: 'fp-1' }
|
||||
})
|
||||
)
|
||||
await expect(
|
||||
store.reserveOwner(
|
||||
reserveRequest({
|
||||
sessionId: 'session-beta',
|
||||
operation: { callerKey: 'client-1', operationId: operationId_, fingerprint: 'fp-9' }
|
||||
})
|
||||
)
|
||||
).rejects.toThrow('agent_session_operation_conflict')
|
||||
})
|
||||
|
||||
it('rolls the in-memory state back when a transaction throws', async () => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
const before = store.getRecord('session-alpha')
|
||||
await expect(
|
||||
store.reserveOwner(reserveRequest({ expectedFence: 1, probe: INDETERMINATE }))
|
||||
).rejects.toThrow('agent_session_ownership_unknown')
|
||||
expect(store.getRecord('session-alpha')).toEqual(before)
|
||||
expect(store.listOperationRows()).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('never keeps a change in memory that failed to commit to disk', async () => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
const before = store.getRecord('session-alpha')
|
||||
// Losing both committed copies must not reset the live store to empty authority.
|
||||
await rm(directory, { recursive: true, force: true })
|
||||
await expect(
|
||||
store.setJournalCheckpoint({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
checkpoint: { epoch: 9, sequence: 9 },
|
||||
now: NOW
|
||||
})
|
||||
).rejects.toThrow()
|
||||
expect(store.getRecord('session-alpha')).toEqual(before)
|
||||
expect(store.getRecord('session-alpha')?.lease.journalCheckpoint).toBeNull()
|
||||
})
|
||||
})
|
||||
|
||||
describe('expiry is not eviction', () => {
|
||||
it('never grants a second owner on a lapsed deadline alone', async () => {
|
||||
const store = await open()
|
||||
const owned = await establishOwner(store)
|
||||
const wellPastDeadline = owned.lease.leaseDeadlineAt + 60 * 60 * 1000
|
||||
for (const probe of [INDETERMINATE, MATCHED] as const) {
|
||||
await expect(
|
||||
store.reserveOwner(
|
||||
reserveRequest({
|
||||
expectedFence: 1,
|
||||
probe,
|
||||
now: wellPastDeadline,
|
||||
operation: {
|
||||
callerKey: 'client-1',
|
||||
operationId: operationId(wellPastDeadline),
|
||||
fingerprint: 'fp-2'
|
||||
}
|
||||
})
|
||||
)
|
||||
).rejects.toThrow(/agent_session_(ownership_unknown|conflict)/)
|
||||
}
|
||||
expect(store.getRecord('session-alpha')?.lease.runtimeFence).toBe(1)
|
||||
|
||||
// Only proof of death moves it.
|
||||
const evicted = await store.evictProvenDeadOwner({
|
||||
sessionId: 'session-alpha',
|
||||
expectedFence: 1,
|
||||
probe: { outcome: 'pid-absent' },
|
||||
now: wellPastDeadline
|
||||
})
|
||||
expect(evicted.lease).toMatchObject({ runtimeFence: 2, claimStatus: 'released' })
|
||||
expect(evicted.lease.deathEvidence?.kind).toBe('pid-absent')
|
||||
})
|
||||
|
||||
it('refuses to evict an owner it cannot prove dead', async () => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
await expect(
|
||||
store.evictProvenDeadOwner({
|
||||
sessionId: 'session-alpha',
|
||||
expectedFence: 1,
|
||||
probe: INDETERMINATE,
|
||||
now: NOW
|
||||
})
|
||||
).rejects.toThrow('agent_session_ownership_unknown')
|
||||
})
|
||||
|
||||
it('stops renewing a lease it can no longer vouch for', async () => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
const renewed = await store.renewLease({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
childProbe: MATCHED,
|
||||
now: NOW + 5_000
|
||||
})
|
||||
expect(renewed.lease.lastRenewedAt).toBe(NOW + 5_000)
|
||||
await expect(
|
||||
store.renewLease({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
childProbe: { outcome: 'identity-matched', matchedOn: [] },
|
||||
now: NOW + 10_000
|
||||
})
|
||||
).rejects.toThrow('agent_session_ownership_unknown')
|
||||
})
|
||||
})
|
||||
|
||||
describe('restart reconciliation', () => {
|
||||
it('survives a restart and grants no writer until adjudicated', async () => {
|
||||
const first = await open()
|
||||
await establishOwner(first)
|
||||
|
||||
const reopened = await open()
|
||||
const loaded = reopened.getRecord('session-alpha')
|
||||
expect(loaded?.lease.unreconciled).toBe(true)
|
||||
expect(loaded?.providerHandleChain).toHaveLength(1)
|
||||
expect(loaded?.accountHome).toEqual({
|
||||
variable: 'CLAUDE_CONFIG_DIR',
|
||||
path: '/home/dev/.claude-work'
|
||||
})
|
||||
// Every mutating path is closed while unreconciled.
|
||||
await expect(
|
||||
reopened.renewLease({ sessionId: 'session-alpha', fence: 1, childProbe: MATCHED, now: NOW })
|
||||
).rejects.toThrow('execution_owner_reconciling')
|
||||
await expect(
|
||||
reopened.reserveOwner(
|
||||
reserveRequest({
|
||||
expectedFence: 1,
|
||||
probe: { outcome: 'pid-absent' },
|
||||
operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' }
|
||||
})
|
||||
)
|
||||
).rejects.toThrow('execution_owner_reconciling')
|
||||
})
|
||||
|
||||
it('re-adopts a live owner without moving the fence', async () => {
|
||||
const first = await open()
|
||||
await establishOwner(first)
|
||||
const reopened = await open()
|
||||
await reopened.reconcileOnRestart({ probe: async () => MATCHED, now: NOW + 1_000 })
|
||||
const record = reopened.getRecord('session-alpha')
|
||||
expect(record?.lease).toMatchObject({
|
||||
unreconciled: false,
|
||||
runtimeFence: 1,
|
||||
claimStatus: 'live'
|
||||
})
|
||||
expect(record?.lease.provenHandleLinkId).toBe('link-1')
|
||||
})
|
||||
|
||||
it('never applies a stale restart probe to a replacement owner', async () => {
|
||||
const writer = await open()
|
||||
await establishOwner(writer)
|
||||
const reconciler = await open()
|
||||
let releaseProbe!: (probe: AgentSessionOwnerProbe) => void
|
||||
let markProbeStarted!: () => void
|
||||
const probeStarted = new Promise<void>((resolve) => (markProbeStarted = resolve))
|
||||
const probeResult = new Promise<AgentSessionOwnerProbe>((resolve) => (releaseProbe = resolve))
|
||||
const reconciliation = reconciler.reconcileOnRestart({
|
||||
probe: async () => {
|
||||
markProbeStarted()
|
||||
return probeResult
|
||||
},
|
||||
now: NOW + 1_000
|
||||
})
|
||||
|
||||
await probeStarted
|
||||
await writer.evictProvenDeadOwner({
|
||||
sessionId: 'session-alpha',
|
||||
expectedFence: 1,
|
||||
probe: { outcome: 'pid-absent' },
|
||||
now: NOW + 100
|
||||
})
|
||||
const replacement = await writer.reserveOwner(
|
||||
reserveRequest({
|
||||
expectedFence: 2,
|
||||
probe: UNUSED,
|
||||
spawnToken: 'spawn-b',
|
||||
operation: {
|
||||
callerKey: 'client-1',
|
||||
operationId: operationId(NOW + 200),
|
||||
fingerprint: 'fp-2'
|
||||
},
|
||||
now: NOW + 200
|
||||
})
|
||||
)
|
||||
await writer.commitProcessIdentity({
|
||||
sessionId: 'session-alpha',
|
||||
fence: replacement.record.lease.runtimeFence,
|
||||
process: processIdentity({ pid: 5252, spawnToken: 'spawn-b' }),
|
||||
now: NOW + 300
|
||||
})
|
||||
await writer.proveOwner({
|
||||
sessionId: 'session-alpha',
|
||||
fence: replacement.record.lease.runtimeFence,
|
||||
link: handleLink({ linkId: 'link-2', origin: 'resumed', mintedAtFence: 3 }),
|
||||
now: NOW + 300
|
||||
})
|
||||
|
||||
releaseProbe({ outcome: 'pid-absent' })
|
||||
expect(await reconciliation).toEqual(new Map())
|
||||
const persisted = JSON.parse(await readFile(agentSessionStorePath(directory), 'utf-8'))
|
||||
expect(persisted.records['session-alpha'].lease).toMatchObject({
|
||||
runtimeFence: 3,
|
||||
claimStatus: 'live',
|
||||
ownerProcess: { pid: 5252, spawnToken: 'spawn-b' },
|
||||
unreconciled: false
|
||||
})
|
||||
})
|
||||
|
||||
it('keeps the fence monotonic across a restart and never reuses a retired fence', async () => {
|
||||
const first = await open()
|
||||
await establishOwner(first)
|
||||
await first.evictProvenDeadOwner({
|
||||
sessionId: 'session-alpha',
|
||||
expectedFence: 1,
|
||||
probe: { outcome: 'exit-observed' },
|
||||
now: NOW
|
||||
})
|
||||
|
||||
const second = await open()
|
||||
await second.reconcileOnRestart({ probe: async () => UNUSED, now: NOW + 1_000 })
|
||||
const afterRestart = second.getRecord('session-alpha')?.lease.runtimeFence ?? 0
|
||||
expect(afterRestart).toBeGreaterThanOrEqual(2)
|
||||
|
||||
const reacquired = await second.reserveOwner(
|
||||
reserveRequest({
|
||||
expectedFence: afterRestart,
|
||||
probe: UNUSED,
|
||||
spawnToken: 'spawn-b',
|
||||
operation: {
|
||||
callerKey: 'client-1',
|
||||
operationId: operationId(NOW + 1_000),
|
||||
fingerprint: 'fp-2'
|
||||
},
|
||||
now: NOW + 1_000
|
||||
})
|
||||
)
|
||||
expect(reacquired.record.lease.runtimeFence).toBe(afterRestart + 1)
|
||||
|
||||
const third = await open()
|
||||
expect(third.getRecord('session-alpha')?.lease.runtimeFence).toBe(afterRestart + 1)
|
||||
})
|
||||
|
||||
it('sends an unverifiable owner to recovery rather than releasing it', async () => {
|
||||
const first = await open()
|
||||
await establishOwner(first)
|
||||
const reopened = await open()
|
||||
await reopened.reconcileOnRestart({ probe: async () => INDETERMINATE, now: NOW + 1_000 })
|
||||
const lease = reopened.getRecord('session-alpha')?.lease
|
||||
expect(lease).toMatchObject({ handoffStage: 'recovering', runtimeFence: 1 })
|
||||
expect(lease?.ownerProcess).not.toBeNull()
|
||||
await expect(
|
||||
reopened.reserveOwner(
|
||||
reserveRequest({
|
||||
expectedFence: 1,
|
||||
probe: { outcome: 'pid-absent' },
|
||||
operation: {
|
||||
callerKey: 'client-1',
|
||||
operationId: operationId(NOW + 1_000),
|
||||
fingerprint: 'fp-2'
|
||||
},
|
||||
now: NOW + 1_000
|
||||
})
|
||||
)
|
||||
).rejects.toThrow('agent_session_ownership_unknown')
|
||||
})
|
||||
|
||||
it('keeps a conflict conflicted across a restart', async () => {
|
||||
const first = await open()
|
||||
await establishOwner(first)
|
||||
await first.markClaimConflicted('session-alpha', NOW)
|
||||
|
||||
const reopened = await open()
|
||||
await reopened.reconcileOnRestart({ probe: async () => ({ outcome: 'pid-absent' }), now: NOW })
|
||||
expect(reopened.getRecord('session-alpha')?.lease).toMatchObject({
|
||||
claimStatus: 'conflicted',
|
||||
handoffStage: 'manual-recovery'
|
||||
})
|
||||
await expect(
|
||||
reopened.reserveOwner(
|
||||
reserveRequest({
|
||||
expectedFence: 1,
|
||||
probe: { outcome: 'pid-absent' },
|
||||
operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' }
|
||||
})
|
||||
)
|
||||
).rejects.toThrow('agent_session_conflict')
|
||||
})
|
||||
|
||||
it('frees a reservation that provably never spawned', async () => {
|
||||
const first = await open()
|
||||
await first.reserveOwner(reserveRequest())
|
||||
const reopened = await open()
|
||||
await reopened.reconcileOnRestart({ probe: async () => UNUSED, now: NOW + 1_000 })
|
||||
expect(reopened.getRecord('session-alpha')?.lease).toMatchObject({
|
||||
claimStatus: 'released',
|
||||
runtimeFence: 2,
|
||||
reservedSpawnToken: null
|
||||
})
|
||||
})
|
||||
|
||||
it('carries the operation ledger across a restart so a retry is still a replay', async () => {
|
||||
const operation = { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-1' }
|
||||
const first = await open()
|
||||
await first.reserveOwner(reserveRequest({ operation }))
|
||||
await first.recordOperationOutcome({
|
||||
callerKey: operation.callerKey,
|
||||
operationId: operation.operationId,
|
||||
outcome: { status: 'succeeded', sessionId: 'session-alpha' }
|
||||
})
|
||||
|
||||
const reopened = await open()
|
||||
expect(reopened.listOperationRows()).toHaveLength(1)
|
||||
const replayed = await reopened.reserveOwner(reserveRequest({ operation }))
|
||||
expect(replayed.disposition).toBe('replayed')
|
||||
expect(replayed.record.sessionId).toBe('session-alpha')
|
||||
})
|
||||
})
|
||||
|
||||
describe('host and workspace isolation', () => {
|
||||
it.each([
|
||||
['WSL', WSL],
|
||||
['SSH', SSH],
|
||||
['another workspace', FOLDER],
|
||||
['another workspace kind', { ...NATIVE, workspaceKind: 'folder' }]
|
||||
] as const)('refuses to move one session id to %s', async (_name, location) => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
await expect(
|
||||
store.reserveOwner(
|
||||
reserveRequest({
|
||||
location,
|
||||
expectedFence: 1,
|
||||
probe: { outcome: 'pid-absent' },
|
||||
operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' }
|
||||
})
|
||||
)
|
||||
).rejects.toThrow('agent_session_conflict')
|
||||
})
|
||||
|
||||
it('keeps native, WSL, and SSH sessions in separate scopes', async () => {
|
||||
const store = await open()
|
||||
await establishOwner(store, { sessionId: '__proto__' })
|
||||
await establishOwner(store, { sessionId: 'session-wsl', location: WSL, spawnToken: 'spawn-b' })
|
||||
await establishOwner(store, { sessionId: 'session-ssh', location: SSH, spawnToken: 'spawn-c' })
|
||||
await establishOwner(store, {
|
||||
sessionId: 'session-folder',
|
||||
location: FOLDER,
|
||||
spawnToken: 'spawn-d'
|
||||
})
|
||||
|
||||
expect(store.listByScope(NATIVE).map((record) => record.sessionId)).toEqual(['__proto__'])
|
||||
expect(store.listByScope(WSL).map((record) => record.sessionId)).toEqual(['session-wsl'])
|
||||
expect(store.listByScope(SSH).map((record) => record.sessionId)).toEqual(['session-ssh'])
|
||||
expect(store.listByScope(FOLDER).map((record) => record.sessionId)).toEqual(['session-folder'])
|
||||
expect((await open()).getRecord('__proto__')).not.toBeNull()
|
||||
})
|
||||
|
||||
it('preserves the workspace kind so a folder workspace is never read back as a worktree', async () => {
|
||||
const first = await open()
|
||||
await establishOwner(first, { sessionId: 'session-folder', location: FOLDER })
|
||||
const reopened = await open()
|
||||
expect(reopened.getRecord('session-folder')?.location).toEqual(FOLDER)
|
||||
})
|
||||
})
|
||||
|
||||
describe('orphans, claim keys, checkpoints, and unreadable rows', () => {
|
||||
it('calls a spawn token with no lease an orphan', async () => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
expect(store.listOrphanSpawnTokens(['spawn-a', 'spawn-z'])).toEqual(['spawn-z'])
|
||||
})
|
||||
|
||||
it('keeps a retired claim key verifiable for the retention window', async () => {
|
||||
const store = await open()
|
||||
await store.retireClaimKey('key-1', NOW)
|
||||
expect(store.isClaimKeyVerifiable('key-1', NOW + AGENT_SESSION_CLAIM_KEY_RETENTION_MS)).toBe(
|
||||
true
|
||||
)
|
||||
expect(
|
||||
store.isClaimKeyVerifiable('key-1', NOW + AGENT_SESSION_CLAIM_KEY_RETENTION_MS + 1)
|
||||
).toBe(false)
|
||||
expect(store.isClaimKeyVerifiable('key-unknown', NOW)).toBe(true)
|
||||
})
|
||||
|
||||
it('refuses a journal checkpoint that moves backwards', async () => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
await store.setJournalCheckpoint({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
checkpoint: { epoch: 2, sequence: 10 },
|
||||
now: NOW
|
||||
})
|
||||
await expect(
|
||||
store.setJournalCheckpoint({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
checkpoint: { epoch: 2, sequence: 9 },
|
||||
now: NOW
|
||||
})
|
||||
).rejects.toThrow('agent_session_checkpoint_stale')
|
||||
await expect(
|
||||
store.setJournalCheckpoint({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
checkpoint: { epoch: 1, sequence: 999 },
|
||||
now: NOW
|
||||
})
|
||||
).rejects.toThrow('agent_session_checkpoint_stale')
|
||||
const advanced = await store.setJournalCheckpoint({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
checkpoint: { epoch: 3, sequence: 0 },
|
||||
now: NOW
|
||||
})
|
||||
expect(advanced.lease.journalCheckpoint).toEqual({ epoch: 3, sequence: 0 })
|
||||
})
|
||||
|
||||
it('rejects a handoff stage change under a different operation id', async () => {
|
||||
const store = await open()
|
||||
await establishOwner(store)
|
||||
await store.setHandoffStage({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
stage: 'preparing',
|
||||
handoffOperationId: 'op-1',
|
||||
now: NOW
|
||||
})
|
||||
await expect(
|
||||
store.setHandoffStage({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
stage: 'old-owner-stopped',
|
||||
handoffOperationId: 'op-2',
|
||||
now: NOW
|
||||
})
|
||||
).rejects.toThrow('agent_session_operation_conflict')
|
||||
})
|
||||
|
||||
it('quarantines a record it cannot validate and refuses to own that session id', async () => {
|
||||
const first = await open()
|
||||
await establishOwner(first)
|
||||
const filePath = agentSessionStorePath(directory)
|
||||
const raw = JSON.parse(await readFile(filePath, 'utf-8'))
|
||||
raw.records['session-alpha'].lease.runtimeFence = 'not-a-number'
|
||||
await writeFile(filePath, JSON.stringify(raw))
|
||||
|
||||
const reopened = await open()
|
||||
expect(reopened.getRecord('session-alpha')).toBeNull()
|
||||
expect(reopened.isSessionUnreadable('session-alpha')).toBe(true)
|
||||
await expect(reopened.reserveOwner(reserveRequest())).rejects.toThrow(
|
||||
'execution_owner_reconciling'
|
||||
)
|
||||
// The row is kept verbatim so a rollback does not delete another build's session.
|
||||
await reopened.retireClaimKey('key-2', NOW)
|
||||
const persisted = JSON.parse(await readFile(filePath, 'utf-8'))
|
||||
expect(persisted.records['session-alpha'].lease.runtimeFence).toBe('not-a-number')
|
||||
})
|
||||
|
||||
it.each([
|
||||
[
|
||||
'invalid checkpoint',
|
||||
(record: AgentSessionRecord) =>
|
||||
Object.assign(record.lease, { journalCheckpoint: { epoch: 'bad', sequence: 1 } })
|
||||
],
|
||||
[
|
||||
'missing live proof',
|
||||
(record: AgentSessionRecord) => Object.assign(record.lease, { provenHandleLinkId: null })
|
||||
]
|
||||
])('quarantines a record with %s', async (_name, corrupt) => {
|
||||
const first = await open()
|
||||
await establishOwner(first)
|
||||
const filePath = agentSessionStorePath(directory)
|
||||
const raw = JSON.parse(await readFile(filePath, 'utf-8'))
|
||||
corrupt(raw.records['session-alpha'])
|
||||
await writeFile(filePath, JSON.stringify(raw))
|
||||
expect((await open()).isSessionUnreadable('session-alpha')).toBe(true)
|
||||
})
|
||||
|
||||
it('recovers the previous committed state when the primary file is corrupt', async () => {
|
||||
const first = await open()
|
||||
await establishOwner(first)
|
||||
// A second commit leaves the first as the backup.
|
||||
await first.setJournalCheckpoint({
|
||||
sessionId: 'session-alpha',
|
||||
fence: 1,
|
||||
checkpoint: { epoch: 1, sequence: 1 },
|
||||
now: NOW
|
||||
})
|
||||
await writeFile(join(directory, AGENT_SESSION_STORE_FILE_NAME), '{ truncated')
|
||||
|
||||
const reopened = await open()
|
||||
expect(reopened.recoveredFromBackup).toBe(true)
|
||||
expect(reopened.getRecord('session-alpha')?.lease.runtimeFence).toBe(1)
|
||||
|
||||
await expect(reopened.retireClaimKey('key-2', NOW)).rejects.toThrow()
|
||||
const backup = JSON.parse(await readFile(`${agentSessionStorePath(directory)}.bak`, 'utf-8'))
|
||||
expect(backup.records['session-alpha'].lease.runtimeFence).toBe(1)
|
||||
})
|
||||
|
||||
it.each([
|
||||
['corrupt', ['{ truncated']],
|
||||
['missing required collections', ['{"schemaVersion":1,"hostId":"local"}']],
|
||||
['invalid operation row', [BAD_OP_STORE]],
|
||||
['invalid retired key', [BAD_KEY_STORE]],
|
||||
['corrupt in both committed copies', ['{ truncated', '{ also truncated']]
|
||||
])('fails closed when the store is %s', async (_name, copies) => {
|
||||
const filePath = agentSessionStorePath(directory)
|
||||
await writeFile(filePath, copies[0])
|
||||
if (copies[1]) {
|
||||
await writeFile(`${filePath}.bak`, copies[1])
|
||||
}
|
||||
await expect(open()).rejects.toThrow('agent_session_store_corrupt')
|
||||
})
|
||||
|
||||
it('refuses to write a store written by a newer schema', async () => {
|
||||
const filePath = agentSessionStorePath(directory)
|
||||
await writeFile(
|
||||
filePath,
|
||||
JSON.stringify({ schemaVersion: 99, hostId: 'local', records: {}, operations: {} })
|
||||
)
|
||||
const store = await open()
|
||||
expect(store.readOnly).toBe(true)
|
||||
await expect(store.reserveOwner(reserveRequest())).rejects.toThrow(
|
||||
'agent_session_legacy_required'
|
||||
)
|
||||
})
|
||||
|
||||
it('migrates an older store schema on its next commit', async () => {
|
||||
const filePath = agentSessionStorePath(directory)
|
||||
await writeFile(
|
||||
filePath,
|
||||
JSON.stringify({ schemaVersion: 0, hostId: 'local', records: {}, operations: {} })
|
||||
)
|
||||
|
||||
const store = await open()
|
||||
await store.retireClaimKey('key-1', NOW)
|
||||
|
||||
const persisted = JSON.parse(await readFile(filePath, 'utf-8'))
|
||||
expect(persisted.schemaVersion).toBe(1)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,337 @@
|
||||
/**
|
||||
* Durable agent-session store for one execution host.
|
||||
*
|
||||
* Holds the session records, their single-writer leases, and the client-operation ledger that
|
||||
* used to live only in memory — all in one file so a reservation and its operation row commit in
|
||||
* the same atomic transaction. Mutations are serialized through a single write queue, so two
|
||||
* concurrent compare-and-swaps in one or multiple Orca processes cannot both commit one pre-state.
|
||||
*
|
||||
* Nothing here is wired into the runtime yet; the store is dormant until later parts adopt it.
|
||||
*/
|
||||
|
||||
import {
|
||||
pruneAgentSessionOperationRows,
|
||||
agentSessionOperationKey,
|
||||
type AgentSessionOperationOutcome,
|
||||
type AgentSessionOperationRow
|
||||
} from '../../shared/agent-session-operation-ledger'
|
||||
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'
|
||||
import {
|
||||
agentSessionScopeKey,
|
||||
type AgentSessionExecutionLocation,
|
||||
type AgentSessionHandoffStage,
|
||||
type AgentSessionJournalCheckpoint,
|
||||
type AgentSessionProcessIdentity,
|
||||
type AgentSessionRecord
|
||||
} from '../../shared/agent-session-record'
|
||||
import {
|
||||
applyAgentSessionRestartAdjudication,
|
||||
agentSessionReconciliationTargetMatches,
|
||||
commitAgentSessionProcessIdentity,
|
||||
evictAgentSessionOwner,
|
||||
proveAgentSessionOwner,
|
||||
renewAgentSessionLease,
|
||||
setAgentSessionHandoffStage,
|
||||
setAgentSessionJournalCheckpoint
|
||||
} from './agent-session-lease-transitions'
|
||||
import {
|
||||
applyAgentSessionReservation,
|
||||
evaluateAgentSessionReserveOperation,
|
||||
requireAgentSessionRecordForReplay,
|
||||
type AgentSessionReserveRequest,
|
||||
type AgentSessionReserveResult
|
||||
} from './agent-session-reservation-admission'
|
||||
import {
|
||||
agentSessionStoreRevision,
|
||||
agentSessionStorePath,
|
||||
loadAgentSessionStore,
|
||||
type AgentSessionStoreState
|
||||
} from './agent-session-record-store-file'
|
||||
import {
|
||||
AgentSessionStoreTransactionQueue,
|
||||
markAgentSessionStoreLeasesUnreconciled
|
||||
} from './agent-session-store-transaction-queue'
|
||||
|
||||
export type {
|
||||
AgentSessionReserveRequest,
|
||||
AgentSessionReserveResult
|
||||
} from './agent-session-reservation-admission'
|
||||
|
||||
export const AGENT_SESSION_LEASE_TTL_MS = 30_000
|
||||
export const AGENT_SESSION_LEASE_RENEW_INTERVAL_MS = 10_000
|
||||
/** Retired claim keys stay verifiable this long so a rotation cannot strand a running agent. */
|
||||
export const AGENT_SESSION_CLAIM_KEY_RETENTION_MS = 30 * 24 * 60 * 60 * 1000
|
||||
|
||||
export class AgentSessionRecordStore {
|
||||
private constructor(private readonly transactions: AgentSessionStoreTransactionQueue) {}
|
||||
|
||||
static async open(args: { directory: string; hostId: string }): Promise<AgentSessionRecordStore> {
|
||||
const filePath = agentSessionStorePath(args.directory)
|
||||
const loaded = await loadAgentSessionStore(filePath, args.hostId)
|
||||
// Why: every persisted lease is unreconciled until this host adjudicates it, so a restart
|
||||
// grants no writer on the strength of what the previous process wrote.
|
||||
const diskRevision = agentSessionStoreRevision(loaded.state)
|
||||
markAgentSessionStoreLeasesUnreconciled(loaded.state)
|
||||
return new AgentSessionRecordStore(
|
||||
new AgentSessionStoreTransactionQueue(
|
||||
filePath,
|
||||
args.hostId,
|
||||
loaded.readOnly,
|
||||
loaded.recoveredFromBackup,
|
||||
loaded.storeFound,
|
||||
loaded.state,
|
||||
diskRevision
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
private get state(): AgentSessionStoreState {
|
||||
return this.transactions.state
|
||||
}
|
||||
|
||||
get readOnly(): boolean {
|
||||
return this.transactions.readOnly
|
||||
}
|
||||
|
||||
get recoveredFromBackup(): boolean {
|
||||
return this.transactions.recoveredFromBackup
|
||||
}
|
||||
|
||||
get hostId(): string {
|
||||
return this.state.hostId
|
||||
}
|
||||
|
||||
getRecord(sessionId: string): AgentSessionRecord | null {
|
||||
return this.state.records.get(sessionId) ?? null
|
||||
}
|
||||
|
||||
listRecords(): AgentSessionRecord[] {
|
||||
return [...this.state.records.values()]
|
||||
}
|
||||
|
||||
listByScope(location: AgentSessionExecutionLocation): AgentSessionRecord[] {
|
||||
const scope = agentSessionScopeKey(location)
|
||||
return this.listRecords().filter((record) => agentSessionScopeKey(record.location) === scope)
|
||||
}
|
||||
|
||||
/** A record this build cannot validate: readable as present, never grantable as a writer. */
|
||||
isSessionUnreadable(sessionId: string): boolean {
|
||||
return this.state.unreadableRecords.has(sessionId)
|
||||
}
|
||||
|
||||
listOperationRows(): AgentSessionOperationRow[] {
|
||||
return [...this.state.operations.values()]
|
||||
}
|
||||
|
||||
isClaimKeyVerifiable(keyId: string, now: number): boolean {
|
||||
const retired = this.state.retiredClaimKeys.find((entry) => entry.keyId === keyId)
|
||||
return !retired || now - retired.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS
|
||||
}
|
||||
|
||||
/** Spawn tokens observed on the host with no matching lease. Stop them; never adopt them. */
|
||||
listOrphanSpawnTokens(observedTokens: readonly string[]): string[] {
|
||||
const leases = this.listRecords().map((record) => record.lease)
|
||||
return observedTokens.filter(
|
||||
(spawnToken) => classifyObservedAgentSessionSpawnToken({ spawnToken, leases }) === 'orphan'
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare-and-swap reservation plus its client-operation row, committed together. A replayed
|
||||
* operation returns the recorded outcome and never reaches the reservation.
|
||||
*/
|
||||
async reserveOwner(request: AgentSessionReserveRequest): Promise<AgentSessionReserveResult> {
|
||||
return this.transact(() => {
|
||||
const decision = evaluateAgentSessionReserveOperation(this.state, request)
|
||||
if (decision.decision === 'refused') {
|
||||
throw new Error(decision.code)
|
||||
}
|
||||
if (decision.decision === 'replay') {
|
||||
const record = requireAgentSessionRecordForReplay(
|
||||
this.state,
|
||||
decision.row,
|
||||
request.sessionId
|
||||
)
|
||||
return { record, disposition: 'replayed' as const, operationRow: decision.row }
|
||||
}
|
||||
const result = applyAgentSessionReservation(this.state, request, AGENT_SESSION_LEASE_TTL_MS)
|
||||
this.state.operations.set(
|
||||
agentSessionOperationKey(request.operation.callerKey, request.operation.operationId),
|
||||
decision.row
|
||||
)
|
||||
this.state.records.set(result.record.sessionId, result.record)
|
||||
return { ...result, operationRow: decision.row }
|
||||
})
|
||||
}
|
||||
|
||||
async commitProcessIdentity(args: {
|
||||
sessionId: string
|
||||
fence: number
|
||||
process: AgentSessionProcessIdentity
|
||||
now: number
|
||||
}): Promise<AgentSessionRecord> {
|
||||
return this.mutate(args.sessionId, (record) =>
|
||||
commitAgentSessionProcessIdentity({ ...args, record })
|
||||
)
|
||||
}
|
||||
|
||||
async proveOwner(args: {
|
||||
sessionId: string
|
||||
fence: number
|
||||
link: AgentSessionProviderHandleLink
|
||||
now: number
|
||||
leaseTtlMs?: number
|
||||
}): Promise<AgentSessionRecord> {
|
||||
return this.mutate(args.sessionId, (record) =>
|
||||
proveAgentSessionOwner({
|
||||
record,
|
||||
fence: args.fence,
|
||||
link: args.link,
|
||||
now: args.now,
|
||||
leaseTtlMs: args.leaseTtlMs ?? AGENT_SESSION_LEASE_TTL_MS
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
async renewLease(args: {
|
||||
sessionId: string
|
||||
fence: number
|
||||
childProbe: AgentSessionOwnerProbe
|
||||
now: number
|
||||
leaseTtlMs?: number
|
||||
}): Promise<AgentSessionRecord> {
|
||||
return this.mutate(args.sessionId, (record) =>
|
||||
renewAgentSessionLease({
|
||||
record,
|
||||
fence: args.fence,
|
||||
childProbe: args.childProbe,
|
||||
now: args.now,
|
||||
leaseTtlMs: args.leaseTtlMs ?? AGENT_SESSION_LEASE_TTL_MS
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
async evictProvenDeadOwner(args: {
|
||||
sessionId: string
|
||||
expectedFence: number
|
||||
probe: AgentSessionOwnerProbe
|
||||
now: number
|
||||
}): Promise<AgentSessionRecord> {
|
||||
return this.mutate(args.sessionId, (record) => evictAgentSessionOwner({ ...args, record }))
|
||||
}
|
||||
|
||||
async setHandoffStage(args: {
|
||||
sessionId: string
|
||||
fence: number
|
||||
stage: AgentSessionHandoffStage | null
|
||||
handoffOperationId: string | null
|
||||
now: number
|
||||
}): Promise<AgentSessionRecord> {
|
||||
return this.mutate(args.sessionId, (record) => setAgentSessionHandoffStage({ ...args, record }))
|
||||
}
|
||||
|
||||
async setJournalCheckpoint(args: {
|
||||
sessionId: string
|
||||
fence: number
|
||||
checkpoint: AgentSessionJournalCheckpoint
|
||||
now: number
|
||||
}): Promise<AgentSessionRecord> {
|
||||
return this.mutate(args.sessionId, (record) =>
|
||||
setAgentSessionJournalCheckpoint({ ...args, record })
|
||||
)
|
||||
}
|
||||
|
||||
/** Adjudicate every lease this host loaded. No lease grants a writer until it appears here. */
|
||||
async reconcileOnRestart(args: {
|
||||
probe: (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>
|
||||
now: number
|
||||
}): Promise<Map<string, AgentSessionRecord>> {
|
||||
const pending = this.listRecords().filter((record) => record.lease.unreconciled)
|
||||
const probes = new Map<string, { record: AgentSessionRecord; probe: AgentSessionOwnerProbe }>()
|
||||
for (const record of pending) {
|
||||
probes.set(record.sessionId, { record, probe: await args.probe(record) })
|
||||
}
|
||||
return this.transact(() => {
|
||||
const reconciled = new Map<string, AgentSessionRecord>()
|
||||
for (const [sessionId, probed] of probes) {
|
||||
const record = this.state.records.get(sessionId)
|
||||
if (
|
||||
!record?.lease.unreconciled ||
|
||||
!agentSessionReconciliationTargetMatches(record, probed.record)
|
||||
) {
|
||||
continue
|
||||
}
|
||||
const next = applyAgentSessionRestartAdjudication({
|
||||
record,
|
||||
probe: probed.probe,
|
||||
now: args.now
|
||||
})
|
||||
this.state.records.set(sessionId, next)
|
||||
reconciled.set(sessionId, next)
|
||||
}
|
||||
this.state.operations = pruneAgentSessionOperationRows(this.state.operations, args.now)
|
||||
return reconciled
|
||||
})
|
||||
}
|
||||
|
||||
async recordOperationOutcome(args: {
|
||||
callerKey: string
|
||||
operationId: string
|
||||
outcome: AgentSessionOperationOutcome
|
||||
}): Promise<void> {
|
||||
await this.transact(() => {
|
||||
const key = agentSessionOperationKey(args.callerKey, args.operationId)
|
||||
const row = this.state.operations.get(key)
|
||||
if (row) {
|
||||
this.state.operations.set(key, { ...row, outcome: args.outcome })
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
async markClaimConflicted(sessionId: string, now: number): Promise<AgentSessionRecord> {
|
||||
return this.mutate(sessionId, (record) => ({
|
||||
...record,
|
||||
updatedAt: now,
|
||||
// Why: a conflicted key must stay conflicted across a restart; it cannot resolve to free
|
||||
// merely because the process that observed the conflict is gone.
|
||||
lease: { ...record.lease, claimStatus: 'conflicted', handoffStage: 'manual-recovery' }
|
||||
}))
|
||||
}
|
||||
|
||||
async retireClaimKey(keyId: string, now: number): Promise<void> {
|
||||
await this.transact(() => {
|
||||
if (!this.state.retiredClaimKeys.some((entry) => entry.keyId === keyId)) {
|
||||
this.state.retiredClaimKeys.push({ keyId, retiredAt: now })
|
||||
}
|
||||
this.state.retiredClaimKeys = this.state.retiredClaimKeys.filter(
|
||||
(entry) => now - entry.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
private async mutate(
|
||||
sessionId: string,
|
||||
apply: (record: AgentSessionRecord) => AgentSessionRecord
|
||||
): Promise<AgentSessionRecord> {
|
||||
return this.transact(() => {
|
||||
const record = this.state.records.get(sessionId)
|
||||
if (!record) {
|
||||
throw new Error(
|
||||
this.isSessionUnreadable(sessionId)
|
||||
? 'execution_owner_reconciling'
|
||||
: 'agent_session_identity_required'
|
||||
)
|
||||
}
|
||||
const next = apply(record)
|
||||
this.state.records.set(sessionId, next)
|
||||
return next
|
||||
})
|
||||
}
|
||||
|
||||
/** Serialize every mutation against the latest committed disk state. */
|
||||
private async transact<T>(apply: () => T): Promise<T> {
|
||||
return this.transactions.transact(apply)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,166 @@
|
||||
/**
|
||||
* Reservation admission: what a reserve request means against the persisted state.
|
||||
*
|
||||
* Pure over a store snapshot so the compare-and-swap, the idempotency replay, and the
|
||||
* location-immutability check can be reasoned about without touching the disk. The store applies
|
||||
* the result inside one transaction; nothing here mutates.
|
||||
*/
|
||||
|
||||
import {
|
||||
evaluateAgentSessionOperation,
|
||||
pruneAgentSessionOperationRows,
|
||||
type AgentSessionOperationDecision,
|
||||
type AgentSessionOperationRow
|
||||
} from '../../shared/agent-session-operation-ledger'
|
||||
import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication'
|
||||
import {
|
||||
AGENT_SESSION_RECORD_SCHEMA_VERSION,
|
||||
agentSessionExecutionLocationsEqual,
|
||||
type AgentSessionAccountHome,
|
||||
type AgentSessionExecutionLocation,
|
||||
type AgentSessionRecord
|
||||
} from '../../shared/agent-session-record'
|
||||
import type { AgentSessionHandleProvider } from '../../shared/agent-session-provider-handle'
|
||||
import {
|
||||
reserveAgentSessionOwner,
|
||||
type AgentSessionReservation
|
||||
} from './agent-session-lease-transitions'
|
||||
import type { AgentSessionStoreState } from './agent-session-record-store-file'
|
||||
|
||||
export type AgentSessionReserveRequest = {
|
||||
sessionId: string
|
||||
location: AgentSessionExecutionLocation
|
||||
provider: AgentSessionHandleProvider
|
||||
accountHome: AgentSessionAccountHome
|
||||
runtimeKind: AgentSessionReservation['runtimeKind']
|
||||
/** Null when the session does not exist yet; otherwise the fence the caller last observed. */
|
||||
expectedFence: number | null
|
||||
spawnToken: string
|
||||
claimKeyId: string
|
||||
handoffOperationId: string | null
|
||||
probe: AgentSessionOwnerProbe
|
||||
operation: { callerKey: string; operationId: string; fingerprint: string }
|
||||
now: number
|
||||
leaseTtlMs?: number
|
||||
}
|
||||
|
||||
export type AgentSessionReserveDisposition =
|
||||
| 'created'
|
||||
| 'reserved'
|
||||
| 'retry-reservation'
|
||||
| 'replayed'
|
||||
|
||||
export type AgentSessionReserveResult = {
|
||||
record: AgentSessionRecord
|
||||
disposition: AgentSessionReserveDisposition
|
||||
operationRow: AgentSessionOperationRow
|
||||
}
|
||||
|
||||
export function evaluateAgentSessionReserveOperation(
|
||||
state: AgentSessionStoreState,
|
||||
request: AgentSessionReserveRequest
|
||||
): AgentSessionOperationDecision {
|
||||
state.operations = pruneAgentSessionOperationRows(state.operations, request.now)
|
||||
return evaluateAgentSessionOperation({
|
||||
rows: state.operations,
|
||||
callerKey: request.operation.callerKey,
|
||||
operationId: request.operation.operationId,
|
||||
fingerprint: request.operation.fingerprint,
|
||||
now: request.now
|
||||
})
|
||||
}
|
||||
|
||||
export function requireAgentSessionRecordForReplay(
|
||||
state: AgentSessionStoreState,
|
||||
row: AgentSessionOperationRow,
|
||||
sessionId: string
|
||||
): AgentSessionRecord {
|
||||
const replayedId = row.outcome.status === 'succeeded' ? row.outcome.sessionId : sessionId
|
||||
const record = state.records.get(replayedId)
|
||||
if (!record) {
|
||||
// Why: the recorded effect is no longer reconstructable, and re-running it would be a second
|
||||
// spawn rather than a replay.
|
||||
throw new Error('agent_session_ownership_unknown')
|
||||
}
|
||||
return record
|
||||
}
|
||||
|
||||
export function applyAgentSessionReservation(
|
||||
state: AgentSessionStoreState,
|
||||
request: AgentSessionReserveRequest,
|
||||
leaseTtlMs: number
|
||||
): {
|
||||
record: AgentSessionRecord
|
||||
disposition: Exclude<AgentSessionReserveDisposition, 'replayed'>
|
||||
} {
|
||||
const reservation: AgentSessionReservation = {
|
||||
runtimeKind: request.runtimeKind,
|
||||
spawnToken: request.spawnToken,
|
||||
claimKeyId: request.claimKeyId,
|
||||
handoffOperationId: request.handoffOperationId,
|
||||
leaseTtlMs: request.leaseTtlMs ?? leaseTtlMs,
|
||||
now: request.now
|
||||
}
|
||||
const existing = state.records.get(request.sessionId)
|
||||
if (!existing) {
|
||||
if (state.unreadableRecords.has(request.sessionId)) {
|
||||
throw new Error('execution_owner_reconciling')
|
||||
}
|
||||
if (request.expectedFence !== null) {
|
||||
throw new Error('agent_session_checkpoint_stale')
|
||||
}
|
||||
return { record: createAgentSessionRecord(request, reservation), disposition: 'created' }
|
||||
}
|
||||
if (
|
||||
!agentSessionExecutionLocationsEqual(existing.location, request.location) ||
|
||||
existing.provider !== request.provider ||
|
||||
existing.accountHome.variable !== request.accountHome.variable ||
|
||||
existing.accountHome.path !== request.accountHome.path
|
||||
) {
|
||||
// Why: location, provider, and account are the session identity; changing one is a fork.
|
||||
throw new Error('agent_session_conflict')
|
||||
}
|
||||
if (request.expectedFence === null) {
|
||||
throw new Error('agent_session_conflict')
|
||||
}
|
||||
return reserveAgentSessionOwner({
|
||||
record: existing,
|
||||
expectedFence: request.expectedFence,
|
||||
probe: request.probe,
|
||||
reservation
|
||||
})
|
||||
}
|
||||
|
||||
function createAgentSessionRecord(
|
||||
request: AgentSessionReserveRequest,
|
||||
reservation: AgentSessionReservation
|
||||
): AgentSessionRecord {
|
||||
return {
|
||||
schemaVersion: AGENT_SESSION_RECORD_SCHEMA_VERSION,
|
||||
sessionId: request.sessionId,
|
||||
location: request.location,
|
||||
provider: request.provider,
|
||||
providerHandleChain: [],
|
||||
accountHome: request.accountHome,
|
||||
createdAt: request.now,
|
||||
updatedAt: request.now,
|
||||
lease: {
|
||||
sessionId: request.sessionId,
|
||||
runtimeKind: reservation.runtimeKind,
|
||||
// Why: fence 1 is the first reservation; 0 is reserved for "no owner has ever existed".
|
||||
runtimeFence: 1,
|
||||
handoffStage: 'new-owner-proving',
|
||||
provenHandleLinkId: null,
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: reservation.spawnToken,
|
||||
leaseDeadlineAt: reservation.now + reservation.leaseTtlMs,
|
||||
lastRenewedAt: reservation.now,
|
||||
handoffOperationId: reservation.handoffOperationId,
|
||||
journalCheckpoint: null,
|
||||
claimKeyId: reservation.claimKeyId,
|
||||
claimStatus: 'reserved',
|
||||
unreconciled: false,
|
||||
deathEvidence: null
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
import { mkdir } from 'node:fs/promises'
|
||||
import { dirname } from 'node:path'
|
||||
import { lock } from 'proper-lockfile'
|
||||
|
||||
const LOCK_RETRIES = {
|
||||
retries: 20,
|
||||
factor: 1.3,
|
||||
minTimeout: 10,
|
||||
maxTimeout: 250,
|
||||
randomize: true
|
||||
}
|
||||
|
||||
/** Serialize whole-file transactions across Orca processes sharing one execution host. */
|
||||
export async function withAgentSessionStoreTransactionLock<T>(
|
||||
filePath: string,
|
||||
apply: () => Promise<T>
|
||||
): Promise<T> {
|
||||
await mkdir(dirname(filePath), { recursive: true })
|
||||
const release = await lock(filePath, { realpath: false, retries: LOCK_RETRIES })
|
||||
try {
|
||||
return await apply()
|
||||
} finally {
|
||||
await release()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
import type { AgentSessionOperationRow } from '../../shared/agent-session-operation-ledger'
|
||||
import type { AgentSessionRecord } from '../../shared/agent-session-record'
|
||||
import {
|
||||
AGENT_SESSION_STORE_SCHEMA_VERSION,
|
||||
agentSessionStoreRevision,
|
||||
loadAgentSessionStore,
|
||||
saveAgentSessionStore,
|
||||
type AgentSessionStoreState
|
||||
} from './agent-session-record-store-file'
|
||||
import { withAgentSessionStoreTransactionLock } from './agent-session-store-transaction-lock'
|
||||
|
||||
function markLoadedLeasesUnreconciled(state: AgentSessionStoreState): void {
|
||||
for (const [sessionId, record] of state.records) {
|
||||
state.records.set(sessionId, {
|
||||
...record,
|
||||
lease: { ...record.lease, unreconciled: true }
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
function mapEntriesMatch<K, V>(left: ReadonlyMap<K, V>, right: ReadonlyMap<K, V>): boolean {
|
||||
if (left.size !== right.size) {
|
||||
return false
|
||||
}
|
||||
for (const [key, value] of left) {
|
||||
if (right.get(key) !== value) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
function agentSessionStoreStateChanged(
|
||||
state: AgentSessionStoreState,
|
||||
records: ReadonlyMap<string, AgentSessionRecord>,
|
||||
operations: ReadonlyMap<string, AgentSessionOperationRow>,
|
||||
retiredClaimKeys: AgentSessionStoreState['retiredClaimKeys']
|
||||
): boolean {
|
||||
return (
|
||||
!mapEntriesMatch(state.records, records) ||
|
||||
!mapEntriesMatch(state.operations, operations) ||
|
||||
state.retiredClaimKeys.length !== retiredClaimKeys.length ||
|
||||
state.retiredClaimKeys.some((entry, index) => entry !== retiredClaimKeys[index])
|
||||
)
|
||||
}
|
||||
|
||||
export class AgentSessionStoreTransactionQueue {
|
||||
private queue: Promise<unknown> = Promise.resolve()
|
||||
private diskRecoveredFromBackup: boolean
|
||||
|
||||
constructor(
|
||||
private readonly filePath: string,
|
||||
readonly hostId: string,
|
||||
readonly readOnly: boolean,
|
||||
readonly recoveredFromBackup: boolean,
|
||||
private diskStoreFound: boolean,
|
||||
public state: AgentSessionStoreState,
|
||||
private diskRevision: string
|
||||
) {
|
||||
this.diskRecoveredFromBackup = recoveredFromBackup
|
||||
}
|
||||
|
||||
transact<T>(apply: () => T): Promise<T> {
|
||||
const run = this.queue.then(() =>
|
||||
withAgentSessionStoreTransactionLock(this.filePath, async () => {
|
||||
if (this.readOnly) {
|
||||
throw new Error('agent_session_legacy_required')
|
||||
}
|
||||
await this.refreshExternallyChangedState()
|
||||
if (this.diskRecoveredFromBackup) {
|
||||
// Why: the missing commit may hold a newer fence, so rollback cannot mint another writer.
|
||||
throw new Error('execution_owner_reconciling')
|
||||
}
|
||||
const records = new Map(this.state.records)
|
||||
const operations = new Map(this.state.operations)
|
||||
const retiredClaimKeys = [...this.state.retiredClaimKeys]
|
||||
try {
|
||||
const result = apply()
|
||||
if (!agentSessionStoreStateChanged(this.state, records, operations, retiredClaimKeys)) {
|
||||
return result
|
||||
}
|
||||
await saveAgentSessionStore(this.filePath, this.state, {
|
||||
recoveredFromBackup: this.diskRecoveredFromBackup
|
||||
})
|
||||
this.state.schemaVersion = AGENT_SESSION_STORE_SCHEMA_VERSION
|
||||
this.diskRevision = agentSessionStoreRevision(this.state)
|
||||
this.diskRecoveredFromBackup = false
|
||||
this.diskStoreFound = true
|
||||
return result
|
||||
} catch (error) {
|
||||
this.state.records = records
|
||||
this.state.operations = operations
|
||||
this.state.retiredClaimKeys = retiredClaimKeys
|
||||
throw error
|
||||
}
|
||||
})
|
||||
)
|
||||
this.queue = run.catch(() => {})
|
||||
return run
|
||||
}
|
||||
|
||||
private async refreshExternallyChangedState(): Promise<void> {
|
||||
const loaded = await loadAgentSessionStore(this.filePath, this.hostId)
|
||||
if (this.diskStoreFound && !loaded.storeFound) {
|
||||
throw new Error('agent_session_store_corrupt')
|
||||
}
|
||||
this.diskStoreFound ||= loaded.storeFound
|
||||
const diskRevision = agentSessionStoreRevision(loaded.state)
|
||||
this.diskRecoveredFromBackup = loaded.recoveredFromBackup
|
||||
if (diskRevision === this.diskRevision) {
|
||||
return
|
||||
}
|
||||
if (loaded.readOnly) {
|
||||
throw new Error('agent_session_legacy_required')
|
||||
}
|
||||
markLoadedLeasesUnreconciled(loaded.state)
|
||||
this.state = loaded.state
|
||||
this.diskRevision = diskRevision
|
||||
}
|
||||
}
|
||||
|
||||
export function markAgentSessionStoreLeasesUnreconciled(state: AgentSessionStoreState): void {
|
||||
markLoadedLeasesUnreconciled(state)
|
||||
}
|
||||
Reference in New Issue
Block a user