Merge remote-tracking branch 'origin/brennanb2025/structured-tui-adoption' into brennanb2025/structured-session-eviction

# Conflicts:
#	src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts
This commit is contained in:
Brennan Benson
2026-08-23 16:57:21 -07:00
19 changed files with 538 additions and 125 deletions
@@ -22,8 +22,10 @@ export type CodexStructuredLaunchResolverDeps = {
/** Absolute path of a workspace on this host. Rejects when the workspace no
* longer resolves, which is the case a stale mobile client hits. */
resolveWorkspacePath: (workspaceId: string) => Promise<string>
/** Overridden in tests; production scans PATH and version-manager dirs. */
resolveCommand?: () => string
/** Overridden in tests; production scans the boot-cached PATH and version-manager dirs. */
resolveCommand?: (options?: { pathEnv?: string | null; homePath?: string }) => string
/** Fresh shell/configured environment for this spawn; never written to the session record. */
resolveEnvironment?: () => Promise<NodeJS.ProcessEnv>
resolveRollout?: typeof resolvePinnedCodexRolloutProof
}
@@ -50,7 +52,13 @@ export function createCodexStructuredLaunchResolver(
if (accountHome.variable !== 'CODEX_HOME') {
throw new Error(`codex sessions pin CODEX_HOME, not ${accountHome.variable}`)
}
const command = (deps.resolveCommand ?? resolveCodexCommand)()
const environment = await deps.resolveEnvironment?.()
const pathEnv = environment?.PATH ?? environment?.Path ?? null
const homePath = environment?.HOME ?? environment?.USERPROFILE
const command = (deps.resolveCommand ?? resolveCodexCommand)({
pathEnv,
...(homePath ? { homePath } : {})
})
const { spawnCmd, spawnArgs } = getSpawnArgsForWindows(command, CODEX_APP_SERVER_ARGS)
const head = agentSessionProviderHandleChainHead(record.providerHandleChain)
const resumeThreadId = head?.handle.provider === 'codex' ? head.handle.threadId : null
@@ -59,7 +67,7 @@ export function createCodexStructuredLaunchResolver(
args: spawnArgs,
cwd: await deps.resolveWorkspacePath(location.workspaceId),
codexHome: accountHome.path,
...(record.launchEnv ? { env: { ...record.launchEnv } } : {}),
...(environment ? { env: { ...environment } as Record<string, string> } : {}),
// An empty chain is a session that has never proved a thread, so it
// starts one; anything else resumes the last link this session proved.
resumeThreadId,
+6 -2
View File
@@ -56,8 +56,12 @@ export async function renameDurable(tmpPath: string, finalPath: string): Promise
* other work between "the new content is durable" and "the new content is visible" — a backup
* rotation that has to happen while the old file is still in place, for instance.
*/
export async function writeTempFileDurable(tmpPath: string, payload: string): Promise<void> {
const handle = await open(tmpPath, 'w')
export async function writeTempFileDurable(
tmpPath: string,
payload: string,
mode?: number
): Promise<void> {
const handle = await open(tmpPath, 'w', mode)
try {
await handle.writeFile(payload, 'utf-8')
await handle.sync()
@@ -11,6 +11,7 @@ import type {
StructuredAgentSessionHostSession
} from './structured-agent-session-host-types'
import type { StructuredAgentSessionHostRuntimeState } from './structured-agent-session-host-runtime-state'
import type { StructuredAgentSessionTaskQueue } from './structured-agent-session-task-queue'
export type StructuredAgentSessionAttachContext = {
deps: StructuredAgentSessionHostDeps
@@ -26,6 +27,8 @@ export type StructuredAgentSessionAttachContext = {
snapshot: (sessionId: string, journal: AgentSessionJournal, fence: number) => void
publish: (sessionId: string, journal: AgentSessionJournal) => void
}
tasks: StructuredAgentSessionTaskQueue
reconcileLeases: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
serialize: <T>(sessionId: string, task: () => Promise<T>) => Promise<T>
now: () => number
}
@@ -1,11 +1,9 @@
// The attach orchestration, lifted out of the host.
// The host's attach, lifted out of the host class.
//
// The host is a coordinator: it serializes work per session and tracks in-flight attaches. THIS is
// the part that decides what attaching means — reconcile the lease, resolve any pending recovery,
// bind an event sink, and publish the right thing to subscribers depending on whether the journal
// was recovered, re-fenced, or unchanged. Keeping it here means the host file stays a surface
// rather than an implementation, and this sequence can be read without scrolling past everything
// else a session can do.
// Attach is the one operation that touches every collaborator the host owns — the lease
// reconciler, the recovery resolver, the event sink, the journal, the subscriber set and the task
// queue — so leaving it inline made the host grow every time any of them did. The host keeps the
// state; this owns the ordering between them.
import { randomUUID } from 'node:crypto'
import type {
@@ -13,69 +11,74 @@ import type {
AgentSessionMutationResult
} from '../../../shared/agent-session-wire'
import type { AgentSessionAttachParams } from './structured-agent-session-attach'
import { refuseAgentSessionMutation } from './structured-agent-session-mutation-admission'
import { performAttach } from './structured-agent-session-attach-flow'
import { pinnedAgentSessionLaunchEnv } from './structured-agent-session-launch-env'
import { refuseAgentSessionMutation } from './structured-agent-session-mutation-admission'
import type { StructuredAgentSessionAttachContext } from './structured-agent-session-attach-context'
export async function attachStructuredAgentSession(
export function attachStructuredAgentSession(
context: StructuredAgentSessionAttachContext,
callerKey: string,
params: AgentSessionAttachParams
): Promise<AgentSessionMutationResult<AgentSessionAttachResult>> {
const sessionId = params.envelope.sessionId
const unreconciled = await context.reconcileLeases(sessionId)
if (unreconciled) {
return refuseAgentSessionMutation(unreconciled)
}
await context.runtimeState.resolveRecovery(sessionId)
const eventSink = context.runtimeState.eventSinkFor(sessionId)
const attached = await performAttach({
store: context.deps.store,
adapter: context.deps.adapter,
journalRoot: context.deps.journalRoot,
eventSink: eventSink.sink,
onAcquiring: () => eventSink.unbind(),
beforeJournalOpen: async () => {
eventSink.unbind()
await eventSink.drained()
},
authority: {
spawnToken: context.deps.mintSpawnToken?.() ?? randomUUID(),
claimKeyId: context.deps.claimKeyId,
handoffOperationId: params.envelope.clientOperationId,
probe: await context.runtimeState.probeOwner(sessionId),
...(await pinnedAgentSessionLaunchEnv(context.deps.resolveLaunchEnv, params))
},
callerKey,
params,
now: () => context.now(),
onAttached: (attached) => {
const fence = context.deps.store.getRecord(sessionId)?.lease.runtimeFence ?? 0
const previousFence = context.sessions.get(sessionId)?.fence
context.sessions.set(sessionId, {
journal: attached.journal,
params,
fence,
hasProviderChild: true
})
if (attached.recovery) {
context.subscribers.reset(sessionId, attached.journal, attached.recovery.reset, fence)
} else if (previousFence !== undefined && previousFence !== fence) {
context.subscribers.snapshot(sessionId, attached.journal, fence)
} else {
context.subscribers.publish(sessionId, attached.journal)
}
eventSink.bind({
journal: attached.journal,
fence,
publish: () => context.subscribers.publish(sessionId, attached.journal)
})
const attaching = context.serialize(sessionId, async () => {
const unreconciled = await context.reconcileLeases(sessionId)
if (unreconciled) {
return refuseAgentSessionMutation(unreconciled)
}
await context.runtimeState.resolveRecovery(sessionId)
const eventSink = context.runtimeState.eventSinkFor(sessionId)
const attached = await performAttach({
store: context.deps.store,
adapter: context.deps.adapter,
journalRoot: context.deps.journalRoot,
eventSink: eventSink.sink,
onAcquiring: () => eventSink.unbind(),
beforeJournalOpen: async () => {
eventSink.unbind()
await eventSink.drained()
},
authority: {
spawnToken: context.deps.mintSpawnToken?.() ?? randomUUID(),
claimKeyId: context.deps.claimKeyId,
handoffOperationId: params.envelope.clientOperationId,
probe: await context.runtimeState.probeOwner(sessionId),
...(await pinnedAgentSessionLaunchEnv(context.deps.resolveLaunchEnv, params))
},
callerKey,
params,
now: () => context.now(),
onAttached: (attached) => {
const fence = context.deps.store.getRecord(sessionId)?.lease.runtimeFence ?? 0
const previousFence = context.sessions.get(sessionId)?.fence
context.sessions.set(sessionId, {
journal: attached.journal,
params,
fence,
hasProviderChild: true
})
if (attached.recovery) {
context.subscribers.reset(sessionId, attached.journal, attached.recovery.reset, fence)
} else if (previousFence !== undefined && previousFence !== fence) {
context.subscribers.snapshot(sessionId, attached.journal, fence)
} else {
context.subscribers.publish(sessionId, attached.journal)
}
eventSink.bind({
journal: attached.journal,
fence,
publish: () => context.subscribers.publish(sessionId, attached.journal)
})
}
})
// Why: a failed attach that left no session behind must not strand a bound sink; the runtime
// caches one per session id and would hand this same closed instance to the next attempt.
if (!attached.ok && !context.sessions.has(sessionId)) {
eventSink.close()
context.runtimeState.discardEventSink(sessionId)
}
return attached
})
if (!attached.ok && !context.sessions.has(sessionId)) {
eventSink.close()
context.runtimeState.discardEventSink(sessionId)
}
return attached
return context.tasks.trackAttach(attaching)
}
@@ -15,7 +15,6 @@ import type { AgentSessionHandleProvider } from '../../../shared/agent-session-p
import type {
AgentSessionAccountHome,
AgentSessionExecutionLocation,
AgentSessionLaunchEnv,
AgentSessionOwnerRuntimeKind,
AgentSessionRecord
} from '../../../shared/agent-session-record'
@@ -63,7 +62,6 @@ export type AgentSessionAttachAuthority = {
claimKeyId: string
handoffOperationId: string | null
probe: AgentSessionOwnerProbe
launchEnv?: AgentSessionLaunchEnv
}
/** The fields that define WHICH session this call would attach to. Deliberately
@@ -180,7 +178,6 @@ export function reserveRequestFor(input: {
location: params.location,
provider: params.provider,
accountHome: params.accountHome,
...(authority.launchEnv ? { launchEnv: authority.launchEnv } : {}),
runtimeKind: params.runtimeKind,
expectedFence: params.envelope.expectedRuntimeFence,
spawnToken: authority.spawnToken,
@@ -166,7 +166,9 @@ export class StructuredAgentSessionHost {
runtimeState: this.runtimeState,
sessions: this.sessions,
subscribers: this.subscribers,
tasks: this.tasks,
reconcileLeases: (sessionId) => this.reconcileLeases(sessionId),
serialize: (sessionId, task) => this.serialize(sessionId, task),
now: () => this.now()
}
}
@@ -207,12 +209,7 @@ export class StructuredAgentSessionHost {
caller: StructuredAgentSessionCaller,
params: AgentSessionAttachParams
): Promise<AgentSessionMutationResult<AgentSessionAttachResult>> {
// Why here and not in the orchestration: the host owns per-session serialization and the
// in-flight attach set, so the module stays testable without a task queue.
const attaching = this.serialize(params.envelope.sessionId, () =>
attachStructuredAgentSession(this.attachContext(), caller.callerKey, params)
)
return this.tasks.trackAttach(attaching)
return attachStructuredAgentSession(this.attachContext(), caller.callerKey, params)
}
/** Reserve the lease an adopted TUI must hold before the write gate will admit its proof. */
@@ -1,19 +0,0 @@
import { describe, expect, it } from 'vitest'
import { hostTestAttachParams } from './structured-agent-session-host-test-data'
import { pinnedAgentSessionLaunchEnv } from './structured-agent-session-launch-env'
describe('pinnedAgentSessionLaunchEnv', () => {
it('layers the pinned account home over the shell environment', async () => {
await expect(
pinnedAgentSessionLaunchEnv(
async () => ({ EXAMPLE_GATEWAY_TOKEN: 'shell-exported', CODEX_HOME: '/shell/home' }),
hostTestAttachParams(null)
)
).resolves.toEqual({
launchEnv: {
EXAMPLE_GATEWAY_TOKEN: 'shell-exported',
CODEX_HOME: '/home/dev/.codex'
}
})
})
})
@@ -1,4 +1,4 @@
import { mkdtemp, rm } from 'node:fs/promises'
import { mkdtemp, readFile, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
@@ -45,14 +45,17 @@ afterEach(async () => {
})
describe('legacy agent session launch environment', () => {
it('durably pins the first environment resolved by a current reservation', async () => {
it('does not persist ambient launch variables', async () => {
const store = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
await store.reserveOwner(request())
await store.reserveOwner(
request({
expectedFence: 1,
spawnToken: 'spawn-b',
launchEnv: { ANTHROPIC_AUTH_TOKEN: 'pinned-token' },
launchEnv: {
PATH: '/custom/bin:/usr/bin',
ANTHROPIC_AUTH_TOKEN: 'fixture-token'
},
operation: {
callerKey: 'client-1',
operationId: `${NOW}-00000000000000000000000000000002`,
@@ -61,10 +64,11 @@ describe('legacy agent session launch environment', () => {
})
)
const raw = await readFile(join(directory, 'agent-sessions.json'), 'utf-8')
expect(raw).not.toContain('ANTHROPIC_AUTH_TOKEN')
expect(raw).not.toContain('"PATH"')
const reopened = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
expect(reopened.getRecord(SESSION)?.launchEnv).toEqual({
ANTHROPIC_AUTH_TOKEN: 'pinned-token'
})
expect(reopened.getRecord(SESSION)).not.toHaveProperty('launchEnv')
})
it('rejects an environment that could not be reloaded before writing it', async () => {
@@ -8,13 +8,14 @@
*/
import { createHash } from 'node:crypto'
import { mkdir, readFile, rm } from 'node:fs/promises'
import { chmod, mkdir, readFile, rm } from 'node:fs/promises'
import { dirname, join } from 'node:path'
import {
agentSessionOperationKey,
isAgentSessionOperationRow,
type AgentSessionOperationRow
} from '../../shared/agent-session-operation-ledger'
import { scrubAgentSessionRecordLaunchEnv } from '../../shared/agent-session-launch-env-scrub'
import type { AgentSessionRecord } from '../../shared/agent-session-record'
import { loadAgentSessionRecord } from '../../shared/agent-session-record-load'
import {
@@ -134,14 +135,16 @@ function parseState(
let needsRewrite = schemaVersion < AGENT_SESSION_STORE_SCHEMA_VERSION
if (typeof file.records === 'object' && file.records !== null) {
for (const [sessionId, value] of Object.entries(file.records)) {
const loaded = loadAgentSessionRecord(value)
const scrubbed = scrubAgentSessionRecordLaunchEnv(value)
const loaded = loadAgentSessionRecord(scrubbed.value)
needsRewrite ||= scrubbed.changed
if (loaded.status !== 'unusable' && loaded.record.sessionId === sessionId) {
state.records.set(sessionId, loaded.record)
needsRewrite ||= loaded.status === 'upgraded'
} else {
state.unreadableRecords.set(sessionId, {
reason: loaded.status === 'unusable' ? loaded.reason : 'record_key_session_id_mismatch',
raw: value
raw: scrubbed.value
})
needsRewrite ||= schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION
}
@@ -162,7 +165,9 @@ function parseState(
}
continue
}
state.unreadableRecords.set(sessionId, { reason: unusable.reason, raw: unusable.raw })
const scrubbed = scrubAgentSessionRecordLaunchEnv(unusable.raw)
needsRewrite ||= scrubbed.changed
state.unreadableRecords.set(sessionId, { reason: unusable.reason, raw: scrubbed.value })
}
}
if (typeof file.operations === 'object' && file.operations !== null) {
@@ -314,10 +319,12 @@ export async function saveAgentSessionStore(
filePath: string,
state: AgentSessionStoreState
): Promise<void> {
await mkdir(dirname(filePath), { recursive: true })
const directory = dirname(filePath)
await mkdir(directory, { recursive: true, mode: 0o700 })
await chmod(directory, 0o700)
const tmpPath = durableWriteTempPath(filePath)
try {
await writeTempFileDurable(tmpPath, serializeState(state))
await writeTempFileDurable(tmpPath, serializeState(state), 0o600)
// A failed rotation aborts the save. Recovery's fence floor assumes the backup is at most one
// committed generation behind; letting the primary advance past a stale backup breaks that.
await copyFileDurable(filePath, backupPath(filePath))
@@ -0,0 +1,206 @@
import { chmod, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import type { AgentSessionRecord } from '../../shared/agent-session-record'
import { createCodexStructuredLaunchResolver } from '../codex/codex-structured-launch-resolution'
import { AgentSessionRecordStore } from './agent-session-record-store'
import { agentSessionStorePath } from './agent-session-record-store-file'
import type { AgentSessionReserveRequest } from './agent-session-reservation-admission'
const NOW = 1_800_000_000_000
let directory: string
function resumableRecord(sessionId: string): AgentSessionRecord {
const threadId = `thread-${sessionId}`
return {
schemaVersion: 2,
sessionId,
location: {
executionHostId: 'local',
wslDistro: null,
workspaceId: `workspace-${sessionId}`,
workspaceKind: 'folder'
},
provider: 'codex',
providerHandleChain: [
{
linkId: `link-${sessionId}`,
handle: { provider: 'codex', threadId },
origin: 'created',
mintedAtFence: 1,
observedAt: NOW
}
],
accountHome: { variable: 'CODEX_HOME', path: `/accounts/${sessionId}` },
launchEnv: {
PATH: '/custom/bin:/usr/bin',
MY_COMPANY_CREDS: 'fixture-credentials',
GITHUB_TOKEN: 'fixture-token'
},
lease: {
sessionId,
runtimeKind: 'native',
runtimeFence: 1,
handoffStage: null,
provenHandleLinkId: null,
ownerProcess: null,
reservedSpawnToken: null,
leaseDeadlineAt: NOW,
lastRenewedAt: NOW,
handoffOperationId: null,
journalCheckpoint: null,
claimKeyId: 'key-1',
claimStatus: 'released',
unreconciled: false,
deathEvidence: null
},
createdAt: NOW,
updatedAt: NOW
}
}
function storePayload(): string {
const records = Object.fromEntries(
['session-live-a', 'session-live-b'].map((sessionId) => [
sessionId,
{ ...resumableRecord(sessionId), futureRecordField: { preserved: true } }
])
)
return JSON.stringify({
schemaVersion: 2,
hostId: 'local',
records,
operations: {},
retiredClaimKeys: [],
unusableRecords: {},
futureStoreField: { preserved: true }
})
}
function reserveRequest(): AgentSessionReserveRequest {
return {
sessionId: 'session-created',
location: {
executionHostId: 'local',
wslDistro: null,
workspaceId: 'workspace-created',
workspaceKind: 'folder'
},
provider: 'codex',
accountHome: { variable: 'CODEX_HOME', path: '/accounts/created' },
launchEnv: { PATH: '/usr/bin' },
runtimeKind: 'native',
expectedFence: null,
spawnToken: 'spawn-created',
claimKeyId: 'key-1',
handoffOperationId: null,
probe: { outcome: 'reservation-unused' },
operation: {
callerKey: 'client-1',
operationId: `${NOW}-00000000000000000000000000000001`,
fingerprint: 'fp-1'
},
now: NOW
}
}
beforeEach(async () => {
directory = await mkdtemp(join(tmpdir(), 'orca-agent-session-security-'))
})
afterEach(async () => {
await rm(directory, { recursive: true, force: true })
})
describe('agent session record store security', () => {
it('scrubs both committed generations without losing resumable sessions', async () => {
const filePath = agentSessionStorePath(directory)
await Promise.all([
writeFile(filePath, storePayload()),
writeFile(`${filePath}.bak`, storePayload())
])
if (process.platform !== 'win32') {
await chmod(directory, 0o755)
await Promise.all([chmod(filePath, 0o644), chmod(`${filePath}.bak`, 0o644)])
}
const store = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
expect(store.readOnly).toBe(false)
expect(
store
.listRecords()
.map((record) => record.sessionId)
.sort()
).toEqual(['session-live-a', 'session-live-b'])
for (const committedPath of [filePath, `${filePath}.bak`]) {
const raw = await readFile(committedPath, 'utf-8')
expect(raw).not.toContain('MY_COMPANY_CREDS')
expect(raw).not.toContain('GITHUB_TOKEN')
const parsed = JSON.parse(raw) as {
records: Record<string, AgentSessionRecord & { futureRecordField?: unknown }>
futureStoreField?: unknown
}
const { records } = parsed
expect(Object.keys(records).sort()).toEqual(['session-live-a', 'session-live-b'])
expect(records['session-live-a']).not.toHaveProperty('launchEnv')
expect(parsed.futureStoreField).toEqual({ preserved: true })
expect(records['session-live-a']?.futureRecordField).toEqual({ preserved: true })
if (process.platform !== 'win32') {
expect((await stat(committedPath)).mode & 0o777).toBe(0o600)
}
}
if (process.platform !== 'win32') {
expect((await stat(directory)).mode & 0o777).toBe(0o700)
}
const resolveCommand = vi.fn(() => '/custom/bin/codex')
const resolveLaunch = createCodexStructuredLaunchResolver({
store,
resolveWorkspacePath: async (workspaceId) => `/workspaces/${workspaceId}`,
resolveCommand,
resolveEnvironment: async () => ({
PATH: '/fresh/bin:/usr/bin',
MY_COMPANY_CREDS: 'fresh-fixture-credentials'
})
})
for (const sessionId of ['session-live-a', 'session-live-b']) {
const launch = await resolveLaunch({
identity: {
sessionId,
workspaceId: `workspace-${sessionId}`,
hostId: 'local',
agent: 'codex',
providerHandle: { kind: 'codex', threadId: `thread-${sessionId}` }
}
})
expect(launch).toMatchObject({
command: '/custom/bin/codex',
codexHome: `/accounts/${sessionId}`,
resumeThreadId: `thread-${sessionId}`,
env: {
PATH: '/fresh/bin:/usr/bin',
MY_COMPANY_CREDS: 'fresh-fixture-credentials'
}
})
}
expect(resolveCommand).toHaveBeenCalledTimes(2)
expect(resolveCommand).toHaveBeenCalledWith({ pathEnv: '/fresh/bin:/usr/bin' })
})
it.skipIf(process.platform === 'win32')(
'creates the directory and store owner-only',
async () => {
const nestedDirectory = join(directory, 'agent-sessions')
await chmod(directory, 0o755)
const store = await AgentSessionRecordStore.open({
directory: nestedDirectory,
hostId: 'local'
})
await store.reserveOwner(reserveRequest())
expect((await stat(nestedDirectory)).mode & 0o777).toBe(0o700)
expect((await stat(agentSessionStorePath(nestedDirectory))).mode & 0o777).toBe(0o600)
}
)
})
@@ -0,0 +1,145 @@
import { chmod, mkdir, readFile, rm } from 'node:fs/promises'
import { dirname } from 'node:path'
import { scrubAgentSessionRecordLaunchEnv } from '../../shared/agent-session-launch-env-scrub'
import { durableWriteTempPath, renameDurable, writeTempFileDurable } from '../durable-file-write'
import {
loadAgentSessionStore,
type LoadedAgentSessionStore
} from './agent-session-record-store-file'
import { withAgentSessionStoreTransactionLock } from './agent-session-store-transaction-lock'
const OWNER_DIRECTORY_MODE = 0o700
const OWNER_FILE_MODE = 0o600
type PreparedScrub = { filePath: string; tmpPath: string; published: boolean }
function scrubRecordMap(value: unknown): { value: unknown; changed: boolean } {
if (typeof value !== 'object' || value === null || Array.isArray(value)) {
return { value, changed: false }
}
let changed = false
const entries = Object.entries(value).map(([sessionId, record]) => {
const scrubbed = scrubAgentSessionRecordLaunchEnv(record)
changed ||= scrubbed.changed
return [sessionId, scrubbed.value]
})
return { value: changed ? Object.fromEntries(entries) : value, changed }
}
function scrubUnreadableRecordMap(value: unknown): { value: unknown; changed: boolean } {
if (typeof value !== 'object' || value === null || Array.isArray(value)) {
return { value, changed: false }
}
let changed = false
const entries = Object.entries(value).map(([sessionId, unusable]) => {
if (typeof unusable !== 'object' || unusable === null || Array.isArray(unusable)) {
return [sessionId, unusable]
}
const row = unusable as Record<string, unknown>
const scrubbed = scrubAgentSessionRecordLaunchEnv(row.raw)
changed ||= scrubbed.changed
return [sessionId, scrubbed.changed ? { ...row, raw: scrubbed.value } : row]
})
return { value: changed ? Object.fromEntries(entries) : value, changed }
}
function scrubStorePayload(raw: string): string | null {
let parsed: unknown
try {
parsed = JSON.parse(raw) as unknown
} catch {
return null
}
if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) {
return null
}
const store = parsed as Record<string, unknown>
if (!Number.isSafeInteger(store.schemaVersion) || (store.schemaVersion as number) < 0) {
return null
}
const records = scrubRecordMap(store.records)
const unreadableRecords = scrubUnreadableRecordMap(store.unusableRecords)
if (!records.changed && !unreadableRecords.changed) {
return null
}
return JSON.stringify({
...store,
...(records.changed ? { records: records.value } : {}),
...(unreadableRecords.changed ? { unusableRecords: unreadableRecords.value } : {})
})
}
async function chmodIfPresent(path: string, mode: number): Promise<void> {
try {
await chmod(path, mode)
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') {
throw error
}
}
}
export async function hardenAgentSessionStorePermissions(filePath: string): Promise<void> {
const directory = dirname(filePath)
await mkdir(directory, { recursive: true, mode: OWNER_DIRECTORY_MODE })
await chmod(directory, OWNER_DIRECTORY_MODE)
await Promise.all([
chmodIfPresent(filePath, OWNER_FILE_MODE),
chmodIfPresent(`${filePath}.bak`, OWNER_FILE_MODE)
])
}
async function prepareScrub(filePath: string): Promise<PreparedScrub | null> {
let raw: string
try {
raw = await readFile(filePath, 'utf-8')
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
return null
}
throw error
}
const payload = scrubStorePayload(raw)
if (payload === null) {
return null
}
const tmpPath = durableWriteTempPath(filePath)
await writeTempFileDurable(tmpPath, payload, OWNER_FILE_MODE)
return { filePath, tmpPath, published: false }
}
/** Scrubs each committed generation independently so backup-only sessions survive. */
export async function scrubAgentSessionStoreFiles(filePath: string): Promise<void> {
const prepared: PreparedScrub[] = []
try {
for (const candidate of [`${filePath}.bak`, filePath]) {
const scrub = await prepareScrub(candidate)
if (scrub) {
prepared.push(scrub)
}
}
// Both replacements are durable first; publishing backup then live never removes the live path.
for (const scrub of prepared) {
await renameDurable(scrub.tmpPath, scrub.filePath)
scrub.published = true
}
} finally {
await Promise.all(
prepared
.filter((scrub) => !scrub.published)
.map((scrub) => rm(scrub.tmpPath, { force: true }).catch(() => {}))
)
}
}
export async function loadProtectedAgentSessionStore(
filePath: string,
hostId: string
): Promise<LoadedAgentSessionStore> {
return withAgentSessionStoreTransactionLock(filePath, async () => {
await hardenAgentSessionStorePermissions(filePath)
await scrubAgentSessionStoreFiles(filePath)
await hardenAgentSessionStorePermissions(filePath)
return loadAgentSessionStore(filePath, hostId)
})
}
@@ -47,9 +47,9 @@ import {
import {
agentSessionStoreRevision,
agentSessionStorePath,
loadAgentSessionStore,
type AgentSessionStoreState
} from './agent-session-record-store-file'
import { loadProtectedAgentSessionStore } from './agent-session-record-store-security'
import {
AgentSessionStoreTransactionQueue,
markAgentSessionStoreLeasesUnreconciled
@@ -65,7 +65,7 @@ export class AgentSessionRecordStore {
static async open(args: { directory: string; hostId: string }): Promise<AgentSessionRecordStore> {
const filePath = agentSessionStorePath(args.directory)
const loaded = await loadAgentSessionStore(filePath, args.hostId)
const loaded = await loadProtectedAgentSessionStore(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)
@@ -34,6 +34,7 @@ export type AgentSessionReserveRequest = {
location: AgentSessionExecutionLocation
provider: AgentSessionHandleProvider
accountHome: AgentSessionAccountHome
/** Legacy internal input accepted for compatibility and deliberately discarded. */
launchEnv?: AgentSessionLaunchEnv
runtimeKind: AgentSessionReservation['runtimeKind']
/** Null when the session does not exist yet; otherwise the fence the caller last observed. */
@@ -129,10 +130,7 @@ export function applyAgentSessionReservation(
if (request.expectedFence === null) {
throw new Error('agent_session_conflict')
}
const pinned =
existing.launchEnv || !request.launchEnv
? existing
: { ...existing, launchEnv: { ...request.launchEnv }, updatedAt: request.now }
const { launchEnv: _discarded, ...pinned } = existing
return reserveAgentSessionOwner({
record: pinned,
expectedFence: request.expectedFence,
@@ -152,7 +150,6 @@ function createAgentSessionRecord(
provider: request.provider,
providerHandleChain: [],
accountHome: request.accountHome,
...(request.launchEnv ? { launchEnv: { ...request.launchEnv } } : {}),
createdAt: request.now,
updatedAt: request.now,
lease: {
@@ -1,4 +1,4 @@
import { mkdir } from 'node:fs/promises'
import { chmod, mkdir } from 'node:fs/promises'
import { dirname } from 'node:path'
import { lock } from 'proper-lockfile'
@@ -15,7 +15,9 @@ export async function withAgentSessionStoreTransactionLock<T>(
filePath: string,
apply: () => Promise<T>
): Promise<T> {
await mkdir(dirname(filePath), { recursive: true })
const directory = dirname(filePath)
await mkdir(directory, { recursive: true, mode: 0o700 })
await chmod(directory, 0o700)
const release = await lock(filePath, { realpath: false, retries: LOCK_RETRIES })
try {
return await apply()
+2
View File
@@ -10627,6 +10627,8 @@ export class OrcaRuntimeService {
// in a plain folder lands in the folder rather than failing to resolve.
resolveWorkspacePath: async (workspaceId) =>
(await this.resolveRuntimeFileTarget(`id:${workspaceId}`)).worktree.path,
resolveCodexOverrides: () =>
resolveTuiAgentLaunchEnv('codex', this.requireStore().getSettings().agentDefaultEnv),
handoffTransport: this.createStructuredAgentSessionHandoffTransport()
})
}
@@ -6,7 +6,7 @@
// that ship. The fake app-server answers the same JSON-RPC calls the real one
// does and pushes the same notifications and blocking requests back.
import { mkdtemp, rm } from 'node:fs/promises'
import { mkdtemp, readFile, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
@@ -213,6 +213,9 @@ let codex: CodexScript
let root: string
let dispatcher: RpcDispatcher
let cleanups: Map<string, () => void>
let bootEnvironmentReads: number
let codexOverrideReads: number
let configuredCodexProfile: string
/** Runs a one-shot method and returns its decoded reply. */
async function call(method: string, params: unknown): Promise<RpcResponse> {
@@ -290,6 +293,9 @@ beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'orca-structured-integration-'))
codex = fakeCodex()
cleanups = new Map()
bootEnvironmentReads = 0
codexOverrideReads = 0
configuredCodexProfile = 'configured'
const runtime = {
getRuntimeId: () => 'runtime-1',
getStructuredAgentSessionCreateSupport: async () => ({ supported: true }),
@@ -309,10 +315,18 @@ beforeEach(async () => {
claimKeyId: 'key-1',
resolveWorkspacePath: async (workspaceId) => `/repos/${workspaceId}`,
resolveCodexCommand: () => '/usr/local/bin/codex',
resolveLaunchEnv: async () => ({
EXAMPLE_GATEWAY_TOKEN: 'shell-exported',
CODEX_HOME: '/shell/home'
}),
resolveEnvironment: async () => {
bootEnvironmentReads += 1
return {
PATH: '/shell/bin:/usr/bin',
EXAMPLE_GATEWAY_TOKEN: 'shell-exported',
CODEX_HOME: '/shell/home'
}
},
resolveCodexOverrides: () => {
codexOverrideReads += 1
return { CODEX_PROFILE: configuredCodexProfile }
},
openCodexConnection: codex.openConnection,
readProcessStartTime: async () => 1_700_000_000_000
}).then(() => undefined),
@@ -394,9 +408,13 @@ describe('a structured codex session over agentSession.*', () => {
)
expect(created.snapshot.items).toEqual([])
expect(codex.live().launch.env).toMatchObject({
CODEX_PROFILE: 'configured',
EXAMPLE_GATEWAY_TOKEN: 'shell-exported',
CODEX_HOME: '/home/dev/.codex'
})
const store = await readFile(join(root, 'agent-sessions', 'agent-sessions.json'), 'utf-8')
expect(store).not.toContain('EXAMPLE_GATEWAY_TOKEN')
expect(store).not.toContain('"launchEnv"')
const stream = await subscribe('sub-first-send')
const body = { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] }
@@ -642,6 +660,25 @@ describe('a structured codex session over agentSession.*', () => {
])
})
it('caches shell exports but re-reads configured overrides for a resume', async () => {
const created = await ok<{ fence: number }>('agentSession.create', createIntentParams())
expect({ bootEnvironmentReads, codexOverrideReads }).toEqual({
bootEnvironmentReads: 1,
codexOverrideReads: 1
})
configuredCodexProfile = 'updated'
const resumed = await ok<{ fence: number }>('agentSession.ensure', attachParams(created.fence))
expect(resumed.fence).toBe(created.fence + 1)
expect(codex.live().resumedThreadId).toBe(THREAD)
expect(codex.live().launch.env).toMatchObject({ CODEX_PROFILE: 'updated' })
expect({ bootEnvironmentReads, codexOverrideReads }).toEqual({
bootEnvironmentReads: 1,
codexOverrideReads: 2
})
})
it('refuses to build a host for a client that never advertised the capability', async () => {
const replies: RpcResponse[] = []
await dispatcher.dispatchStreaming(
@@ -41,13 +41,14 @@ export type StructuredAgentSessionRuntimeDeps = {
/** Key id this host's claims are minted under. */
claimKeyId: string
resolveWorkspacePath: (workspaceId: string) => Promise<string>
resolveCodexCommand?: () => string
resolveCodexCommand?: (options?: { pathEnv?: string | null; homePath?: string }) => string
/** Transport for the Codex child. Overridden only to drive the whole runtime
* against a scripted app-server; production spawns the real one. */
openCodexConnection?: CodexStructuredSessionAdapterDeps['openConnection']
/** Scripted app-servers carry fake pids the real start-time read cannot answer for. */
readProcessStartTime?: CodexStructuredSessionAdapterDeps['readProcessStartTime']
resolveLaunchEnv?: () => Promise<NodeJS.ProcessEnv>
resolveEnvironment?: () => Promise<NodeJS.ProcessEnv>
resolveCodexOverrides?: () => NodeJS.ProcessEnv
onError?: (input: { scope: string; error: unknown }) => void
handoffTransport?: StructuredAgentSessionHandoffTransport
}
@@ -92,6 +93,11 @@ export async function stopStructuredAgentSessionRuntime(): Promise<void> {
}
async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<InstalledRuntime> {
const bootEnvironment = (deps.resolveEnvironment ?? resolveLoginShellEnvironment)()
const resolveEnvironment = async (): Promise<NodeJS.ProcessEnv> => ({
...(await bootEnvironment),
...deps.resolveCodexOverrides?.()
})
const store = await AgentSessionRecordStore.open({
directory: join(deps.stateDirectory, RECORD_STORE_DIR_NAME),
hostId: deps.hostId
@@ -102,6 +108,7 @@ async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<Install
resolveLaunch: createCodexStructuredLaunchResolver({
store,
resolveWorkspacePath: deps.resolveWorkspacePath,
resolveEnvironment,
...(deps.resolveCodexCommand ? { resolveCommand: deps.resolveCodexCommand } : {})
}),
...(deps.openCodexConnection ? { openConnection: deps.openCodexConnection } : {}),
@@ -113,8 +120,6 @@ async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<Install
journalRoot: deps.stateDirectory,
claimKeyId: deps.claimKeyId,
probeOwner: createStructuredAgentSessionOwnerProbe(deps.hostId),
resolveLaunchEnv: async () =>
(await (deps.resolveLaunchEnv ?? resolveLoginShellEnvironment)()) as Record<string, string>,
onEventSinkError: ({ sessionId, error }) =>
deps.onError?.({ scope: `structured-agent-session-journal:${sessionId}`, error }),
...(deps.handoffTransport ? { handoffTransport: deps.handoffTransport } : {})
@@ -0,0 +1,14 @@
export function scrubAgentSessionRecordLaunchEnv(value: unknown): {
value: unknown
changed: boolean
} {
if (typeof value !== 'object' || value === null || Array.isArray(value)) {
return { value, changed: false }
}
const record = value as Record<string, unknown>
if (!Object.hasOwn(record, 'launchEnv')) {
return { value, changed: false }
}
const { launchEnv: _discarded, ...rest } = record
return { value: rest, changed: true }
}
+1
View File
@@ -118,6 +118,7 @@ export type AgentSessionRecord = {
accountHome: AgentSessionAccountHome
/** Provider options acknowledged for the next turn, restored across owner replacement. */
options?: Record<string, string>
/** Legacy read-compatible field; current stores scrub and omit it. */
launchEnv?: AgentSessionLaunchEnv
lease: AgentSessionLease
createdAt: number