From 4a39d186da72a496b29a9ec5c523bbbf0a3460d8 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 27 Oct 2026 10:27:58 -0700 Subject: [PATCH] Extract conversation switch foundations without activating them (#34384) Co-authored-by: m4air --- .../daemon-pty-spawn-preparations.test.ts | 31 ++ .../daemon/daemon-pty-spawn-preparations.ts | 69 +++- .../structured-agent-session-host-types.ts | 7 + ...-agent-session-switch-command-selection.ts | 192 +++++++++++ ...tructured-agent-session-switch-commands.ts | 167 ++++++++++ .../agent-session-record-publication.ts | 52 +++ .../runtime/agent-session-record-store.ts | 105 +++--- .../agent-session-store-transactions.ts | 10 +- ...tured-agent-session-read-owner-contract.ts | 1 + ...uctured-agent-session-read-owner-create.ts | 300 ++++++++++++++++++ .../structured-agent-session-read-owner.ts | 296 +---------------- .../agent-session-lease-adjudication.ts | 26 ++ src/shared/agent-session-native-content.ts | 5 + src/shared/agent-session-switch-execution.ts | 126 ++++++++ src/shared/pty-fenced-termination.ts | 6 + ...uctured-agent-session-mutation-envelope.ts | 14 + .../structured-agent-session-params.ts | 139 ++++---- ...tructured-agent-session-question-params.ts | 56 ++++ .../rpc-contract/terminal-unary-params.ts | 26 ++ src/shared/runtime-rpc-envelope.ts | 7 + src/shared/session-tab-placement.ts | 65 ++++ src/shared/terminal-tab-close.ts | 108 +++++++ 22 files changed, 1383 insertions(+), 425 deletions(-) create mode 100644 src/main/daemon/daemon-pty-spawn-preparations.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-switch-command-selection.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-switch-commands.ts create mode 100644 src/main/runtime/agent-session-record-publication.ts create mode 100644 src/renderer/src/components/native-chat/structured-agent-session-read-owner-create.ts create mode 100644 src/shared/agent-session-native-content.ts create mode 100644 src/shared/agent-session-switch-execution.ts create mode 100644 src/shared/rpc-contract/structured-agent-session-mutation-envelope.ts create mode 100644 src/shared/rpc-contract/structured-agent-session-question-params.ts diff --git a/src/main/daemon/daemon-pty-spawn-preparations.test.ts b/src/main/daemon/daemon-pty-spawn-preparations.test.ts new file mode 100644 index 00000000000..8d33bdec6f0 --- /dev/null +++ b/src/main/daemon/daemon-pty-spawn-preparations.test.ts @@ -0,0 +1,31 @@ +import { afterEach, expect, it, vi } from 'vitest' +import { DaemonPtySpawnPreparations } from './daemon-pty-spawn-preparations' + +afterEach(() => vi.restoreAllMocks()) +it('anchors the commit on receiver preparation, so network and queued delay consume authority without timer delivery', () => { + let now = 10 + vi.spyOn(performance, 'now').mockImplementation(() => now) + const preparations = new DaemonPtySpawnPreparations(async () => {}) + const receipt = preparations.prepareSwitch('pane', 'connection', 'prepare', 'click') + now = 20 + const commit = preparations.commitSwitch('pane', 'connection', 'commit', { + invocationId: 'click', + anchor: receipt.anchor, + remainingMs: 100 + }) + expect(preparations.isCanceled(commit)).toBe(false) + // a+r = 110; receipt-relative 120 would incorrectly allow this continuation. + now = 111 + expect(preparations.isCanceled(commit)).toBe(true) + preparations.finish('pane', commit) +}) +it('rejects another connection and a reused anchor while the actual selected preparation remains owned', () => { + const preparations = new DaemonPtySpawnPreparations(async () => {}) + const receipt = preparations.prepareSwitch('pane', 'connection', 'prepare', 'click') + const execution = { invocationId: 'click', anchor: receipt.anchor, remainingMs: 1000 } + expect(() => preparations.commitSwitch('pane', 'different', 'commit', execution)).toThrow() + const selected = preparations.commitSwitch('pane', 'connection', 'commit', execution) + expect(() => preparations.commitSwitch('pane', 'connection', 'duplicate', execution)).toThrow() + expect(preparations.isCanceled(selected)).toBe(false) + preparations.finish('pane', selected) +}) diff --git a/src/main/daemon/daemon-pty-spawn-preparations.ts b/src/main/daemon/daemon-pty-spawn-preparations.ts index febecde25af..88421345e9c 100644 --- a/src/main/daemon/daemon-pty-spawn-preparations.ts +++ b/src/main/daemon/daemon-pty-spawn-preparations.ts @@ -2,6 +2,7 @@ import { TerminalAttachCanceledError } from './types' export type PendingPtySpawnPreparation = { canceled: boolean + switchAnchor?: { token: string; invocationId: string; at: number; deadline: number } // Unlike the polled canceled flag, the signal interrupts in-flight cwd probes on dead shares. readonly controller: AbortController cancelTimer?: ReturnType @@ -52,6 +53,72 @@ export class DaemonPtySpawnPreparations { return preparation } + prepareSwitch(sessionId: string, clientId: string, requestId: string, invocationId: string) { + const preparation = this.register(sessionId, clientId, requestId, undefined) + preparation.switchAnchor = { + token: requestId, + invocationId, + at: performance.now(), + deadline: Infinity + } + preparation.cancelTimer = setTimeout(() => { + this.cancelPreparation(preparation) + this.finish(sessionId, preparation) + }, 720_000) + preparation.cancelTimer.unref?.() + return { anchor: requestId } + } + + commitSwitch( + sessionId: string, + clientId: string, + requestId: string, + execution: { invocationId: string; anchor?: string; remainingMs?: number } + ): PendingPtySpawnPreparation { + const preparation = [...(this.pending.get(sessionId) ?? [])].find( + (entry) => + entry.clientId === clientId && + entry.switchAnchor?.token === execution.anchor && + entry.switchAnchor?.invocationId === execution.invocationId + ) + const anchor = preparation?.switchAnchor + if ( + !preparation || + !anchor || + preparation.canceled || + !Number.isFinite(execution.remainingMs) || + execution.remainingMs! <= 0 || + execution.remainingMs! > 720_000 || + anchor.deadline !== Infinity + ) { + throw new TerminalAttachCanceledError(sessionId) + } + anchor.deadline = anchor.at + execution.remainingMs! + preparation.requestId = requestId + if (performance.now() >= anchor.deadline) { + this.cancelPreparation(preparation) + this.finish(sessionId, preparation) + throw new TerminalAttachCanceledError(sessionId) + } + if (preparation.cancelTimer) { + clearTimeout(preparation.cancelTimer) + } + preparation.cancelTimer = setTimeout( + () => this.cancelPreparation(preparation), + anchor.deadline - performance.now() + ) + preparation.cancelTimer.unref?.() + return preparation + } + + isCanceled(preparation: PendingPtySpawnPreparation): boolean { + return ( + preparation.canceled || + (preparation.switchAnchor !== undefined && + performance.now() >= preparation.switchAnchor.deadline) + ) + } + async prepareUnlessCanceled( sessionId: string, preparation: PendingPtySpawnPreparation @@ -62,7 +129,7 @@ export class DaemonPtySpawnPreparations { ? Promise.race([this.preparePtySpawn(), cancellation.promise]) : this.preparePtySpawn() await preparationTask - if (preparation.canceled) { + if (this.isCanceled(preparation)) { throw new TerminalAttachCanceledError(sessionId) } } 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 0fde3b27a30..63de9765418 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 @@ -47,6 +47,7 @@ import type { StructuredAgentId } from '../../../shared/agent-session-provider-h import type { StructuredAgentRegistry } from './structured-agent-registry' import type { StructuredAgentSessionStartupLimits } from './structured-agent-session-startup-attempt-contract' import type { AgentChatPermissionChoice } from '../../../shared/agent-chat-permission-mode' +import type { AgentSessionSwitchResult } from '../../../shared/agent-session-switch-execution' export type StructuredAgentSessionCaller = { callerKey: string } @@ -116,6 +117,10 @@ export type StructuredAgentSessionEndedChild = StructuredAgentSessionProviderChi /** Conversation state outlives its provider children. */ export type StructuredAgentSessionHostSession = { + switchCommand?: { + invocationId: string + completion: Promise + } /** Enter only through session-map set, which binds delivery. */ readonly journal: AgentSessionJournal params: AgentSessionAttachParams @@ -161,6 +166,8 @@ export type StructuredAgentSessionHostDeps = { hasOpenDispatch?: ( record: Pick ) => boolean + /** Explicit switch admission also includes direct chat assignments, even at rest. */ + hasOwedSwitchWork?: (record: AgentSessionRecord) => boolean /** Advisory tab-hidden notice; failures are logged. */ onSessionTabHidden?: (sessionId: string) => void /** Throws while a terminal may still write the record's head conversation; run before a spawn. */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-switch-command-selection.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-switch-command-selection.ts new file mode 100644 index 00000000000..b2d7e4d8c21 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-switch-command-selection.ts @@ -0,0 +1,192 @@ +import { randomUUID } from 'node:crypto' +import type { StructuredAgentSessionConversationLifetimeHost } from './structured-agent-session-conversation-lifetime-host' +import { + AgentSessionRefusalError, + agentSessionRefusalError +} from '../../../shared/agent-session-wire-refusals' +import { AGENT_SESSION_NOT_ATTACHED } from './structured-agent-session-mutation-admission' +import type { + AgentSessionSwitchDeadline, + AgentSessionSwitchResult, + AgentSessionSwitchReceipt +} from '../../../shared/agent-session-switch-execution' + +// Commands outlive the read handle they may close themselves. +export type SwitchCommand = { + invocationId: string + generation: string + settled: boolean + present: (view: Extract) => void + presentation: Promise + shown: Promise + acknowledge: (shown: boolean) => void + opened: Promise + openedResult: (result: AgentSessionSwitchResult) => void + deadline: AgentSessionSwitchDeadline + revokeOnSend: boolean + retain?: () => Promise + retentionAllowed: boolean + completion: Promise + physical?: Promise + cleanupTimer?: ReturnType +} + +export function selectStructuredSessionSwitchCommand(args: { + host: StructuredAgentSessionConversationLifetimeHost + commands: Map + isDisposed: () => boolean + id: string + invocationId: string + deadline: AgentSessionSwitchDeadline + run: () => Promise + revokeOnSend: boolean + phase?: { observe?: boolean; receipt?: AgentSessionSwitchReceipt } + onRetained?: (check: () => void) => Promise + allowUncreated: boolean +}): SwitchCommand { + const { + host, + commands, + isDisposed, + id, + invocationId, + deadline, + run, + revokeOnSend, + phase, + onRetained, + allowUncreated + } = args + const { sessions, serialize } = host + const deps = () => host.context().deps + deadline.check() + const prior = commands.get(id) + if (prior) { + if (prior.invocationId !== invocationId && !prior.settled) { + throw agentSessionRefusalError('agent_session_operation_invalid', { + reason: 'handoffInFlight' + }) + } + if (prior.invocationId === invocationId) { + return prior + } + prior.retentionAllowed = false + clearTimeout(prior.cleanupTimer) + prior.deadline.revoke() + commands.delete(id) + } + if (phase?.observe || phase?.receipt) { + throw agentSessionRefusalError('agent_session_operation_unknown', { + reason: 'outcomeUnknown' + }) + } + const session = sessions.get(id) + if (!deps().store.getRecord(id) && !allowUncreated) { + throw new AgentSessionRefusalError(AGENT_SESSION_NOT_ATTACHED) + } + let present!: (view: Extract) => void + let acknowledge!: (shown: boolean) => void + let openedResult!: (result: AgentSessionSwitchResult) => void + const presentation = new Promise>( + (resolve) => { + present = resolve + } + ) + const shown = new Promise((resolve) => { + acknowledge = resolve + }) + const opened = new Promise((resolve) => { + openedResult = resolve + }) + const command: SwitchCommand = { + invocationId, + generation: randomUUID(), + settled: false, + deadline, + revokeOnSend, + retentionAllowed: true, + retain: async () => { + const check = () => { + if (!command.retentionAllowed || commands.get(id) !== command) { + throw agentSessionRefusalError('agent_session_operation_expired', { + reason: 'operationExpired' + }) + } + } + check() + await onRetained?.(check) + }, + present, + acknowledge, + presentation, + shown, + opened, + openedResult, + completion: Promise.resolve().then(async () => { + if (deps().store.getRecord(id)) { + await serialize(id, async () => { + deadline.check() + const opened = await host.open(id) + deadline.check() + if (opened) { + opened.switchCommand = command + } + }) + } + try { + const result = await run() + if (revokeOnSend && (result.status === 'refused' || result.status === 'uncertain')) { + await command.retain?.() + } + return result + } catch (error) { + if (revokeOnSend) { + await command.retain?.().catch(() => {}) + } + throw error + } + }) + } + commands.set(id, command) + const retainOnExpiry = () => { + if (revokeOnSend) { + void command.retain?.().catch(() => {}) + } + } + deadline.signal.addEventListener('abort', retainOnExpiry, { once: true }) + if (session) { + session.switchCommand = command + } + host.context().publishStatus?.(id) + void command.completion + .finally(() => { + command.settled = true + deadline.dispose() + for (const [key, current] of commands) { + if (current === command && sessions.get(key)?.switchCommand === command) { + delete sessions.get(key)!.switchCommand + } + } + host.context().publishStatus?.(id) + if (isDisposed()) { + return + } + command.cleanupTimer = setTimeout( + () => { + void Promise.resolve(command.physical).then(() => { + command.retentionAllowed = false + deadline.signal.removeEventListener('abort', retainOnExpiry) + for (const [key, current] of commands) { + if (current === command) { + commands.delete(key) + } + } + }) + }, + Math.max(1, deadline.remaining()) + ) + command.cleanupTimer.unref?.() + }) + .catch(() => {}) + return command +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-switch-commands.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-switch-commands.ts new file mode 100644 index 00000000000..fc02bba47fb --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-switch-commands.ts @@ -0,0 +1,167 @@ +import type { StructuredAgentSessionConversationLifetimeHost } from './structured-agent-session-conversation-lifetime-host' +import { agentSessionRefusalError } from '../../../shared/agent-session-wire-refusals' +import type { + AgentSessionSwitchDeadline, + AgentSessionSwitchResult, + AgentSessionSwitchReceipt +} from '../../../shared/agent-session-switch-execution' +import { + selectStructuredSessionSwitchCommand, + type SwitchCommand +} from './structured-agent-session-switch-command-selection' + +export function createStructuredAgentSessionSwitchCommands( + host: StructuredAgentSessionConversationLifetimeHost, + isDisposed: () => boolean +) { + const { serialize } = host + const deps = () => host.context().deps + const commands = new Map() + const unsubscribeAcceptance = deps().store.onAcceptedSend((id) => { + const command = commands.get(id) + if (!command?.revokeOnSend || command.settled) { + return + } + command.deadline.revoke( + agentSessionRefusalError('agent_session_operation_invalid', { + reason: 'messagesUnsettled' + }) + ) + }) + + return { + controls: { + runSwitch: async ( + id: string, + invocationId: string, + deadline: AgentSessionSwitchDeadline, + run: () => Promise, + revokeOnSend = true, + phase?: { + observe?: boolean + receipt?: AgentSessionSwitchReceipt + }, + onRetained?: (check: () => void) => Promise, + allowUncreated = false, + onConverged?: (generation: string) => void + ): Promise => { + const selected = await deadline.wait( + serialize(id, async () => { + return selectStructuredSessionSwitchCommand({ + host, + commands, + isDisposed, + id, + invocationId, + deadline, + run, + revokeOnSend, + phase, + onRetained, + allowUncreated + }) + }) + ) + if (selected.deadline !== deadline) { + deadline.dispose() + } + if (phase?.receipt) { + if (phase.receipt.generation !== selected.generation) { + throw agentSessionRefusalError('agent_session_operation_invalid', { + reason: 'handoffInFlight' + }) + } + if (phase.receipt.converged) { + onConverged?.(selected.generation) + return { status: 'completed', sessionId: id } + } + if (phase.receipt.retained) { + selected.deadline.revoke( + agentSessionRefusalError('agent_session_operation_invalid', { + reason: 'messagesUnsettled' + }) + ) + await selected.retain?.() + return { status: 'completed', sessionId: id, sourceRetained: true } + } + selected.deadline.check() + selected.acknowledge(phase.receipt.shown) + return Promise.race([selected.completion, selected.opened]) + } + return phase?.observe + ? selected.completion + : Promise.race([selected.completion, selected.presentation]) + }, + adoptSwitch: (from: string, to: string) => + serialize(to, async () => { + const command = commands.get(from) + const prior = commands.get(to) + command?.deadline.check() + if (!command || (prior && prior !== command && !prior.settled)) { + throw agentSessionRefusalError('agent_session_operation_invalid', { + reason: 'handoffInFlight' + }) + } + commands.set(to, command) + }), + switchGeneration: (id: string) => commands.get(id)?.generation, + retainSwitch: async (id: string) => { + await commands.get(id)?.retain?.() + }, + presentSwitch: async ( + id: string, + view: Omit< + Extract, + 'status' | 'generation' + > + ) => { + const command = commands.get(id) + if (!command) { + throw agentSessionRefusalError('agent_session_operation_expired', { + reason: 'operationExpired' + }) + } + command.deadline.check() + command.present({ ...view, status: 'present', generation: command.generation }) + if (!(await command.deadline.wait(command.shown))) { + throw agentSessionRefusalError('agent_session_operation_unknown', { + reason: 'tabUnconfirmed' + }) + } + command.deadline.check() + }, + openedSwitch: (id: string, result: AgentSessionSwitchResult) => + commands.get(id)?.openedResult(result), + switchExecuting: (id: string) => Boolean(commands.get(id) && !commands.get(id)?.settled), + trackSwitchPhysical: (id: string, work: Promise): Promise => { + const command = commands.get(id) + if (command) { + command.physical = work.then( + () => {}, + () => {} + ) + } + return work + }, + waitSwitchPhysical: async (id: string): Promise => { + await commands.get(id)?.physical + } + }, + revoke: (id: string) => { + const command = commands.get(id) + if (command) { + command.retentionAllowed = false + command.deadline.revoke() + } + }, + dispose: () => { + unsubscribeAcceptance() + for (const command of commands.values()) { + command.retentionAllowed = false + clearTimeout(command.cleanupTimer) + command.deadline.revoke() + } + commands.clear() + } + } +} diff --git a/src/main/runtime/agent-session-record-publication.ts b/src/main/runtime/agent-session-record-publication.ts new file mode 100644 index 00000000000..bb532bfbde4 --- /dev/null +++ b/src/main/runtime/agent-session-record-publication.ts @@ -0,0 +1,52 @@ +import type { AgentSessionStoreState } from './agent-session-store-state' +import type { AgentSessionStoreTransactions } from './agent-session-store-transactions' + +export abstract class AgentSessionRecordPublication { + constructor(protected readonly transactions: AgentSessionStoreTransactions) {} + abstract holdsRecords(): boolean + private readonly deathEvidenceListeners = new Set<(sessionId: string) => void>() + + protected readonly firstRecordListeners = new Set<() => void>() + + /** Notify committed death evidence from every transition; listeners must not throw. */ + onDeathEvidence(listener: (sessionId: string) => void): () => void { + this.deathEvidenceListeners.add(listener) + return () => this.deathEvidenceListeners.delete(listener) + } + + /** Notify first committed record; listeners must not throw. */ + onFirstRecord(listener: () => void): () => void { + this.firstRecordListeners.add(listener) + return () => this.firstRecordListeners.delete(listener) + } + + /** Serialize draft mutations; publish to readers only after row commit. */ + protected transact = async ( + apply: (draft: AgentSessionStoreState) => T, + options?: { inMemoryWhenReadOnly?: boolean } + ): Promise => { + let proven: string[] = [] + let heldBefore = true + const result = await this.transactions.transact((draft) => { + heldBefore = this.holdsRecords() + if (this.deathEvidenceListeners.size === 0) { + return apply(draft) + } + const before = new Map( + [...draft.records].map(([id, record]) => [id, record.lease.deathEvidence]) + ) + const applied = apply(draft) + proven = [...draft.records] + .filter(([id, { lease }]) => lease.deathEvidence && lease.deathEvidence !== before.get(id)) + .map(([id]) => id) + return applied + }, options) + for (const sessionId of proven) { + this.deathEvidenceListeners.forEach((listener) => listener(sessionId)) + } + if (!heldBefore && this.holdsRecords()) { + this.firstRecordListeners.forEach((listener) => listener()) + } + return result + } +} diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index df25d64f32a..13bb3da1566 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -1,5 +1,3 @@ -/** Single-writer records/operation ledger in the host chat-journal database. */ - import { backfillAgentSessionPermissionModes } from './agent-session/ownership/agent-session-permission-backfill' import { createAgentSessionConversationReceipts } from './agent-session-conversation-receipts' import { pinAgentSessionRecordLaunchDirectory } from './agent-session-record-launch-directory' @@ -86,19 +84,21 @@ import { setAgentSessionRecordConversationName, setAgentSessionRecordUserNameThrough } from './agent-session-record-conversation-name' +import { AgentSessionRecordPublication } from './agent-session-record-publication' + +/** Single-writer records/operation ledger in the host chat-journal database. */ export const AGENT_SESSION_LEASE_TTL_MS = 30_000, AGENT_SESSION_LEASE_RENEW_INTERVAL_MS = 10_000 -export class AgentSessionRecordStore { - private readonly deathEvidenceListeners = new Set<(sessionId: string) => void>() - private readonly firstRecordListeners = new Set<() => void>() +export class AgentSessionRecordStore extends AgentSessionRecordPublication { readonly conversationReceipts: ReturnType private constructor( - private readonly transactions: AgentSessionStoreTransactions, + transactions: AgentSessionStoreTransactions, readonly hostId: string ) { + super(transactions) this.founding = createAgentSessionFoundingStore(transactions, this.firstRecordListeners) this.conversationReceipts = createAgentSessionConversationReceipts(transactions) } @@ -155,8 +155,36 @@ export class AgentSessionRecordStore { this.state.sessionTabs?.tabIdFor(sessionId) ?? null /** Persist chat-tab identity separately from rollback-sensitive profile tabs; explicit tabId restores hidden tabs under their old ids. */ - setSessionTabVisibility(sessionId: string, visible: boolean, tabId?: string): Promise { - return this.transact((draft) => setAgentSessionTabVisibility(draft, sessionId, visible, tabId)) + setSessionTabVisibility( + sessionId: string, + visible: boolean, + tabId?: string, + check?: () => void + ): Promise { + return this.transact((draft) => { + check?.() + return setAgentSessionTabVisibility(draft, sessionId, visible, tabId) + }) + } + + setSwitchedSessionTabVisibility( + sessionId: string, + tabId: string, + visible: boolean, + admit: () => boolean + ): Promise { + return this.transact((draft) => { + if ( + (visible + ? draft.sessionTabs?.tabIdFor(sessionId) !== undefined + : draft.sessionTabs?.tabIdFor(sessionId) !== tabId) || + !admit() + ) { + return false + } + setAgentSessionTabVisibility(draft, sessionId, visible, tabId) + return true + }) } /** Publish the complete tab index in one write; a partial index would drop remaining chats on restart. */ @@ -199,10 +227,14 @@ export class AgentSessionRecordStore { isClaimKeyVerifiable = (keyId: string, now: number): boolean => isAgentSessionClaimKeyVerifiable(this.state, keyId, now) - reserveOwner = (request: AgentSessionReserveRequest): Promise => - this.transact((draft) => - commitAgentSessionReservation(draft, request, AGENT_SESSION_LEASE_TTL_MS) - ) + reserveOwner = ( + request: AgentSessionReserveRequest, + check?: () => void + ): Promise => + this.transact((draft) => { + check?.() + return commitAgentSessionReservation(draft, request, AGENT_SESSION_LEASE_TTL_MS) + }) readonly founding: ReturnType @@ -302,6 +334,13 @@ export class AgentSessionRecordStore { submissionReceipt = (sessionId: string) => this.transactions.receipt(() => {}, { submissionSessionId: sessionId }) + onAcceptedSend = (listener: (sessionId: string) => void): (() => void) => { + this.transactions.acceptedSendListeners.add(listener) + return () => { + this.transactions.acceptedSendListeners.delete(listener) + } + } + admittedSendCount = (sessionId: string) => this.transactions.admittedSendCount(sessionId) commitOperationReceipt = (receipt: JournalOperationReceipt) => @@ -334,46 +373,4 @@ export class AgentSessionRecordStore { return next }) } - - /** Notify committed death evidence from every transition; listeners must not throw. */ - onDeathEvidence(listener: (sessionId: string) => void): () => void { - this.deathEvidenceListeners.add(listener) - return () => this.deathEvidenceListeners.delete(listener) - } - - /** Notify first committed record; listeners must not throw. */ - onFirstRecord(listener: () => void): () => void { - this.firstRecordListeners.add(listener) - return () => this.firstRecordListeners.delete(listener) - } - - /** Serialize draft mutations; publish to readers only after row commit. */ - private transact = async ( - apply: (draft: AgentSessionStoreState) => T, - options?: { inMemoryWhenReadOnly?: boolean } - ): Promise => { - let proven: string[] = [] - let heldBefore = true - const result = await this.transactions.transact((draft) => { - heldBefore = this.holdsRecords() - if (this.deathEvidenceListeners.size === 0) { - return apply(draft) - } - const before = new Map( - [...draft.records].map(([id, record]) => [id, record.lease.deathEvidence]) - ) - const applied = apply(draft) - proven = [...draft.records] - .filter(([id, { lease }]) => lease.deathEvidence && lease.deathEvidence !== before.get(id)) - .map(([id]) => id) - return applied - }, options) - for (const sessionId of proven) { - this.deathEvidenceListeners.forEach((listener) => listener(sessionId)) - } - if (!heldBefore && this.holdsRecords()) { - this.firstRecordListeners.forEach((listener) => listener()) - } - return result - } } diff --git a/src/main/runtime/agent-session-store-transactions.ts b/src/main/runtime/agent-session-store-transactions.ts index 30508c42e87..d9a3a0d54f8 100644 --- a/src/main/runtime/agent-session-store-transactions.ts +++ b/src/main/runtime/agent-session-store-transactions.ts @@ -1,6 +1,3 @@ -// Commit validated draft rows atomically before publication; failures discard the draft. -// The promise queue preserves FIFO and defers BEGIN beyond the caller's open journal transaction. - import type { JournalHostDatabase } from '../native-chat/agent-session-journal/journal-host-database' import type { JournalOperationReceipt } from '../native-chat/agent-session-journal/journal-row-writer' import { journalOpenRefusalError } from '../native-chat/agent-session-journal/journal-open-failure' @@ -33,6 +30,9 @@ import { type AgentSessionStoreRowWrites } from './agent-session-store-draft' +// Commit validated draft rows atomically before publication; failures discard the draft. +// The promise queue preserves FIFO and defers BEGIN beyond the caller's open journal transaction. + // Identity-based diffs miss in-place edits; freeze rows in dev/tests to catch them. const FREEZE_ROWS = process.env.NODE_ENV === 'development' || process.env.NODE_ENV === 'test' @@ -86,6 +86,7 @@ export class AgentSessionStoreTransactions { private queue: Promise = Promise.resolve() // Reads/reopens do not admit work; a count survives handle eviction without replaying history. private readonly admittedSends = new Map() + readonly acceptedSendListeners = new Set<(sessionId: string) => void>() private published: AgentSessionStoreState /** Loaded takeover evidence; immutable for this runtime. */ readonly replacedRuntimes: ReadonlyMap @@ -179,6 +180,9 @@ export class AgentSessionStoreTransactions { staged?.adopt() if (admitted) { this.admittedSends.set(admitted, this.admittedSendCount(admitted) + 1) + for (const listener of this.acceptedSendListeners) { + listener(admitted) + } } staged = null admitted = undefined diff --git a/src/renderer/src/components/native-chat/structured-agent-session-read-owner-contract.ts b/src/renderer/src/components/native-chat/structured-agent-session-read-owner-contract.ts index ffc0dc3d8f9..350b67a8627 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-read-owner-contract.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-read-owner-contract.ts @@ -16,6 +16,7 @@ export type StructuredAgentSessionReadSnapshot = { export type StructuredAgentSessionReadOwner = { publishCreated: (page: AgentSessionHistoryPage) => void activate: () => () => void + resync?: () => void dispose: () => void getSnapshot: () => StructuredAgentSessionReadSnapshot loadOlder: () => Promise diff --git a/src/renderer/src/components/native-chat/structured-agent-session-read-owner-create.ts b/src/renderer/src/components/native-chat/structured-agent-session-read-owner-create.ts new file mode 100644 index 00000000000..be37d9ca458 --- /dev/null +++ b/src/renderer/src/components/native-chat/structured-agent-session-read-owner-create.ts @@ -0,0 +1,300 @@ +import type { AgentJournalRenderItem } from '../../../../shared/agent-session-journal-types' +import { + type AgentProviderSessionMetadata, + agentProviderSessionsEqual +} from '../../../../shared/agent-session-resume' +import { + AGENT_SESSION_HISTORY_MAX_LIMIT, + type AgentSessionHistoryResult +} from '../../../../shared/agent-session-wire' +import { + EMPTY_STRUCTURED_AGENT_SESSION, + oldestStructuredAgentSessionCursor, + reduceStructuredAgentSession, + type StructuredAgentSessionAction +} from '../../../../shared/structured-agent-session-reducer' +import type { RuntimeClientTarget } from '../../../../shared/runtime-client-target' +import { callStructuredAgentSession } from '@/runtime/structured-agent-session-client' +import { NATIVE_CHAT_INITIAL_LIMIT, type NativeChatOlderPageResult } from './native-chat-pagination' +import { + OLDER_PAGE_ANCHOR_ATTEMPTS, + readStructuredAgentSessionOlderPage +} from './structured-agent-session-older-page-read' +import { startStructuredAgentSessionReadTransport } from './structured-agent-session-read-transport' +import { createStructuredAgentSessionPRLinkObserver } from './structured-agent-session-pr-link-observer' +import type { + StructuredAgentSessionReadSnapshot, + StructuredAgentSessionReadOwner +} from './structured-agent-session-read-owner-contract' +import { + adoptStructuredReadOwner, + forgetStructuredReadOwner +} from './structured-agent-session-read-owner-registry' + +export function countsTowardInitialHistory(item: AgentJournalRenderItem): boolean { + return item.body.kind !== 'status' || !item.body.providerFrame +} + +export function createReadOwner( + key: string, + sessionId: string, + target: RuntimeClientTarget +): StructuredAgentSessionReadOwner { + let snapshot: StructuredAgentSessionReadSnapshot = { + state: EMPTY_STRUCTURED_AGENT_SESSION, + loadingOlder: false, + olderHistoryGeneration: 0, + current: false + } + let stopActiveRun: (() => void) | null = null + const retiredHistoryRead = (): boolean => true + let captureActiveHistoryReadGuard = (): (() => boolean) => retiredHistoryRead + const activations = new Set() + const listeners = new Set<() => void>() + const observePRLinks = createStructuredAgentSessionPRLinkObserver(sessionId, target) + + const emit = (): void => { + for (const listener of listeners) { + listener() + } + } + const setSnapshot = (next: StructuredAgentSessionReadSnapshot): void => { + if (next === snapshot) { + return + } + snapshot = next + emit() + } + const apply = (action: StructuredAgentSessionAction): void => { + const state = reduceStructuredAgentSession(snapshot.state, action, Date.now()) + // Published even when the frame changed nothing: a caught-up resume is an empty batch. + const current = + snapshot.current || + (action.type === 'event' && + action.event.type !== 'end' && + !(action.event.type === 'batch' && action.event.hasNewer)) + if (state !== snapshot.state || current !== snapshot.current) { + setSnapshot({ ...snapshot, state, current }) + } + } + const markStale = (): void => { + if (snapshot.current) { + setSnapshot({ ...snapshot, current: false }) + } + } + const setProviderSession = (providerSession: AgentProviderSessionMetadata | undefined): void => { + if (!agentProviderSessionsEqual(undefined, snapshot.providerSession, providerSession)) { + setSnapshot({ ...snapshot, providerSession }) + } + } + const clearLoadingOlder = (): void => { + if (snapshot.loadingOlder) { + setSnapshot({ ...snapshot, loadingOlder: false }) + } + } + const invalidateOlderPages = (): void => { + setSnapshot({ + ...snapshot, + loadingOlder: false, + olderHistoryGeneration: snapshot.olderHistoryGeneration + 1 + }) + } + const hydrate = async (shouldStop: () => boolean): Promise => { + const result = await callStructuredAgentSession( + target, + 'agentSession.history', + { sessionId, direction: 'tail', limit: AGENT_SESSION_HISTORY_MAX_LIMIT } + ) + if (shouldStop()) { + return + } + setProviderSession(result.providerSession) + if (!result.ok) { + if (shouldStop()) { + return + } + apply({ + type: 'event', + event: { + type: 'reset', + sessionId, + reset: result.reset, + page: result.page, + fence: result.fence ?? 0 + } + }) + return + } + if (shouldStop()) { + return + } + apply({ type: 'history-page', page: result.page }) + if (shouldStop()) { + return + } + let restored = snapshot.state.items.filter(countsTowardInitialHistory).length + let anchorSlides = 0 + while (snapshot.state.hasOlder && restored < NATIVE_CHAT_INITIAL_LIMIT) { + const oldest = oldestStructuredAgentSessionCursor(snapshot.state) + if (!oldest || shouldStop()) { + break + } + const missing = NATIVE_CHAT_INITIAL_LIMIT - restored + const older = await callStructuredAgentSession( + target, + 'agentSession.history', + { + sessionId, + direction: 'before', + cursor: oldest, + limit: Math.min(AGENT_SESSION_HISTORY_MAX_LIMIT, missing) + } + ) + if (shouldStop()) { + return + } + if (!older.ok || older.page.window.oldest?.sequence === oldest.sequence) { + break + } + if (shouldStop()) { + return + } + // Bound re-anchoring when head trimming makes older pages discontiguous. + if (oldestStructuredAgentSessionCursor(snapshot.state)?.sequence !== oldest.sequence) { + anchorSlides += 1 + if (anchorSlides >= OLDER_PAGE_ANCHOR_ATTEMPTS) { + break + } + continue + } + apply({ type: 'older-page', requestedCursor: oldest, page: older.page }) + if (shouldStop()) { + return + } + restored = snapshot.state.items.filter(countsTowardInitialHistory).length + } + } + + let olderPage: { + shouldStop: () => boolean + promise: Promise + } | null = null + + const start = (): void => { + if (snapshot.state.epoch === null) { + apply({ type: 'loading' }) + } + const transport = startStructuredAgentSessionReadTransport({ + applyRequests: apply, + applyEvent: (event, options) => { + apply({ type: 'event', event, ...options }) + observePRLinks(event) + }, + applyError: (message, refusal) => { + // After loading, transport failures retain history while retrying; host refusals are explicit answers to show. + if (snapshot.state.epoch !== null && refusal === undefined) { + return + } + apply({ type: 'error', message, refusal }) + }, + getCursor: () => snapshot.state.cursor, + onContactLost: markStale, + onHistoryReadInvalidated: invalidateOlderPages, + hydrate: snapshot.state.epoch === null ? hydrate : undefined, + sessionId, + target + }) + captureActiveHistoryReadGuard = transport.captureHistoryReadGuard + stopActiveRun = () => { + captureActiveHistoryReadGuard = () => retiredHistoryRead + transport.dispose() + stopActiveRun = null + markStale() + } + } + + let owner: StructuredAgentSessionReadOwner + const deleteIfUnused = (): void => { + if (activations.size === 0 && listeners.size === 0) { + forgetStructuredReadOwner(key, owner) + } + } + owner = { + publishCreated: (page) => { + if (snapshot.state.epoch === null) { + apply({ type: 'history-page', page }) + } + }, + activate: () => { + // Remounts and StrictMode re-run setup on an owner whose cleanup just forgot it. + adoptStructuredReadOwner(key, owner) + const token = Symbol(sessionId) + activations.add(token) + if (activations.size === 1) { + start() + } + return () => { + activations.delete(token) + if (activations.size === 0) { + stopActiveRun?.() + deleteIfUnused() + } + } + }, + resync: () => { + if (activations.size === 0) { + return + } + stopActiveRun?.() + start() + }, + dispose: () => { + activations.clear() + listeners.clear() + stopActiveRun?.() + }, + getSnapshot: () => snapshot, + loadOlder: () => { + // Share concurrent page requests so another caller’s refusal cannot look like exhaustion. + if (olderPage && !olderPage.shouldStop()) { + return olderPage.promise + } + const shouldStop = captureActiveHistoryReadGuard() + if (shouldStop()) { + return Promise.resolve('superseded') + } + if (!oldestStructuredAgentSessionCursor(snapshot.state) || !snapshot.state.hasOlder) { + return Promise.resolve('exhausted') + } + setSnapshot({ ...snapshot, loadingOlder: true }) + const promise = readStructuredAgentSessionOlderPage({ + target, + sessionId, + getState: () => snapshot.state, + apply, + shouldStop + }) + const page = { shouldStop, promise } + olderPage = page + void page.promise.finally(() => { + if (olderPage !== page) { + return + } + olderPage = null + if (!shouldStop()) { + clearLoadingOlder() + } + }) + return page.promise + }, + subscribe: (listener) => { + adoptStructuredReadOwner(key, owner) + listeners.add(listener) + return () => { + listeners.delete(listener) + deleteIfUnused() + } + } + } + return owner +} diff --git a/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts b/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts index 68e145c4e6e..f11bb24bf62 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts @@ -1,308 +1,22 @@ -import type { AgentJournalRenderItem } from '../../../../shared/agent-session-journal-types' -import type { AgentProviderSessionMetadata } from '../../../../shared/agent-session-resume' -import { agentProviderSessionsEqual } from '../../../../shared/agent-session-resume' -import { - AGENT_SESSION_HISTORY_MAX_LIMIT, - type AgentSessionHistoryResult -} from '../../../../shared/agent-session-wire' -import { - EMPTY_STRUCTURED_AGENT_SESSION, - oldestStructuredAgentSessionCursor, - reduceStructuredAgentSession, - type StructuredAgentSessionAction -} from '../../../../shared/structured-agent-session-reducer' import type { RuntimeClientTarget } from '../../../../shared/runtime-client-target' -import { callStructuredAgentSession } from '@/runtime/structured-agent-session-client' -import { NATIVE_CHAT_INITIAL_LIMIT, type NativeChatOlderPageResult } from './native-chat-pagination' +import type { StructuredAgentSessionReadOwner } from './structured-agent-session-read-owner-contract' import { - OLDER_PAGE_ANCHOR_ATTEMPTS, - readStructuredAgentSessionOlderPage -} from './structured-agent-session-older-page-read' -import { startStructuredAgentSessionReadTransport } from './structured-agent-session-read-transport' -import { createStructuredAgentSessionPRLinkObserver } from './structured-agent-session-pr-link-observer' + getOrCreateStructuredReadOwner, + structuredReadOwnerKey +} from './structured-agent-session-read-owner-registry' +import { createReadOwner } from './structured-agent-session-read-owner-create' -import type { - StructuredAgentSessionReadSnapshot, - StructuredAgentSessionReadOwner -} from './structured-agent-session-read-owner-contract' export type { StructuredAgentSessionReadSnapshot, StructuredAgentSessionReadOwner } from './structured-agent-session-read-owner-contract' -import { - adoptStructuredReadOwner, - getOrCreateStructuredReadOwner, - forgetStructuredReadOwner, - structuredReadOwnerKey -} from './structured-agent-session-read-owner-registry' export { findStructuredAgentSessionReadOwner, retireStructuredAgentSessionReadOwner, resetStructuredAgentSessionReadOwnersForTests } from './structured-agent-session-read-owner-registry' -function countsTowardInitialHistory(item: AgentJournalRenderItem): boolean { - return item.body.kind !== 'status' || !item.body.providerFrame -} - -function createReadOwner( - key: string, - sessionId: string, - target: RuntimeClientTarget -): StructuredAgentSessionReadOwner { - let snapshot: StructuredAgentSessionReadSnapshot = { - state: EMPTY_STRUCTURED_AGENT_SESSION, - loadingOlder: false, - olderHistoryGeneration: 0, - current: false - } - let stopActiveRun: (() => void) | null = null - const retiredHistoryRead = (): boolean => true - let captureActiveHistoryReadGuard = (): (() => boolean) => retiredHistoryRead - const activations = new Set() - const listeners = new Set<() => void>() - const observePRLinks = createStructuredAgentSessionPRLinkObserver(sessionId, target) - - const emit = (): void => { - for (const listener of listeners) { - listener() - } - } - const setSnapshot = (next: StructuredAgentSessionReadSnapshot): void => { - if (next === snapshot) { - return - } - snapshot = next - emit() - } - const apply = (action: StructuredAgentSessionAction): void => { - const state = reduceStructuredAgentSession(snapshot.state, action, Date.now()) - // Published even when the frame changed nothing: a caught-up resume is an empty batch. - const current = - snapshot.current || - (action.type === 'event' && - action.event.type !== 'end' && - !(action.event.type === 'batch' && action.event.hasNewer)) - if (state !== snapshot.state || current !== snapshot.current) { - setSnapshot({ ...snapshot, state, current }) - } - } - const markStale = (): void => { - if (snapshot.current) { - setSnapshot({ ...snapshot, current: false }) - } - } - const setProviderSession = (providerSession: AgentProviderSessionMetadata | undefined): void => { - if (!agentProviderSessionsEqual(undefined, snapshot.providerSession, providerSession)) { - setSnapshot({ ...snapshot, providerSession }) - } - } - const clearLoadingOlder = (): void => { - if (snapshot.loadingOlder) { - setSnapshot({ ...snapshot, loadingOlder: false }) - } - } - const invalidateOlderPages = (): void => { - setSnapshot({ - ...snapshot, - loadingOlder: false, - olderHistoryGeneration: snapshot.olderHistoryGeneration + 1 - }) - } - const hydrate = async (shouldStop: () => boolean): Promise => { - const result = await callStructuredAgentSession( - target, - 'agentSession.history', - { sessionId, direction: 'tail', limit: AGENT_SESSION_HISTORY_MAX_LIMIT } - ) - if (shouldStop()) { - return - } - setProviderSession(result.providerSession) - if (!result.ok) { - if (shouldStop()) { - return - } - apply({ - type: 'event', - event: { - type: 'reset', - sessionId, - reset: result.reset, - page: result.page, - fence: result.fence ?? 0 - } - }) - return - } - if (shouldStop()) { - return - } - apply({ type: 'history-page', page: result.page }) - if (shouldStop()) { - return - } - let restored = snapshot.state.items.filter(countsTowardInitialHistory).length - let anchorSlides = 0 - while (snapshot.state.hasOlder && restored < NATIVE_CHAT_INITIAL_LIMIT) { - const oldest = oldestStructuredAgentSessionCursor(snapshot.state) - if (!oldest || shouldStop()) { - break - } - const missing = NATIVE_CHAT_INITIAL_LIMIT - restored - const older = await callStructuredAgentSession( - target, - 'agentSession.history', - { - sessionId, - direction: 'before', - cursor: oldest, - limit: Math.min(AGENT_SESSION_HISTORY_MAX_LIMIT, missing) - } - ) - if (shouldStop()) { - return - } - if (!older.ok || older.page.window.oldest?.sequence === oldest.sequence) { - break - } - if (shouldStop()) { - return - } - // Bound re-anchoring when head trimming makes older pages discontiguous. - if (oldestStructuredAgentSessionCursor(snapshot.state)?.sequence !== oldest.sequence) { - anchorSlides += 1 - if (anchorSlides >= OLDER_PAGE_ANCHOR_ATTEMPTS) { - break - } - continue - } - apply({ type: 'older-page', requestedCursor: oldest, page: older.page }) - if (shouldStop()) { - return - } - restored = snapshot.state.items.filter(countsTowardInitialHistory).length - } - } - - let olderPage: { - shouldStop: () => boolean - promise: Promise - } | null = null - - const start = (): void => { - if (snapshot.state.epoch === null) { - apply({ type: 'loading' }) - } - const transport = startStructuredAgentSessionReadTransport({ - applyRequests: apply, - applyEvent: (event, options) => { - apply({ type: 'event', event, ...options }) - observePRLinks(event) - }, - applyError: (message, refusal) => { - // After loading, transport failures retain history while retrying; host refusals are explicit answers to show. - if (snapshot.state.epoch !== null && refusal === undefined) { - return - } - apply({ type: 'error', message, refusal }) - }, - getCursor: () => snapshot.state.cursor, - onContactLost: markStale, - onHistoryReadInvalidated: invalidateOlderPages, - hydrate: snapshot.state.epoch === null ? hydrate : undefined, - sessionId, - target - }) - captureActiveHistoryReadGuard = transport.captureHistoryReadGuard - stopActiveRun = () => { - captureActiveHistoryReadGuard = () => retiredHistoryRead - transport.dispose() - stopActiveRun = null - markStale() - } - } - - let owner: StructuredAgentSessionReadOwner - const deleteIfUnused = (): void => { - if (activations.size === 0 && listeners.size === 0) { - forgetStructuredReadOwner(key, owner) - } - } - owner = { - publishCreated: (page) => { - if (snapshot.state.epoch === null) { - apply({ type: 'history-page', page }) - } - }, - activate: () => { - // Remounts and StrictMode re-run setup on an owner whose cleanup just forgot it. - adoptStructuredReadOwner(key, owner) - const token = Symbol(sessionId) - activations.add(token) - if (activations.size === 1) { - start() - } - return () => { - activations.delete(token) - if (activations.size === 0) { - stopActiveRun?.() - deleteIfUnused() - } - } - }, - dispose: () => { - activations.clear() - listeners.clear() - stopActiveRun?.() - }, - getSnapshot: () => snapshot, - loadOlder: () => { - // Share concurrent page requests so another caller’s refusal cannot look like exhaustion. - if (olderPage && !olderPage.shouldStop()) { - return olderPage.promise - } - const shouldStop = captureActiveHistoryReadGuard() - if (shouldStop()) { - return Promise.resolve('superseded') - } - if (!oldestStructuredAgentSessionCursor(snapshot.state) || !snapshot.state.hasOlder) { - return Promise.resolve('exhausted') - } - setSnapshot({ ...snapshot, loadingOlder: true }) - const promise = readStructuredAgentSessionOlderPage({ - target, - sessionId, - getState: () => snapshot.state, - apply, - shouldStop - }) - const page = { shouldStop, promise } - olderPage = page - void page.promise.finally(() => { - if (olderPage !== page) { - return - } - olderPage = null - if (!shouldStop()) { - clearLoadingOlder() - } - }) - return page.promise - }, - subscribe: (listener) => { - adoptStructuredReadOwner(key, owner) - listeners.add(listener) - return () => { - listeners.delete(listener) - deleteIfUnused() - } - } - } - return owner -} - export function getStructuredAgentSessionReadOwner( sessionId: string, target: RuntimeClientTarget diff --git a/src/shared/agent-session-lease-adjudication.ts b/src/shared/agent-session-lease-adjudication.ts index 1470dcd8e5a..747b6918894 100644 --- a/src/shared/agent-session-lease-adjudication.ts +++ b/src/shared/agent-session-lease-adjudication.ts @@ -125,6 +125,32 @@ export function agentSessionOwnerObservationVerdict( : 'unverifiable' } +/** Founding is not an observed death; every reservation/recovery leaves a different tuple. */ +export function agentSessionLeaseNeverAcquired(lease: AgentSessionLease): boolean { + return ( + lease.runtimeKind === 'native' && + lease.runtimeFence === 1 && + lease.minimumNextFence === undefined && + !lease.unreconciled && + lease.claimStatus === 'released' && + lease.ownerProcess === null && + lease.reservedSpawnToken === null && + lease.handoffStage === null && + lease.handoffOperationId === null && + lease.provenHandleLinkId === null && + lease.deathEvidence === null + ) +} + +export function agentSessionLeasePossiblyWriting( + lease: AgentSessionLease | AgentSessionOwnerObservation +): boolean { + return ( + !('runtimeKind' in lease && agentSessionLeaseNeverAcquired(lease)) && + agentSessionOwnerObservationVerdict(lease) !== 'exited' + ) +} + /** Nothing holds this lease: released, no handoff in flight, and reconciled since the last restart. */ export function agentSessionLeaseIsReleased(lease: AgentSessionLease): boolean { return !lease.unreconciled && lease.claimStatus === 'released' && lease.handoffStage === null diff --git a/src/shared/agent-session-native-content.ts b/src/shared/agent-session-native-content.ts new file mode 100644 index 00000000000..a52a49118a5 --- /dev/null +++ b/src/shared/agent-session-native-content.ts @@ -0,0 +1,5 @@ +import type { AgentJournalItemBody } from './agent-session-journal-types' + +export function isNativeConversationMessage(body: AgentJournalItemBody): boolean { + return body.kind === 'message' && (body.role === 'user' || body.role === 'assistant') +} diff --git a/src/shared/agent-session-switch-execution.ts b/src/shared/agent-session-switch-execution.ts new file mode 100644 index 00000000000..a92dac631d5 --- /dev/null +++ b/src/shared/agent-session-switch-execution.ts @@ -0,0 +1,126 @@ +import type { AgentSessionWireRefusal } from './agent-session-wire' +import { agentSessionRefusalError } from './agent-session-wire-refusals' +import { waitForPromiseWithSignal } from './abort-signal-reason' + +export const AGENT_SESSION_SWITCH_HOST_MS = 720_000 +export const AGENT_SESSION_SWITCH_CLIENT_MS = 735_000 + +export type AgentSessionSwitchResult = + | { + status: 'completed' + sessionId: string + terminal?: string + sourceRetained?: boolean + startupPending?: boolean + } + | { + status: 'present' + surface?: 'terminal' + leafId?: string + sessionId: string + generation: string + workspaceId: string + tabId: string + agent: 'codex' + placement?: { groupId?: string; afterTabId?: string } + } + | { + status: 'refused' | 'uncertain' + refusal: AgentSessionWireRefusal + sessionId: string + failureItemId?: string + } + +export type AgentSessionSwitchReceipt = { + generation: string + shown: boolean + retained?: boolean + converged?: boolean +} + +/** Expiry revokes new effects; already dispatched owners still reconcile their real outcome. */ +export function agentSessionSwitchDeadline( + now = () => performance.now(), + budget = AGENT_SESSION_SWITCH_HOST_MS +) { + let expiresAt = now() + budget + let dispatched = false + let delegated = false + const controller = new AbortController() + let timer = setTimeout(() => controller.abort(expired()), budget) + function expired() { + return agentSessionRefusalError('agent_session_operation_expired', { + reason: 'operationExpired' + }) + } + return { + signal: controller.signal, + delegateSend: () => { + if (controller.signal.aborted) { + throw expired() + } + const dispatchExpiresAt = expiresAt + if (!dispatched && !delegated) { + if (now() >= expiresAt) { + throw expired() + } + delegated = true + clearTimeout(timer) + // Main now owns unsent expiry; allow its actual-send receipt to cross IPC. + timer = setTimeout(() => controller.abort(expired()), expiresAt - now() + budget) + } + return { + expiresAt: dispatchExpiresAt, + responseTimeoutMs: budget, + ...(dispatched ? { responseExpiresAt: expiresAt } : {}) + } + }, + beforeSend: (sentAt = now()) => { + if (controller.signal.aborted || (!dispatched && sentAt >= expiresAt)) { + throw expired() + } + if (dispatched) { + return + } + dispatched = true + expiresAt = sentAt + budget + clearTimeout(timer) + timer = setTimeout(() => controller.abort(expired()), Math.max(0, expiresAt - now())) + if (now() >= expiresAt) { + controller.abort(expired()) + } + }, + revoke: (reason: unknown = expired()) => controller.abort(reason), + remaining: () => Math.max(0, expiresAt - now()), + check: () => { + if (controller.signal.aborted) { + throw controller.signal.reason + } + if (now() >= expiresAt) { + controller.abort(expired()) + throw controller.signal.reason + } + }, + wait: (promise: Promise) => waitForPromiseWithSignal(promise, controller.signal), + dispose: () => clearTimeout(timer) + } +} +export type AgentSessionSwitchDeadline = ReturnType + +/** In-process authority; never serialize the callbacks as a transport permit. */ +export type AgentSessionSwitchAuthority = { + invocationId: string + check: () => void + remaining: () => number + signal: AbortSignal + /** Existing physical acquisition only, excluding placement and presentation. */ + trackPhysical?: (work: Promise) => Promise +} + +export type SessionSwitchProjection = { + invocationId: string + generation: string + sessionId: string + tabId: string + phase: 'prepare' | 'retire' | 'retain' +} diff --git a/src/shared/pty-fenced-termination.ts b/src/shared/pty-fenced-termination.ts index 4d277958b7d..9037e537494 100644 --- a/src/shared/pty-fenced-termination.ts +++ b/src/shared/pty-fenced-termination.ts @@ -3,6 +3,12 @@ export type PtyFencedTerminationRequest = { expectedIncarnationId: string requireNoAgentOwner: boolean requireUnattached: boolean + switchCondition?: { + token: string + remainingMs: number + ownedSpawn?: true + claimGeneration?: string + } } export type PtyFencedTerminationOutcome = diff --git a/src/shared/rpc-contract/structured-agent-session-mutation-envelope.ts b/src/shared/rpc-contract/structured-agent-session-mutation-envelope.ts new file mode 100644 index 00000000000..412b0e62377 --- /dev/null +++ b/src/shared/rpc-contract/structured-agent-session-mutation-envelope.ts @@ -0,0 +1,14 @@ +import { Identifier, SessionId } from './structured-agent-session-identifiers' +import { z } from 'zod' + +export const MutationEnvelope = z + .object({ + sessionId: SessionId, + clientOperationId: Identifier('Invalid client operation id'), + /** Null is the "must not exist yet" case; every other call fences. */ + expectedRuntimeFence: z.number().int().positive().nullable(), + payloadFingerprint: z + .string() + .regex(/^[0-9a-f]{64}$/, 'Payload fingerprint must be a sha256 hex digest') + }) + .strict() diff --git a/src/shared/rpc-contract/structured-agent-session-params.ts b/src/shared/rpc-contract/structured-agent-session-params.ts index f385509b481..546fe88172f 100644 --- a/src/shared/rpc-contract/structured-agent-session-params.ts +++ b/src/shared/rpc-contract/structured-agent-session-params.ts @@ -1,63 +1,49 @@ import { SendBody, - AgentSessionOptions, MAX_OPTION_LABEL, - MAX_PROMPT_BYTES -} from './structured-agent-session-message-params' -export { - SendBlock, - MAX_BLOCKS, - MAX_OPTION_LABEL, - MAX_PROMPT_BYTES + AgentSessionOptions } from './structured-agent-session-message-params' +import { TerminalClose } from './terminal-unary-params' import { Identifier, JournalCursor, MAX_ID_LENGTH, SessionId } from './structured-agent-session-identifiers' -export { - Identifier, - JournalCursor, - MAX_ID_LENGTH, - SessionId -} from './structured-agent-session-identifiers' import { z } from 'zod' import { CREATE_TAB_FIELDS } from './structured-agent-session-create-tab-params' import { isStructuredAgentId } from '../agent-session-provider-handle-encoding' import { normalizeExecutionHostId } from '../execution-host' -import { - AGENT_SESSION_QUESTION_ANSWER_MAX_BYTES, - AGENT_SESSION_RESPONSE_OPTION_ID_MAX_LENGTH -} from '../agent-session-question-answer' +import { AGENT_SESSION_RESPONSE_OPTION_ID_MAX_LENGTH } from '../agent-session-question-answer' import { AGENT_SESSION_HISTORY_DIRECTIONS, AGENT_SESSION_HISTORY_MAX_LIMIT } from '../agent-session-wire' import { AGENT_SESSION_THREAD_GOAL_OBJECTIVE_MAX_LENGTH } from '../agent-session-thread-goal-wire' +import { MutationEnvelope } from './structured-agent-session-mutation-envelope' + +export { + SendBlock, + MAX_BLOCKS, + MAX_OPTION_LABEL, + MAX_PROMPT_BYTES +} from './structured-agent-session-message-params' + +export { + Identifier, + JournalCursor, + MAX_ID_LENGTH, + SessionId +} from './structured-agent-session-identifiers' // Four Claude questions with all four generated choices occupy 610 chars when fully percent-encoded. export const MAX_RESPONSE_OPTION_ID_LENGTH = AGENT_SESSION_RESPONSE_OPTION_ID_MAX_LENGTH /** Matches the journal's bounds on one grouped prompt. */ -const MAX_QUESTION_ANSWER_QUESTIONS = 4 -const MAX_QUESTION_ANSWER_OPTIONS = 64 /** One relaunch cannot offer more chats than a profile plausibly holds. */ export const MAX_RESTART_RESUME_SESSIONS = 512 -export const MutationEnvelope = z - .object({ - sessionId: SessionId, - clientOperationId: Identifier('Invalid client operation id'), - /** Null is the "must not exist yet" case; every other call fences. */ - expectedRuntimeFence: z.number().int().positive().nullable(), - payloadFingerprint: z - .string() - .regex(/^[0-9a-f]{64}$/, 'Payload fingerprint must be a sha256 hex digest') - }) - .strict() - export const ProviderHandle = z.discriminatedUnion('kind', [ z.object({ kind: z.literal('codex'), threadId: Identifier('Invalid thread id') }).strict(), z @@ -216,50 +202,6 @@ export const RespondParams = z }) .strict() -const QuestionAnswer = z - .object({ - // Codex question ids are model-written and untrimmed; the host matches them exactly. - questionId: z - .string() - .min(1, 'Invalid question id') - .max(MAX_RESPONSE_OPTION_ID_LENGTH, 'Invalid question id'), - optionIds: z - .array(Identifier('Invalid option id', MAX_RESPONSE_OPTION_ID_LENGTH)) - .max(MAX_QUESTION_ANSWER_OPTIONS), - // Hashed verbatim by both peers, so no trim or transform here. - other: z - .string() - .max(AGENT_SESSION_QUESTION_ANSWER_MAX_BYTES) - .refine( - (value) => Buffer.byteLength(value, 'utf8') <= AGENT_SESSION_QUESTION_ANSWER_MAX_BYTES, - 'Answer is too large' - ) - .optional() - }) - .strict() - -export const RespondToQuestionParams = z - .object({ - envelope: MutationEnvelope, - itemId: Identifier('Invalid item id'), - expectedRevision: z.number().int().positive(), - /** An answer packed into one id, from clients that predate `answers`. */ - optionId: Identifier('Invalid option id', MAX_RESPONSE_OPTION_ID_LENGTH).optional(), - answers: z.array(QuestionAnswer).min(1).max(MAX_QUESTION_ANSWER_QUESTIONS).optional() - }) - .strict() - .superRefine((value, ctx) => { - if ((value.optionId === undefined) === (value.answers === undefined)) { - ctx.addIssue({ code: 'custom', message: 'Send exactly one of an option id or answers' }) - } - if ( - value.answers !== undefined && - Buffer.byteLength(JSON.stringify(value.answers), 'utf8') > MAX_PROMPT_BYTES - ) { - ctx.addIssue({ code: 'custom', message: 'Answer is too large' }) - } - }) - export const SetOptionParams = z.strictObject({ envelope: MutationEnvelope, key: Identifier('Invalid option key'), @@ -267,6 +209,47 @@ export const SetOptionParams = z.strictObject({ }) export const OptionsParams = z.object({ sessionId: SessionId }).strict() +export const SwitchToChatParams = OptionsParams.extend({ + execution: z + .object({ + invocationId: Identifier('Invalid invocation'), + observe: z.boolean().optional(), + receipt: z + .object({ + generation: Identifier('Invalid generation'), + shown: z.boolean(), + retained: z.boolean().optional(), + converged: z.boolean().optional() + }) + .strict() + .optional(), + create: CreateIntentParams.omit({ firstMessage: true }).optional(), + source: TerminalClose.required().optional() + }) + .strict() + .optional() +}).strict() +export const SwitchToTerminalParams = OptionsParams.extend({ + execution: z + .object({ + invocationId: Identifier('Invalid invocation'), + observe: z.boolean().optional(), + receipt: z + .object({ + generation: Identifier('Invalid generation'), + shown: z.boolean(), + retained: z.boolean().optional(), + converged: z.boolean().optional() + }) + .strict() + .optional(), + sourceTabId: Identifier('Invalid tab').optional(), + groupId: Identifier('Invalid group').optional(), + commandOverride: z.string().max(8192).optional() + }) + .strict() + .optional() +}).strict() export const ItemParams = OptionsParams.extend({ itemId: z.string().min(1) }).strict() // Why optional: only a host advertising the workspace-sleep-mark capability is sent `markAsleep`. export const SleepWorkspaceParams = z @@ -365,3 +348,7 @@ export const RewindParams = z expectedEpoch: Identifier('Invalid journal epoch') }) .strict() + +export { RespondToQuestionParams } from './structured-agent-session-question-params' + +export { MutationEnvelope } from './structured-agent-session-mutation-envelope' diff --git a/src/shared/rpc-contract/structured-agent-session-question-params.ts b/src/shared/rpc-contract/structured-agent-session-question-params.ts new file mode 100644 index 00000000000..8dfa95e5fd9 --- /dev/null +++ b/src/shared/rpc-contract/structured-agent-session-question-params.ts @@ -0,0 +1,56 @@ +import { MAX_PROMPT_BYTES } from './structured-agent-session-message-params' +import { Identifier } from './structured-agent-session-identifiers' +import { z } from 'zod' +import { + AGENT_SESSION_QUESTION_ANSWER_MAX_BYTES, + AGENT_SESSION_RESPONSE_OPTION_ID_MAX_LENGTH as MAX_RESPONSE_OPTION_ID_LENGTH +} from '../agent-session-question-answer' +import { MutationEnvelope } from './structured-agent-session-mutation-envelope' + +const MAX_QUESTION_ANSWER_QUESTIONS = 4 + +const MAX_QUESTION_ANSWER_OPTIONS = 64 + +const QuestionAnswer = z + .object({ + // Codex question ids are model-written and untrimmed; the host matches them exactly. + questionId: z + .string() + .min(1, 'Invalid question id') + .max(MAX_RESPONSE_OPTION_ID_LENGTH, 'Invalid question id'), + optionIds: z + .array(Identifier('Invalid option id', MAX_RESPONSE_OPTION_ID_LENGTH)) + .max(MAX_QUESTION_ANSWER_OPTIONS), + // Hashed verbatim by both peers, so no trim or transform here. + other: z + .string() + .max(AGENT_SESSION_QUESTION_ANSWER_MAX_BYTES) + .refine( + (value) => Buffer.byteLength(value, 'utf8') <= AGENT_SESSION_QUESTION_ANSWER_MAX_BYTES, + 'Answer is too large' + ) + .optional() + }) + .strict() + +export const RespondToQuestionParams = z + .object({ + envelope: MutationEnvelope, + itemId: Identifier('Invalid item id'), + expectedRevision: z.number().int().positive(), + /** An answer packed into one id, from clients that predate `answers`. */ + optionId: Identifier('Invalid option id', MAX_RESPONSE_OPTION_ID_LENGTH).optional(), + answers: z.array(QuestionAnswer).min(1).max(MAX_QUESTION_ANSWER_QUESTIONS).optional() + }) + .strict() + .superRefine((value, ctx) => { + if ((value.optionId === undefined) === (value.answers === undefined)) { + ctx.addIssue({ code: 'custom', message: 'Send exactly one of an option id or answers' }) + } + if ( + value.answers !== undefined && + Buffer.byteLength(JSON.stringify(value.answers), 'utf8') > MAX_PROMPT_BYTES + ) { + ctx.addIssue({ code: 'custom', message: 'Answer is too large' }) + } + }) diff --git a/src/shared/rpc-contract/terminal-unary-params.ts b/src/shared/rpc-contract/terminal-unary-params.ts index 279e337eaff..6be4912c509 100644 --- a/src/shared/rpc-contract/terminal-unary-params.ts +++ b/src/shared/rpc-contract/terminal-unary-params.ts @@ -20,12 +20,38 @@ export const WindowsShellOverrideParam = z // Canonicalize host-side to the .exe spellings exact-matched by spawn, even if the client did not. .transform((shell) => canonicalizeWindowsShellOverride(shell) ?? shell) +const TerminalSurfaceCloseIdentity = z.object({ + worktreeId: requiredString('Missing worktree id'), + tabId: requiredString('Missing tab id'), + leafId: requiredString('Missing leaf id'), + createdAt: z.number().finite().nullable(), + generation: z.number().int().nonnegative(), + hasLeaf: z.boolean(), + ptyId: z.string().nullable(), + incarnationId: z.string().nullable() +}) +const TerminalSurfaceCloseCapture = z.object({ + identity: TerminalSurfaceCloseIdentity, + holders: z + .array( + z.object({ + hostId: OptionalExecutionHostId.unwrap(), + identity: TerminalSurfaceCloseIdentity.nullable() + }) + ) + .max(64) +}) + export const TerminalHandle = z.object({ terminal: requiredString('Missing terminal handle'), // Additive fence understood by newer hosts; legacy hosts safely ignore it. expectedIncarnationId: requiredString('Missing PTY incarnation').optional() }) +export const TerminalClose = TerminalHandle.extend({ + originalPresentation: TerminalSurfaceCloseCapture.optional() +}) + export const TerminalReadKittyFile = TerminalHandle.extend({ path: z .string() diff --git a/src/shared/runtime-rpc-envelope.ts b/src/shared/runtime-rpc-envelope.ts index 203e2cdae5e..a702748fe29 100644 --- a/src/shared/runtime-rpc-envelope.ts +++ b/src/shared/runtime-rpc-envelope.ts @@ -98,3 +98,10 @@ export function isKeepaliveFrame(frame: unknown): frame is RuntimeRpcKeepaliveFr (frame as { _keepalive: unknown })._keepalive === true ) } + +/** Same-machine IPC observation only; remote hosts own their own monotonic operation budget. */ +export type RuntimeRpcDispatchObservation = { + expiresAt: number + responseTimeoutMs: number + responseExpiresAt?: number +} diff --git a/src/shared/session-tab-placement.ts b/src/shared/session-tab-placement.ts index 1b4729a6156..0727f809675 100644 --- a/src/shared/session-tab-placement.ts +++ b/src/shared/session-tab-placement.ts @@ -1,5 +1,58 @@ +import { insertTabIdIntoOrder, partitionPinnedTabOrder } from './workspace-layout/tab-order' + +/** Call-local neighbors survive an anchor retiring while creation is awaited. */ +export type CapturedSessionTabPlacement = { preceding: string[]; following: string[] } + +export function captureSessionTabPlacement( + order: readonly string[], + afterTabId: string | null | undefined +): CapturedSessionTabPlacement | undefined { + const index = afterTabId ? order.indexOf(afterTabId) : -1 + return index < 0 + ? undefined + : { + preceding: order.slice(0, index + 1), + following: order.slice(index + 1) + } +} + +/** Apply only insertion to current order, keeping unrelated concurrent edits. */ +export function placeSessionTabAtCapture( + order: readonly string[], + createdId: string, + captured: CapturedSessionTabPlacement +): string[] { + const next = order.filter((id) => id !== createdId) + const present = new Set(next) + const previous = captured.preceding.toReversed().find((id) => present.has(id)) + const following = captured.following.find((id) => present.has(id)) + const index = + following !== undefined + ? next.indexOf(following) + : previous !== undefined + ? next.indexOf(previous) + 1 + : Math.min(captured.preceding.length, next.length) + next.splice(index, 0, createdId) + return next +} + +/** Captured placement uses the same pin partitions as an ordinary tab insertion. */ +export function placeUnpinnedSessionTabAtCapture( + order: readonly string[], + tabs: readonly { id: string; isPinned?: boolean }[], + createdId: string, + captured: CapturedSessionTabPlacement +): string[] { + const placed = placeSessionTabAtCapture(order, createdId, captured) + const preceding = placed[placed.indexOf(createdId) - 1] + return preceding + ? insertTabIdIntoOrder(placed, tabs, createdId, false, preceding) + : partitionPinnedTabOrder(placed, tabs, createdId) +} + export type SessionTabPlacementOptions = { afterParentGroup?: boolean + captured?: CapturedSessionTabPlacement } /** Places a created tab after the anchor, or after its parent group when enabled. */ @@ -12,6 +65,18 @@ export function placeCreatedSessionTab tab.id !== created.id) const anchor = afterTabId ? next.findIndex((tab) => tab.id === afterTabId) : -1 if (anchor < 0) { + if (options.captured) { + const order = placeSessionTabAtCapture( + next.map((tab) => tab.id), + created.id, + options.captured + ) + const byId = new Map([...next, created].map((tab) => [tab.id, tab])) + return order.flatMap((id) => { + const tab = byId.get(id) + return tab ? [tab] : [] + }) + } next.push(created) return next } diff --git a/src/shared/terminal-tab-close.ts b/src/shared/terminal-tab-close.ts index 75a77fbfc2b..61f33b35ffe 100644 --- a/src/shared/terminal-tab-close.ts +++ b/src/shared/terminal-tab-close.ts @@ -1,11 +1,119 @@ +import type { ExecutionHostId } from './execution-host' +import type { WorkspaceSessionState } from './workspace-session-state-types' +import type { TerminalPaneLayoutNode } from './terminal-tab-types' + export type TerminalTabCloseRequest = { requestId: string tabId: string localPtyTeardownOwnedExternally?: boolean force?: boolean + originalSurface?: TerminalSurfaceCloseIdentity } export type TerminalTabCloseResponse = { requestId: string error?: string } + +/** Captured presentation facts only narrow the authenticated terminal close. */ +export type TerminalSurfaceCloseIdentity = { + worktreeId: string + tabId: string + leafId: string + createdAt: number | null + generation: number + hasLeaf: boolean + ptyId: string | null + incarnationId: string | null +} + +export type TerminalSurfaceCloseCapture = { + identity: TerminalSurfaceCloseIdentity + holders: { hostId: ExecutionHostId; identity: TerminalSurfaceCloseIdentity | null }[] +} + +export function isRetainedTerminalSurfaceIdentity( + original: TerminalSurfaceCloseIdentity, + current: TerminalSurfaceCloseIdentity | null +): boolean { + return ( + current === null || + (current.worktreeId === original.worktreeId && + current.tabId === original.tabId && + current.leafId === original.leafId && + current.createdAt === original.createdAt && + current.generation === original.generation && + (!current.hasLeaf || original.hasLeaf) && + (current.ptyId === null || current.ptyId === original.ptyId) && + (current.incarnationId === null || current.incarnationId === original.incarnationId)) + ) +} + +export function terminalTreeHasLeaf( + node: TerminalPaneLayoutNode | null | undefined, + leafId: string +): boolean { + return ( + !!node && + (node.type === 'leaf' + ? node.leafId === leafId + : terminalTreeHasLeaf(node.first, leafId) || terminalTreeHasLeaf(node.second, leafId)) + ) +} + +export function captureTerminalSurfaceIdentity( + state: + | (Pick & + Partial>) + | null + | undefined, + worktreeId: string, + tabId: string, + leafId: string +): TerminalSurfaceCloseIdentity | null { + const row = state?.tabsByWorktree[worktreeId]?.find((tab) => tab.id === tabId) + if (!row || !state) { + return null + } + const layout = state.terminalLayoutsByTabId[tabId] + return { + worktreeId, + tabId, + leafId, + createdAt: row.createdAt, + generation: row.generation ?? 0, + hasLeaf: terminalTreeHasLeaf(layout?.root, leafId), + ptyId: layout?.ptyIdsByLeafId?.[leafId] ?? null, + incarnationId: state.terminalPtyIncarnationsByPaneKey?.[`${tabId}:${leafId}`] ?? null + } +} + +export function matchesTerminalSurfaceCapture( + original: TerminalSurfaceCloseCapture, + hostIds: readonly ExecutionHostId[], + read: (hostId: ExecutionHostId) => Parameters[0] +): boolean { + return ( + original.identity.hasLeaf && + original.holders.length > 0 && + original.holders.every((holder) => + isRetainedTerminalSurfaceIdentity(original.identity, holder.identity) + ) && + [...new Set([...original.holders.map((holder) => holder.hostId), ...hostIds])].every( + (hostId) => { + const holder = original.holders.find((entry) => entry.hostId === hostId) + const next = captureTerminalSurfaceIdentity( + read(hostId), + original.identity.worktreeId, + original.identity.tabId, + original.identity.leafId + ) + return holder + ? holder.identity + ? isRetainedTerminalSurfaceIdentity(holder.identity, next) + : next === null + : next === null + } + ) + ) +}