From 1b8301e6e359ba0ce96ec09654b5d67bc9b6dbfe Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 19 Aug 2026 14:22:12 -0700 Subject: [PATCH] feat(native-chat): give a structured session a surface-holder lifetime MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../mobile-structured-session-hold.test.ts | 63 +++++ .../session/mobile-structured-session-hold.ts | 37 +++ .../use-mobile-structured-agent-session.ts | 3 + .../mobile-e2ee-v2-physical-channel.test.ts | 2 +- .../mobile-e2ee-v2-physical-channel.ts | 10 +- .../rpc-client-structured-session.test.ts | 2 +- mobile/src/transport/rpc-client.ts | 10 +- ...ured-agent-session-attach-orchestration.ts | 7 +- ...uctured-agent-session-eviction-deadline.ts | 51 ++++ .../structured-agent-session-eviction.test.ts | 13 +- .../structured-agent-session-eviction.ts | 8 + ...tructured-agent-session-handoff-restart.ts | 12 + .../structured-agent-session-hold-resume.ts | 34 +++ .../structured-agent-session-holders.ts | 45 ++++ .../structured-agent-session-holds.test.ts | 213 +++++++++++++++ .../structured-agent-session-holds.ts | 91 +++++++ .../structured-agent-session-host-lifetime.ts | 86 ++++++ .../structured-agent-session-host-types.ts | 6 + .../structured-agent-session-host.test.ts | 5 +- .../structured-agent-session-host.ts | 71 +++-- .../structured-agent-session-lease-release.ts | 32 +++ ...tured-agent-session-legacy-restart.test.ts | 11 +- .../structured-agent-session-read-restore.ts | 4 +- ...uctured-agent-session-readable-restorer.ts | 8 +- ...tured-agent-session-recovery-exits.test.ts | 13 +- .../structured-agent-session-release-clock.ts | 75 ++++++ ...tructured-agent-session-restart-restore.ts | 33 +-- ...ctured-agent-session-resume-eligibility.ts | 41 +++ ...red-agent-session-surface-lifetime.test.ts | 246 ++++++++++++++++++ .../structured-agent-session-tui-adoption.ts | 8 +- .../structured-tui-transcript-catchup.test.ts | 7 +- ...gent-session-surface-release-transition.ts | 60 +++++ .../methods/structured-agent-session-gate.ts | 58 +++++ .../structured-agent-session-hold.test.ts | 224 ++++++++++++++++ .../methods/structured-agent-session-hold.ts | 59 +++++ .../structured-agent-session-schemas.ts | 7 + .../methods/structured-agent-session.test.ts | 2 +- .../rpc/methods/structured-agent-session.ts | 86 +++--- src/main/runtime/runtime-rpc.ts | 2 + ...use-structured-agent-session-hold.test.tsx | 100 +++++++ .../use-structured-agent-session-hold.ts | 55 ++++ .../use-structured-agent-session.test.tsx | 10 +- .../use-structured-agent-session.ts | 4 + src/shared/protocol-version.ts | 8 + ...ss-version-agent-session-wire.unit.test.ts | 10 + 45 files changed, 1810 insertions(+), 122 deletions(-) create mode 100644 mobile/src/session/mobile-structured-session-hold.test.ts create mode 100644 mobile/src/session/mobile-structured-session-hold.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-eviction-deadline.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-hold-resume.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-holders.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-holds.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-holds.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-lease-release.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-release-clock.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-resume-eligibility.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts create mode 100644 src/main/runtime/agent-session-surface-release-transition.ts create mode 100644 src/main/runtime/rpc/methods/structured-agent-session-gate.ts create mode 100644 src/main/runtime/rpc/methods/structured-agent-session-hold.test.ts create mode 100644 src/main/runtime/rpc/methods/structured-agent-session-hold.ts create mode 100644 src/renderer/src/components/native-chat/use-structured-agent-session-hold.test.tsx create mode 100644 src/renderer/src/components/native-chat/use-structured-agent-session-hold.ts diff --git a/mobile/src/session/mobile-structured-session-hold.test.ts b/mobile/src/session/mobile-structured-session-hold.test.ts new file mode 100644 index 00000000000..0a4c5971706 --- /dev/null +++ b/mobile/src/session/mobile-structured-session-hold.test.ts @@ -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): 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() + }) +}) diff --git a/mobile/src/session/mobile-structured-session-hold.ts b/mobile/src/session/mobile-structured-session-hold.ts new file mode 100644 index 00000000000..8dd165352f3 --- /dev/null +++ b/mobile/src/session/mobile-structured-session-hold.ts @@ -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]) +} diff --git a/mobile/src/session/use-mobile-structured-agent-session.ts b/mobile/src/session/use-mobile-structured-agent-session.ts index cf8fe08522c..cc84c32433c 100644 --- a/mobile/src/session/use-mobile-structured-agent-session.ts +++ b/mobile/src/session/use-mobile-structured-agent-session.ts @@ -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 } { 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 diff --git a/mobile/src/transport/mobile-e2ee-v2-physical-channel.test.ts b/mobile/src/transport/mobile-e2ee-v2-physical-channel.test.ts index a5c3db33122..8f02f8372d5 100644 --- a/mobile/src/transport/mobile-e2ee-v2-physical-channel.test.ts +++ b/mobile/src/transport/mobile-e2ee-v2-physical-channel.test.ts @@ -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() diff --git a/mobile/src/transport/mobile-e2ee-v2-physical-channel.ts b/mobile/src/transport/mobile-e2ee-v2-physical-channel.ts index 40e0395cba0..ce8f980bbff 100644 --- a/mobile/src/transport/mobile-e2ee-v2-physical-channel.ts +++ b/mobile/src/transport/mobile-e2ee-v2-physical-channel.ts @@ -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() diff --git a/mobile/src/transport/rpc-client-structured-session.test.ts b/mobile/src/transport/rpc-client-structured-session.test.ts index 857f50f4265..c4741a6ae6f 100644 --- a/mobile/src/transport/rpc-client-structured-session.test.ts +++ b/mobile/src/transport/rpc-client-structured-session.test.ts @@ -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() }) diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index 451d11fc99e..be8efd6da24 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -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 } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts index 08368fb6aad..0653e7abd6a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts @@ -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) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction-deadline.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction-deadline.ts new file mode 100644 index 00000000000..f34caba3593 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction-deadline.ts @@ -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 | undefined + try { + await Promise.race([ + Promise.resolve(step.run(context)), + new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new StructuredAgentSessionEvictionTimeoutError(step.name, timeoutMs)), + timeoutMs + ) + timer.unref?.() + }) + ]) + } finally { + if (timer) { + clearTimeout(timer) + } + } + } + })) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts index 849611e9fea..3abe8299e06 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts @@ -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[] = [] diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts index 93953b6835b..da2d856bec9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts @@ -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 } 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() } ] diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts index 68ec81fa5ea..7d448ca755c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts @@ -99,6 +99,18 @@ export async function restoreStructuredAgentSessionHandoff( } async function restoreOnce(input: RestartAccess, record: AgentSessionRecord): Promise { + 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 diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-hold-resume.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-hold-resume.ts new file mode 100644 index 00000000000..630558d40fe --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-hold-resume.ts @@ -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 { + 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) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-holders.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-holders.ts new file mode 100644 index 00000000000..a1bf2e30b5f --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-holders.ts @@ -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>() + + /** 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) + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-holds.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-holds.test.ts new file mode 100644 index 00000000000..c6c9e579773 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-holds.test.ts @@ -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 + 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(() => {}) } 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(() => {}) }], + 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']) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-holds.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-holds.ts new file mode 100644 index 00000000000..1e7ec8e09af --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-holds.ts @@ -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 + /** Whether evicting this session would actually free anything. */ + hasProviderChild: (sessionId: string) => boolean + isTurnActive: (sessionId: string) => boolean + evict: (sessionId: string) => Promise + 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 { + 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() + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts new file mode 100644 index 00000000000..6f37f677e6f --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts @@ -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 + 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 { + 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 + evict: (sessionId: string) => Promise + } +): 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 }) + }) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts index 4c90e5d647a..70c7a8734fc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts @@ -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 | undefined> | Record | 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 } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts index 549b1702315..75eed906a3d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts @@ -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({ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 65468968d52..13ebaefaaa7 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -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 => 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 { + 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 { 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 { + this.holds.dispose() this.runtimeState.stopLeaseRenewal() this.handoffs.stopTuiHistoryCatchup() await this.tasks.drainAttaches() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-lease-release.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-lease-release.ts new file mode 100644 index 00000000000..63d29754f61 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-lease-release.ts @@ -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 { + 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 + }) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-legacy-restart.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-legacy-restart.test.ts index 3bacb960b13..13edd1f9028 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-legacy-restart.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-legacy-restart.test.ts @@ -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 (_sessionId: string, task: () => Promise) => 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) }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.ts index 7599fe134ce..e8b06535329 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.ts @@ -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( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts index e6e934024dd..1fa42a39c17 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts @@ -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 resolveRecovery: (sessionId: string) => Promise - resume: (params: AgentSessionAttachParams) => Promise serialize: (sessionId: string, task: () => Promise) => Promise hasSession: (sessionId: string) => boolean onReadable: (sessionId: string, restored: RestoredStructuredAgentSessionRead) => void restoreHandoff: (sessionId: string) => Promise - now: () => number } ) {} restore(): Promise { 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 diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-exits.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-exits.test.ts index 80c0d22db21..40a06ac9091 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-exits.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-exits.test.ts @@ -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, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-release-clock.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-release-clock.ts new file mode 100644 index 00000000000..452e6a6fae0 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-release-clock.ts @@ -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 + onError?: (input: { sessionId: string; error: unknown }) => void + graceMs?: number +} + +export class StructuredAgentSessionReleaseClock { + private readonly timers = new Map>() + 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 }) + }) + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts index 721fe187cc4..7149552ead7 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts @@ -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 resolveRecovery: (sessionId: string) => Promise - operationId: () => string - resume: (params: AgentSessionAttachParams) => Promise serialize: (sessionId: string, task: () => Promise) => Promise 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 } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-resume-eligibility.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-resume-eligibility.ts new file mode 100644 index 00000000000..0baf6fe9952 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-resume-eligibility.ts @@ -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('-', '')}` +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts new file mode 100644 index 00000000000..2cdf5e24ccb --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts @@ -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 +let closeSession: Mock> +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): 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 { + 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 { + 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 { + 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 { + 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') + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-tui-adoption.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-tui-adoption.ts index d9b4cc1a410..a450d13dc9e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-tui-adoption.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-tui-adoption.ts @@ -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) diff --git a/src/main/native-chat/agent-session-wire/structured-tui-transcript-catchup.test.ts b/src/main/native-chat/agent-session-wire/structured-tui-transcript-catchup.test.ts index 8969ed0d68c..b94d671fa9a 100644 --- a/src/main/native-chat/agent-session-wire/structured-tui-transcript-catchup.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-tui-transcript-catchup.test.ts @@ -99,7 +99,12 @@ async function createCatchupFixture() { function createCatchup(input: Awaited>) { 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() diff --git a/src/main/runtime/agent-session-surface-release-transition.ts b/src/main/runtime/agent-session-surface-release-transition.ts new file mode 100644 index 00000000000..d4da43f1933 --- /dev/null +++ b/src/main/runtime/agent-session-surface-release-transition.ts @@ -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 { + return store.transitionHandoff(args.sessionId, (record) => + releaseAgentSessionOwnerAfterSurfaceClose({ ...args, record }) + ) +} diff --git a/src/main/runtime/rpc/methods/structured-agent-session-gate.ts b/src/main/runtime/rpc/methods/structured-agent-session-gate.ts new file mode 100644 index 00000000000..2dd317e08b0 --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-gate.ts @@ -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 { + // 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'}` + } +} diff --git a/src/main/runtime/rpc/methods/structured-agent-session-hold.test.ts b/src/main/runtime/rpc/methods/structured-agent-session-hold.test.ts new file mode 100644 index 00000000000..932d17fd192 --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-hold.test.ts @@ -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> +let requests = 0 + +async function call(method: string, params: unknown): Promise { + 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 { + 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 { + 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') + }) +}) diff --git a/src/main/runtime/rpc/methods/structured-agent-session-hold.ts b/src/main/runtime/rpc/methods/structured-agent-session-hold.ts new file mode 100644 index 00000000000..9718145a2ad --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-hold.ts @@ -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 } + } + }) +] diff --git a/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts b/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts index 41f42e8805a..0010c52fcd1 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts @@ -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, diff --git a/src/main/runtime/rpc/methods/structured-agent-session.test.ts b/src/main/runtime/rpc/methods/structured-agent-session.test.ts index c4a86a87a14..b1fffa8d31c 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.test.ts @@ -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 () => { diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index b2e96cfb3ad..4ab47bb6752 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -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 { - // 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 ] diff --git a/src/main/runtime/runtime-rpc.ts b/src/main/runtime/runtime-rpc.ts index 286d6cd0194..81f749c98e6 100644 --- a/src/main/runtime/runtime-rpc.ts +++ b/src/main/runtime/runtime-rpc.ts @@ -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', diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-hold.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-hold.test.tsx new file mode 100644 index 00000000000..b915958a266 --- /dev/null +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-hold.test.tsx @@ -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((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() + }) +}) diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-hold.ts b/src/renderer/src/components/native-chat/use-structured-agent-session-hold.ts new file mode 100644 index 00000000000..13fa5b1adf5 --- /dev/null +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-hold.ts @@ -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]) +} diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session.test.tsx index 3090fc638c7..55a70cb2968 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session.test.tsx @@ -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( diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session.ts b/src/renderer/src/components/native-chat/use-structured-agent-session.ts index 94301dc0489..395880ca5c0 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session.ts +++ b/src/renderer/src/components/native-chat/use-structured-agent-session.ts @@ -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(null) diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index a5deb5f4458..4da22088b43 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -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, diff --git a/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts b/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts index 63b396aee8d..7be0c5c734d 100644 --- a/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts +++ b/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts @@ -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))