mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
perf: batch structured session lease renewal
This commit is contained in:
@@ -0,0 +1,108 @@
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { performance } from 'node:perf_hooks'
|
||||
import { AgentSessionRecordStore } from '../../src/main/runtime/agent-session-record-store'
|
||||
import { readProcessStartTimesMs } from '../../src/main/runtime/agent-session-process-identity-probe'
|
||||
import { StructuredAgentSessionLeaseRenewer } from '../../src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer'
|
||||
|
||||
const RECORD_COUNT = 50
|
||||
const NOW = 1_800_000_000_000
|
||||
|
||||
async function seedStore(store: AgentSessionRecordStore, directory: string): Promise<void> {
|
||||
for (let index = 0; index < RECORD_COUNT; index += 1) {
|
||||
const sessionId = `benchmark-session-${index}`
|
||||
const spawnToken = `benchmark-spawn-${index}`
|
||||
const reserved = await store.reserveOwner({
|
||||
sessionId,
|
||||
location: {
|
||||
executionHostId: 'local',
|
||||
wslDistro: null,
|
||||
workspaceId: 'benchmark-workspace',
|
||||
workspaceKind: 'folder'
|
||||
},
|
||||
provider: 'codex',
|
||||
accountHome: { variable: 'CODEX_HOME', path: directory },
|
||||
runtimeKind: 'native',
|
||||
expectedFence: null,
|
||||
spawnToken,
|
||||
claimKeyId: 'benchmark-key',
|
||||
handoffOperationId: null,
|
||||
probe: { outcome: 'reservation-unused' },
|
||||
operation: {
|
||||
callerKey: 'benchmark',
|
||||
operationId: `${NOW}-${index.toString(16).padStart(32, '0')}`,
|
||||
fingerprint: `create-${index}`
|
||||
},
|
||||
now: NOW
|
||||
})
|
||||
const fence = reserved.record.lease.runtimeFence
|
||||
await store.commitProcessIdentity({
|
||||
sessionId,
|
||||
fence,
|
||||
process: {
|
||||
hostId: 'local',
|
||||
pid: process.pid,
|
||||
processStartTimeMs: null,
|
||||
spawnToken
|
||||
},
|
||||
now: NOW
|
||||
})
|
||||
await store.proveOwner({
|
||||
sessionId,
|
||||
fence,
|
||||
link: {
|
||||
linkId: `benchmark-link-${index}`,
|
||||
handle: { provider: 'codex', threadId: `benchmark-thread-${index}` },
|
||||
origin: 'created',
|
||||
mintedAtFence: fence,
|
||||
observedAt: NOW
|
||||
},
|
||||
now: NOW
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const directory = await mkdtemp(join(tmpdir(), 'orca-lease-renewal-benchmark-'))
|
||||
try {
|
||||
const store = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
|
||||
await seedStore(store, directory)
|
||||
const renewer = new StructuredAgentSessionLeaseRenewer({
|
||||
store,
|
||||
probe: async () => ({ outcome: 'identity-matched', matchedOn: ['spawn-token'] }),
|
||||
now: () => NOW + 10_000
|
||||
})
|
||||
const startedAt = performance.now()
|
||||
await renewer.renewNow()
|
||||
const elapsedMs = performance.now() - startedAt
|
||||
const renewed = store
|
||||
.listRecords()
|
||||
.filter((record) => record.lease.lastRenewedAt === NOW + 10_000).length
|
||||
const processProbeStartedAt = performance.now()
|
||||
const processStartTimes = await readProcessStartTimesMs(
|
||||
Array.from({ length: RECORD_COUNT }, () => process.pid),
|
||||
'darwin'
|
||||
)
|
||||
console.log(
|
||||
JSON.stringify({
|
||||
records: RECORD_COUNT,
|
||||
renewed,
|
||||
elapsedMs,
|
||||
processTableElapsedMs: performance.now() - processProbeStartedAt,
|
||||
processTableMatches:
|
||||
processStartTimes.get(process.pid) === null ||
|
||||
processStartTimes.get(process.pid) === undefined
|
||||
? 0
|
||||
: RECORD_COUNT
|
||||
})
|
||||
)
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
|
||||
void main().catch((error: unknown) => {
|
||||
console.error(error)
|
||||
process.exitCode = 1
|
||||
})
|
||||
@@ -21,6 +21,7 @@ export class StructuredAgentSessionHostRuntimeState {
|
||||
this.leaseRenewer = new StructuredAgentSessionLeaseRenewer({
|
||||
store: deps.store,
|
||||
probe: (record) => this.probeRecord(record),
|
||||
...(deps.probeOwners ? { probeMany: deps.probeOwners } : {}),
|
||||
now: () => deps.now?.() ?? Date.now(),
|
||||
...(onLeaseRenewed ? { onRenewed: onLeaseRenewed } : {}),
|
||||
...(onDeadTuiOwner ? { onDeadTuiOwner } : {}),
|
||||
|
||||
@@ -25,6 +25,9 @@ export type StructuredAgentSessionHostDeps = {
|
||||
journalRoot: string
|
||||
claimKeyId: string
|
||||
probeOwner?: (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>
|
||||
probeOwners?: (
|
||||
records: readonly AgentSessionRecord[]
|
||||
) => Promise<Map<string, AgentSessionOwnerProbe>>
|
||||
/** Recovery-exit stop requests only; a lease moves only on a later proven-absent probe. */
|
||||
stopOwnerProcess?: (pid: number, signal: 'SIGTERM' | 'SIGKILL') => void
|
||||
/** Host spawn-token process scan; null means the platform cannot enumerate, never "none". */
|
||||
|
||||
+48
@@ -2,6 +2,10 @@ import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
agentSessionLeaseFixture,
|
||||
agentSessionRecordFixture
|
||||
} from '../../../shared/agent-session-record.test-fixture'
|
||||
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
|
||||
import { StructuredAgentSessionLeaseRenewer } from './structured-agent-session-lease-renewer'
|
||||
|
||||
@@ -66,6 +70,50 @@ afterEach(async () => {
|
||||
})
|
||||
|
||||
describe('structured agent-session lease renewal', () => {
|
||||
it('commits one store transaction for the whole live-record sweep', async () => {
|
||||
const records = ['a', 'b'].map((suffix, index) => {
|
||||
const sessionId = `session-${suffix}`
|
||||
return agentSessionRecordFixture(
|
||||
agentSessionLeaseFixture({
|
||||
sessionId,
|
||||
runtimeKind: 'native',
|
||||
runtimeFence: index + 1,
|
||||
ownerProcess: {
|
||||
hostId: 'local',
|
||||
pid: 4200 + index,
|
||||
processStartTimeMs: NOW - 1_000,
|
||||
spawnToken: `spawn-${suffix}`
|
||||
},
|
||||
reservedSpawnToken: `spawn-${suffix}`,
|
||||
lastRenewedAt: NOW,
|
||||
leaseDeadlineAt: NOW + 30_000
|
||||
})
|
||||
)
|
||||
})
|
||||
const renewLeases = vi.fn(async (_renewals: readonly unknown[]) => records)
|
||||
const probeMany = vi.fn(
|
||||
async () =>
|
||||
new Map(
|
||||
records.map((record) => [
|
||||
record.sessionId,
|
||||
{ outcome: 'identity-matched', matchedOn: ['spawn-token'] } as const
|
||||
])
|
||||
)
|
||||
)
|
||||
const renewer = new StructuredAgentSessionLeaseRenewer({
|
||||
store: { listRecords: () => records, renewLeases } as unknown as AgentSessionRecordStore,
|
||||
probe: vi.fn(),
|
||||
probeMany,
|
||||
now: () => NOW + 10_000
|
||||
})
|
||||
|
||||
await renewer.renewNow()
|
||||
|
||||
expect(probeMany).toHaveBeenCalledOnce()
|
||||
expect(renewLeases).toHaveBeenCalledOnce()
|
||||
expect(renewLeases.mock.calls[0]?.[0]).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('drives renewal on the production interval', async () => {
|
||||
vi.useFakeTimers()
|
||||
const store = await liveStore()
|
||||
|
||||
@@ -18,6 +18,9 @@ export class StructuredAgentSessionLeaseRenewer {
|
||||
private readonly input: {
|
||||
store: AgentSessionRecordStore
|
||||
probe: (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>
|
||||
probeMany?: (
|
||||
records: readonly AgentSessionRecord[]
|
||||
) => Promise<Map<string, AgentSessionOwnerProbe>>
|
||||
now: () => number
|
||||
onRenewed?: (record: AgentSessionRecord) => Promise<void>
|
||||
onDeadTuiOwner?: (record: AgentSessionRecord, probe: AgentSessionOwnerProbe) => Promise<void>
|
||||
@@ -47,49 +50,87 @@ export class StructuredAgentSessionLeaseRenewer {
|
||||
}
|
||||
this.running = true
|
||||
try {
|
||||
await Promise.all(
|
||||
this.input.store
|
||||
.listRecords()
|
||||
.filter(
|
||||
(record) =>
|
||||
!record.lease.unreconciled &&
|
||||
record.lease.claimStatus === 'live' &&
|
||||
record.lease.ownerProcess !== null &&
|
||||
// A native record parked in recovery has no transport the host can vouch
|
||||
// for; renewing it keeps an orphan pid's lease reading as a healthy owner.
|
||||
!(
|
||||
record.lease.runtimeKind === 'native' &&
|
||||
(record.lease.handoffStage === 'recovering' ||
|
||||
record.lease.handoffStage === 'manual-recovery')
|
||||
)
|
||||
const records = this.input.store.listRecords().filter(
|
||||
(record) =>
|
||||
!record.lease.unreconciled &&
|
||||
record.lease.claimStatus === 'live' &&
|
||||
record.lease.ownerProcess !== null &&
|
||||
// A native record parked in recovery has no transport the host can vouch
|
||||
// for; renewing it keeps an orphan pid's lease reading as a healthy owner.
|
||||
!(
|
||||
record.lease.runtimeKind === 'native' &&
|
||||
(record.lease.handoffStage === 'recovering' ||
|
||||
record.lease.handoffStage === 'manual-recovery')
|
||||
)
|
||||
.map((record) => this.renewRecord(record))
|
||||
)
|
||||
const probes = await this.probe(records)
|
||||
const renewals: {
|
||||
sessionId: string
|
||||
fence: number
|
||||
childProbe: AgentSessionOwnerProbe
|
||||
now: number
|
||||
}[] = []
|
||||
const now = this.input.now()
|
||||
for (const record of records) {
|
||||
const probe = probes.get(record.sessionId)
|
||||
if (!probe) {
|
||||
continue
|
||||
}
|
||||
if (
|
||||
record.lease.runtimeKind === 'tui' &&
|
||||
isProvenDeadProbe(probe) &&
|
||||
this.input.onDeadTuiOwner
|
||||
) {
|
||||
try {
|
||||
await this.input.onDeadTuiOwner(record, probe)
|
||||
} catch (error) {
|
||||
this.input.onError?.({ sessionId: record.sessionId, error })
|
||||
}
|
||||
continue
|
||||
}
|
||||
renewals.push({
|
||||
sessionId: record.sessionId,
|
||||
fence: record.lease.runtimeFence,
|
||||
childProbe: probe,
|
||||
now
|
||||
})
|
||||
}
|
||||
try {
|
||||
const renewed = await this.input.store.renewLeases(renewals)
|
||||
await Promise.all(renewed.map((record) => this.input.onRenewed?.(record)))
|
||||
} catch (error) {
|
||||
for (const renewal of renewals) {
|
||||
this.input.onError?.({ sessionId: renewal.sessionId, error })
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
this.running = false
|
||||
}
|
||||
}
|
||||
|
||||
private async renewRecord(record: AgentSessionRecord): Promise<void> {
|
||||
private async probe(
|
||||
records: readonly AgentSessionRecord[]
|
||||
): Promise<Map<string, AgentSessionOwnerProbe>> {
|
||||
try {
|
||||
const probe = await this.input.probe(record)
|
||||
if (
|
||||
record.lease.runtimeKind === 'tui' &&
|
||||
isProvenDeadProbe(probe) &&
|
||||
this.input.onDeadTuiOwner
|
||||
) {
|
||||
await this.input.onDeadTuiOwner(record, probe)
|
||||
return
|
||||
if (this.input.probeMany) {
|
||||
return await this.input.probeMany(records)
|
||||
}
|
||||
const renewed = await this.input.store.renewLease({
|
||||
sessionId: record.sessionId,
|
||||
fence: record.lease.runtimeFence,
|
||||
childProbe: probe,
|
||||
now: this.input.now()
|
||||
})
|
||||
await this.input.onRenewed?.(renewed)
|
||||
const settled = await Promise.allSettled(records.map((record) => this.input.probe(record)))
|
||||
const probes = new Map<string, AgentSessionOwnerProbe>()
|
||||
for (const [index, result] of settled.entries()) {
|
||||
const record = records[index]
|
||||
if (result.status === 'fulfilled') {
|
||||
probes.set(record.sessionId, result.value)
|
||||
} else {
|
||||
this.input.onError?.({ sessionId: record.sessionId, error: result.reason })
|
||||
}
|
||||
}
|
||||
return probes
|
||||
} catch (error) {
|
||||
this.input.onError?.({ sessionId: record.sessionId, error })
|
||||
for (const record of records) {
|
||||
this.input.onError?.({ sessionId: record.sessionId, error })
|
||||
}
|
||||
return new Map()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
import { mkdtemp, readFile, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import type { AgentSessionRecord } from '../../shared/agent-session-record'
|
||||
import { AgentSessionRecordStore } from './agent-session-record-store'
|
||||
import { agentSessionStorePath } from './agent-session-record-store-file'
|
||||
|
||||
const NOW = 1_800_000_000_000
|
||||
const MATCHED = { outcome: 'identity-matched', matchedOn: ['spawn-token'] } as const
|
||||
const directories: string[] = []
|
||||
|
||||
async function establishOwner(
|
||||
store: AgentSessionRecordStore,
|
||||
directory: string,
|
||||
suffix: string
|
||||
): Promise<AgentSessionRecord> {
|
||||
const sessionId = `session-${suffix}`
|
||||
const spawnToken = `spawn-${suffix}`
|
||||
const reserved = await store.reserveOwner({
|
||||
sessionId,
|
||||
location: {
|
||||
executionHostId: 'local',
|
||||
wslDistro: null,
|
||||
workspaceId: 'workspace-1',
|
||||
workspaceKind: 'folder'
|
||||
},
|
||||
provider: 'codex',
|
||||
accountHome: { variable: 'CODEX_HOME', path: directory },
|
||||
runtimeKind: 'native',
|
||||
expectedFence: null,
|
||||
spawnToken,
|
||||
claimKeyId: 'key-1',
|
||||
handoffOperationId: null,
|
||||
probe: { outcome: 'reservation-unused' },
|
||||
operation: {
|
||||
callerKey: 'renewal-test',
|
||||
operationId: `${NOW}-${suffix.padStart(32, '0')}`,
|
||||
fingerprint: `create-${suffix}`
|
||||
},
|
||||
now: NOW
|
||||
})
|
||||
const fence = reserved.record.lease.runtimeFence
|
||||
await store.commitProcessIdentity({
|
||||
sessionId,
|
||||
fence,
|
||||
process: { hostId: 'local', pid: 4242, processStartTimeMs: NOW - 1_000, spawnToken },
|
||||
now: NOW
|
||||
})
|
||||
return store.proveOwner({
|
||||
sessionId,
|
||||
fence,
|
||||
link: {
|
||||
linkId: `link-${suffix}`,
|
||||
handle: { provider: 'codex', threadId: `thread-${suffix}` },
|
||||
origin: 'created',
|
||||
mintedAtFence: fence,
|
||||
observedAt: NOW
|
||||
},
|
||||
now: NOW
|
||||
})
|
||||
}
|
||||
|
||||
async function liveStore(): Promise<{ directory: string; store: AgentSessionRecordStore }> {
|
||||
const directory = await mkdtemp(join(tmpdir(), 'orca-lease-renewal-batch-'))
|
||||
directories.push(directory)
|
||||
const store = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
|
||||
await establishOwner(store, directory, 'a')
|
||||
await establishOwner(store, directory, 'b')
|
||||
return { directory, store }
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(directories.splice(0).map((path) => rm(path, { recursive: true, force: true })))
|
||||
})
|
||||
|
||||
describe('agent-session lease renewal batch', () => {
|
||||
it('refuses a stale fence without changing the record', async () => {
|
||||
const { store } = await liveStore()
|
||||
|
||||
await expect(
|
||||
store.renewLeases([
|
||||
{ sessionId: 'session-a', fence: 0, childProbe: MATCHED, now: NOW + 10_000 }
|
||||
])
|
||||
).rejects.toThrow('agent_session_checkpoint_stale')
|
||||
expect(store.getRecord('session-a')?.lease.lastRenewedAt).toBe(NOW)
|
||||
})
|
||||
|
||||
it('leaves every session durable when a later renewal was superseded', async () => {
|
||||
const { directory, store } = await liveStore()
|
||||
await store.evictProvenDeadOwner({
|
||||
sessionId: 'session-b',
|
||||
expectedFence: 1,
|
||||
probe: { outcome: 'pid-absent' },
|
||||
now: NOW + 5_000
|
||||
})
|
||||
const beforeDisk = await readFile(agentSessionStorePath(directory), 'utf-8')
|
||||
const beforeFirst = store.getRecord('session-a')
|
||||
|
||||
await expect(
|
||||
store.renewLeases([
|
||||
{ sessionId: 'session-a', fence: 1, childProbe: MATCHED, now: NOW + 10_000 },
|
||||
{ sessionId: 'session-b', fence: 1, childProbe: MATCHED, now: NOW + 10_000 }
|
||||
])
|
||||
).rejects.toThrow('agent_session_checkpoint_stale')
|
||||
|
||||
expect(store.getRecord('session-a')).toEqual(beforeFirst)
|
||||
expect(await readFile(agentSessionStorePath(directory), 'utf-8')).toBe(beforeDisk)
|
||||
const reopened = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
|
||||
expect(
|
||||
reopened
|
||||
.listRecords()
|
||||
.map((record) => record.sessionId)
|
||||
.sort()
|
||||
).toEqual(['session-a', 'session-b'])
|
||||
expect(reopened.listRecords().every((record) => record.providerHandleChain.length > 0)).toBe(
|
||||
true
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,38 @@
|
||||
import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication'
|
||||
import type { AgentSessionRecord } from '../../shared/agent-session-record'
|
||||
import { renewAgentSessionLease } from './agent-session-lease-transitions'
|
||||
import type { AgentSessionStoreState } from './agent-session-record-store-file'
|
||||
|
||||
export type AgentSessionLeaseRenewal = {
|
||||
sessionId: string
|
||||
fence: number
|
||||
childProbe: AgentSessionOwnerProbe
|
||||
now: number
|
||||
leaseTtlMs?: number
|
||||
}
|
||||
|
||||
export function renewAgentSessionLeases(
|
||||
state: AgentSessionStoreState,
|
||||
renewals: readonly AgentSessionLeaseRenewal[],
|
||||
defaultLeaseTtlMs: number
|
||||
): AgentSessionRecord[] {
|
||||
return renewals.map((args) => {
|
||||
const record = state.records.get(args.sessionId)
|
||||
if (!record) {
|
||||
throw new Error(
|
||||
state.unreadableRecords.has(args.sessionId)
|
||||
? 'execution_owner_reconciling'
|
||||
: 'agent_session_identity_required'
|
||||
)
|
||||
}
|
||||
const renewed = renewAgentSessionLease({
|
||||
record,
|
||||
fence: args.fence,
|
||||
childProbe: args.childProbe,
|
||||
now: args.now,
|
||||
leaseTtlMs: args.leaseTtlMs ?? defaultLeaseTtlMs
|
||||
})
|
||||
state.records.set(args.sessionId, renewed)
|
||||
return renewed
|
||||
})
|
||||
}
|
||||
@@ -2,6 +2,7 @@ import { describe, expect, it, vi } from 'vitest'
|
||||
import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record'
|
||||
import {
|
||||
PROCESS_START_TIME_TOLERANCE_MS,
|
||||
probeAgentSessionProcessIdentities,
|
||||
probeAgentSessionProcessIdentity,
|
||||
probeAgentSessionReservation,
|
||||
readProcessStartTimeMs,
|
||||
@@ -28,6 +29,32 @@ function deps(overrides: AgentSessionProcessProbeDeps = {}): AgentSessionProcess
|
||||
}
|
||||
|
||||
describe('owner identity probe', () => {
|
||||
it('shares one process-table read across a batch without weakening identity checks', async () => {
|
||||
const readProcessStartTimes = vi.fn(
|
||||
async () =>
|
||||
new Map([
|
||||
[4242, START_TIME],
|
||||
[4243, START_TIME + PROCESS_START_TIME_TOLERANCE_MS + 1]
|
||||
])
|
||||
)
|
||||
const probes = await probeAgentSessionProcessIdentities({
|
||||
identities: [IDENTITY, { ...IDENTITY, pid: 4243, spawnToken: 'spawn-b' }],
|
||||
deps: {
|
||||
isPidPresent: () => true,
|
||||
readEchoedSpawnToken: async () => null,
|
||||
readProcessStartTimesMs: readProcessStartTimes,
|
||||
platform: 'darwin'
|
||||
}
|
||||
})
|
||||
|
||||
expect(readProcessStartTimes).toHaveBeenCalledOnce()
|
||||
expect(readProcessStartTimes).toHaveBeenCalledWith([4242, 4243], 'darwin')
|
||||
expect(probes).toEqual([
|
||||
{ outcome: 'identity-matched', matchedOn: ['process-start-time'] },
|
||||
{ outcome: 'identity-mismatch', field: 'process-start-time' }
|
||||
])
|
||||
})
|
||||
|
||||
it('reports an observed exit without touching the host', () => {
|
||||
return expect(
|
||||
probeAgentSessionProcessIdentity({
|
||||
|
||||
@@ -76,6 +76,39 @@ async function readDarwinProcessStartTimeMs(pid: number): Promise<number | null>
|
||||
}
|
||||
}
|
||||
|
||||
async function readDarwinProcessStartTimesMs(
|
||||
pids: readonly number[]
|
||||
): Promise<Map<number, number | null>> {
|
||||
const observed = new Map<number, number | null>()
|
||||
if (pids.length === 0) {
|
||||
return observed
|
||||
}
|
||||
try {
|
||||
const result = await runProcess({
|
||||
program: 'ps',
|
||||
args: ['-o', 'pid=,lstart=', '-p', pids.join(',')],
|
||||
timeoutMs: PROCESS_START_TIME_TIMEOUT_MS
|
||||
})
|
||||
if (result.timedOut || result.code !== 0) {
|
||||
return observed
|
||||
}
|
||||
for (const line of result.stdout.split('\n')) {
|
||||
const match = /^\s*(\d+)\s+(.+?)\s*$/.exec(line)
|
||||
if (!match) {
|
||||
continue
|
||||
}
|
||||
const pid = Number(match[1])
|
||||
const parsed = Date.parse(match[2])
|
||||
if (Number.isSafeInteger(pid) && Number.isFinite(parsed)) {
|
||||
observed.set(pid, parsed)
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// A missing process table is unknown, never evidence that every owner exited.
|
||||
}
|
||||
return observed
|
||||
}
|
||||
|
||||
async function readWindowsProcessStartTimeMs(pid: number): Promise<number | null> {
|
||||
try {
|
||||
const script =
|
||||
@@ -116,6 +149,57 @@ export async function readProcessStartTimeMs(
|
||||
return null
|
||||
}
|
||||
|
||||
export async function readProcessStartTimesMs(
|
||||
pids: readonly number[],
|
||||
platform: NodeJS.Platform = process.platform
|
||||
): Promise<Map<number, number | null>> {
|
||||
const uniquePids = [...new Set(pids)]
|
||||
if (platform === 'darwin') {
|
||||
const table = await readDarwinProcessStartTimesMs(uniquePids)
|
||||
return new Map(uniquePids.map((pid) => [pid, table.get(pid) ?? null]))
|
||||
}
|
||||
return new Map(
|
||||
await Promise.all(
|
||||
uniquePids.map(async (pid) => [pid, await readProcessStartTimeMs(pid, platform)] as const)
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
export type AgentSessionProcessBatchProbeDeps = Omit<
|
||||
AgentSessionProcessProbeDeps,
|
||||
'readProcessStartTimeMs'
|
||||
> & {
|
||||
readProcessStartTimesMs?: (
|
||||
pids: readonly number[],
|
||||
platform?: NodeJS.Platform
|
||||
) => Promise<Map<number, number | null>>
|
||||
}
|
||||
|
||||
export async function probeAgentSessionProcessIdentities(args: {
|
||||
identities: readonly AgentSessionProcessIdentity[]
|
||||
deps?: AgentSessionProcessBatchProbeDeps
|
||||
}): Promise<AgentSessionOwnerProbe[]> {
|
||||
const deps = args.deps ?? {}
|
||||
const platform = deps.platform ?? process.platform
|
||||
const pids = args.identities
|
||||
.filter((identity) => identity.processStartTimeMs !== null)
|
||||
.map((identity) => identity.pid)
|
||||
const readStartTimes = deps.readProcessStartTimesMs ?? readProcessStartTimesMs
|
||||
const startTimes = await readStartTimes(pids, platform).catch(() => new Map())
|
||||
return Promise.all(
|
||||
args.identities.map((identity) =>
|
||||
probeAgentSessionProcessIdentity({
|
||||
identity,
|
||||
deps: {
|
||||
...deps,
|
||||
platform,
|
||||
readProcessStartTimeMs: async (pid) => startTimes.get(pid) ?? null
|
||||
}
|
||||
})
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Probe one recorded owner. `observedExit` short-circuits everything: Orca watching that exact
|
||||
* process exit is the strongest evidence available.
|
||||
|
||||
@@ -26,10 +26,13 @@ import {
|
||||
commitAgentSessionProcessIdentity,
|
||||
evictAgentSessionOwner,
|
||||
proveAgentSessionOwner,
|
||||
renewAgentSessionLease,
|
||||
setAgentSessionJournalCheckpoint,
|
||||
type AgentSessionProcessIdentityCommit
|
||||
} from './agent-session-lease-transitions'
|
||||
import {
|
||||
renewAgentSessionLeases,
|
||||
type AgentSessionLeaseRenewal
|
||||
} from './agent-session-lease-renewal'
|
||||
import { applyAgentSessionRestartAdjudication } from './agent-session-restart-lease-transitions'
|
||||
import { agentSessionReconciliationTargetMatches } from './agent-session-reconciliation-target'
|
||||
import { replaceAgentSessionRecordOptions } from './agent-session-record-options'
|
||||
@@ -198,21 +201,14 @@ export class AgentSessionRecordStore {
|
||||
})
|
||||
}
|
||||
|
||||
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 renewLease(args: AgentSessionLeaseRenewal): Promise<AgentSessionRecord> {
|
||||
const [renewed] = await this.renewLeases([args])
|
||||
return renewed
|
||||
}
|
||||
|
||||
async renewLeases(renewals: readonly AgentSessionLeaseRenewal[]): Promise<AgentSessionRecord[]> {
|
||||
return this.transact(() =>
|
||||
renewAgentSessionLeases(this.state, renewals, AGENT_SESSION_LEASE_TTL_MS)
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -20,6 +20,7 @@ import type { StructuredAgentSessionHandoffTransport } from '../native-chat/agen
|
||||
import { setStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry'
|
||||
import { AgentSessionRecordStore } from './agent-session-record-store'
|
||||
import {
|
||||
probeAgentSessionProcessIdentities,
|
||||
probeAgentSessionProcessIdentity,
|
||||
probeAgentSessionReservation
|
||||
} from './agent-session-process-identity-probe'
|
||||
@@ -113,6 +114,7 @@ async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<Install
|
||||
journalRoot: deps.stateDirectory,
|
||||
claimKeyId: deps.claimKeyId,
|
||||
probeOwner: createStructuredAgentSessionOwnerProbe(deps.hostId),
|
||||
probeOwners: createStructuredAgentSessionOwnerProbes(deps.hostId),
|
||||
resolveLaunchEnv: async () =>
|
||||
(await (deps.resolveLaunchEnv ?? resolveLoginShellEnvironment)()) as Record<string, string>,
|
||||
onEventSinkError: ({ sessionId, error }) =>
|
||||
@@ -182,6 +184,38 @@ export function createStructuredAgentSessionOwnerProbe(
|
||||
}
|
||||
}
|
||||
|
||||
export function createStructuredAgentSessionOwnerProbes(
|
||||
hostId: string
|
||||
): (records: readonly AgentSessionRecord[]) => Promise<Map<string, AgentSessionOwnerProbe>> {
|
||||
const probeOne = createStructuredAgentSessionOwnerProbe(hostId)
|
||||
return async (records) => {
|
||||
const results = new Map<string, AgentSessionOwnerProbe>()
|
||||
const localOwners: {
|
||||
record: AgentSessionRecord
|
||||
owner: NonNullable<AgentSessionRecord['lease']['ownerProcess']>
|
||||
}[] = []
|
||||
for (const record of records) {
|
||||
const owner = record.lease.ownerProcess
|
||||
if (owner?.hostId === hostId) {
|
||||
localOwners.push({ record, owner })
|
||||
} else {
|
||||
results.set(record.sessionId, await probeOne(record))
|
||||
}
|
||||
}
|
||||
const probes = await probeAgentSessionProcessIdentities({
|
||||
identities: localOwners.map(({ owner }) => owner),
|
||||
deps: { readEchoedSpawnToken: readEchoedAgentSessionSpawnToken }
|
||||
})
|
||||
for (const [index, { record }] of localOwners.entries()) {
|
||||
results.set(
|
||||
record.sessionId,
|
||||
probes[index] ?? { outcome: 'indeterminate', reason: 'owner probe returned no result' }
|
||||
)
|
||||
}
|
||||
return results
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The only provider-side trace a reservation can leave in its own record: a handle link minted at
|
||||
* this fence. `proveAgentSessionOwner` refuses to append one before an identity is committed, so a
|
||||
|
||||
@@ -106,6 +106,7 @@ describe('NativeChatStructuredSession', () => {
|
||||
it('wires local structured file links through the native chat opener', () => {
|
||||
render(
|
||||
<NativeChatStructuredSession
|
||||
isVisible
|
||||
tabId="structured-tab-1"
|
||||
sessionId="session-1"
|
||||
target={{ kind: 'local' }}
|
||||
@@ -121,6 +122,7 @@ describe('NativeChatStructuredSession', () => {
|
||||
it('routes a bare model command to the native option picker', async () => {
|
||||
render(
|
||||
<NativeChatStructuredSession
|
||||
isVisible
|
||||
tabId="structured-tab-1"
|
||||
sessionId="session-1"
|
||||
target={{ kind: 'local' }}
|
||||
|
||||
@@ -29,6 +29,7 @@ export function NativeChatStructuredSession(props: {
|
||||
sessionId: string
|
||||
target: RuntimeClientTarget
|
||||
agent: AgentType
|
||||
isVisible: boolean
|
||||
allowFileUriLinks: boolean
|
||||
onSwitchToTerminal?: () => void
|
||||
orchestrationDispatchStatus?: AgentStatusOrchestrationContext['dispatchStatus']
|
||||
|
||||
@@ -38,6 +38,7 @@ export type NativeChatStructuredViewProps = NativeChatOrchestrationProps & {
|
||||
sessionId: string
|
||||
target: RuntimeClientTarget
|
||||
agent: AgentType
|
||||
isVisible: boolean
|
||||
allowFileUriLinks: boolean
|
||||
onSwitchToTerminal?: () => void
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
// @vitest-environment happy-dom
|
||||
|
||||
import { act, renderHook, waitFor } from '@testing-library/react'
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { act, cleanup, renderHook, waitFor } from '@testing-library/react'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type {
|
||||
AgentJournalCursor,
|
||||
AgentJournalRenderItem
|
||||
@@ -77,6 +77,8 @@ function page(
|
||||
}
|
||||
|
||||
describe('useStructuredAgentSessionRead history window', () => {
|
||||
afterEach(cleanup)
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
mocks.subscribe.mockResolvedValue({ unsubscribe: vi.fn() })
|
||||
@@ -155,4 +157,36 @@ describe('useStructuredAgentSessionRead history window', () => {
|
||||
expect(result.current.state.items).toHaveLength(301)
|
||||
expect(result.current.state.items[0]?.itemId).toBe('oldest')
|
||||
})
|
||||
|
||||
it('refreshes only visible structured sessions when the app regains focus', async () => {
|
||||
const hasFocus = vi.spyOn(document, 'hasFocus').mockReturnValue(true)
|
||||
mocks.call.mockResolvedValue({ ok: true, page: page('tail', [], false) })
|
||||
const visible = renderHook(() =>
|
||||
useStructuredAgentSessionRead({
|
||||
sessionId: 'session-visible',
|
||||
target: LOCAL_TARGET,
|
||||
isVisible: true
|
||||
})
|
||||
)
|
||||
const hidden = renderHook(() =>
|
||||
useStructuredAgentSessionRead({
|
||||
sessionId: 'session-hidden',
|
||||
target: LOCAL_TARGET,
|
||||
isVisible: false
|
||||
})
|
||||
)
|
||||
await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(2))
|
||||
|
||||
act(() => window.dispatchEvent(new Event('focus')))
|
||||
|
||||
await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(3))
|
||||
expect(mocks.call).toHaveBeenLastCalledWith(LOCAL_TARGET, 'agentSession.history', {
|
||||
sessionId: 'session-visible',
|
||||
direction: 'tail',
|
||||
limit: AGENT_SESSION_HISTORY_MAX_LIMIT
|
||||
})
|
||||
visible.unmount()
|
||||
hidden.unmount()
|
||||
hasFocus.mockRestore()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -49,12 +49,23 @@ function createReconnectScheduler(args: { shouldStop: () => boolean; reconnect:
|
||||
export function useStructuredAgentSessionRead(args: {
|
||||
sessionId: string
|
||||
target: RuntimeClientTarget
|
||||
isVisible?: boolean
|
||||
}) {
|
||||
const { sessionId, target } = args
|
||||
const { sessionId, target, isVisible = true } = args
|
||||
const [state, dispatch] = useReducer(reduceStructuredAgentSession, EMPTY_STRUCTURED_AGENT_SESSION)
|
||||
const stateRef = useRef(state)
|
||||
const resumeCursorRef = useRef(state.cursor)
|
||||
const [loadingOlder, setLoadingOlder] = useState(false)
|
||||
const refreshOnFocusRef = useRef<() => void>(() => {})
|
||||
|
||||
useEffect(() => {
|
||||
if (!isVisible) {
|
||||
return
|
||||
}
|
||||
const refresh = (): void => refreshOnFocusRef.current()
|
||||
window.addEventListener('focus', refresh)
|
||||
return () => window.removeEventListener('focus', refresh)
|
||||
}, [isVisible])
|
||||
|
||||
useEffect(() => {
|
||||
stateRef.current = state
|
||||
@@ -198,7 +209,7 @@ export function useStructuredAgentSessionRead(args: {
|
||||
}
|
||||
})
|
||||
}
|
||||
window.addEventListener('focus', refreshOnFocus)
|
||||
refreshOnFocusRef.current = refreshOnFocus
|
||||
void refreshTail()
|
||||
.then(() => open())
|
||||
.catch((error) => {
|
||||
@@ -207,7 +218,9 @@ export function useStructuredAgentSessionRead(args: {
|
||||
})
|
||||
return () => {
|
||||
stopped = true
|
||||
window.removeEventListener('focus', refreshOnFocus)
|
||||
if (refreshOnFocusRef.current === refreshOnFocus) {
|
||||
refreshOnFocusRef.current = () => {}
|
||||
}
|
||||
reconnectScheduler.dispose()
|
||||
coalescer.dispose()
|
||||
unsubscribe()
|
||||
|
||||
@@ -97,7 +97,8 @@ describe('useStructuredAgentSession options', () => {
|
||||
useStructuredAgentSession({
|
||||
sessionId: 'session-1',
|
||||
target: LOCAL_TARGET,
|
||||
agent: 'codex'
|
||||
agent: 'codex',
|
||||
isVisible: true
|
||||
})
|
||||
)
|
||||
await waitFor(() => expect(result.current.optionSnapshot).toHaveLength(2))
|
||||
@@ -128,7 +129,8 @@ describe('useStructuredAgentSession options', () => {
|
||||
useStructuredAgentSession({
|
||||
sessionId: 'session-1',
|
||||
target: LOCAL_TARGET,
|
||||
agent: 'codex'
|
||||
agent: 'codex',
|
||||
isVisible: true
|
||||
})
|
||||
)
|
||||
await waitFor(() => expect(result.current.optionSnapshot).toHaveLength(2))
|
||||
@@ -174,7 +176,8 @@ describe('useStructuredAgentSession options', () => {
|
||||
useStructuredAgentSession({
|
||||
sessionId: 'session-1',
|
||||
target: LOCAL_TARGET,
|
||||
agent: 'codex'
|
||||
agent: 'codex',
|
||||
isVisible: true
|
||||
})
|
||||
)
|
||||
await waitFor(() => expect(result.current.optionSnapshot).toHaveLength(2))
|
||||
@@ -228,7 +231,8 @@ describe('useStructuredAgentSession options', () => {
|
||||
useStructuredAgentSession({
|
||||
sessionId: 'session-1',
|
||||
target: LOCAL_TARGET,
|
||||
agent: 'codex'
|
||||
agent: 'codex',
|
||||
isVisible: true
|
||||
})
|
||||
)
|
||||
await waitFor(() => expect(result.current.optionSnapshot).toHaveLength(2))
|
||||
@@ -273,7 +277,8 @@ describe('useStructuredAgentSession options', () => {
|
||||
useStructuredAgentSession({
|
||||
sessionId: 'session-1',
|
||||
target: LOCAL_TARGET,
|
||||
agent: 'codex'
|
||||
agent: 'codex',
|
||||
isVisible: true
|
||||
})
|
||||
)
|
||||
await waitFor(() => expect(result.current.optionSnapshot).toHaveLength(2))
|
||||
|
||||
@@ -37,12 +37,17 @@ export function useStructuredAgentSession(args: {
|
||||
sessionId: string
|
||||
target: RuntimeClientTarget
|
||||
agent: AgentType
|
||||
isVisible: boolean
|
||||
}) {
|
||||
const { agent, sessionId, target } = args
|
||||
const { agent, isVisible, sessionId, target } = args
|
||||
// Declared first: the hold is what gives a restored session its provider child back, and the
|
||||
// read below is useless for sending until it lands.
|
||||
useStructuredAgentSessionHold({ sessionId, target, surface: 'desktop-chat' })
|
||||
const { state, loadingOlder, loadOlder } = useStructuredAgentSessionRead({ sessionId, target })
|
||||
const { state, loadingOlder, loadOlder } = useStructuredAgentSessionRead({
|
||||
sessionId,
|
||||
target,
|
||||
isVisible
|
||||
})
|
||||
const stateRef = useRef(state)
|
||||
const [writeError, setWriteError] = useState<string | null>(null)
|
||||
const operationIds = useRef(new Map<string, string>())
|
||||
|
||||
@@ -380,6 +380,7 @@ export default function TabGroupPanel({
|
||||
tabId={activeTab.id}
|
||||
sessionId={activeTab.entityId}
|
||||
agent={activeTab.agentSessionAgent ?? 'codex'}
|
||||
isVisible={isVisible}
|
||||
target={structuredRuntimeTarget}
|
||||
allowFileUriLinks={structuredFileLinksEnabled}
|
||||
/>
|
||||
|
||||
@@ -3175,6 +3175,7 @@ function TerminalPane(
|
||||
tabId={unifiedTabId ?? tabId}
|
||||
sessionId={structuredSessionId}
|
||||
agent="codex"
|
||||
isVisible={isRendererVisible}
|
||||
target={{ kind: 'local' }}
|
||||
allowFileUriLinks
|
||||
orchestrationDispatchStatus={chatPaneDispatchStatus}
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
# Structured lease-renewal scale report
|
||||
|
||||
## Outcome
|
||||
|
||||
The 50-record renewal sweep now performs one durable store transaction and one Darwin process-table command. Structured-chat focus refresh is registered only while a chat pane is visible.
|
||||
|
||||
## Measurements
|
||||
|
||||
Measured on base `8572614c38` with the same 50-record durable benchmark:
|
||||
|
||||
| Path | Before | After |
|
||||
| ---------------------------------- | -------------------------: | ---------------------------: |
|
||||
| Durable renewal sweep, median of 3 | 5,358 ms | 30.1 ms |
|
||||
| Durable renewal sweep, samples | 6,772 / 5,358 / 4,445 ms | 51.3 / 26.1 / 30.1 ms |
|
||||
| Darwin process liveness | 1,268 ms, 50 `ps` children | 11.0 ms median, 1 `ps` child |
|
||||
|
||||
The durable sweep improved about 178x at 50 records. The process probe now asks one targeted `ps` invocation for all unique owner PIDs and retains the same PID-presence, start-time tolerance, spawn-token, contradiction, and indeterminate-result rules.
|
||||
|
||||
## Semantics
|
||||
|
||||
- `renewLeases` applies every per-record fence check inside one transaction. A stale or superseded fence aborts the full sweep; the transaction queue restores every in-memory map and does not publish a new file.
|
||||
- Persistence continues to write and fsync a temp file, copy the still-live primary to the backup, then atomically rename the temp over the primary. The live primary is never renamed aside or absent.
|
||||
- Renewal still considers every live, reconciled owner every 10 seconds. I did not add a near-expiry skip: the 10-second pass also re-proves liveness and drives dead-TUI recovery, while moving to a 20-second cadence would leave only one timer interval before the 30-second deadline. With the measured 30 ms batch, keeping the existing safety cadence is cheap and preserves semantics.
|
||||
- Focus refresh is scoped to visible structured chats. Hidden mounted chats keep their existing stream subscription but install no `window.focus` handler and issue no focus-triggered history RPC.
|
||||
|
||||
## Concurrent app-server cap
|
||||
|
||||
No hard cap belongs in this fix. Fifty idle servers consume meaningful memory but remain viable on the measured 64 GiB machine, and a fixed global number would arbitrarily reject valid workflows on larger hosts while still being unsafe on smaller ones. If product policy needs admission control later, it should be a resource-aware budget with explicit queued/parked UX at the app-server launch boundary, not a hidden numeric cap in lease renewal.
|
||||
|
||||
## Verification
|
||||
|
||||
- Targeted suites: 91 tests passed across the record store, renewal batching, process identity, structured runtime, structured read/focus, and structured session behavior.
|
||||
- New correctness coverage proves stale-fence refusal, superseded-record refusal, full in-memory and on-disk rollback, and successful reopening of every record.
|
||||
- RED proof: temporarily changing the renewer back to one `renewLeases([record])` call per record made `commits one store transaction for the whole live-record sweep` fail with `expected ... once, but got 2 times`; restoring the batch made it pass.
|
||||
- `pnpm run typecheck`, `pnpm exec oxlint`, and targeted `pnpm exec oxfmt --check` were run. A repo-wide `oxfmt --check .` also ran and reported 40 pre-existing unrelated files, including owner-authored untracked artifacts; none were modified.
|
||||
Reference in New Issue
Block a user