mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 00:03:52 +00:00
feat(native-chat): give a structured session a surface-holder lifetime
Nothing told the host that a chat WANTED a session, and nothing told it when one stopped. Both halves cost real processes: closing a chat left its codex app-server running until the app quit (9 orphans, 420 MB, oldest 21 hours), and every launch resumed a child for every record on disk — `released` with no handoff in flight is the normal end state of a cleanly closed chat, so a healthy profile respawned everything it had ever opened, in parallel, with no client attached. A surface takes a HOLD when it binds and drops it when it goes away: the desktop chat view, the paired mobile session view. Holders are a SET, not a count — every release path can fire twice or not at all, and an integer cannot tell a duplicate release from a lost one. Transport close is the backstop, not the mechanism: a phone that loses signal never sends its release, so the hold registers a cleanup against its connection, and a live subscription holds the session on its own. Eviction runs the existing steps, unreordered, under a per-step deadline: a step that times out aborts the rest, which leaves the session indexed and the child LOADED rather than killed. The clock starts when the last holder leaves and re-arms while a turn is running, so a session is never stopped mid-answer. Teardown now also hands the lease back — a record left claiming a live owner is one nothing can resume. Startup no longer resumes anything. It reconciles, exits recovery, and restores the journal for READING; a provider child exists because a surface asked for the session. Paired builds that predate holds keep working: they do not advertise `agent-session.structured.hold.v1`, and for them a subscription is the resuming kind of hold.
This commit is contained in:
@@ -0,0 +1,63 @@
|
||||
import { createElement } from 'react'
|
||||
import { act, create, type ReactTestRenderer } from 'react-test-renderer'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { RpcClient } from '../transport/rpc-client'
|
||||
import { useMobileStructuredSessionHold } from './mobile-structured-session-hold'
|
||||
|
||||
function renderHold(props: { client: RpcClient | null; sessionId: string | null }): {
|
||||
renderer: ReactTestRenderer
|
||||
} {
|
||||
function Probe(): null {
|
||||
useMobileStructuredSessionHold(props)
|
||||
return null
|
||||
}
|
||||
let renderer!: ReactTestRenderer
|
||||
act(() => {
|
||||
renderer = create(createElement(Probe))
|
||||
})
|
||||
return { renderer }
|
||||
}
|
||||
|
||||
function client(sendRequest: ReturnType<typeof vi.fn>): RpcClient {
|
||||
return { sendRequest } as unknown as RpcClient
|
||||
}
|
||||
|
||||
let rendered: ReactTestRenderer | null = null
|
||||
|
||||
afterEach(() => {
|
||||
act(() => rendered?.unmount())
|
||||
rendered = null
|
||||
})
|
||||
|
||||
describe('a paired session view', () => {
|
||||
it('holds the session while it is open and releases it when it closes', async () => {
|
||||
const sendRequest = vi.fn(async () => ({ ok: true, result: {} }))
|
||||
const { renderer } = renderHold({ client: client(sendRequest), sessionId: 'session-alpha' })
|
||||
|
||||
const hold = sendRequest.mock.calls.find(([method]) => method === 'agentSession.hold')
|
||||
expect(hold?.[1]).toMatchObject({ sessionId: 'session-alpha' })
|
||||
|
||||
await act(async () => renderer.unmount())
|
||||
|
||||
const release = sendRequest.mock.calls.find(([method]) => method === 'agentSession.release')
|
||||
expect(release?.[1]).toEqual(hold?.[1])
|
||||
})
|
||||
|
||||
it('asks for nothing before there is a session to hold', () => {
|
||||
const sendRequest = vi.fn(async () => ({ ok: true, result: {} }))
|
||||
rendered = renderHold({ client: client(sendRequest), sessionId: null }).renderer
|
||||
|
||||
expect(sendRequest).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
// An older host answers method_not_found; a chat that cannot be held is still a chat that reads.
|
||||
it('survives a host that does not know the method', () => {
|
||||
const sendRequest = vi.fn(async () => {
|
||||
throw new Error('method_not_found')
|
||||
})
|
||||
|
||||
expect(() => {
|
||||
rendered = renderHold({ client: client(sendRequest), sessionId: 'session-alpha' }).renderer
|
||||
}).not.toThrow()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,37 @@
|
||||
// The paired session view telling the host it is looking at this chat.
|
||||
//
|
||||
// Same contract as the desktop chat's hold, over the mobile transport. It matters more here: a
|
||||
// phone is the surface most likely to disappear without saying anything, so the host's own
|
||||
// connection-close backstop is what usually ends up releasing this — the explicit release only
|
||||
// covers the polite case of navigating away with the socket still up.
|
||||
//
|
||||
// A host that predates the method answers `method_not_found`; the session still reads.
|
||||
|
||||
import { useEffect } from 'react'
|
||||
import type { RpcClient } from '../transport/rpc-client'
|
||||
|
||||
let holderOrdinal = 0
|
||||
|
||||
export function useMobileStructuredSessionHold(args: {
|
||||
client: RpcClient | null
|
||||
sessionId: string | null
|
||||
}): void {
|
||||
const { client, sessionId } = args
|
||||
useEffect(() => {
|
||||
if (!client || !sessionId) {
|
||||
return
|
||||
}
|
||||
holderOrdinal += 1
|
||||
const holderId = `mobile-session:${holderOrdinal}`
|
||||
// Chained, not raced: unmounting during the hold's round trip would otherwise release a hold
|
||||
// that has not been taken yet.
|
||||
const held = client
|
||||
.sendRequest('agentSession.hold', { sessionId, holderId })
|
||||
.catch(() => undefined)
|
||||
return () => {
|
||||
void held.then(() =>
|
||||
client.sendRequest('agentSession.release', { sessionId, holderId }).catch(() => undefined)
|
||||
)
|
||||
}
|
||||
}, [client, sessionId])
|
||||
}
|
||||
@@ -15,6 +15,7 @@ import {
|
||||
type StructuredAgentSessionState
|
||||
} from '../../../src/shared/structured-agent-session-reducer'
|
||||
import type { RpcClient } from '../transport/rpc-client'
|
||||
import { useMobileStructuredSessionHold } from './mobile-structured-session-hold'
|
||||
import {
|
||||
createMobileStructuredReconnectState,
|
||||
noteStructuredBackground,
|
||||
@@ -59,6 +60,8 @@ export function useMobileStructuredAgentSession(args: {
|
||||
loadOlder: () => Promise<boolean>
|
||||
} {
|
||||
const { client, sessionId } = args
|
||||
// Declared before the read: the hold is what gives a restored session its provider child back.
|
||||
useMobileStructuredSessionHold({ client, sessionId })
|
||||
const [state, dispatch] = useReducer(reduceStructuredAgentSession, EMPTY_STRUCTURED_AGENT_SESSION)
|
||||
const stateRef = useRef(state)
|
||||
stateRef.current = state
|
||||
|
||||
@@ -143,7 +143,7 @@ describe('mobile E2EE v2 physical channel', () => {
|
||||
expect(JSON.parse(new TextDecoder().decode(capabilities!))).toEqual({
|
||||
type: 'e2ee_client_capabilities',
|
||||
v: 1,
|
||||
clientCapabilities: ['agent-session.structured.v1']
|
||||
clientCapabilities: ['agent-session.structured.v1', 'agent-session.structured.hold.v1']
|
||||
})
|
||||
expect(ctx.onAuthenticated).toHaveBeenCalledOnce()
|
||||
expect(ctx.onError).not.toHaveBeenCalled()
|
||||
|
||||
@@ -2,7 +2,10 @@ import {
|
||||
createWsOutboundBackpressureQueue,
|
||||
type WsOutboundBackpressureQueue
|
||||
} from '../../../src/shared/ws-outbound-backpressure-queue'
|
||||
import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../../src/shared/protocol-version'
|
||||
import {
|
||||
STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY,
|
||||
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY
|
||||
} from '../../../src/shared/protocol-version'
|
||||
import type { MobileE2EEV2ClientSession } from './mobile-e2ee-v2-client-session'
|
||||
|
||||
type ChannelState = 'awaiting-ready' | 'awaiting-authenticated' | 'ready'
|
||||
@@ -117,7 +120,10 @@ export class MobileE2EEV2PhysicalChannel {
|
||||
plaintext: JSON.stringify({
|
||||
type: 'e2ee_client_capabilities',
|
||||
v: 1,
|
||||
clientCapabilities: [STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY]
|
||||
clientCapabilities: [
|
||||
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
|
||||
STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY
|
||||
]
|
||||
})
|
||||
})
|
||||
this.args.onAuthenticated()
|
||||
|
||||
@@ -99,7 +99,7 @@ describe('structured session RPC transport', () => {
|
||||
expect(auth).toEqual({
|
||||
type: 'e2ee_auth',
|
||||
deviceToken: 'token',
|
||||
clientCapabilities: ['agent-session.structured.v1']
|
||||
clientCapabilities: ['agent-session.structured.v1', 'agent-session.structured.hold.v1']
|
||||
})
|
||||
client.close()
|
||||
})
|
||||
|
||||
@@ -46,7 +46,10 @@ import {
|
||||
} from './rpc-session-liveness-watchdog'
|
||||
import { isStaleForegroundDial } from './rpc-stale-dial'
|
||||
import { websocketPayloadToUint8 } from './websocket-payload-bytes'
|
||||
import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../../src/shared/protocol-version'
|
||||
import {
|
||||
STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY,
|
||||
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY
|
||||
} from '../../../src/shared/protocol-version'
|
||||
import type { RpcClient, SendRequestOptions, SubscribeOptions } from './rpc-client-contract'
|
||||
|
||||
export type { RpcClient, SendRequestOptions } from './rpc-client-contract'
|
||||
@@ -374,7 +377,10 @@ export function connect(
|
||||
sendEncrypted({
|
||||
type: 'e2ee_auth',
|
||||
deviceToken,
|
||||
clientCapabilities: [STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY]
|
||||
clientCapabilities: [
|
||||
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
|
||||
STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY
|
||||
]
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
+6
-1
@@ -53,7 +53,12 @@ export async function attachStructuredAgentSession(
|
||||
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 })
|
||||
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) {
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
// A bound on how long teardown may take, expressed as a wrapper around the eviction steps rather
|
||||
// than a change to them.
|
||||
//
|
||||
// The step list is ordered and abort-on-failure for reasons that have nothing to do with time, and
|
||||
// a deadline must not disturb either. Wrapping each step's `run` keeps the order, and a timeout
|
||||
// surfaces as that step failing — which is exactly the behavior wanted here: the rest of the
|
||||
// eviction aborts, the session stays indexed, and the child stays LOADED. A stuck app-server that
|
||||
// is still holding a conversation is a better outcome than one killed out from under it; the next
|
||||
// close retries.
|
||||
|
||||
import type { StructuredAgentSessionEvictionStep } from './structured-agent-session-eviction'
|
||||
|
||||
export const STRUCTURED_AGENT_SESSION_EVICTION_STEP_TIMEOUT_MS = 10_000
|
||||
|
||||
export class StructuredAgentSessionEvictionTimeoutError extends Error {
|
||||
constructor(
|
||||
readonly step: string,
|
||||
readonly timeoutMs: number
|
||||
) {
|
||||
super(`agent session eviction step "${step}" did not finish within ${timeoutMs}ms`)
|
||||
this.name = 'StructuredAgentSessionEvictionTimeoutError'
|
||||
}
|
||||
}
|
||||
|
||||
export function withStructuredAgentSessionEvictionDeadline(
|
||||
steps: readonly StructuredAgentSessionEvictionStep[],
|
||||
timeoutMs = STRUCTURED_AGENT_SESSION_EVICTION_STEP_TIMEOUT_MS
|
||||
): readonly StructuredAgentSessionEvictionStep[] {
|
||||
return steps.map((step) => ({
|
||||
name: step.name,
|
||||
run: async (context) => {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
try {
|
||||
await Promise.race([
|
||||
Promise.resolve(step.run(context)),
|
||||
new Promise<never>((_resolve, reject) => {
|
||||
timer = setTimeout(
|
||||
() => reject(new StructuredAgentSessionEvictionTimeoutError(step.name, timeoutMs)),
|
||||
timeoutMs
|
||||
)
|
||||
timer.unref?.()
|
||||
})
|
||||
])
|
||||
} finally {
|
||||
if (timer) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
}
|
||||
}
|
||||
}))
|
||||
}
|
||||
@@ -25,7 +25,10 @@ function context(): StructuredAgentSessionEvictionContext & { order: string[] }
|
||||
})
|
||||
} as unknown as StructuredAgentSessionEvictionContext['adapter'],
|
||||
forget: vi.fn(() => order.push('forget')),
|
||||
discardSink: vi.fn(() => order.push('discardSink'))
|
||||
discardSink: vi.fn(() => order.push('discardSink')),
|
||||
releaseLease: vi.fn(async () => {
|
||||
order.push('releaseLease')
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -46,6 +49,7 @@ describe('structured agent session eviction', () => {
|
||||
'unbind',
|
||||
'close',
|
||||
'discardSink',
|
||||
'releaseLease',
|
||||
'forget'
|
||||
])
|
||||
})
|
||||
@@ -57,6 +61,7 @@ describe('structured agent session eviction', () => {
|
||||
'stop-publishing',
|
||||
'close-sink',
|
||||
'discard-sink',
|
||||
'release-lease',
|
||||
'forget-session'
|
||||
])
|
||||
})
|
||||
@@ -83,7 +88,8 @@ describe('rows the provider emits while closing', () => {
|
||||
}
|
||||
} as never,
|
||||
forget: () => {},
|
||||
discardSink: () => state.discardEventSink(sessionId)
|
||||
discardSink: () => state.discardEventSink(sessionId),
|
||||
releaseLease: async () => {}
|
||||
})
|
||||
|
||||
expect(published).toEqual(['final-flush'])
|
||||
@@ -131,7 +137,8 @@ describe('eviction against the real sink cache', () => {
|
||||
eventSink: state.eventSinkFor(sessionId),
|
||||
adapter: { closeSession: async () => {} } as never,
|
||||
forget: () => {},
|
||||
discardSink: () => state.discardEventSink(sessionId)
|
||||
discardSink: () => state.discardEventSink(sessionId),
|
||||
releaseLease: async () => {}
|
||||
})
|
||||
|
||||
const published: string[] = []
|
||||
|
||||
@@ -27,6 +27,9 @@ export type StructuredAgentSessionEvictionContext = {
|
||||
forget: () => void
|
||||
/** Drops the cached sink so a later attach mints a fresh one. */
|
||||
discardSink: () => void
|
||||
/** Hands the lease back now that this host's child is proven gone. No-ops when the record is
|
||||
* not this host's to release. */
|
||||
releaseLease: () => Promise<void>
|
||||
}
|
||||
|
||||
export type StructuredAgentSessionEvictionStep = {
|
||||
@@ -54,6 +57,11 @@ export const STRUCTURED_AGENT_SESSION_EVICTION_STEPS: readonly StructuredAgentSe
|
||||
// accepts every provider event and publishes none. Attach's own failure path already pairs
|
||||
// these two; eviction has to as well.
|
||||
{ name: 'discard-sink', run: (context) => context.discardSink() },
|
||||
// Why here and not last: the durable lease still names a process this host just stopped, and a
|
||||
// record left claiming a live owner is one nothing can resume — the next surface to open the
|
||||
// chat would find a session it may not acquire. Placed BEFORE forget so a release that cannot
|
||||
// be written aborts while the session is still indexed, which is what makes the retry real.
|
||||
{ name: 'release-lease', run: (context) => context.releaseLease() },
|
||||
{ name: 'forget-session', run: (context) => context.forget() }
|
||||
]
|
||||
|
||||
|
||||
@@ -99,6 +99,18 @@ export async function restoreStructuredAgentSessionHandoff(
|
||||
}
|
||||
|
||||
async function restoreOnce(input: RestartAccess, record: AgentSessionRecord): Promise<void> {
|
||||
if (
|
||||
record.lease.handoffStage === null &&
|
||||
record.lease.claimStatus === 'released' &&
|
||||
record.lease.ownerProcess === null
|
||||
) {
|
||||
// Reconcile already proved the owner gone: no transfer is in flight and there is no process to
|
||||
// recover, whatever kind the last owner was. Falling through would latch manual recovery on a
|
||||
// host with no TUI transport — a state a user then has to clear by hand, for nothing. (Startup
|
||||
// used to reach this path only for sessions it had not eagerly resumed, which is why removing
|
||||
// that resume is what made it visible.)
|
||||
return
|
||||
}
|
||||
if (canRestoreLiveTuiOwner(record)) {
|
||||
await restoreRecoverableLiveTui(input, record)
|
||||
return
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
// Giving a held session its provider child back.
|
||||
//
|
||||
// This is the replacement for the startup resume, and the difference is only in WHO asks: the same
|
||||
// eligibility rule, run when a surface binds instead of when the app launches. A refusal is not an
|
||||
// error here — the surface still holds the session and can still read it; it just cannot send until
|
||||
// the lease frees up — so this resolves either way and leaves the refusal to the read path.
|
||||
|
||||
import type { AgentSessionAttachParams } from './structured-agent-session-attach'
|
||||
import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types'
|
||||
import { adapterSupportsRecord } from './structured-agent-session-provider-support'
|
||||
import {
|
||||
structuredAgentSessionResumeOperationId,
|
||||
structuredAgentSessionResumeParams
|
||||
} from './structured-agent-session-resume-eligibility'
|
||||
|
||||
export async function resumeHeldStructuredAgentSession(input: {
|
||||
sessionId: string
|
||||
deps: StructuredAgentSessionHostDeps
|
||||
now: () => number
|
||||
attach: (params: AgentSessionAttachParams) => Promise<{ ok: boolean }>
|
||||
}): Promise<void> {
|
||||
const record = input.deps.store.getRecord(input.sessionId)
|
||||
if (!record || !adapterSupportsRecord(input.deps.adapter, record)) {
|
||||
return
|
||||
}
|
||||
const params = structuredAgentSessionResumeParams(
|
||||
record,
|
||||
structuredAgentSessionResumeOperationId(input.now())
|
||||
)
|
||||
if (!params) {
|
||||
return
|
||||
}
|
||||
await input.attach(params)
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
// Who WANTS this session alive, as a set of ids rather than a count.
|
||||
//
|
||||
// A refcount is the obvious shape and the wrong one. Every path that decrements it — a chat tab
|
||||
// closing, a transport dying, a client retrying a release it already sent — can fire twice or not
|
||||
// at all, and an integer cannot tell those apart: a duplicate release evicts a session somebody is
|
||||
// still looking at, and a lost one leaks the child forever. A set answers both idempotently,
|
||||
// because it records WHICH surface holds the session, not how many do.
|
||||
|
||||
export class StructuredAgentSessionHolders {
|
||||
private readonly bySession = new Map<string, Set<string>>()
|
||||
|
||||
/** True when the session gained its FIRST holder — the edge that ends a pending release. */
|
||||
add(sessionId: string, holderId: string): boolean {
|
||||
const holders = this.bySession.get(sessionId)
|
||||
if (!holders) {
|
||||
this.bySession.set(sessionId, new Set([holderId]))
|
||||
return true
|
||||
}
|
||||
holders.add(holderId)
|
||||
return false
|
||||
}
|
||||
|
||||
/** True when the session lost its LAST holder — the edge that starts one. */
|
||||
remove(sessionId: string, holderId: string): boolean {
|
||||
const holders = this.bySession.get(sessionId)
|
||||
if (!holders?.delete(holderId) || holders.size > 0) {
|
||||
return false
|
||||
}
|
||||
this.bySession.delete(sessionId)
|
||||
return true
|
||||
}
|
||||
|
||||
isHeld(sessionId: string): boolean {
|
||||
return (this.bySession.get(sessionId)?.size ?? 0) > 0
|
||||
}
|
||||
|
||||
holderIds(sessionId: string): string[] {
|
||||
return [...(this.bySession.get(sessionId) ?? [])]
|
||||
}
|
||||
|
||||
/** Drops every holder of one session without evaluating the edge, for a session that is gone. */
|
||||
forget(sessionId: string): void {
|
||||
this.bySession.delete(sessionId)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,213 @@
|
||||
// The parts a session's lifetime is assembled from: the holder set, the release clock, and the
|
||||
// deadline that keeps teardown from hanging.
|
||||
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { StructuredAgentSessionHolders } from './structured-agent-session-holders'
|
||||
import { StructuredAgentSessionReleaseClock } from './structured-agent-session-release-clock'
|
||||
import { StructuredAgentSessionHolds } from './structured-agent-session-holds'
|
||||
import {
|
||||
STRUCTURED_AGENT_SESSION_EVICTION_STEPS,
|
||||
evictStructuredAgentSession
|
||||
} from './structured-agent-session-eviction'
|
||||
import {
|
||||
StructuredAgentSessionEvictionTimeoutError,
|
||||
withStructuredAgentSessionEvictionDeadline
|
||||
} from './structured-agent-session-eviction-deadline'
|
||||
|
||||
const clocks: StructuredAgentSessionReleaseClock[] = []
|
||||
|
||||
function clock(deps: {
|
||||
isTurnActive?: () => boolean
|
||||
isHeld?: () => boolean
|
||||
evict: (sessionId: string) => Promise<void>
|
||||
onError?: (input: { sessionId: string; error: unknown }) => void
|
||||
}): StructuredAgentSessionReleaseClock {
|
||||
const created = new StructuredAgentSessionReleaseClock({
|
||||
isTurnActive: deps.isTurnActive ?? (() => false),
|
||||
isHeld: deps.isHeld ?? (() => false),
|
||||
evict: deps.evict,
|
||||
...(deps.onError ? { onError: deps.onError } : {}),
|
||||
graceMs: 1
|
||||
})
|
||||
clocks.push(created)
|
||||
return created
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
for (const created of clocks.splice(0)) {
|
||||
created.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
describe('the holder set', () => {
|
||||
it('reports the first and last holder, and nothing in between', () => {
|
||||
const holders = new StructuredAgentSessionHolders()
|
||||
|
||||
expect(holders.add('session-1', 'a')).toBe(true)
|
||||
expect(holders.add('session-1', 'b')).toBe(false)
|
||||
expect(holders.remove('session-1', 'a')).toBe(false)
|
||||
expect(holders.remove('session-1', 'b')).toBe(true)
|
||||
expect(holders.isHeld('session-1')).toBe(false)
|
||||
})
|
||||
|
||||
// The reason this is a set and not a count: every release path can fire twice or not at all.
|
||||
it('absorbs a duplicate hold and a duplicate release', () => {
|
||||
const holders = new StructuredAgentSessionHolders()
|
||||
|
||||
holders.add('session-1', 'a')
|
||||
holders.add('session-1', 'a')
|
||||
expect(holders.remove('session-1', 'a')).toBe(true)
|
||||
expect(holders.remove('session-1', 'a')).toBe(false)
|
||||
expect(holders.holderIds('session-1')).toEqual([])
|
||||
})
|
||||
|
||||
it('keeps one session holders out of another session holders', () => {
|
||||
const holders = new StructuredAgentSessionHolders()
|
||||
|
||||
holders.add('session-1', 'a')
|
||||
holders.add('session-2', 'a')
|
||||
holders.remove('session-1', 'a')
|
||||
|
||||
expect(holders.isHeld('session-1')).toBe(false)
|
||||
expect(holders.isHeld('session-2')).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('the release clock', () => {
|
||||
it('waits out a running turn instead of evicting into it', async () => {
|
||||
const evict = vi.fn(async () => {})
|
||||
let turnRunning = true
|
||||
const releasing = clock({ isTurnActive: () => turnRunning, evict })
|
||||
|
||||
releasing.arm('session-1')
|
||||
await new Promise((resolve) => setTimeout(resolve, 30))
|
||||
expect(evict).not.toHaveBeenCalled()
|
||||
|
||||
turnRunning = false
|
||||
await vi.waitFor(() => expect(evict).toHaveBeenCalledWith('session-1'))
|
||||
})
|
||||
|
||||
it('stands down when a holder arrives during the wait', async () => {
|
||||
const evict = vi.fn(async () => {})
|
||||
const releasing = clock({ isHeld: () => true, evict })
|
||||
|
||||
releasing.arm('session-1')
|
||||
await new Promise((resolve) => setTimeout(resolve, 30))
|
||||
|
||||
expect(evict).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('reports a failed eviction rather than swallowing it', async () => {
|
||||
const onError = vi.fn()
|
||||
const releasing = clock({
|
||||
evict: async () => {
|
||||
throw new Error('child would not stop')
|
||||
},
|
||||
onError
|
||||
})
|
||||
|
||||
releasing.arm('session-1')
|
||||
|
||||
await vi.waitFor(() =>
|
||||
expect(onError).toHaveBeenCalledWith({
|
||||
sessionId: 'session-1',
|
||||
error: expect.objectContaining({ message: 'child would not stop' })
|
||||
})
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
describe('holds', () => {
|
||||
it('resumes a session on its first hold and not on a retained one', async () => {
|
||||
const resume = vi.fn(async () => {})
|
||||
let child = false
|
||||
const holds = new StructuredAgentSessionHolds({
|
||||
resume,
|
||||
hasProviderChild: () => child,
|
||||
isTurnActive: () => false,
|
||||
evict: async () => {},
|
||||
graceMs: 1
|
||||
})
|
||||
|
||||
await holds.hold('session-1', 'stream-1', { resume: false })
|
||||
expect(resume).not.toHaveBeenCalled()
|
||||
|
||||
await holds.hold('session-1', 'chat-1')
|
||||
expect(resume).toHaveBeenCalledOnce()
|
||||
|
||||
child = true
|
||||
await holds.hold('session-1', 'chat-2')
|
||||
expect(resume).toHaveBeenCalledOnce()
|
||||
holds.dispose()
|
||||
})
|
||||
|
||||
it('never arms the clock for a session with nothing to stop', async () => {
|
||||
const evict = vi.fn(async () => {})
|
||||
const holds = new StructuredAgentSessionHolds({
|
||||
resume: async () => {},
|
||||
hasProviderChild: () => false,
|
||||
isTurnActive: () => false,
|
||||
evict,
|
||||
graceMs: 1
|
||||
})
|
||||
|
||||
await holds.hold('session-1', 'chat-1', { resume: false })
|
||||
holds.release('session-1', 'chat-1')
|
||||
await new Promise((resolve) => setTimeout(resolve, 20))
|
||||
|
||||
expect(evict).not.toHaveBeenCalled()
|
||||
expect(holds.isReleasePending('session-1')).toBe(false)
|
||||
holds.dispose()
|
||||
})
|
||||
})
|
||||
|
||||
describe('the teardown deadline', () => {
|
||||
it('leaves the child loaded instead of forcing it, and keeps the session indexed', async () => {
|
||||
const forget = vi.fn()
|
||||
const releaseLease = vi.fn(async () => {})
|
||||
|
||||
await expect(
|
||||
evictStructuredAgentSession(
|
||||
{
|
||||
sessionId: 'session-1',
|
||||
eventSink: {
|
||||
unbind: vi.fn(),
|
||||
drained: vi.fn(async () => {}),
|
||||
close: vi.fn()
|
||||
} as never,
|
||||
adapter: { closeSession: () => new Promise<void>(() => {}) } as never,
|
||||
forget,
|
||||
discardSink: vi.fn(),
|
||||
releaseLease
|
||||
},
|
||||
withStructuredAgentSessionEvictionDeadline(STRUCTURED_AGENT_SESSION_EVICTION_STEPS, 5)
|
||||
)
|
||||
).rejects.toMatchObject({ step: 'stop-provider-child' })
|
||||
|
||||
expect(forget).not.toHaveBeenCalled()
|
||||
expect(releaseLease).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('names the step that ran out of time', async () => {
|
||||
const [step] = withStructuredAgentSessionEvictionDeadline(
|
||||
[{ name: 'slow-step', run: () => new Promise<void>(() => {}) }],
|
||||
5
|
||||
)
|
||||
|
||||
await expect(step?.run({} as never)).rejects.toBeInstanceOf(
|
||||
StructuredAgentSessionEvictionTimeoutError
|
||||
)
|
||||
})
|
||||
|
||||
it('does not delay a step that finishes', async () => {
|
||||
const ran: string[] = []
|
||||
const steps = withStructuredAgentSessionEvictionDeadline(
|
||||
[{ name: 'fast-step', run: () => void ran.push('fast-step') }],
|
||||
5_000
|
||||
)
|
||||
|
||||
await steps[0]?.run({} as never)
|
||||
|
||||
expect(ran).toEqual(['fast-step'])
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,91 @@
|
||||
// The lifetime of a structured session, tied to the surfaces that want one.
|
||||
//
|
||||
// Nothing used to tell the host that a chat WANTED a session, and nothing told it when a chat
|
||||
// stopped wanting one. Both halves of that gap cost real processes: sessions nobody had opened got
|
||||
// an app-server at every launch, and sessions the user closed kept theirs until the app quit.
|
||||
//
|
||||
// A surface takes a hold when it binds and drops it when it goes away. The first hold on a session
|
||||
// with no child resumes it — that, and not the shape of a lease on disk, is what makes a provider
|
||||
// process exist. The last hold leaving starts the release clock. Transport close is the BACKSTOP,
|
||||
// not the mechanism: a client that vanishes mid-flight never sends its release, so the caller
|
||||
// registers one against the connection and the holder set absorbs the duplicate.
|
||||
|
||||
import {
|
||||
StructuredAgentSessionReleaseClock,
|
||||
type StructuredAgentSessionReleaseClockDeps
|
||||
} from './structured-agent-session-release-clock'
|
||||
import { StructuredAgentSessionHolders } from './structured-agent-session-holders'
|
||||
|
||||
export type StructuredAgentSessionHoldsDeps = {
|
||||
/** Acquires a provider child for a session that has none. A no-op when one is already live. */
|
||||
resume: (sessionId: string) => Promise<void>
|
||||
/** Whether evicting this session would actually free anything. */
|
||||
hasProviderChild: (sessionId: string) => boolean
|
||||
isTurnActive: (sessionId: string) => boolean
|
||||
evict: (sessionId: string) => Promise<void>
|
||||
onError?: (input: { sessionId: string; error: unknown }) => void
|
||||
graceMs?: number
|
||||
}
|
||||
|
||||
export type StructuredAgentSessionHoldOptions = {
|
||||
/** False for a hold that only RETAINS — a subscription stream, which must not make a child
|
||||
* exist just by reading history. */
|
||||
resume?: boolean
|
||||
}
|
||||
|
||||
export class StructuredAgentSessionHolds {
|
||||
private readonly holders = new StructuredAgentSessionHolders()
|
||||
private readonly clock: StructuredAgentSessionReleaseClock
|
||||
|
||||
constructor(private readonly deps: StructuredAgentSessionHoldsDeps) {
|
||||
const clockDeps: StructuredAgentSessionReleaseClockDeps = {
|
||||
isTurnActive: deps.isTurnActive,
|
||||
isHeld: (sessionId) => this.holders.isHeld(sessionId),
|
||||
evict: (sessionId) => this.deps.evict(sessionId),
|
||||
...(deps.onError ? { onError: deps.onError } : {}),
|
||||
...(deps.graceMs === undefined ? {} : { graceMs: deps.graceMs })
|
||||
}
|
||||
this.clock = new StructuredAgentSessionReleaseClock(clockDeps)
|
||||
}
|
||||
|
||||
async hold(
|
||||
sessionId: string,
|
||||
holderId: string,
|
||||
options: StructuredAgentSessionHoldOptions = {}
|
||||
): Promise<void> {
|
||||
this.holders.add(sessionId, holderId)
|
||||
// Unconditional, not only on the first-holder edge: a second surface arriving during the grace
|
||||
// window must cancel the pending release too.
|
||||
this.clock.cancel(sessionId)
|
||||
if (options.resume !== false && !this.deps.hasProviderChild(sessionId)) {
|
||||
await this.deps.resume(sessionId)
|
||||
}
|
||||
}
|
||||
|
||||
release(sessionId: string, holderId: string): void {
|
||||
if (!this.holders.remove(sessionId, holderId)) {
|
||||
return
|
||||
}
|
||||
if (this.deps.hasProviderChild(sessionId)) {
|
||||
this.clock.arm(sessionId)
|
||||
}
|
||||
}
|
||||
|
||||
/** Drops the holders of a session that is gone, whoever evicted it. */
|
||||
forget(sessionId: string): void {
|
||||
this.clock.cancel(sessionId)
|
||||
this.holders.forget(sessionId)
|
||||
}
|
||||
|
||||
isHeld(sessionId: string): boolean {
|
||||
return this.holders.isHeld(sessionId)
|
||||
}
|
||||
|
||||
isReleasePending(sessionId: string): boolean {
|
||||
return this.clock.isArmed(sessionId)
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
this.clock.dispose()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
// The host's half of a session's lifetime: what a close does, and what a hold is wired to.
|
||||
//
|
||||
// Lifted out of the host for the same reason attaching was — the host is a coordinator, and the
|
||||
// sequence that stops a provider child and hands its lease back reads better next to the holder
|
||||
// bookkeeping that decides when to run it than buried among the twenty other things a session can
|
||||
// do.
|
||||
|
||||
import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection'
|
||||
import {
|
||||
evictStructuredAgentSession,
|
||||
STRUCTURED_AGENT_SESSION_EVICTION_STEPS,
|
||||
type StructuredAgentSessionEvictionContext
|
||||
} from './structured-agent-session-eviction'
|
||||
import { withStructuredAgentSessionEvictionDeadline } from './structured-agent-session-eviction-deadline'
|
||||
import { StructuredAgentSessionHolds } from './structured-agent-session-holds'
|
||||
import type { StructuredAgentSessionHostRuntimeState } from './structured-agent-session-host-runtime-state'
|
||||
import type {
|
||||
StructuredAgentSessionHostDeps,
|
||||
StructuredAgentSessionHostSession
|
||||
} from './structured-agent-session-host-types'
|
||||
import { releaseStoredStructuredAgentSessionOwner } from './structured-agent-session-lease-release'
|
||||
|
||||
export type StructuredAgentSessionLifetimeContext = {
|
||||
deps: StructuredAgentSessionHostDeps
|
||||
runtimeState: StructuredAgentSessionHostRuntimeState
|
||||
sessions: Map<string, StructuredAgentSessionHostSession>
|
||||
now: () => number
|
||||
}
|
||||
|
||||
function hasProviderChild(
|
||||
context: StructuredAgentSessionLifetimeContext,
|
||||
sessionId: string
|
||||
): boolean {
|
||||
return context.sessions.get(sessionId)?.hasProviderChild === true
|
||||
}
|
||||
|
||||
/** Runs the eviction steps under a deadline. A step that fails — or runs out of time — aborts the
|
||||
* rest, which leaves the session indexed and the child loaded so the next close is a real retry. */
|
||||
export async function evictHeldStructuredAgentSession(
|
||||
context: StructuredAgentSessionLifetimeContext,
|
||||
sessionId: string
|
||||
): Promise<void> {
|
||||
if (!context.sessions.has(sessionId)) {
|
||||
return
|
||||
}
|
||||
const eviction: StructuredAgentSessionEvictionContext = {
|
||||
sessionId,
|
||||
eventSink: context.runtimeState.eventSinkFor(sessionId),
|
||||
adapter: context.deps.adapter,
|
||||
forget: () => context.sessions.delete(sessionId),
|
||||
discardSink: () => context.runtimeState.discardEventSink(sessionId),
|
||||
releaseLease: () =>
|
||||
releaseStoredStructuredAgentSessionOwner({
|
||||
store: context.deps.store,
|
||||
sessionId,
|
||||
hasProviderChild: hasProviderChild(context, sessionId),
|
||||
now: context.now()
|
||||
})
|
||||
}
|
||||
await evictStructuredAgentSession(
|
||||
eviction,
|
||||
withStructuredAgentSessionEvictionDeadline(STRUCTURED_AGENT_SESSION_EVICTION_STEPS)
|
||||
)
|
||||
}
|
||||
|
||||
export function createStructuredAgentSessionHolds(
|
||||
context: StructuredAgentSessionLifetimeContext,
|
||||
input: {
|
||||
resume: (sessionId: string) => Promise<void>
|
||||
evict: (sessionId: string) => Promise<void>
|
||||
}
|
||||
): StructuredAgentSessionHolds {
|
||||
return new StructuredAgentSessionHolds({
|
||||
resume: input.resume,
|
||||
evict: input.evict,
|
||||
hasProviderChild: (sessionId) => hasProviderChild(context, sessionId),
|
||||
isTurnActive: (sessionId) => {
|
||||
const session = context.sessions.get(sessionId)
|
||||
return session
|
||||
? activeStructuredAgentSessionTurnId(session.journal.snapshot().items) !== null
|
||||
: false
|
||||
},
|
||||
onError: (error) => context.deps.onEventSinkError?.(error),
|
||||
...(context.deps.releaseGraceMs === undefined ? {} : { graceMs: context.deps.releaseGraceMs })
|
||||
})
|
||||
}
|
||||
@@ -12,6 +12,10 @@ export type StructuredAgentSessionHostSession = {
|
||||
journal: AgentSessionJournal
|
||||
params: AgentSessionAttachParams
|
||||
fence: number
|
||||
/** Whether THIS host generation is running the provider process behind the session. A journal
|
||||
* restored for reading has none, and neither has a session a TUI owns — so neither may be
|
||||
* evicted to free a child, and neither may have its lease released as an observed exit. */
|
||||
hasProviderChild: boolean
|
||||
}
|
||||
|
||||
export type StructuredAgentSessionHostDeps = {
|
||||
@@ -27,6 +31,8 @@ export type StructuredAgentSessionHostDeps = {
|
||||
provider: AgentSessionRecord['provider']
|
||||
) => Promise<Record<string, string> | undefined> | Record<string, string> | undefined
|
||||
now?: () => number
|
||||
/** How long a session outlives its last surface. Tests drive this; production takes the default. */
|
||||
releaseGraceMs?: number
|
||||
onEventSinkError?: (input: { sessionId: string; error: unknown }) => void
|
||||
handoffTransport?: StructuredAgentSessionHandoffTransport
|
||||
}
|
||||
|
||||
@@ -709,7 +709,7 @@ describe('restart', () => {
|
||||
expect(listRecords).toHaveBeenCalledTimes(restoreReads)
|
||||
})
|
||||
|
||||
it('clears stale TUI recovery when restart successfully reacquires the native owner', async () => {
|
||||
it('clears stale TUI recovery at restart, and reacquires the native owner when a surface holds it', async () => {
|
||||
await attach()
|
||||
await store.transitionHandoff(SESSION, (record) => ({
|
||||
...record,
|
||||
@@ -723,6 +723,9 @@ describe('restart', () => {
|
||||
acquire.mockClear()
|
||||
|
||||
await host.restoreReadableSessions()
|
||||
// The recovery stage clears on evidence at startup; the child comes back only once a surface
|
||||
// holds the session (see structured-agent-session-surface-lifetime.test.ts).
|
||||
await host.hold(SESSION, 'surface-1')
|
||||
|
||||
expect(acquire).toHaveBeenCalledOnce()
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
|
||||
@@ -47,9 +47,15 @@ import {
|
||||
import { StructuredAgentSessionHostRuntimeState } from './structured-agent-session-host-runtime-state'
|
||||
import { attachStructuredAgentSession } from './structured-agent-session-attach-orchestration'
|
||||
import {
|
||||
evictStructuredAgentSession,
|
||||
type StructuredAgentSessionEvictionContext
|
||||
} from './structured-agent-session-eviction'
|
||||
createStructuredAgentSessionHolds,
|
||||
evictHeldStructuredAgentSession,
|
||||
type StructuredAgentSessionLifetimeContext
|
||||
} from './structured-agent-session-host-lifetime'
|
||||
import type {
|
||||
StructuredAgentSessionHolds,
|
||||
StructuredAgentSessionHoldOptions
|
||||
} from './structured-agent-session-holds'
|
||||
import { resumeHeldStructuredAgentSession } from './structured-agent-session-hold-resume'
|
||||
import type { StructuredAgentSessionAttachContext } from './structured-agent-session-attach-context'
|
||||
import { listStructuredAgentSessionTabs } from './structured-agent-session-host-tabs'
|
||||
import { StructuredAgentSessionReadableRestorer } from './structured-agent-session-readable-restorer'
|
||||
@@ -74,6 +80,7 @@ export class StructuredAgentSessionHost {
|
||||
private readonly handoffs: StructuredAgentSessionHostHandoff
|
||||
private readonly readableRestorer: StructuredAgentSessionReadableRestorer
|
||||
private readonly restartRestore = new StructuredAgentSessionRestartRestoreGate()
|
||||
private readonly holds: StructuredAgentSessionHolds
|
||||
|
||||
constructor(readonly deps: StructuredAgentSessionHostDeps) {
|
||||
this.runtimeState = new StructuredAgentSessionHostRuntimeState(
|
||||
@@ -99,21 +106,20 @@ export class StructuredAgentSessionHost {
|
||||
subscribers: this.subscribers,
|
||||
now: this.now
|
||||
})
|
||||
this.holds = createStructuredAgentSessionHolds(this.lifetimeContext(), {
|
||||
resume: (sessionId) => this.resumeForHold(sessionId),
|
||||
evict: (sessionId) => this.close(sessionId)
|
||||
})
|
||||
this.readableRestorer = new StructuredAgentSessionReadableRestorer({
|
||||
store: deps.store,
|
||||
journalRoot: deps.journalRoot,
|
||||
supportsRecord: (record) => providerSupport.adapterSupportsRecord(deps.adapter, record),
|
||||
reconcile: this.reconcileLeases,
|
||||
resolveRecovery: (sessionId) => this.runtimeState.resolveRecovery(sessionId),
|
||||
resume: (params) =>
|
||||
this.attach({ callerKey: 'trusted-local:host-restart' }, params).then(
|
||||
(result) => result.ok
|
||||
),
|
||||
serialize: (sessionId, task) => this.serialize(sessionId, task),
|
||||
hasSession: (sessionId) => this.sessions.has(sessionId),
|
||||
onReadable: (sessionId, restored) => this.sessions.set(sessionId, restored),
|
||||
restoreHandoff: (sessionId) => this.handoffs.restore(sessionId),
|
||||
now: this.now
|
||||
restoreHandoff: (sessionId) => this.handoffs.restore(sessionId)
|
||||
})
|
||||
this.runtimeState.startLeaseRenewal()
|
||||
}
|
||||
@@ -122,6 +128,37 @@ export class StructuredAgentSessionHost {
|
||||
|
||||
hasSession = (sessionId: string): boolean => this.sessions.has(sessionId)
|
||||
|
||||
/** A surface bound to this session and wants it live. The FIRST hold on a session with no
|
||||
* provider child is what resumes one; a retained hold (a subscription) only keeps it. */
|
||||
hold = (
|
||||
sessionId: string,
|
||||
holderId: string,
|
||||
options?: StructuredAgentSessionHoldOptions
|
||||
): Promise<void> => this.holds.hold(sessionId, holderId, options)
|
||||
|
||||
/** That surface is gone. The child outlives it by the release grace, and by any running turn. */
|
||||
release = (sessionId: string, holderId: string): void => this.holds.release(sessionId, holderId)
|
||||
|
||||
isHeld = (sessionId: string): boolean => this.holds.isHeld(sessionId)
|
||||
|
||||
private resumeForHold(sessionId: string): Promise<void> {
|
||||
return resumeHeldStructuredAgentSession({
|
||||
sessionId,
|
||||
deps: this.deps,
|
||||
now: () => this.now(),
|
||||
attach: (params) => this.attach({ callerKey: 'trusted-local:surface-hold' }, params)
|
||||
})
|
||||
}
|
||||
|
||||
private lifetimeContext(): StructuredAgentSessionLifetimeContext {
|
||||
return {
|
||||
deps: this.deps,
|
||||
runtimeState: this.runtimeState,
|
||||
sessions: this.sessions,
|
||||
now: () => this.now()
|
||||
}
|
||||
}
|
||||
|
||||
/** The host's half of attaching, named so it cannot grow dependencies unnoticed. */
|
||||
private attachContext(): StructuredAgentSessionAttachContext {
|
||||
return {
|
||||
@@ -138,17 +175,10 @@ export class StructuredAgentSessionHost {
|
||||
* on disk, so the same session can be attached again. */
|
||||
close(sessionId: string): Promise<void> {
|
||||
return this.serialize(sessionId, async () => {
|
||||
if (!this.sessions.has(sessionId)) {
|
||||
return
|
||||
}
|
||||
const context: StructuredAgentSessionEvictionContext = {
|
||||
sessionId,
|
||||
eventSink: this.runtimeState.eventSinkFor(sessionId),
|
||||
adapter: this.deps.adapter,
|
||||
forget: () => this.sessions.delete(sessionId),
|
||||
discardSink: () => this.runtimeState.discardEventSink(sessionId)
|
||||
}
|
||||
await evictStructuredAgentSession(context)
|
||||
await evictHeldStructuredAgentSession(this.lifetimeContext(), sessionId)
|
||||
// Whoever asked for the close, the surfaces that were holding this session are looking at a
|
||||
// session that no longer exists. A failed eviction throws above and keeps them.
|
||||
this.holds.forget(sessionId)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -190,6 +220,7 @@ export class StructuredAgentSessionHost {
|
||||
this.runtimeState.flushEventSink(sessionId)
|
||||
|
||||
async flushAllStreamedEvents(): Promise<void> {
|
||||
this.holds.dispose()
|
||||
this.runtimeState.stopLeaseRenewal()
|
||||
this.handoffs.stopTuiHistoryCatchup()
|
||||
await this.tasks.drainAttaches()
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
// Handing the durable lease back after eviction stopped this host's child.
|
||||
//
|
||||
// Guarded on `hasProviderChild` for a reason that is not bookkeeping: a session restored only for
|
||||
// reading, or one a TUI owns, names an owner process this host never started and may still be
|
||||
// alive. Writing `exit-observed` against that record would release a lease out from under a running
|
||||
// process and let a second writer in.
|
||||
|
||||
import {
|
||||
isSurfaceReleasableAgentSessionRecord,
|
||||
releaseStoredAgentSessionOwnerAfterSurfaceClose
|
||||
} from '../../runtime/agent-session-surface-release-transition'
|
||||
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
|
||||
|
||||
export async function releaseStoredStructuredAgentSessionOwner(input: {
|
||||
store: AgentSessionRecordStore
|
||||
sessionId: string
|
||||
hasProviderChild: boolean
|
||||
now: number
|
||||
}): Promise<void> {
|
||||
if (!input.hasProviderChild) {
|
||||
return
|
||||
}
|
||||
const record = input.store.getRecord(input.sessionId)
|
||||
if (!record || !isSurfaceReleasableAgentSessionRecord(record)) {
|
||||
return
|
||||
}
|
||||
await releaseStoredAgentSessionOwnerAfterSurfaceClose(input.store, {
|
||||
sessionId: input.sessionId,
|
||||
expectedFence: record.lease.runtimeFence,
|
||||
now: input.now
|
||||
})
|
||||
}
|
||||
+6
-5
@@ -4,6 +4,7 @@ import {
|
||||
LEGACY_AGENT_SESSION_ID
|
||||
} from '../../../shared/agent-session-record-legacy.test-fixture'
|
||||
import { loadAgentSessionRecord } from '../../../shared/agent-session-record-load'
|
||||
import { isResumableStructuredAgentSessionRecord } from './structured-agent-session-resume-eligibility'
|
||||
import { restoreStructuredAgentSessionsOnRestart } from './structured-agent-session-restart-restore'
|
||||
|
||||
describe('legacy structured session restart', () => {
|
||||
@@ -12,7 +13,7 @@ describe('legacy structured session restart', () => {
|
||||
if (loaded.status !== 'upgraded') {
|
||||
throw new Error(`expected upgrade, got ${loaded.status}`)
|
||||
}
|
||||
const resume = vi.fn(async () => true)
|
||||
const serialize = vi.fn(async <T>(_sessionId: string, task: () => Promise<T>) => task())
|
||||
|
||||
await restoreStructuredAgentSessionsOnRestart({
|
||||
store: { getRecord: () => loaded.record } as never,
|
||||
@@ -20,9 +21,7 @@ describe('legacy structured session restart', () => {
|
||||
records: [loaded.record],
|
||||
reconcile: async () => null,
|
||||
resolveRecovery: async () => 'not-applicable',
|
||||
operationId: () => 'legacy-upgrade-operation',
|
||||
resume,
|
||||
serialize: async () => undefined as never,
|
||||
serialize: serialize as never,
|
||||
hasSession: () => false,
|
||||
onReadable: vi.fn(),
|
||||
restoreHandoff: vi.fn(async () => undefined)
|
||||
@@ -32,6 +31,8 @@ describe('legacy structured session restart', () => {
|
||||
sessionId: LEGACY_AGENT_SESSION_ID,
|
||||
lease: { claimStatus: 'conflicted', handoffStage: 'manual-recovery' }
|
||||
})
|
||||
expect(resume).not.toHaveBeenCalled()
|
||||
// Restart no longer resumes ANYTHING, so the guard that matters now is the one a surface hits:
|
||||
// a conflicted record under manual recovery is not a record a hold may acquire.
|
||||
expect(isResumableStructuredAgentSessionRecord(loaded.record)).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -20,6 +20,7 @@ export type RestoredStructuredAgentSessionRead = {
|
||||
journal: AgentSessionJournal
|
||||
params: AgentSessionAttachParams
|
||||
fence: number
|
||||
hasProviderChild: false
|
||||
}
|
||||
|
||||
export async function restoreStructuredAgentSessionRead(
|
||||
@@ -47,7 +48,8 @@ export async function restoreStructuredAgentSessionRead(
|
||||
identity: journalIdentityFor(record, params),
|
||||
journalDir
|
||||
})
|
||||
return { journal, params, fence: record.lease.runtimeFence }
|
||||
// Read restore opens the journal and nothing else: no adapter call, so no provider child.
|
||||
return { journal, params, fence: record.lease.runtimeFence, hasProviderChild: false }
|
||||
}
|
||||
|
||||
export function attachParamsForRecord(
|
||||
|
||||
+1
-7
@@ -1,8 +1,6 @@
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
|
||||
import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire'
|
||||
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
|
||||
import type { AgentSessionAttachParams } from './structured-agent-session-attach'
|
||||
import type { RestoredStructuredAgentSessionRead } from './structured-agent-session-read-restore'
|
||||
import { restoreStructuredAgentSessionsOnRestart } from './structured-agent-session-restart-restore'
|
||||
|
||||
@@ -16,21 +14,17 @@ export class StructuredAgentSessionReadableRestorer {
|
||||
supportsRecord: (record: AgentSessionRecord) => boolean
|
||||
reconcile: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
|
||||
resolveRecovery: (sessionId: string) => Promise<unknown>
|
||||
resume: (params: AgentSessionAttachParams) => Promise<boolean>
|
||||
serialize: <T>(sessionId: string, task: () => Promise<T>) => Promise<T>
|
||||
hasSession: (sessionId: string) => boolean
|
||||
onReadable: (sessionId: string, restored: RestoredStructuredAgentSessionRead) => void
|
||||
restoreHandoff: (sessionId: string) => Promise<void>
|
||||
now: () => number
|
||||
}
|
||||
) {}
|
||||
|
||||
restore(): Promise<void> {
|
||||
this.restorePromise ??= restoreStructuredAgentSessionsOnRestart({
|
||||
...this.input,
|
||||
records: this.input.store.listRecords().filter(this.input.supportsRecord),
|
||||
operationId: () =>
|
||||
`${Math.trunc(this.input.now()).toString().padStart(13, '0')}-${randomUUID().replaceAll('-', '')}`
|
||||
records: this.input.store.listRecords().filter(this.input.supportsRecord)
|
||||
}).catch((error) => {
|
||||
this.restorePromise = null
|
||||
throw error
|
||||
|
||||
+12
-1
@@ -140,7 +140,7 @@ describe('recovery exits', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('heals a stranded native owner during startup restore without waiting for a client', async () => {
|
||||
it('heals a stranded native owner during startup restore, and spawns nothing until a surface asks', async () => {
|
||||
expect((await host.attach(CALLER, hostTestAttachParams(null))).ok).toBe(true)
|
||||
await reopenStore()
|
||||
|
||||
@@ -159,7 +159,18 @@ describe('recovery exits', () => {
|
||||
|
||||
await host.restoreReadableSessions()
|
||||
|
||||
// Healing is startup's job; spawning is not. The orphan is stopped and the lease is free, but
|
||||
// nothing has asked to look at this session, so no replacement child exists yet.
|
||||
expect(stopOwnerProcess).toHaveBeenCalledWith(4242, 'SIGTERM')
|
||||
expect(acquire).toHaveBeenCalledTimes(1)
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'released',
|
||||
handoffStage: null,
|
||||
ownerProcess: null
|
||||
})
|
||||
|
||||
await host.hold(SESSION, 'surface-1')
|
||||
|
||||
expect(acquire).toHaveBeenCalledTimes(2)
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
runtimeFence: 3,
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
// The delay between "nothing holds this session" and "stop its provider child".
|
||||
//
|
||||
// TWO reasons it is not immediate. A surface that reconnects — a mobile socket dropping on a
|
||||
// network switch, a renderer remounting a tab — releases and re-holds within a second, and killing
|
||||
// an app-server in that window costs the user a respawn plus a resume for nothing. And a turn the
|
||||
// user already asked for must finish: the provider is mid-answer, the journal has an open turn
|
||||
// marker, and stopping the child there strands both.
|
||||
//
|
||||
// So the clock arms when the last holder leaves, and a tick that finds a turn still running RE-ARMS
|
||||
// instead of evicting. That is what makes the wait start at the later of the two events rather than
|
||||
// at whichever came first.
|
||||
|
||||
export const STRUCTURED_AGENT_SESSION_RELEASE_GRACE_MS = 15_000
|
||||
|
||||
export type StructuredAgentSessionReleaseClockDeps = {
|
||||
/** Never evict mid-turn; a true answer re-arms the clock instead. */
|
||||
isTurnActive: (sessionId: string) => boolean
|
||||
/** Re-checked at fire time: a holder may have arrived while the timer ran. */
|
||||
isHeld: (sessionId: string) => boolean
|
||||
evict: (sessionId: string) => Promise<void>
|
||||
onError?: (input: { sessionId: string; error: unknown }) => void
|
||||
graceMs?: number
|
||||
}
|
||||
|
||||
export class StructuredAgentSessionReleaseClock {
|
||||
private readonly timers = new Map<string, ReturnType<typeof setTimeout>>()
|
||||
private readonly graceMs: number
|
||||
|
||||
constructor(private readonly deps: StructuredAgentSessionReleaseClockDeps) {
|
||||
this.graceMs = deps.graceMs ?? STRUCTURED_AGENT_SESSION_RELEASE_GRACE_MS
|
||||
}
|
||||
|
||||
arm(sessionId: string): void {
|
||||
this.cancel(sessionId)
|
||||
const timer = setTimeout(() => {
|
||||
this.timers.delete(sessionId)
|
||||
this.fire(sessionId)
|
||||
}, this.graceMs)
|
||||
// A pending release must never be the reason a process stays alive at quit.
|
||||
timer.unref?.()
|
||||
this.timers.set(sessionId, timer)
|
||||
}
|
||||
|
||||
cancel(sessionId: string): void {
|
||||
const timer = this.timers.get(sessionId)
|
||||
if (timer) {
|
||||
clearTimeout(timer)
|
||||
this.timers.delete(sessionId)
|
||||
}
|
||||
}
|
||||
|
||||
isArmed(sessionId: string): boolean {
|
||||
return this.timers.has(sessionId)
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
for (const timer of this.timers.values()) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
this.timers.clear()
|
||||
}
|
||||
|
||||
private fire(sessionId: string): void {
|
||||
if (this.deps.isHeld(sessionId)) {
|
||||
return
|
||||
}
|
||||
if (this.deps.isTurnActive(sessionId)) {
|
||||
this.arm(sessionId)
|
||||
return
|
||||
}
|
||||
void this.deps.evict(sessionId).catch((error: unknown) => {
|
||||
this.deps.onError?.({ sessionId, error })
|
||||
})
|
||||
}
|
||||
}
|
||||
+14
-19
@@ -1,9 +1,20 @@
|
||||
// What a restart owes a persisted session, and what it does NOT.
|
||||
//
|
||||
// It owes reconciliation — every lease loaded from disk names an owner from a process generation
|
||||
// that no longer exists, and adjudicating that is startup's job. It owes an exit from any recovery
|
||||
// stage the evidence now permits. And it owes a READABLE session: the journal open, history
|
||||
// answerable, the tab restorable.
|
||||
//
|
||||
// It does not owe a provider child. This used to resume every record whose lease was `released`
|
||||
// with no handoff in flight, which is the normal end state of a chat the user closed cleanly — so a
|
||||
// healthy profile started an app-server per session it had ever used, in parallel, at every launch,
|
||||
// with no client attached and nothing on screen. A child now exists because a surface asked for the
|
||||
// session (see `structured-agent-session-holds`), not because a record survived on disk.
|
||||
|
||||
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
|
||||
import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire'
|
||||
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
|
||||
import type { AgentSessionAttachParams } from './structured-agent-session-attach'
|
||||
import {
|
||||
attachParamsForRecord,
|
||||
restoreStructuredAgentSessionRead,
|
||||
type RestoredStructuredAgentSessionRead
|
||||
} from './structured-agent-session-read-restore'
|
||||
@@ -14,8 +25,6 @@ export async function restoreStructuredAgentSessionsOnRestart(input: {
|
||||
records: AgentSessionRecord[]
|
||||
reconcile: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
|
||||
resolveRecovery: (sessionId: string) => Promise<unknown>
|
||||
operationId: () => string
|
||||
resume: (params: AgentSessionAttachParams) => Promise<boolean>
|
||||
serialize: <T>(sessionId: string, task: () => Promise<T>) => Promise<T>
|
||||
hasSession: (sessionId: string) => boolean
|
||||
onReadable: (sessionId: string, restored: RestoredStructuredAgentSessionRead) => void
|
||||
@@ -28,23 +37,9 @@ export async function restoreStructuredAgentSessionsOnRestart(input: {
|
||||
// A session latched in recovery exits here at startup, without waiting for a client.
|
||||
await input.resolveRecovery(sessionId)
|
||||
}
|
||||
const current = input.store.getRecord(sessionId)
|
||||
if (
|
||||
!unreconciled &&
|
||||
current?.lease.claimStatus === 'released' &&
|
||||
current.lease.handoffStage === null &&
|
||||
(await input.resume(
|
||||
attachParamsForRecord(current, {
|
||||
clientOperationId: input.operationId(),
|
||||
expectedRuntimeFence: current.lease.runtimeFence,
|
||||
runtimeKind: 'native'
|
||||
})
|
||||
))
|
||||
) {
|
||||
return
|
||||
}
|
||||
await input.serialize(sessionId, async () => {
|
||||
if (input.hasSession(sessionId)) {
|
||||
// A surface that took a hold mid-restore already attached this one.
|
||||
await input.restoreHandoff(sessionId)
|
||||
return
|
||||
}
|
||||
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
// When a record may be handed back a provider child.
|
||||
//
|
||||
// This used to be inline in the restart restore, which is what made it a STARTUP rule: every record
|
||||
// whose lease looked like this got a child, at launch, whether or not anything was going to look at
|
||||
// it. `released` + no handoff is the normal end state of a chat the user closed cleanly, so a
|
||||
// healthy profile respawned everything it had ever opened. The predicate itself was never wrong —
|
||||
// it answers "may this be resumed", not "should it be" — so it lives here now and the caller that
|
||||
// knows a surface is asking is the only one that acts on it.
|
||||
|
||||
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { AgentSessionAttachParams } from './structured-agent-session-attach'
|
||||
import { attachParamsForRecord } from './structured-agent-session-read-restore'
|
||||
|
||||
export function isResumableStructuredAgentSessionRecord(record: AgentSessionRecord): boolean {
|
||||
return (
|
||||
!record.lease.unreconciled &&
|
||||
record.lease.claimStatus === 'released' &&
|
||||
record.lease.handoffStage === null
|
||||
)
|
||||
}
|
||||
|
||||
/** Attach params for a resume, or null when this record's lease is somebody else's problem. */
|
||||
export function structuredAgentSessionResumeParams(
|
||||
record: AgentSessionRecord,
|
||||
clientOperationId: string
|
||||
): AgentSessionAttachParams | null {
|
||||
if (!isResumableStructuredAgentSessionRecord(record)) {
|
||||
return null
|
||||
}
|
||||
return attachParamsForRecord(record, {
|
||||
clientOperationId,
|
||||
expectedRuntimeFence: record.lease.runtimeFence,
|
||||
runtimeKind: 'native'
|
||||
})
|
||||
}
|
||||
|
||||
/** `<13-digit ms>-<32 hex>`, the only operation-id shape the durable ledger admits. */
|
||||
export function structuredAgentSessionResumeOperationId(now: number): string {
|
||||
return `${Math.trunc(now).toString().padStart(13, '0')}-${randomUUID().replaceAll('-', '')}`
|
||||
}
|
||||
+246
@@ -0,0 +1,246 @@
|
||||
// The lifetime of a provider child, from the surfaces that hold the session.
|
||||
//
|
||||
// Two leaks meet here and each has to be tested against the real host, not a double: a chat that
|
||||
// closes without stopping its app-server, and a launch that starts one for every record on disk.
|
||||
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest'
|
||||
import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication'
|
||||
import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire'
|
||||
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
|
||||
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
|
||||
import type { StructuredAgentSessionEventSink } from './structured-agent-session-event-sink'
|
||||
import { StructuredAgentSessionHost } from './structured-agent-session-host'
|
||||
import {
|
||||
HOST_TEST_NOW as NOW,
|
||||
HOST_TEST_SESSION as SESSION,
|
||||
HOST_TEST_THREAD as THREAD,
|
||||
hostTestAttachParams,
|
||||
resetHostTestOperationIds
|
||||
} from './structured-agent-session-host-test-data'
|
||||
|
||||
const CALLER = { callerKey: 'client-1' }
|
||||
const SURFACE = 'desktop-chat:1'
|
||||
/** Short enough to keep the suite fast, long enough that an eviction is a decision and not a race. */
|
||||
const GRACE_MS = 5
|
||||
|
||||
let root: string
|
||||
let store: AgentSessionRecordStore
|
||||
let host: StructuredAgentSessionHost
|
||||
let acquire: Mock<StructuredAgentSessionAdapter['acquire']>
|
||||
let closeSession: Mock<NonNullable<StructuredAgentSessionAdapter['closeSession']>>
|
||||
let sink: StructuredAgentSessionEventSink | null
|
||||
let hostErrors: unknown[]
|
||||
|
||||
function adapter(): StructuredAgentSessionAdapter {
|
||||
return {
|
||||
acquire,
|
||||
closeSession,
|
||||
releaseAcquisition: vi.fn(async () => undefined),
|
||||
dispatch: vi.fn(async () => ({ state: 'rejected' as const, reason: 'unused' })),
|
||||
cancelTurn: vi.fn(async () => ({ cancelled: false })),
|
||||
answerPrompt: vi.fn(async () => undefined),
|
||||
setOption: vi.fn(async () => undefined)
|
||||
}
|
||||
}
|
||||
|
||||
function openHost(probeOwner?: (record: never) => Promise<AgentSessionOwnerProbe>): void {
|
||||
host = new StructuredAgentSessionHost({
|
||||
store,
|
||||
adapter: adapter(),
|
||||
journalRoot: root,
|
||||
claimKeyId: 'key-1',
|
||||
mintSpawnToken: () => `spawn-${acquire.mock.calls.length}`,
|
||||
releaseGraceMs: GRACE_MS,
|
||||
now: () => NOW,
|
||||
onEventSinkError: ({ error }) => hostErrors.push(error),
|
||||
...(probeOwner ? { probeOwner: probeOwner as never } : {})
|
||||
})
|
||||
}
|
||||
|
||||
/** A fresh app generation over the same durable store, with its owner proven gone. */
|
||||
async function reboot(): Promise<void> {
|
||||
await host.flushAllStreamedEvents()
|
||||
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
|
||||
openHost(async () => ({ outcome: 'pid-absent' }))
|
||||
acquire.mockClear()
|
||||
closeSession.mockClear()
|
||||
}
|
||||
|
||||
async function attach(): Promise<void> {
|
||||
expect(await host.attach(CALLER, hostTestAttachParams(null))).toMatchObject({ ok: true })
|
||||
}
|
||||
|
||||
function emitTurnLifecycle(state: 'running' | 'completed', ordinal: number): void {
|
||||
sink?.appendItem(
|
||||
{ provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal },
|
||||
{ kind: 'status', text: state, turnLifecycle: { turnId: 'turn-1', state } }
|
||||
)
|
||||
}
|
||||
|
||||
/** Eviction is a sequence, not an event: the child stops first and the session is forgotten last. */
|
||||
function waitForEviction(): Promise<void> {
|
||||
return vi.waitFor(() => {
|
||||
expect(closeSession).toHaveBeenCalledWith(SESSION)
|
||||
expect(host.hasSession(SESSION)).toBe(false)
|
||||
})
|
||||
}
|
||||
|
||||
/** Long enough for several grace windows to elapse, so "not evicted" means the clock declined. */
|
||||
function waitOutSeveralGraceWindows(): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, GRACE_MS * 20))
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-surface-lifetime-'))
|
||||
resetHostTestOperationIds()
|
||||
sink = null
|
||||
hostErrors = []
|
||||
acquire = vi.fn(async ({ fence, spawnToken, events }) => {
|
||||
sink = events ?? null
|
||||
return {
|
||||
process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000, spawnToken },
|
||||
link: {
|
||||
linkId: `link-${fence}`,
|
||||
handle: { provider: 'codex' as const, threadId: THREAD },
|
||||
origin: store.getRecord(SESSION)?.providerHandleChain.length
|
||||
? ('resumed' as const)
|
||||
: ('created' as const),
|
||||
mintedAtFence: fence,
|
||||
observedAt: NOW
|
||||
}
|
||||
}
|
||||
})
|
||||
closeSession = vi.fn(async () => undefined)
|
||||
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
|
||||
openHost()
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await host.flushAllStreamedEvents()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe('a chat that closes', () => {
|
||||
it('releases the provider child it was holding', async () => {
|
||||
await attach()
|
||||
await host.hold(SESSION, SURFACE)
|
||||
|
||||
host.release(SESSION, SURFACE)
|
||||
|
||||
await waitForEviction()
|
||||
expect(hostErrors).toEqual([])
|
||||
// The record and its journal stay; only the process and the claim on it go.
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'released',
|
||||
ownerProcess: null,
|
||||
deathEvidence: { kind: 'exit-observed' }
|
||||
})
|
||||
})
|
||||
|
||||
it('keeps the child while another surface still holds the session', async () => {
|
||||
await attach()
|
||||
await host.hold(SESSION, SURFACE)
|
||||
await host.hold(SESSION, 'paired-phone:1')
|
||||
|
||||
host.release(SESSION, SURFACE)
|
||||
await waitOutSeveralGraceWindows()
|
||||
|
||||
expect(closeSession).not.toHaveBeenCalled()
|
||||
expect(host.hasSession(SESSION)).toBe(true)
|
||||
})
|
||||
|
||||
it('does not lose the session to a release the client sent twice', async () => {
|
||||
await attach()
|
||||
await host.hold(SESSION, SURFACE)
|
||||
await host.hold(SESSION, 'paired-phone:1')
|
||||
|
||||
// A retried release must retire ONE holder, which is what a set gets right and a count does not.
|
||||
host.release(SESSION, SURFACE)
|
||||
host.release(SESSION, SURFACE)
|
||||
await waitOutSeveralGraceWindows()
|
||||
|
||||
expect(closeSession).not.toHaveBeenCalled()
|
||||
expect(host.hasSession(SESSION)).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('a session with a turn in flight', () => {
|
||||
it('is not evicted while the turn runs, and is once it ends', async () => {
|
||||
await attach()
|
||||
await host.hold(SESSION, SURFACE)
|
||||
emitTurnLifecycle('running', 1)
|
||||
await host.flushStreamedEvents(SESSION)
|
||||
|
||||
host.release(SESSION, SURFACE)
|
||||
await waitOutSeveralGraceWindows()
|
||||
|
||||
expect(closeSession).not.toHaveBeenCalled()
|
||||
expect(host.hasSession(SESSION)).toBe(true)
|
||||
|
||||
emitTurnLifecycle('completed', 2)
|
||||
await host.flushStreamedEvents(SESSION)
|
||||
|
||||
await waitForEviction()
|
||||
})
|
||||
})
|
||||
|
||||
describe('startup', () => {
|
||||
it('restores a session for reading without spawning a provider child', async () => {
|
||||
await attach()
|
||||
await reboot()
|
||||
|
||||
await host.restoreReadableSessions()
|
||||
|
||||
// The record is readable — the tab comes back, history answers — and nothing is running.
|
||||
expect(acquire).not.toHaveBeenCalled()
|
||||
expect(host.listSessionTabs()).toEqual([
|
||||
{ sessionId: SESSION, workspaceId: 'workspace-1', agent: 'codex' }
|
||||
])
|
||||
expect(host.history({ sessionId: SESSION, direction: 'tail' }).ok).toBe(true)
|
||||
expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('released')
|
||||
})
|
||||
|
||||
it('gives the child back to a chat a surface actually opens', async () => {
|
||||
await attach()
|
||||
await reboot()
|
||||
await host.restoreReadableSessions()
|
||||
|
||||
await host.hold(SESSION, SURFACE)
|
||||
|
||||
expect(acquire).toHaveBeenCalledOnce()
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'live',
|
||||
runtimeKind: 'native',
|
||||
ownerProcess: { pid: 4242 }
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('a session evicted and opened again', () => {
|
||||
it('publishes provider events to the reattached chat', async () => {
|
||||
await attach()
|
||||
await host.hold(SESSION, SURFACE)
|
||||
host.release(SESSION, SURFACE)
|
||||
await waitForEviction()
|
||||
|
||||
await host.hold(SESSION, 'desktop-chat:2')
|
||||
const events: AgentSessionSubscribeEvent[] = []
|
||||
const unsubscribe = host.subscribe({
|
||||
id: 'subscriber-1',
|
||||
sessionId: SESSION,
|
||||
emit: (event) => events.push(event)
|
||||
})
|
||||
sink?.appendItem(
|
||||
{ provider: 'codex', threadId: THREAD, turnId: 'turn-2', ordinal: 1 },
|
||||
{ kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'back again' }] }
|
||||
)
|
||||
sink?.publish()
|
||||
await host.flushStreamedEvents(SESSION)
|
||||
unsubscribe()
|
||||
|
||||
expect(JSON.stringify(events)).toContain('back again')
|
||||
})
|
||||
})
|
||||
@@ -87,7 +87,13 @@ async function adopt(
|
||||
journalRoot: input.deps.journalRoot,
|
||||
adapter: input.deps.adapter
|
||||
})
|
||||
const session = { journal: attached.journal, params, fence: record.lease.runtimeFence }
|
||||
// The TUI owns the process here; this host publishes the journal only.
|
||||
const session = {
|
||||
journal: attached.journal,
|
||||
params,
|
||||
fence: record.lease.runtimeFence,
|
||||
hasProviderChild: false
|
||||
}
|
||||
input.publish(sessionId, session)
|
||||
input.retain(sessionId, owner)
|
||||
input.snapshot(sessionId, session)
|
||||
|
||||
@@ -99,7 +99,12 @@ async function createCatchupFixture() {
|
||||
function createCatchup(input: Awaited<ReturnType<typeof createCatchupFixture>>) {
|
||||
return new StructuredTuiTranscriptCatchup({
|
||||
store,
|
||||
session: () => ({ journal: input.journal, params: {} as never, fence: input.fence }),
|
||||
session: () => ({
|
||||
hasProviderChild: false,
|
||||
journal: input.journal,
|
||||
params: {} as never,
|
||||
fence: input.fence
|
||||
}),
|
||||
schedule: async (_sessionId, task) => task(),
|
||||
publish: vi.fn(),
|
||||
reset: vi.fn()
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
// Releasing the lease when the LAST surface lets go of a session.
|
||||
//
|
||||
// Every other release in the wire needs a probe, because every other release is about a process
|
||||
// somebody else started and nobody watched die. This one is different: the host stopped its own
|
||||
// child through the adapter and the adapter proved the exit before this runs, so the evidence is
|
||||
// `exit-observed` rather than an adjudicated absence.
|
||||
//
|
||||
// The fence still moves. A released lease at the old fence would let a mutation a client queued
|
||||
// against the dead generation land on the next one.
|
||||
|
||||
import type { AgentSessionRecord } from '../../shared/agent-session-record'
|
||||
import { assertFence, withLease } from './agent-session-lease-transitions'
|
||||
import type { AgentSessionRecordStore } from './agent-session-record-store'
|
||||
|
||||
/** Whether this record is one THIS host may release on its own proof. A TUI owner, a session
|
||||
* mid-handoff, and a lease nobody holds are all somebody else's transition. */
|
||||
export function isSurfaceReleasableAgentSessionRecord(record: AgentSessionRecord): boolean {
|
||||
return (
|
||||
record.lease.runtimeKind === 'native' &&
|
||||
record.lease.claimStatus === 'live' &&
|
||||
record.lease.handoffStage === null &&
|
||||
record.lease.ownerProcess !== null
|
||||
)
|
||||
}
|
||||
|
||||
export function releaseAgentSessionOwnerAfterSurfaceClose(args: {
|
||||
record: AgentSessionRecord
|
||||
expectedFence: number
|
||||
now: number
|
||||
}): AgentSessionRecord {
|
||||
const { record } = args
|
||||
assertFence(record.lease, args.expectedFence)
|
||||
if (!isSurfaceReleasableAgentSessionRecord(record)) {
|
||||
throw new Error('agent_session_ownership_unknown')
|
||||
}
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
runtimeFence: record.lease.runtimeFence + 1,
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: null,
|
||||
processlessAt: null,
|
||||
claimStatus: 'released',
|
||||
lastRenewedAt: args.now,
|
||||
deathEvidence: {
|
||||
kind: 'exit-observed',
|
||||
detail: 'the last surface holding this session released it',
|
||||
observedAt: args.now
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/** Applied through the store's generic transition, the same way handoff records move. */
|
||||
export function releaseStoredAgentSessionOwnerAfterSurfaceClose(
|
||||
store: AgentSessionRecordStore,
|
||||
args: { sessionId: string; expectedFence: number; now: number }
|
||||
): Promise<AgentSessionRecord> {
|
||||
return store.transitionHandoff(args.sessionId, (record) =>
|
||||
releaseAgentSessionOwnerAfterSurfaceClose({ ...args, record })
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
// Who may see `agentSession.*` at all.
|
||||
//
|
||||
// Shared by every structured method file so one gate governs the whole surface: a client that does
|
||||
// not advertise `agent-session.structured.v1` is told the surface does not exist rather than being
|
||||
// handed a session it cannot render or drive — and, just as importantly, cannot make the host EXIST
|
||||
// by calling into it, which is an observable side effect.
|
||||
|
||||
import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version'
|
||||
import { getStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry'
|
||||
import type { StructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-host'
|
||||
import type { StructuredAgentSessionCaller } from '../../../native-chat/agent-session-wire/structured-agent-session-host-types'
|
||||
import type { RpcContext } from '../core'
|
||||
|
||||
/**
|
||||
* In-process callers are the same build as the host, so they carry no negotiated
|
||||
* capability list; every remote client must say it can read structured sessions.
|
||||
*/
|
||||
export function supportsStructuredSessions(ctx: RpcContext): boolean {
|
||||
return (
|
||||
ctx.clientKind === undefined ||
|
||||
(ctx.clientCapabilities?.includes(STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY) ?? false)
|
||||
)
|
||||
}
|
||||
|
||||
export function requireStructuredCapability(ctx: RpcContext): void {
|
||||
if (!supportsStructuredSessions(ctx)) {
|
||||
throw new Error('structured_agent_session_unsupported')
|
||||
}
|
||||
}
|
||||
|
||||
export function requireStructuredHost(ctx: RpcContext): StructuredAgentSessionHost {
|
||||
requireStructuredCapability(ctx)
|
||||
const host = getStructuredAgentSessionHost()
|
||||
if (!host) {
|
||||
throw new Error('structured_agent_session_unsupported')
|
||||
}
|
||||
return host
|
||||
}
|
||||
|
||||
/** Attach is the only way a session comes into being, so it is the only call
|
||||
* that builds the host. Every other method addresses a session that must
|
||||
* already be attached, and correctly reports absent when none is. */
|
||||
export async function ensureStructuredHostInstalled(ctx: RpcContext): Promise<void> {
|
||||
// Gated first: a client that cannot read structured sessions must not be able
|
||||
// to make the host exist, which is an observable side effect of the surface.
|
||||
if (!supportsStructuredSessions(ctx) || getStructuredAgentSessionHost()) {
|
||||
return
|
||||
}
|
||||
await ctx.runtime.ensureStructuredAgentSessionHost()
|
||||
}
|
||||
|
||||
/** Mirrors the existing agent-session host-authority derivation so one client
|
||||
* gets one operation namespace across both surfaces. */
|
||||
export function structuredCallerFor(ctx: RpcContext): StructuredAgentSessionCaller {
|
||||
return {
|
||||
callerKey: ctx.clientId?.trim() || `trusted-local:${ctx.clientKind ?? 'runtime'}`
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,224 @@
|
||||
// The wire half of a session's lifetime: who takes a hold, and what happens when they vanish.
|
||||
//
|
||||
// Run against the REAL subscription registry rather than a stub, because the backstop being tested
|
||||
// IS that registry's connection sweep — a stubbed `registerSubscriptionCleanup` would prove that
|
||||
// the handler called a function, which is not the claim.
|
||||
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest'
|
||||
import type { StructuredAgentSessionAdapter } from '../../../native-chat/agent-session-wire/structured-agent-session-adapter'
|
||||
import { StructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-host'
|
||||
import {
|
||||
HOST_TEST_NOW as NOW,
|
||||
HOST_TEST_SESSION as SESSION,
|
||||
HOST_TEST_THREAD as THREAD,
|
||||
hostTestAttachParams,
|
||||
resetHostTestOperationIds
|
||||
} from '../../../native-chat/agent-session-wire/structured-agent-session-host-test-data'
|
||||
import { setStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry'
|
||||
import {
|
||||
STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY,
|
||||
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY
|
||||
} from '../../../../shared/protocol-version'
|
||||
import { AgentSessionRecordStore } from '../../agent-session-record-store'
|
||||
import { OrcaRuntimeService } from '../../orca-runtime'
|
||||
import type { RpcResponse } from '../core'
|
||||
import { RpcDispatcher } from '../dispatcher'
|
||||
import { STRUCTURED_AGENT_SESSION_METHODS } from './structured-agent-session'
|
||||
|
||||
const CONNECTION = 'connection-1'
|
||||
const GRACE_MS = 5
|
||||
const CLIENT = {
|
||||
clientId: 'device-1',
|
||||
clientKind: 'mobile' as const,
|
||||
clientCapabilities: [
|
||||
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
|
||||
STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY
|
||||
],
|
||||
connectionId: CONNECTION
|
||||
}
|
||||
|
||||
let root: string
|
||||
let store: AgentSessionRecordStore
|
||||
let host: StructuredAgentSessionHost
|
||||
let runtime: OrcaRuntimeService
|
||||
let dispatcher: RpcDispatcher
|
||||
let closeSession: Mock<NonNullable<StructuredAgentSessionAdapter['closeSession']>>
|
||||
let requests = 0
|
||||
|
||||
async function call(method: string, params: unknown): Promise<RpcResponse> {
|
||||
const replies: RpcResponse[] = []
|
||||
requests += 1
|
||||
await dispatcher.dispatchStreaming(
|
||||
{ id: `request-${requests}`, authToken: 'token', method, params },
|
||||
(raw) => replies.push(JSON.parse(raw) as RpcResponse),
|
||||
CLIENT
|
||||
)
|
||||
return replies[0] as RpcResponse
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-hold-wire-'))
|
||||
resetHostTestOperationIds()
|
||||
requests = 0
|
||||
closeSession = vi.fn(async () => undefined)
|
||||
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
|
||||
host = new StructuredAgentSessionHost({
|
||||
store,
|
||||
adapter: {
|
||||
acquire: async ({ fence, spawnToken }) => ({
|
||||
process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000, spawnToken },
|
||||
link: {
|
||||
linkId: `link-${fence}`,
|
||||
handle: { provider: 'codex', threadId: THREAD },
|
||||
origin: store.getRecord(SESSION)?.providerHandleChain.length ? 'resumed' : 'created',
|
||||
mintedAtFence: fence,
|
||||
observedAt: NOW
|
||||
}
|
||||
}),
|
||||
closeSession,
|
||||
dispatch: async () => ({ state: 'rejected', reason: 'unused' }),
|
||||
cancelTurn: async () => ({ cancelled: false }),
|
||||
answerPrompt: async () => undefined,
|
||||
setOption: async () => undefined
|
||||
},
|
||||
journalRoot: root,
|
||||
claimKeyId: 'key-1',
|
||||
mintSpawnToken: () => 'spawn-a',
|
||||
releaseGraceMs: GRACE_MS,
|
||||
now: () => NOW
|
||||
})
|
||||
setStructuredAgentSessionHost(host)
|
||||
runtime = new OrcaRuntimeService()
|
||||
dispatcher = new RpcDispatcher({ runtime, methods: STRUCTURED_AGENT_SESSION_METHODS })
|
||||
expect(await host.attach({ callerKey: 'client-1' }, hostTestAttachParams(null))).toMatchObject({
|
||||
ok: true
|
||||
})
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
setStructuredAgentSessionHost(null)
|
||||
await host.flushAllStreamedEvents()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe('a client that holds a session', () => {
|
||||
it('keeps the provider child while the hold stands', async () => {
|
||||
expect(
|
||||
await call('agentSession.hold', { sessionId: SESSION, holderId: 'chat-1' })
|
||||
).toMatchObject({ ok: true })
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, GRACE_MS * 20))
|
||||
|
||||
expect(closeSession).not.toHaveBeenCalled()
|
||||
expect(host.hasSession(SESSION)).toBe(true)
|
||||
})
|
||||
|
||||
it('releases it when the client says so', async () => {
|
||||
await call('agentSession.hold', { sessionId: SESSION, holderId: 'chat-1' })
|
||||
|
||||
expect(
|
||||
await call('agentSession.release', { sessionId: SESSION, holderId: 'chat-1' })
|
||||
).toMatchObject({ ok: true })
|
||||
|
||||
await vi.waitFor(() => expect(host.hasSession(SESSION)).toBe(false))
|
||||
expect(closeSession).toHaveBeenCalledWith(SESSION)
|
||||
})
|
||||
})
|
||||
|
||||
// A phone that loses signal mid-conversation never sends a release. Before this, the app-server it
|
||||
// was talking to lived until the desktop quit.
|
||||
describe('a client that disappears without cleanup', () => {
|
||||
it('still releases the session when its transport closes', async () => {
|
||||
await call('agentSession.hold', { sessionId: SESSION, holderId: 'chat-1' })
|
||||
|
||||
runtime.cleanupSubscriptionsForConnection(CONNECTION)
|
||||
|
||||
await vi.waitFor(() => expect(host.hasSession(SESSION)).toBe(false))
|
||||
expect(closeSession).toHaveBeenCalledWith(SESSION)
|
||||
})
|
||||
|
||||
it('does not release a hold another connection is still holding', async () => {
|
||||
await call('agentSession.hold', { sessionId: SESSION, holderId: 'chat-1' })
|
||||
await dispatcher.dispatchStreaming(
|
||||
{
|
||||
id: 'request-other',
|
||||
authToken: 'token',
|
||||
method: 'agentSession.hold',
|
||||
params: { sessionId: SESSION, holderId: 'chat-1' }
|
||||
},
|
||||
() => {},
|
||||
{ ...CLIENT, clientId: 'device-2', connectionId: 'connection-2' }
|
||||
)
|
||||
|
||||
runtime.cleanupSubscriptionsForConnection(CONNECTION)
|
||||
await new Promise((resolve) => setTimeout(resolve, GRACE_MS * 20))
|
||||
|
||||
expect(closeSession).not.toHaveBeenCalled()
|
||||
expect(host.hasSession(SESSION)).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
// Paired builds ship on their own schedule, and the ones already in the field never learned to
|
||||
// hold. Subscribing is the only thing they do to say a chat is open, and the host no longer resumes
|
||||
// every record at launch on their behalf.
|
||||
describe('a paired client that predates holds', () => {
|
||||
/** Leaves the session where a restart leaves one: readable, with no provider child. */
|
||||
async function releaseAndRestoreForReading(): Promise<void> {
|
||||
await host.close(SESSION)
|
||||
expect(host.hasSession(SESSION)).toBe(false)
|
||||
await host.restoreReadableSessions()
|
||||
expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('released')
|
||||
}
|
||||
|
||||
async function subscribe(capabilities: string[]): Promise<void> {
|
||||
await dispatcher.dispatchStreaming(
|
||||
{
|
||||
id: 'request-subscribe',
|
||||
authToken: 'token',
|
||||
method: 'agentSession.subscribe',
|
||||
params: { sessionId: SESSION }
|
||||
},
|
||||
() => {},
|
||||
{ ...CLIENT, clientCapabilities: capabilities }
|
||||
)
|
||||
}
|
||||
|
||||
it('keeps its session alive for as long as it is subscribed', async () => {
|
||||
await subscribe([STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY])
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, GRACE_MS * 20))
|
||||
|
||||
expect(host.isHeld(SESSION)).toBe(true)
|
||||
expect(closeSession).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('releases it when that subscription connection goes away', async () => {
|
||||
await subscribe([STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY])
|
||||
|
||||
runtime.cleanupSubscriptionsForConnection(CONNECTION)
|
||||
|
||||
await vi.waitFor(() => expect(host.hasSession(SESSION)).toBe(false))
|
||||
})
|
||||
|
||||
it('gets its provider child back by subscribing to a session nothing is running', async () => {
|
||||
await releaseAndRestoreForReading()
|
||||
|
||||
await subscribe([STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY])
|
||||
|
||||
await vi.waitFor(() => expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('live'))
|
||||
})
|
||||
|
||||
it('does not let a current client stream alone resume a released session', async () => {
|
||||
await releaseAndRestoreForReading()
|
||||
|
||||
await subscribe([
|
||||
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
|
||||
STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY
|
||||
])
|
||||
|
||||
expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('released')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,59 @@
|
||||
// `agentSession.hold` / `agentSession.release` — a surface saying it is bound to a session.
|
||||
//
|
||||
// The holder identity is scoped to the CONNECTION, not taken from the client verbatim: two clients
|
||||
// are free to name their surfaces the same thing, and a hold that collides is one that a stranger
|
||||
// can release.
|
||||
//
|
||||
// The registered cleanup is the backstop, and the ONLY thing that covers a client which vanishes —
|
||||
// a phone that loses signal mid-turn never gets to send its release. Registering it before taking
|
||||
// the hold is deliberate: re-registering an id runs the previous cleanup synchronously, so the
|
||||
// stale release lands before this hold rather than after it.
|
||||
|
||||
import { defineMethod, type RpcAnyMethod, type RpcContext } from '../core'
|
||||
import {
|
||||
ensureStructuredHostInstalled,
|
||||
requireStructuredHost
|
||||
} from './structured-agent-session-gate'
|
||||
import { HoldParams } from './structured-agent-session-schemas'
|
||||
|
||||
const HOLD_CLEANUP_PREFIX = 'agentSession.hold'
|
||||
|
||||
function holderKeyFor(ctx: RpcContext, holderId: string): string {
|
||||
const client = ctx.clientId?.trim() || (ctx.clientKind ?? 'runtime')
|
||||
return `${ctx.connectionId ?? 'local'}:${client}:${holderId}`
|
||||
}
|
||||
|
||||
function holdCleanupIdFor(sessionId: string, holderKey: string): string {
|
||||
return `${HOLD_CLEANUP_PREFIX}:${holderKey}:${sessionId}`
|
||||
}
|
||||
|
||||
export const STRUCTURED_AGENT_SESSION_HOLD_METHODS: RpcAnyMethod[] = [
|
||||
defineMethod({
|
||||
name: 'agentSession.hold',
|
||||
params: HoldParams,
|
||||
handler: async (params, ctx) => {
|
||||
await ensureStructuredHostInstalled(ctx)
|
||||
const host = requireStructuredHost(ctx)
|
||||
const holderKey = holderKeyFor(ctx, params.holderId)
|
||||
ctx.runtime.registerSubscriptionCleanup(
|
||||
holdCleanupIdFor(params.sessionId, holderKey),
|
||||
() => host.release(params.sessionId, holderKey),
|
||||
ctx.connectionId
|
||||
)
|
||||
await host.hold(params.sessionId, holderKey)
|
||||
return { held: true as const }
|
||||
}
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'agentSession.release',
|
||||
params: HoldParams,
|
||||
handler: async (params, ctx) => {
|
||||
const host = requireStructuredHost(ctx)
|
||||
const holderKey = holderKeyFor(ctx, params.holderId)
|
||||
host.release(params.sessionId, holderKey)
|
||||
// Retires the backstop too; its release is a no-op against a holder already gone.
|
||||
ctx.runtime.cleanupSubscription(holdCleanupIdFor(params.sessionId, holderKey))
|
||||
return { released: true as const }
|
||||
}
|
||||
})
|
||||
]
|
||||
@@ -183,6 +183,13 @@ export const SetOptionParams = z
|
||||
|
||||
export const OptionsParams = z.object({ sessionId: SessionId }).strict()
|
||||
|
||||
/** One surface's claim on one session. The id names the surface, not the client: a chat tab and a
|
||||
* paired phone looking at the same session are two holders, and either leaving must not release
|
||||
* the other's. */
|
||||
export const HoldParams = z
|
||||
.object({ sessionId: SessionId, holderId: Identifier('Invalid holder id') })
|
||||
.strict()
|
||||
|
||||
export const HistoryParams = z
|
||||
.object({
|
||||
sessionId: SessionId,
|
||||
|
||||
@@ -194,7 +194,7 @@ describe('capability gating', () => {
|
||||
}
|
||||
// Bump deliberately: the whole agentSession.* surface is behind the structured capability,
|
||||
// so an additive method is invisible to old clients and needs no protocol bump.
|
||||
expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(16)
|
||||
expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(18)
|
||||
})
|
||||
|
||||
it('hides the surface from a declared client that did not advertise it', async () => {
|
||||
|
||||
@@ -6,15 +6,20 @@
|
||||
// is the whole visibility rule, because nothing else on the runtime publishes a
|
||||
// structured session.
|
||||
|
||||
import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version'
|
||||
import { STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version'
|
||||
import {
|
||||
agentSessionFingerprintConflict,
|
||||
computeAgentSessionPayloadFingerprint
|
||||
} from '../../../../shared/agent-session-mutation-envelope'
|
||||
import { getStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry'
|
||||
import type { StructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-host'
|
||||
import type { StructuredAgentSessionCaller } from '../../../native-chat/agent-session-wire/structured-agent-session-host-types'
|
||||
import { defineMethod, defineStreamingMethod, type RpcAnyMethod, type RpcContext } from '../core'
|
||||
import {
|
||||
ensureStructuredHostInstalled as ensureHostInstalled,
|
||||
requireStructuredCapability,
|
||||
requireStructuredHost as requireHost,
|
||||
structuredCallerFor as callerFor,
|
||||
supportsStructuredSessions
|
||||
} from './structured-agent-session-gate'
|
||||
import { STRUCTURED_AGENT_SESSION_HOLD_METHODS } from './structured-agent-session-hold'
|
||||
import {
|
||||
AttachParams,
|
||||
AdoptTerminalParams,
|
||||
@@ -35,32 +40,6 @@ import { hasMobileClipboardImagePath } from '../mobile-clipboard-image-provenanc
|
||||
|
||||
const SUBSCRIPTION_PREFIX = 'agentSession'
|
||||
|
||||
/**
|
||||
* In-process callers are the same build as the host, so they carry no negotiated
|
||||
* capability list; every remote client must say it can read structured sessions.
|
||||
*/
|
||||
function supportsStructuredSessions(ctx: RpcContext): boolean {
|
||||
return (
|
||||
ctx.clientKind === undefined ||
|
||||
(ctx.clientCapabilities?.includes(STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY) ?? false)
|
||||
)
|
||||
}
|
||||
|
||||
function requireStructuredCapability(ctx: RpcContext): void {
|
||||
if (!supportsStructuredSessions(ctx)) {
|
||||
throw new Error('structured_agent_session_unsupported')
|
||||
}
|
||||
}
|
||||
|
||||
function requireHost(ctx: RpcContext): StructuredAgentSessionHost {
|
||||
requireStructuredCapability(ctx)
|
||||
const host = getStructuredAgentSessionHost()
|
||||
if (!host) {
|
||||
throw new Error('structured_agent_session_unsupported')
|
||||
}
|
||||
return host
|
||||
}
|
||||
|
||||
function assertMobileImageProvenance(
|
||||
ctx: RpcContext,
|
||||
body: { blocks: { type: string; path?: string; url?: string }[] }
|
||||
@@ -78,26 +57,6 @@ function assertMobileImageProvenance(
|
||||
}
|
||||
}
|
||||
|
||||
/** Attach is the only way a session comes into being, so it is the only call
|
||||
* that builds the host. Every other method addresses a session that must
|
||||
* already be attached, and correctly reports absent when none is. */
|
||||
async function ensureHostInstalled(ctx: RpcContext): Promise<void> {
|
||||
// Gated first: a client that cannot read structured sessions must not be able
|
||||
// to make the host exist, which is an observable side effect of the surface.
|
||||
if (!supportsStructuredSessions(ctx) || getStructuredAgentSessionHost()) {
|
||||
return
|
||||
}
|
||||
await ctx.runtime.ensureStructuredAgentSessionHost()
|
||||
}
|
||||
|
||||
/** Mirrors the existing agent-session host-authority derivation so one client
|
||||
* gets one operation namespace across both surfaces. */
|
||||
function callerFor(ctx: RpcContext): StructuredAgentSessionCaller {
|
||||
return {
|
||||
callerKey: ctx.clientId?.trim() || `trusted-local:${ctx.clientKind ?? 'runtime'}`
|
||||
}
|
||||
}
|
||||
|
||||
function subscriptionIdFor(ctx: RpcContext, sessionId: string): string {
|
||||
const base = `${SUBSCRIPTION_PREFIX}:${ctx.connectionId ?? 'local'}:${sessionId}`
|
||||
// Shared control multiplexes several streams over one socket; the frame id
|
||||
@@ -276,6 +235,21 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [
|
||||
handler: async (params, ctx, emit) => {
|
||||
const host = requireHost(ctx)
|
||||
const subscriptionId = subscriptionIdFor(ctx, params.sessionId)
|
||||
// A live stream is a surface too. This is what keeps a client that never learned to hold —
|
||||
// an older paired build — from having its session evicted while it is reading, and what
|
||||
// releases one whose socket dies without a word.
|
||||
//
|
||||
// Retain-only for a client that holds its own surfaces: reading history must never be what
|
||||
// starts a provider process. A paired build that predates holds gets the resuming kind,
|
||||
// because subscribing IS the only thing it does to say a chat is open, and the host no longer
|
||||
// resumes every record at launch on its behalf.
|
||||
const streamHolder = `subscription:${subscriptionId}`
|
||||
const streamResumes =
|
||||
ctx.clientKind === 'mobile' &&
|
||||
!(
|
||||
ctx.clientCapabilities?.includes(STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY) ??
|
||||
false
|
||||
)
|
||||
let closed = false
|
||||
let dispose = (): void => {}
|
||||
ctx.runtime.registerSubscriptionCleanup(
|
||||
@@ -283,6 +257,7 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [
|
||||
() => {
|
||||
closed = true
|
||||
dispose()
|
||||
host.release(params.sessionId, streamHolder)
|
||||
},
|
||||
ctx.connectionId
|
||||
)
|
||||
@@ -299,6 +274,14 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [
|
||||
})
|
||||
if (closed) {
|
||||
dispose()
|
||||
} else {
|
||||
// Fire-and-forget, but never unhandled: a resume that refuses leaves the stream holding a
|
||||
// readable session, which is exactly what the client sees anyway.
|
||||
void host
|
||||
.hold(params.sessionId, streamHolder, { resume: streamResumes })
|
||||
.catch((error: unknown) =>
|
||||
console.warn('[agent-session] stream hold failed', params.sessionId, error)
|
||||
)
|
||||
}
|
||||
}
|
||||
}),
|
||||
@@ -317,5 +300,6 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [
|
||||
ctx.runtime.cleanupSubscriptionsByPrefix(`${base}:`)
|
||||
return { unsubscribed: true }
|
||||
}
|
||||
})
|
||||
}),
|
||||
...STRUCTURED_AGENT_SESSION_HOLD_METHODS
|
||||
]
|
||||
|
||||
@@ -193,7 +193,9 @@ const MOBILE_RPC_METHOD_ALLOWLIST = new Set([
|
||||
'agentSession.cancel',
|
||||
'agentSession.close',
|
||||
'agentSession.history',
|
||||
'agentSession.hold',
|
||||
'agentSession.options',
|
||||
'agentSession.release',
|
||||
'agentSession.respondToApproval',
|
||||
'agentSession.respondToQuestion',
|
||||
'agentSession.send',
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
// @vitest-environment happy-dom
|
||||
|
||||
// A structured chat is a view on a terminal tab, so closing the tab is an unmount and nothing else.
|
||||
// If that unmount does not reach main, the codex app-server behind the chat has no other way to
|
||||
// learn the chat is gone.
|
||||
|
||||
import { renderHook, waitFor } from '@testing-library/react'
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const mocks = vi.hoisted(() => ({ call: vi.fn() }))
|
||||
|
||||
vi.mock('@/runtime/structured-agent-session-client', () => ({
|
||||
callStructuredAgentSession: mocks.call,
|
||||
subscribeStructuredAgentSession: vi.fn()
|
||||
}))
|
||||
|
||||
import { useStructuredAgentSessionHold } from './use-structured-agent-session-hold'
|
||||
|
||||
const LOCAL_TARGET = { kind: 'local' } as const
|
||||
|
||||
function callsTo(method: string): unknown[] {
|
||||
return mocks.call.mock.calls.filter((call) => call[1] === method).map((call) => call[2])
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
mocks.call.mockReset()
|
||||
mocks.call.mockResolvedValue(undefined)
|
||||
})
|
||||
|
||||
describe('a mounted structured chat', () => {
|
||||
it('holds the session while it is on screen and releases it on unmount', async () => {
|
||||
const { unmount } = renderHook(() =>
|
||||
useStructuredAgentSessionHold({
|
||||
sessionId: 'session-alpha',
|
||||
target: LOCAL_TARGET,
|
||||
surface: 'desktop-chat'
|
||||
})
|
||||
)
|
||||
|
||||
await waitFor(() => expect(callsTo('agentSession.hold')).toHaveLength(1))
|
||||
const held = callsTo('agentSession.hold')[0] as { sessionId: string; holderId: string }
|
||||
expect(held.sessionId).toBe('session-alpha')
|
||||
expect(callsTo('agentSession.release')).toHaveLength(0)
|
||||
|
||||
unmount()
|
||||
|
||||
await waitFor(() =>
|
||||
expect(callsTo('agentSession.release')).toEqual([
|
||||
{ sessionId: 'session-alpha', holderId: held.holderId }
|
||||
])
|
||||
)
|
||||
})
|
||||
|
||||
it('does not release a hold that has not landed yet', async () => {
|
||||
let settleHold = (): void => {}
|
||||
mocks.call.mockImplementation((_target: unknown, method: string) =>
|
||||
method === 'agentSession.hold'
|
||||
? new Promise<void>((resolve) => {
|
||||
settleHold = resolve
|
||||
})
|
||||
: Promise.resolve()
|
||||
)
|
||||
const { unmount } = renderHook(() =>
|
||||
useStructuredAgentSessionHold({
|
||||
sessionId: 'session-alpha',
|
||||
target: LOCAL_TARGET,
|
||||
surface: 'desktop-chat'
|
||||
})
|
||||
)
|
||||
await waitFor(() => expect(callsTo('agentSession.hold')).toHaveLength(1))
|
||||
|
||||
unmount()
|
||||
// The hold is still in flight; releasing now would leave the late hold with nothing to undo it.
|
||||
expect(callsTo('agentSession.release')).toHaveLength(0)
|
||||
|
||||
settleHold()
|
||||
|
||||
await waitFor(() => expect(callsTo('agentSession.release')).toHaveLength(1))
|
||||
})
|
||||
|
||||
it('keeps one hold across re-renders that rebuild the target object', async () => {
|
||||
const { rerender, unmount } = renderHook(
|
||||
(props: { sessionId: string }) =>
|
||||
useStructuredAgentSessionHold({
|
||||
sessionId: props.sessionId,
|
||||
target: { kind: 'local' },
|
||||
surface: 'desktop-chat'
|
||||
}),
|
||||
{ initialProps: { sessionId: 'session-alpha' } }
|
||||
)
|
||||
await waitFor(() => expect(callsTo('agentSession.hold')).toHaveLength(1))
|
||||
|
||||
rerender({ sessionId: 'session-alpha' })
|
||||
rerender({ sessionId: 'session-alpha' })
|
||||
|
||||
expect(callsTo('agentSession.hold')).toHaveLength(1)
|
||||
expect(callsTo('agentSession.release')).toHaveLength(0)
|
||||
unmount()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,55 @@
|
||||
// The desktop chat telling main that this session is on screen.
|
||||
//
|
||||
// A structured chat is an in-place view on a terminal tab, so closing the tab unmounts this and
|
||||
// nothing else in the close path knows a provider process is involved: `closeUnifiedTab` retires
|
||||
// the PTY and drops the tab, main hears nothing, and a codex app-server outlives the chat for the
|
||||
// rest of the app's life. Mount/unmount is the honest signal — it covers closing the tab, closing
|
||||
// the window, and switching the pane back to the terminal, none of which share a code path.
|
||||
//
|
||||
// The release CHAINS off the hold rather than racing it: an unmount during the hold's round trip
|
||||
// would otherwise release a hold that has not landed yet, and the late hold would never be undone.
|
||||
|
||||
import { useEffect, useRef } from 'react'
|
||||
import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client'
|
||||
import { callStructuredAgentSession } from '@/runtime/structured-agent-session-client'
|
||||
|
||||
let holderOrdinal = 0
|
||||
|
||||
export function structuredAgentSessionHolderId(surface: string): string {
|
||||
holderOrdinal += 1
|
||||
return `${surface}:${holderOrdinal}`
|
||||
}
|
||||
|
||||
export function useStructuredAgentSessionHold(args: {
|
||||
sessionId: string
|
||||
target: RuntimeClientTarget
|
||||
surface: string
|
||||
}): void {
|
||||
const { sessionId, surface, target } = args
|
||||
// Keyed by VALUE, not identity: callers build the target inline, so an identity dependency would
|
||||
// release and re-take the hold on every render of the pane.
|
||||
const targetKey = target.kind === 'local' ? 'local' : `environment:${target.environmentId}`
|
||||
const targetRef = useRef(target)
|
||||
// Synced in an effect declared first (so it lands before the hold below) rather than in render:
|
||||
// a render React discards must not leak its target into the next commit.
|
||||
useEffect(() => {
|
||||
targetRef.current = target
|
||||
}, [target])
|
||||
useEffect(() => {
|
||||
const runtimeTarget = targetRef.current
|
||||
const holderId = structuredAgentSessionHolderId(surface)
|
||||
const held = callStructuredAgentSession(runtimeTarget, 'agentSession.hold', {
|
||||
sessionId,
|
||||
holderId
|
||||
// An older host has no such method; the session still reads, it just is not held.
|
||||
}).catch(() => undefined)
|
||||
return () => {
|
||||
void held.then(() =>
|
||||
callStructuredAgentSession(runtimeTarget, 'agentSession.release', {
|
||||
sessionId,
|
||||
holderId
|
||||
}).catch(() => undefined)
|
||||
)
|
||||
}
|
||||
}, [sessionId, surface, targetKey])
|
||||
}
|
||||
@@ -146,8 +146,9 @@ describe('useStructuredAgentSession options', () => {
|
||||
it('mints a fresh operation when the same option is retried after a typed refusal', async () => {
|
||||
let attempts = 0
|
||||
mocks.call.mockImplementation((_target, method) => {
|
||||
if (method === 'agentSession.options') {
|
||||
return Promise.resolve(OPTIONS)
|
||||
if (method !== 'agentSession.setOption') {
|
||||
// The hook also holds the session while it is mounted; only option writes are attempts.
|
||||
return Promise.resolve(method === 'agentSession.options' ? OPTIONS : null)
|
||||
}
|
||||
attempts += 1
|
||||
return Promise.resolve(
|
||||
@@ -197,8 +198,9 @@ describe('useStructuredAgentSession options', () => {
|
||||
it('reuses an option operation after a pending admission refusal', async () => {
|
||||
let attempts = 0
|
||||
mocks.call.mockImplementation((_target, method) => {
|
||||
if (method === 'agentSession.options') {
|
||||
return Promise.resolve(OPTIONS)
|
||||
if (method !== 'agentSession.setOption') {
|
||||
// The hook also holds the session while it is mounted; only option writes are attempts.
|
||||
return Promise.resolve(method === 'agentSession.options' ? OPTIONS : null)
|
||||
}
|
||||
attempts += 1
|
||||
return Promise.resolve(
|
||||
|
||||
@@ -25,6 +25,7 @@ import {
|
||||
structuredSessionOperationId,
|
||||
useStructuredAgentSessionOutbox
|
||||
} from './use-structured-agent-session-outbox'
|
||||
import { useStructuredAgentSessionHold } from './use-structured-agent-session-hold'
|
||||
import { useStructuredAgentSessionRead } from './use-structured-agent-session-read'
|
||||
import { projectStructuredAgentSessionMessages } from './structured-agent-session-message-projection'
|
||||
|
||||
@@ -38,6 +39,9 @@ export function useStructuredAgentSession(args: {
|
||||
agent: AgentType
|
||||
}) {
|
||||
const { agent, 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 stateRef = useRef(state)
|
||||
const [writeError, setWriteError] = useState<string | null>(null)
|
||||
|
||||
@@ -104,6 +104,13 @@ export const AGENT_SESSION_OMP_RESUME_PATH_RUNTIME_CAPABILITY =
|
||||
// can neither display nor drive. The host also refuses every agentSession.*
|
||||
// method from a connection that does not advertise this.
|
||||
export const STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY = 'agent-session.structured.v1' as const
|
||||
// Why: a client that advertises this takes an explicit hold on every session surface it opens, so
|
||||
// the host knows when the last surface is gone and may stop the provider child. A client WITHOUT it
|
||||
// (every paired build that shipped before holds existed) is held for the life of its subscription
|
||||
// instead, and that subscription is what resumes its session — which is how those builds keep being
|
||||
// able to send after the host stopped resuming every record at launch.
|
||||
export const STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY =
|
||||
'agent-session.structured.hold.v1' as const
|
||||
// Why: the first structured mobile client hardcoded Codex rendering, so hosts may publish Claude tabs only after this narrower capability is negotiated.
|
||||
export const CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY =
|
||||
'agent-session.structured.claude.v1' as const
|
||||
@@ -150,6 +157,7 @@ export const RUNTIME_CAPABILITIES = [
|
||||
AGENT_SESSION_HOST_AUTHORITY_RUNTIME_CAPABILITY,
|
||||
AGENT_SESSION_OMP_RESUME_PATH_RUNTIME_CAPABILITY,
|
||||
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
|
||||
STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY,
|
||||
CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
|
||||
FILE_MUTATION_OWNERSHIP_RUNTIME_CAPABILITY,
|
||||
WORKTREE_VISIBILITY_DEFAULTS_RUNTIME_CAPABILITY,
|
||||
|
||||
@@ -50,6 +50,8 @@ const STRUCTURED_CALLS: { method: string; hostMethod: string | null }[] = [
|
||||
{ method: 'agentSession.handoffStatus', hostMethod: 'handoffStatus' },
|
||||
{ method: 'agentSession.handoff', hostMethod: 'requestHandoff' },
|
||||
{ method: 'agentSession.options', hostMethod: 'readOptions' },
|
||||
{ method: 'agentSession.hold', hostMethod: 'hold' },
|
||||
{ method: 'agentSession.release', hostMethod: 'release' },
|
||||
{ method: 'agentSession.history', hostMethod: 'history' },
|
||||
{ method: 'agentSession.subscribe', hostMethod: 'subscribe' },
|
||||
// Teardown runs through the runtime's subscription registry rather than the
|
||||
@@ -163,6 +165,9 @@ function paramsFor(method: string): unknown {
|
||||
}
|
||||
case 'agentSession.history':
|
||||
return { sessionId: SESSION, direction: 'tail' }
|
||||
case 'agentSession.hold':
|
||||
case 'agentSession.release':
|
||||
return { sessionId: SESSION, holderId: 'surface-1' }
|
||||
default:
|
||||
return { sessionId: SESSION }
|
||||
}
|
||||
@@ -244,6 +249,8 @@ describe('cross-version structured agent sessions', () => {
|
||||
send: vi.fn(async () => ({ ok: true, replayed: false })),
|
||||
cancel: vi.fn(async () => ({ ok: true, replayed: false })),
|
||||
close: vi.fn(async () => undefined),
|
||||
hold: vi.fn(async () => undefined),
|
||||
release: vi.fn(() => undefined),
|
||||
respondToPrompt: vi.fn(async () => ({ ok: true, replayed: false })),
|
||||
setOption: vi.fn(async () => ({ ok: true, replayed: false })),
|
||||
requestHandoff: vi.fn(async () => ({ status: { owner: 'native' } })),
|
||||
@@ -629,6 +636,9 @@ describe('cross-version structured agent sessions', () => {
|
||||
|
||||
const restarted = await bootHost('b')
|
||||
await restarted.restoreReadableSessions()
|
||||
// Restart restores the session for READING. The chat the client still has open takes its
|
||||
// hold, and that is what gives the session a provider child again.
|
||||
await answer('agentSession.hold', { sessionId: SESSION, holderId: 'surface-1' })
|
||||
const resumedFence = store.getRecord(SESSION)?.lease.runtimeFence ?? 0
|
||||
expect(resumedFence).toBeGreaterThan(created.fence)
|
||||
const second = await answer('agentSession.send', sendParams('after restart', resumedFence))
|
||||
|
||||
Reference in New Issue
Block a user