diff --git a/src/main/daemon/daemon-health.test.ts b/src/main/daemon/daemon-health.test.ts index ccfb7d1f0e9..8d266caae9b 100644 --- a/src/main/daemon/daemon-health.test.ts +++ b/src/main/daemon/daemon-health.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' import { tmpdir } from 'node:os' import { basename, join } from 'node:path' import { createServer, connect, type Server } from 'node:net' @@ -429,12 +429,14 @@ describe('killStaleDaemon pid identity guards', () => { // expected and not a real kill. We only care that no actual termination // signal is sent. const killSpy = vi.spyOn(process, 'kill').mockImplementation(() => true) + const pidPath = getDaemonPidPath(dir) try { await expect(killStaleDaemon(dir, socketPath, tokenPath)).resolves.toBe(false) const terminationSignals = killSpy.mock.calls.filter( ([, sig]) => sig === 'SIGTERM' || sig === 'SIGKILL' ) expect(terminationSignals).toEqual([]) + expect(readFileSync(pidPath, 'utf8')).toContain(`"pid":${process.pid}`) } finally { killSpy.mockRestore() } diff --git a/src/main/daemon/daemon-health.ts b/src/main/daemon/daemon-health.ts index 71bc7f3279b..06e940134a1 100644 --- a/src/main/daemon/daemon-health.ts +++ b/src/main/daemon/daemon-health.ts @@ -665,12 +665,14 @@ export async function killStaleDaemon( ): Promise { const pidPath = getDaemonPidPath(runtimeDir, protocolVersion) let killedDaemon = false + let identifiedDaemon = false try { const parsedPid = parseDaemonPidFile(readFileSync(pidPath, 'utf8')) if ( parsedPid && (await isDaemonProcess(parsedPid.pid, socketPath, tokenPath, parsedPid.startedAtMs)) ) { + identifiedDaemon = true const { pid, startedAtMs } = parsedPid process.kill(pid, 'SIGTERM') const deadline = Date.now() + KILL_WAIT_MS @@ -708,10 +710,12 @@ export async function killStaleDaemon( // PID file missing or process already dead } - try { - unlinkSync(pidPath) - } catch { - // Best-effort + if (identifiedDaemon) { + try { + unlinkSync(pidPath) + } catch { + // Best-effort + } } const socketIsLive = await canConnectSocket(socketPath) diff --git a/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts b/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts index e4855640ada..19e82a5faef 100644 --- a/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts +++ b/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts @@ -10,7 +10,7 @@ export class DaemonPtyAdapterSubscriptionFanout { constructor( private readonly adapters: readonly DaemonPtyAdapter[], - onAdapterExit: (id: string) => void + onAdapterExit: (adapter: DaemonPtyAdapter, id: string) => void ) { for (const adapter of adapters) { this.unsubscribers.push( @@ -20,7 +20,7 @@ export class DaemonPtyAdapterSubscriptionFanout { } }), adapter.onExit((payload) => { - onAdapterExit(payload.id) + onAdapterExit(adapter, payload.id) for (const listener of this.exitListeners) { listener(payload) } diff --git a/src/main/daemon/daemon-pty-router-routing-safety.test.ts b/src/main/daemon/daemon-pty-router-routing-safety.test.ts new file mode 100644 index 00000000000..acab8477c26 --- /dev/null +++ b/src/main/daemon/daemon-pty-router-routing-safety.test.ts @@ -0,0 +1,266 @@ +import { describe, expect, it, vi } from 'vitest' +import type { DaemonEndpointIdentity } from './daemon-hello-protocol' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { DaemonPtyRouter } from './daemon-pty-router' +import type { PtyProcessInfo, PtySpawnOptions } from '../providers/types' + +type AdapterHarness = { + adapter: DaemonPtyAdapter + emitIdentityChange: (previous: DaemonEndpointIdentity, current: DaemonEndpointIdentity) => void + setIdentity: (identity: DaemonEndpointIdentity) => void +} + +function identity(label: string, pid: number): DaemonEndpointIdentity { + return { + pid, + startedAtMs: pid * 1_000, + launchNonce: `${label}-${pid}` + } +} + +function createAdapter( + label: string, + sessions: string[] = [], + reconcileResult: { alive: string[]; killed: string[] } = { alive: [], killed: [] } +): AdapterHarness { + let daemonIdentity = identity(label, label === 'current' ? 20 : 10) + const identityListeners: ((event: { + previous: DaemonEndpointIdentity + current: DaemonEndpointIdentity + }) => void)[] = [] + const adapter = { + protocolVersion: label === 'current' ? 30 : 29, + getLastAuthenticatedDaemonIdentity: vi.fn(() => ({ ...daemonIdentity })), + onDaemonIdentityChanged: vi.fn( + ( + listener: (event: { + previous: DaemonEndpointIdentity + current: DaemonEndpointIdentity + }) => void + ) => { + identityListeners.push(listener) + return () => {} + } + ), + listProcesses: vi.fn( + async (): Promise => sessions.map((id) => ({ id, cwd: '', title: label })) + ), + spawn: vi.fn(async (opts: PtySpawnOptions) => ({ + id: opts.sessionId ?? `${label}-fresh` + })), + probePtyLiveness: vi.fn(async (id: string) => sessions.includes(id)), + hasPty: vi.fn((id: string) => sessions.includes(id)), + shutdown: vi.fn(async () => {}), + sendSignal: vi.fn(async () => {}), + getBufferSnapshot: vi.fn(async () => null), + ackColdRestore: vi.fn(), + reconcileOnStartup: vi.fn(async () => reconcileResult), + onData: vi.fn(() => () => {}), + onExit: vi.fn(() => () => {}), + onBackgroundStreamEvent: vi.fn(() => () => {}), + onWriteUnavailable: vi.fn(() => () => {}), + dispose: vi.fn(), + disconnectOnly: vi.fn(async () => {}) + } as unknown as DaemonPtyAdapter + + return { + adapter, + setIdentity: (next) => { + daemonIdentity = next + }, + emitIdentityChange: (previous, current) => { + for (const listener of identityListeners) { + listener({ previous, current }) + } + } + } +} + +describe('DaemonPtyRouter routing safety', () => { + it('fences unknown existing ids until a later complete legacy listing', async () => { + const current = createAdapter('current') + const legacy = createAdapter('legacy', ['legacy-session']) + vi.mocked(legacy.adapter.listProcesses).mockRejectedValueOnce(new Error('listing failed')) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.discoverLegacySessions() + await expect( + router.spawn({ + sessionId: 'fresh-session', + isNewSession: true, + cols: 80, + rows: 24 + }) + ).resolves.toMatchObject({ id: 'fresh-session' }) + await expect(router.spawn({ sessionId: 'legacy-session', cols: 80, rows: 24 })).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + await expect(router.shutdown('legacy-session', { immediate: true })).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + await expect(router.sendSignal('legacy-session', 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + + await router.discoverLegacySessions() + await router.spawn({ sessionId: 'legacy-session', cols: 80, rows: 24 }) + + expect(current.adapter.spawn).toHaveBeenCalledExactlyOnceWith({ + sessionId: 'fresh-session', + isNewSession: true, + cols: 80, + rows: 24 + }) + expect(legacy.adapter.spawn).toHaveBeenCalledExactlyOnceWith({ + sessionId: 'legacy-session', + cols: 80, + rows: 24 + }) + warn.mockRestore() + }) + + it('fails existing operations unknown when an owner probe is inconclusive', async () => { + const current = createAdapter('current') + const legacy = createAdapter('legacy') + vi.mocked(legacy.adapter.probePtyLiveness).mockResolvedValue(null) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await expect( + router.spawn({ sessionId: 'uncertain-session', cols: 80, rows: 24 }) + ).rejects.toThrow('daemon_session_routing_unavailable') + await expect(router.shutdown('uncertain-session', { immediate: true })).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + await expect(router.sendSignal('uncertain-session', 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + + expect(current.adapter.spawn).not.toHaveBeenCalled() + expect(current.adapter.shutdown).not.toHaveBeenCalled() + expect(current.adapter.sendSignal).not.toHaveBeenCalled() + }) + + it('probes every adapter once and memoizes the unique owner', async () => { + const current = createAdapter('current') + const legacy = createAdapter('legacy', ['surviving-session']) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.sendSignal('surviving-session', 'SIGINT') + await router.shutdown('surviving-session', { immediate: true }) + + expect(current.adapter.probePtyLiveness).toHaveBeenCalledExactlyOnceWith('surviving-session') + expect(legacy.adapter.probePtyLiveness).toHaveBeenCalledExactlyOnceWith('surviving-session') + expect(legacy.adapter.sendSignal).toHaveBeenCalledWith('surviving-session', 'SIGINT') + expect(legacy.adapter.shutdown).toHaveBeenCalledWith('surviving-session', { + immediate: true + }) + }) + + it('keeps cross-generation id collisions ambiguous and away from current', async () => { + const sessionId = 'colliding-session' + const current = createAdapter('current', [sessionId], { + alive: [sessionId], + killed: [] + }) + const legacy = createAdapter('legacy', [sessionId], { + alive: [sessionId], + killed: [] + }) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.reconcileOnStartup(new Set()) + expect(router.getSessionRouteState(sessionId)).toBe('ambiguous') + await expect(router.spawn({ sessionId, cols: 80, rows: 24 })).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + await expect(router.shutdown(sessionId, { immediate: true })).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + + expect(current.adapter.spawn).not.toHaveBeenCalled() + expect(current.adapter.shutdown).not.toHaveBeenCalled() + }) + + it('scopes unavailable tombstones to the authenticated daemon incarnation', async () => { + const sessionId = 'legacy-session' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + const previous = identity('legacy', 10) + const replacement = identity('legacy', 11) + + await router.discoverLegacySessions() + legacy.setIdentity(replacement) + legacy.emitIdentityChange(previous, replacement) + + expect(router.getSessionRouteState(sessionId)).toBe('unavailable') + await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + expect(current.adapter.sendSignal).not.toHaveBeenCalled() + + await router.discoverLegacySessions() + expect(router.getSessionRouteState(sessionId)).toBe('owned') + await router.sendSignal(sessionId, 'SIGTERM') + expect(legacy.adapter.sendSignal).toHaveBeenCalledWith(sessionId, 'SIGTERM') + }) + + it('acknowledges a colliding id through the adapter that produced its snapshot', async () => { + const sessionId = 'colliding-snapshot' + const current = createAdapter('current', [sessionId], { + alive: [sessionId], + killed: [] + }) + const legacy = createAdapter('legacy', [sessionId]) + vi.mocked(legacy.adapter.getBufferSnapshot).mockResolvedValue({ + data: 'legacy frame', + cols: 80, + rows: 24, + seq: 1, + source: 'headless' + }) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.discoverLegacySessions() + await router.getBufferSnapshot(sessionId) + await router.reconcileOnStartup(new Set()) + router.ackColdRestore(sessionId) + + expect(legacy.adapter.ackColdRestore).toHaveBeenCalledExactlyOnceWith(sessionId) + expect(current.adapter.ackColdRestore).not.toHaveBeenCalled() + }) + + it('passes folder and git-worktree keys through the same reconciliation path', async () => { + const current = createAdapter('current') + const legacy = createAdapter('legacy') + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + const validWorkspaceKeys = new Set(['folder:folder-1', 'repo-1::/workspace/project']) + + await router.reconcileOnStartup(validWorkspaceKeys) + + expect(current.adapter.reconcileOnStartup).toHaveBeenCalledExactlyOnceWith(validWorkspaceKeys) + expect(legacy.adapter.reconcileOnStartup).toHaveBeenCalledExactlyOnceWith(validWorkspaceKeys) + }) +}) diff --git a/src/main/daemon/daemon-pty-router-session-routing.ts b/src/main/daemon/daemon-pty-router-session-routing.ts new file mode 100644 index 00000000000..cc90ada1961 --- /dev/null +++ b/src/main/daemon/daemon-pty-router-session-routing.ts @@ -0,0 +1,147 @@ +import type { PtyProviderBufferSnapshot, PtySpawnOptions, PtySpawnResult } from '../providers/types' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { DaemonSessionRouteTable } from './daemon-session-route-table' +import { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowledgement-routes' + +export class DaemonPtyRouterSessionRouting { + private readonly routes: DaemonSessionRouteTable + private readonly snapshotAcks = new DaemonSnapshotAcknowledgementRoutes() + private readonly identityUnsubscribes: (() => void)[] = [] + + constructor( + private readonly current: DaemonPtyAdapter, + private readonly adapters: readonly DaemonPtyAdapter[] + ) { + this.routes = new DaemonSessionRouteTable(adapters) + for (const adapter of adapters) { + this.identityUnsubscribes.push( + adapter.onDaemonIdentityChanged(({ previous }) => { + this.routes.markOwnerUnavailable(adapter, previous) + }) + ) + } + } + + recordDiscovery(adapter: DaemonPtyAdapter, sessions: readonly { id: string }[]): void { + this.routes.recordCompleteDiscovery( + adapter, + sessions.map((session) => session.id) + ) + } + + recordDiscoveryFailure(adapter: DaemonPtyAdapter): void { + this.routes.recordDiscoveryFailure(adapter) + } + + spawnTarget(opts: PtySpawnOptions): DaemonPtyAdapter | Promise { + if (!opts.sessionId || opts.isNewSession) { + return this.routes.ownerForFreshSpawn(opts.sessionId, this.current) + } + return this.routes.resolveOwner(opts.sessionId) + } + + recordSpawn(result: PtySpawnResult, target: DaemonPtyAdapter): void { + if (result.exitedBeforeSpawnReply) { + return + } + this.routes.transfer(result.id, target) + } + + async owner(sessionId: string): Promise { + return await this.routes.resolveOwner(sessionId) + } + + ownerSync(sessionId: string): DaemonPtyAdapter { + return this.routes.resolveOwnerSync(sessionId) + } + + ownerOrCurrent(sessionId?: string): DaemonPtyAdapter { + return sessionId ? this.ownerSync(sessionId) : this.current + } + + hasPty(sessionId: string): boolean { + const route = this.routes.get(sessionId) + if (route?.state === 'unavailable') { + return false + } + if (route?.state === 'owned') { + return route.owner.hasPty(sessionId) + } + return this.adapters.some((adapter) => adapter.hasPty(sessionId)) + } + + async probePtyLiveness(sessionId: string): Promise { + return await this.routes.probeLiveness(sessionId) + } + + providesAgentSessionOwnerListings(sessionId: string): boolean { + return this.routes.getOwned(sessionId)?.providesAgentSessionOwnerListings(sessionId) === true + } + + canProvideAuthoritativeBufferSnapshot(sessionId: string): boolean { + return this.ownerSync(sessionId).canProvideAuthoritativeBufferSnapshot(sessionId) + } + + async getBufferSnapshot( + sessionId: string, + opts?: { scrollbackRows?: number } + ): Promise { + const owner = await this.owner(sessionId) + const snapshot = await owner.getBufferSnapshot(sessionId, opts) + this.snapshotAcks.record(sessionId, snapshot, owner) + return snapshot + } + + acknowledgeBufferSnapshot(sessionId: string): void { + this.snapshotAcks.acknowledge(sessionId) + } + + recordShutdown( + sessionId: string, + owner: DaemonPtyAdapter, + keepHistory: boolean | undefined, + migrateHistory: boolean + ): void { + if (migrateHistory) { + this.routes.transfer(sessionId, this.current) + return + } + if (!keepHistory) { + this.routes.markUnavailable(sessionId, owner) + this.snapshotAcks.drop(sessionId) + } + } + + recordReconciledAlive(sessionId: string, owner: DaemonPtyAdapter): void { + this.routes.recordOwned(sessionId, owner) + } + + recordReconciledKilled(sessionId: string, owner: DaemonPtyAdapter): void { + this.routes.markUnavailable(sessionId, owner) + this.snapshotAcks.drop(sessionId) + } + + recordExit(sessionId: string, owner: DaemonPtyAdapter): void { + this.routes.markUnavailable(sessionId, owner) + } + + markAdapterUnavailable(adapter: DaemonPtyAdapter): void { + if (this.adapters.includes(adapter)) { + this.routes.markOwnerUnavailable(adapter) + } + } + + getRouteState(sessionId: string): 'owned' | 'unavailable' | 'ambiguous' | null { + return this.routes.get(sessionId)?.state ?? null + } + + clearAdapterTombstone(sessionId: string): void { + this.routes.getOwned(sessionId)?.clearTombstone(sessionId) + } + + dispose(): void { + for (const unsubscribe of this.identityUnsubscribes.splice(0)) { + unsubscribe() + } + } +} diff --git a/src/main/daemon/daemon-pty-router.test.ts b/src/main/daemon/daemon-pty-router.test.ts index 17ad0dd835c..5cafdc03877 100644 --- a/src/main/daemon/daemon-pty-router.test.ts +++ b/src/main/daemon/daemon-pty-router.test.ts @@ -13,6 +13,7 @@ type AdapterMock = DaemonPtyAdapter & { emitData: (id: string, data: string, sequenceChars?: number) => void emitBackground: (event: PtyBackgroundStreamEvent) => void emitExit: (id: string, code: number, incarnationId?: string) => void + emitIdentityChanged: () => void triggerWriteUnavailable: (id: string) => void } @@ -39,6 +40,10 @@ function createAdapter( const writeUnavailableListeners: ((payload: { id: string }) => void)[] = [] const exitListeners: ((payload: { id: string; code: number; incarnationId?: string }) => void)[] = [] + const identityListeners: ((event: { + previous: { pid: number; startedAtMs: number; launchNonce: string } + current: { pid: number; startedAtMs: number; launchNonce: string } + }) => void)[] = [] return { protocolVersion, supportsGitCredentialGuardHost: () => @@ -50,6 +55,16 @@ function createAdapter( providesAgentSessionOwnerListings: () => protocolVersion >= AGENT_SESSION_CLAIM_DAEMON_PROTOCOL_VERSION, canProvideAuthoritativeBufferSnapshot: () => protocolVersion >= 20, + getLastAuthenticatedDaemonIdentity: vi.fn(() => null), + onDaemonIdentityChanged: vi.fn((listener) => { + identityListeners.push(listener) + return () => { + const idx = identityListeners.indexOf(listener) + if (idx !== -1) { + identityListeners.splice(idx, 1) + } + } + }), spawn: vi.fn(async (opts: PtySpawnOptions): Promise => { const id = opts.sessionId ?? `${label}-new` sessions.push(id) @@ -150,6 +165,14 @@ function createAdapter( listener({ id, code, ...(incarnationId ? { incarnationId } : {}) }) } }, + emitIdentityChanged: () => { + for (const listener of identityListeners) { + listener({ + previous: { pid: 1, startedAtMs: 1, launchNonce: `${label}-old` }, + current: { pid: 2, startedAtMs: 2, launchNonce: `${label}-new` } + }) + } + }, triggerWriteUnavailable: (id: string) => { for (const listener of writeUnavailableListeners) { listener({ id }) @@ -266,7 +289,12 @@ describe('DaemonPtyRouter', () => { ) const router = new DaemonPtyRouter({ current, legacy: [] }) - const spawning = router.spawn({ cols: 80, rows: 24, sessionId: 'raced-session' }) + const spawning = router.spawn({ + cols: 80, + rows: 24, + sessionId: 'raced-session', + isNewSession: true + }) finishSpawn?.({ id: 'raced-session', incarnationId: 'raced-incarnation', @@ -274,10 +302,7 @@ describe('DaemonPtyRouter', () => { }) await expect(spawning).resolves.toMatchObject({ exitedBeforeSpawnReply: true }) - const internals = router as unknown as { - sessionAdapters: Map - } - expect(internals.sessionAdapters.has('raced-session')).toBe(false) + expect(router.getSessionRouteState('raced-session')).toBeNull() }) it('routes a replacement when only an older incarnation exits during spawn', async () => { @@ -291,15 +316,17 @@ describe('DaemonPtyRouter', () => { ) const router = new DaemonPtyRouter({ current, legacy: [] }) - const spawning = router.spawn({ cols: 80, rows: 24, sessionId: 'reused-session' }) + const spawning = router.spawn({ + cols: 80, + rows: 24, + sessionId: 'reused-session', + isNewSession: true + }) current.emitExit('reused-session', 0, 'incarnation-old') finishSpawn?.({ id: 'reused-session', incarnationId: 'incarnation-current' }) await spawning - const internals = router as unknown as { - sessionAdapters: Map - } - expect(internals.sessionAdapters.get('reused-session')).toBe(current) + expect(router.getSessionRouteState('reused-session')).toBe('owned') }) it('preserves canonical claimed-owner exit proof from the adapter', async () => { @@ -317,6 +344,7 @@ describe('DaemonPtyRouter', () => { cols: 80, rows: 24, sessionId: 'requested-session', + isNewSession: true, agentSessionEnsure: {} as never }) finishSpawn?.({ @@ -329,10 +357,7 @@ describe('DaemonPtyRouter', () => { id: 'canonical-session', exitedBeforeSpawnReply: true }) - const internals = router as unknown as { - sessionAdapters: Map - } - expect(internals.sessionAdapters.has('canonical-session')).toBe(false) + expect(router.getSessionRouteState('canonical-session')).toBeNull() }) it('reports snapshot capability for the adapter that owns each session', async () => { @@ -445,7 +470,7 @@ describe('DaemonPtyRouter', () => { }) }) - it('drops a legacy mapping after the routed session exits', async () => { + it('tombstones a legacy route after the routed session exits', async () => { const current = createAdapter('current') const legacy = createAdapter('legacy', ['legacy-session']) const router = new DaemonPtyRouter({ current, legacy: [legacy] }) @@ -453,9 +478,12 @@ describe('DaemonPtyRouter', () => { await router.discoverLegacySessions() legacy.emitExit('legacy-session', 0) - await router.spawn({ sessionId: 'legacy-session', cols: 80, rows: 24 }) + await expect(router.spawn({ sessionId: 'legacy-session', cols: 80, rows: 24 })).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) - expect(current.spawn).toHaveBeenCalledWith({ sessionId: 'legacy-session', cols: 80, rows: 24 }) + expect(router.getSessionRouteState('legacy-session')).toBe('unavailable') + expect(current.spawn).not.toHaveBeenCalled() }) it('uses mapped adapter liveness instead of routing-cache presence for hasPty', async () => { @@ -559,7 +587,7 @@ describe('DaemonPtyRouter', () => { expect(current.listProcesses).toHaveBeenCalledTimes(2) }) - it('pins colliding unmapped legacy ids falling through to the current daemon', async () => { + it('blocks colliding ids when legacy discovery is incomplete', async () => { const sessionId = 'cross-generation-collision' const current = createAdapter('current', [sessionId]) const legacy = createAdapter('legacy', [sessionId]) @@ -568,10 +596,14 @@ describe('DaemonPtyRouter', () => { const router = new DaemonPtyRouter({ current, legacy: [legacy] }) await router.discoverLegacySessions() - router.write(sessionId, 'misrouted\n') + expect(() => router.write(sessionId, 'blocked\n')).toThrow('daemon_session_routing_unavailable') + await expect(router.spawn({ sessionId, cols: 80, rows: 24 })).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) - expect(current.write).toHaveBeenCalledWith(sessionId, 'misrouted\n') + expect(current.write).not.toHaveBeenCalled() expect(legacy.write).not.toHaveBeenCalled() + expect(current.spawn).not.toHaveBeenCalled() warn.mockRestore() }) diff --git a/src/main/daemon/daemon-pty-router.ts b/src/main/daemon/daemon-pty-router.ts index 0f07da4dc18..b3130451af1 100644 --- a/src/main/daemon/daemon-pty-router.ts +++ b/src/main/daemon/daemon-pty-router.ts @@ -9,51 +9,62 @@ import type { PtySpawnResult } from '../providers/types' import type { PtyProcessInspection } from '../providers/pty-process-inspection' -import { probePtyOwners } from './daemon-pty-liveness-probe' import { shouldHandoffDaemonHistory } from './daemon-history-handoff' import type { DaemonPtyRouterDataEvent, DaemonPtyRouterExitEvent } from './daemon-pty-router-events' +import { DaemonPtyRouterSessionRouting } from './daemon-pty-router-session-routing' export class DaemonPtyRouter implements IPtyProvider { private current: DaemonPtyAdapter private legacy: DaemonPtyAdapter[] - private sessionAdapters = new Map() + private readonly routing: DaemonPtyRouterSessionRouting private readonly subscriptions: DaemonPtyAdapterSubscriptionFanout constructor(opts: { current: DaemonPtyAdapter; legacy: DaemonPtyAdapter[] }) { this.current = opts.current this.legacy = opts.legacy - this.subscriptions = new DaemonPtyAdapterSubscriptionFanout(this.allAdapters(), (id) => { - this.sessionAdapters.delete(id) - }) + this.routing = new DaemonPtyRouterSessionRouting(this.current, this.allAdapters()) + this.subscriptions = new DaemonPtyAdapterSubscriptionFanout( + this.allAdapters(), + (adapter, id) => { + this.routing.recordExit(id, adapter) + } + ) + } + + markAdapterRoutesUnavailable(adapter: DaemonPtyAdapter): void { + this.routing.markAdapterUnavailable(adapter) + } + + getSessionRouteState(sessionId: string): 'owned' | 'unavailable' | 'ambiguous' | null { + return this.routing.getRouteState(sessionId) + } + + private allAdapters(): DaemonPtyAdapter[] { + return [this.current, ...this.legacy] } async discoverLegacySessions(): Promise { for (const adapter of this.legacy) { try { const sessions = await adapter.listProcesses() - for (const session of sessions) { - this.sessionAdapters.set(session.id, adapter) - } + this.routing.recordDiscovery(adapter, sessions) } catch (error) { + this.routing.recordDiscoveryFailure(adapter) console.warn('[daemon] Failed to discover legacy daemon sessions', error) } } } async spawn(opts: PtySpawnOptions): Promise { - const adapter = opts.sessionId ? this.sessionAdapters.get(opts.sessionId) : undefined - const target = adapter ?? this.current + const targetResult = this.routing.spawnTarget(opts) + const target = targetResult instanceof Promise ? await targetResult : targetResult const result = await target.spawn(opts) - // Why: the adapter filters intentional recovery exits and canonical-ID races before publishing proof. - if (!result.exitedBeforeSpawnReply) { - this.sessionAdapters.set(result.id, target) - } + this.routing.recordSpawn(result, target) return result } supportsGitCredentialGuardHost(sessionId?: string): boolean { - const adapter = sessionId ? this.adapterFor(sessionId) : this.current - return adapter.supportsGitCredentialGuardHost() + return this.routing.ownerOrCurrent(sessionId).supportsGitCredentialGuardHost() } supportsAgentSessionClaims(): boolean { @@ -62,10 +73,9 @@ export class DaemonPtyRouter implements IPtyProvider { } providesAgentSessionOwnerListings(ptyId: string): boolean { - const adapter = this.sessionAdapters.get(ptyId) // Why: an unmapped id may belong to any preserved daemon generation; // only an established route can make an omitted owner authoritative. - return adapter?.providesAgentSessionOwnerListings(ptyId) === true + return this.routing.providesAgentSessionOwnerListings(ptyId) } supportsAgentSessionCreateOperations(): boolean { @@ -74,111 +84,103 @@ export class DaemonPtyRouter implements IPtyProvider { } async attach(id: string): Promise { - await this.adapterFor(id).attach(id) + await (await this.routing.owner(id)).attach(id) } hasPty(id: string): boolean { - const routed = this.sessionAdapters.get(id) - if (routed) { - return routed.hasPty(id) - } - return this.current.hasPty(id) || this.legacy.some((adapter) => adapter.hasPty(id)) + return this.routing.hasPty(id) } async probePtyLiveness(id: string): Promise { - return await probePtyOwners(id, this.sessionAdapters.get(id), this.allAdapters()) + return await this.routing.probePtyLiveness(id) } write(id: string, data: string): void { - this.adapterFor(id).write(id, data) + this.routing.ownerSync(id).write(id, data) } resize(id: string, cols: number, rows: number): void { - this.adapterFor(id).resize(id, cols, rows) + this.routing.ownerSync(id).resize(id, cols, rows) } pauseProducer(id: string): void { - this.adapterFor(id).pauseProducer(id) + this.routing.ownerSync(id).pauseProducer(id) } resumeProducer(id: string): void { - this.adapterFor(id).resumeProducer(id) + this.routing.ownerSync(id).resumeProducer(id) } setPtyBackgrounded(id: string, background: boolean): void { - this.adapterFor(id).setPtyBackgrounded(id, background) + this.routing.ownerSync(id).setPtyBackgrounded(id, background) } async shutdown( id: string, opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number } ): Promise { - const adapter = this.adapterFor(id) + const adapter = await this.routing.owner(id) const migrateHistory = shouldHandoffDaemonHistory(opts.keepHistory, adapter, this.current) await adapter.shutdown(id, opts) - if (!opts.keepHistory || migrateHistory) { - if (migrateHistory) { - adapter.ackColdRestore(id) - } - if (this.sessionAdapters.get(id) === adapter) { - this.sessionAdapters.delete(id) - } + if (migrateHistory) { + adapter.ackColdRestore(id) } + this.routing.recordShutdown(id, adapter, opts.keepHistory, migrateHistory) } async sendSignal(id: string, signal: string): Promise { - await this.adapterFor(id).sendSignal(id, signal) + await (await this.routing.owner(id)).sendSignal(id, signal) } async getCwd(id: string): Promise { - return this.adapterFor(id).getCwd(id) + return (await this.routing.owner(id)).getCwd(id) } async getInitialCwd(id: string): Promise { - return this.adapterFor(id).getInitialCwd(id) + return (await this.routing.owner(id)).getInitialCwd(id) } async getAppliedSize(id: string): Promise<{ cols: number; rows: number } | null> { - return (await this.adapterFor(id).getAppliedSize?.(id)) ?? null + return (await (await this.routing.owner(id)).getAppliedSize?.(id)) ?? null } async getBufferSnapshot( id: string, opts?: { scrollbackRows?: number } ): Promise { - return await this.adapterFor(id).getBufferSnapshot(id, opts) + return await this.routing.getBufferSnapshot(id, opts) } canProvideAuthoritativeBufferSnapshot(id: string): boolean { - return this.adapterFor(id).canProvideAuthoritativeBufferSnapshot(id) + return this.routing.canProvideAuthoritativeBufferSnapshot(id) } async clearBuffer(id: string): Promise { - await this.adapterFor(id).clearBuffer(id) + await (await this.routing.owner(id)).clearBuffer(id) } async closeStartupQueryAuthority(id: string): Promise { - return (await this.adapterFor(id).closeStartupQueryAuthority?.(id)) ?? 0 + return (await (await this.routing.owner(id)).closeStartupQueryAuthority?.(id)) ?? 0 } acknowledgeDataEvent(id: string, charCount: number): void { - this.adapterFor(id).acknowledgeDataEvent(id, charCount) + this.routing.ownerSync(id).acknowledgeDataEvent(id, charCount) } async hasChildProcesses(id: string): Promise { - return this.adapterFor(id).hasChildProcesses(id) + return (await this.routing.owner(id)).hasChildProcesses(id) } async getForegroundProcess(id: string): Promise { - return this.adapterFor(id).getForegroundProcess(id) + return (await this.routing.owner(id)).getForegroundProcess(id) } async inspectProcess(id: string): Promise { - return this.adapterForInspection(id).inspectProcess(id) + return (await this.routing.owner(id)).inspectProcess(id) } async confirmForegroundProcess(id: string): Promise { - return this.adapterFor(id).confirmForegroundProcess(id) + return (await this.routing.owner(id)).confirmForegroundProcess(id) } async serialize(ids: string[]): Promise { @@ -227,11 +229,11 @@ export class DaemonPtyRouter implements IPtyProvider { } ackColdRestore(sessionId: string): void { - this.adapterFor(sessionId).ackColdRestore(sessionId) + this.routing.acknowledgeBufferSnapshot(sessionId) } clearTombstone(sessionId: string): void { - this.adapterFor(sessionId).clearTombstone(sessionId) + this.routing.clearAdapterTombstone(sessionId) } async reconcileOnStartup(validWorktreeIds: Set): Promise<{ @@ -251,10 +253,10 @@ export class DaemonPtyRouter implements IPtyProvider { killed.push(id) } for (const id of result.alive) { - this.sessionAdapters.set(id, adapter) + this.routing.recordReconciledAlive(id, adapter) } for (const id of result.killed) { - this.sessionAdapters.delete(id) + this.routing.recordReconciledKilled(id, adapter) } } return { alive, killed } @@ -262,6 +264,7 @@ export class DaemonPtyRouter implements IPtyProvider { dispose(): void { this.subscriptions.dispose() + this.routing.dispose() for (const adapter of this.allAdapters()) { adapter.dispose() } @@ -276,10 +279,12 @@ export class DaemonPtyRouter implements IPtyProvider { // adapters' listener arrays (one pair per adapter per restart). disposeRouterOnly(): void { this.subscriptions.dispose() + this.routing.dispose() } async disconnectOnly(): Promise { this.subscriptions.dispose() + this.routing.dispose() await Promise.all([...this.allAdapters()].map((adapter) => adapter.disconnectOnly())) } @@ -300,23 +305,4 @@ export class DaemonPtyRouter implements IPtyProvider { getAllAdapters(): readonly DaemonPtyAdapter[] { return this.allAdapters() } - - private adapterFor(sessionId: string): DaemonPtyAdapter { - return this.sessionAdapters.get(sessionId) ?? this.current - } - - private adapterForInspection(sessionId: string): DaemonPtyAdapter { - const adapter = - this.sessionAdapters.get(sessionId) ?? - this.allAdapters().find((candidate) => candidate.hasPty(sessionId)) - if (!adapter) { - throw new Error('terminal_gone') - } - this.sessionAdapters.set(sessionId, adapter) - return adapter - } - - private allAdapters(): DaemonPtyAdapter[] { - return [this.current, ...this.legacy] - } } diff --git a/src/main/daemon/daemon-session-route-table.ts b/src/main/daemon/daemon-session-route-table.ts new file mode 100644 index 00000000000..fb5bc6c2dda --- /dev/null +++ b/src/main/daemon/daemon-session-route-table.ts @@ -0,0 +1,277 @@ +import type { DaemonEndpointIdentity } from './daemon-hello-protocol' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { DAEMON_SESSION_ROUTING_UNAVAILABLE_MARKER } from '../../shared/daemon-session-routing-error' + +export type DaemonSessionRoute = + | { + state: 'owned' + owner: DaemonPtyAdapter + incarnation: DaemonEndpointIdentity | null + } + | { + state: 'unavailable' + owner: DaemonPtyAdapter + incarnation: DaemonEndpointIdentity | null + } + | { + state: 'ambiguous' + candidates: ReadonlySet + } + +export class DaemonSessionOwnerUnknownError extends Error { + constructor(sessionId: string) { + super(`${DAEMON_SESSION_ROUTING_UNAVAILABLE_MARKER}: owner unknown for "${sessionId}"`) + this.name = 'DaemonSessionOwnerUnknownError' + } +} + +export class DaemonSessionUnavailableError extends Error { + constructor(sessionId: string) { + super(`${DAEMON_SESSION_ROUTING_UNAVAILABLE_MARKER}: route unavailable for "${sessionId}"`) + this.name = 'DaemonSessionUnavailableError' + } +} + +export class DaemonSessionGoneError extends Error { + constructor(sessionId: string) { + super(`terminal_gone: no daemon owns "${sessionId}"`) + this.name = 'DaemonSessionGoneError' + } +} + +export class DaemonSessionRouteTable { + private readonly routes = new Map() + private readonly incompleteDiscoveries = new Set() + + constructor(private readonly adapters: readonly DaemonPtyAdapter[]) {} + + get(sessionId: string): DaemonSessionRoute | undefined { + return this.routes.get(sessionId) + } + + getOwned(sessionId: string): DaemonPtyAdapter | undefined { + const route = this.routes.get(sessionId) + return route?.state === 'owned' ? route.owner : undefined + } + + recordOwned(sessionId: string, owner: DaemonPtyAdapter): void { + const route = this.routes.get(sessionId) + if (!route) { + this.routes.set(sessionId, ownedRoute(owner)) + return + } + if (route.state === 'owned' && route.owner === owner) { + this.routes.set(sessionId, ownedRoute(owner)) + return + } + if (route.state === 'unavailable') { + const incarnation = owner.getLastAuthenticatedDaemonIdentity() + if (route.owner === owner && !sameIncarnation(route.incarnation, incarnation)) { + this.routes.set(sessionId, ownedRoute(owner)) + } + return + } + const candidates = + route.state === 'ambiguous' ? new Set(route.candidates) : new Set([route.owner]) + candidates.add(owner) + this.routes.set(sessionId, { state: 'ambiguous', candidates }) + } + + transfer(sessionId: string, owner: DaemonPtyAdapter): void { + this.routes.set(sessionId, ownedRoute(owner)) + } + + recordDiscoveryFailure(owner: DaemonPtyAdapter): void { + this.incompleteDiscoveries.add(owner) + } + + recordCompleteDiscovery(owner: DaemonPtyAdapter, sessionIds: readonly string[]): void { + for (const sessionId of sessionIds) { + this.recordOwned(sessionId, owner) + } + this.incompleteDiscoveries.delete(owner) + } + + markUnavailable(sessionId: string, owner: DaemonPtyAdapter): void { + const route = this.routes.get(sessionId) + if (!route) { + this.routes.set(sessionId, { + state: 'unavailable', + owner, + incarnation: owner.getLastAuthenticatedDaemonIdentity() + }) + return + } + if (route.state === 'owned' && route.owner === owner) { + this.routes.set(sessionId, { + state: 'unavailable', + owner, + incarnation: route.incarnation + }) + return + } + if (route.state === 'ambiguous' && route.candidates.has(owner)) { + this.routes.set(sessionId, { + state: 'unavailable', + owner, + incarnation: owner.getLastAuthenticatedDaemonIdentity() + }) + } + } + + markOwnerUnavailable( + owner: DaemonPtyAdapter, + incarnation = owner.getLastAuthenticatedDaemonIdentity() + ): void { + for (const [sessionId, route] of this.routes) { + if ( + route.state === 'owned' && + route.owner === owner && + sameIncarnation(route.incarnation, incarnation) + ) { + this.routes.set(sessionId, { + state: 'unavailable', + owner, + incarnation: route.incarnation + }) + } + } + } + + ownerForFreshSpawn(sessionId: string | undefined, current: DaemonPtyAdapter): DaemonPtyAdapter { + if (!sessionId) { + return current + } + const route = this.routes.get(sessionId) + if (!route) { + return current + } + if (route.state === 'owned' && route.owner === current) { + return current + } + if (route.state === 'unavailable') { + throw new DaemonSessionUnavailableError(sessionId) + } + throw new DaemonSessionOwnerUnknownError(sessionId) + } + + async resolveOwner(sessionId: string): Promise { + const route = this.routes.get(sessionId) + if (route?.state === 'owned') { + return route.owner + } + if (route?.state === 'unavailable') { + throw new DaemonSessionUnavailableError(sessionId) + } + if ((!route || route.state === 'ambiguous') && this.incompleteDiscoveries.size > 0) { + throw new DaemonSessionOwnerUnknownError(sessionId) + } + const candidates = route?.state === 'ambiguous' ? [...route.candidates] : this.adapters + const results = await Promise.all( + candidates.map(async (owner) => ({ + owner, + result: await probeOwner(owner, sessionId) + })) + ) + const owners = results.filter(({ result }) => result === true).map(({ owner }) => owner) + if (owners.length === 1 && results.every(({ result }) => result !== null)) { + this.transfer(sessionId, owners[0]) + return owners[0] + } + if (owners.length > 1) { + this.routes.set(sessionId, { state: 'ambiguous', candidates: new Set(owners) }) + } + if ( + owners.length === 0 && + results.every(({ result }) => result === false) && + this.incompleteDiscoveries.size === 0 + ) { + throw new DaemonSessionGoneError(sessionId) + } + throw new DaemonSessionOwnerUnknownError(sessionId) + } + + async probeLiveness(sessionId: string): Promise { + const route = this.routes.get(sessionId) + if (route?.state === 'unavailable' || route?.state === 'ambiguous') { + return null + } + if (route?.state === 'owned') { + return await probeOwner(route.owner, sessionId) + } + try { + await this.resolveOwner(sessionId) + return true + } catch (error) { + return error instanceof DaemonSessionGoneError ? false : null + } + } + + resolveOwnerSync(sessionId: string): DaemonPtyAdapter { + const route = this.routes.get(sessionId) + if (route?.state === 'owned') { + return route.owner + } + if (route?.state === 'unavailable') { + throw new DaemonSessionUnavailableError(sessionId) + } + if (route?.state === 'ambiguous' || this.incompleteDiscoveries.size > 0) { + throw new DaemonSessionOwnerUnknownError(sessionId) + } + const results = this.adapters.map((owner) => ({ + owner, + result: hasPty(owner, sessionId) + })) + const owners = results.filter(({ result }) => result === true).map(({ owner }) => owner) + if (results.some(({ result }) => result === null)) { + throw new DaemonSessionOwnerUnknownError(sessionId) + } + if (owners.length === 1) { + this.transfer(sessionId, owners[0]) + return owners[0] + } + if (owners.length > 1) { + this.routes.set(sessionId, { state: 'ambiguous', candidates: new Set(owners) }) + throw new DaemonSessionOwnerUnknownError(sessionId) + } + throw new DaemonSessionGoneError(sessionId) + } +} + +function ownedRoute(owner: DaemonPtyAdapter): Extract { + return { + state: 'owned', + owner, + incarnation: owner.getLastAuthenticatedDaemonIdentity() + } +} + +async function probeOwner(owner: DaemonPtyAdapter, sessionId: string): Promise { + try { + return await owner.probePtyLiveness(sessionId) + } catch { + return null + } +} + +function hasPty(owner: DaemonPtyAdapter, sessionId: string): boolean | null { + try { + return owner.hasPty(sessionId) + } catch { + return null + } +} + +function sameIncarnation( + left: DaemonEndpointIdentity | null, + right: DaemonEndpointIdentity | null +): boolean { + if (!left || !right) { + return left === right + } + return ( + left.pid === right.pid && + left.startedAtMs === right.startedAtMs && + left.launchNonce === right.launchNonce + ) +} diff --git a/src/main/daemon/daemon-snapshot-acknowledgement-routes.ts b/src/main/daemon/daemon-snapshot-acknowledgement-routes.ts new file mode 100644 index 00000000000..576212b9aeb --- /dev/null +++ b/src/main/daemon/daemon-snapshot-acknowledgement-routes.ts @@ -0,0 +1,49 @@ +import type { IPtyProvider, PtyProviderBufferSnapshot } from '../providers/types' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' + +export class DaemonSnapshotAcknowledgementRoutes { + private readonly producers = new Map() + + async capture( + sessionId: string, + opts: { scrollbackRows?: number } | undefined, + provider: IPtyProvider, + adapters: readonly DaemonPtyAdapter[] + ): Promise { + const snapshot = (await provider.getBufferSnapshot?.(sessionId, opts)) ?? null + this.recordAdapter(sessionId, snapshot, provider, adapters) + return snapshot + } + + record( + sessionId: string, + snapshot: PtyProviderBufferSnapshot | null, + producer: DaemonPtyAdapter + ): void { + if (snapshot) { + this.producers.set(sessionId, producer) + } + } + + recordAdapter( + sessionId: string, + snapshot: PtyProviderBufferSnapshot | null, + provider: IPtyProvider, + adapters: readonly DaemonPtyAdapter[] + ): void { + const producer = adapters.find((adapter) => adapter === provider) + if (producer) { + this.record(sessionId, snapshot, producer) + } + } + + acknowledge(sessionId: string): void { + const producer = this.producers.get(sessionId) + this.producers.delete(sessionId) + producer?.ackColdRestore(sessionId) + } + + drop(sessionId: string): void { + this.producers.delete(sessionId) + } +} diff --git a/src/main/daemon/degraded-daemon-pty-provider.test.ts b/src/main/daemon/degraded-daemon-pty-provider.test.ts index fbbfdfe4408..56287fd7299 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.test.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.test.ts @@ -294,6 +294,34 @@ describe('DegradedDaemonPtyProvider', () => { }) }) + it('acknowledges a colliding id through the daemon that produced its snapshot', async () => { + const sessionId = 'colliding-session' + const current = createDaemonAdapter('current', [sessionId]) + const legacy = createDaemonAdapter('legacy', [sessionId]) + const fallback = createProvider('fallback') + legacy.getBufferSnapshot = vi.fn(async () => ({ + data: 'legacy frame', + cols: 80, + rows: 24, + seq: 1, + source: 'headless' as const + })) + const provider = new DegradedDaemonPtyProvider({ + current, + legacy: [legacy], + fallback + }) + + await provider.discoverDaemonSessions() + await provider.getBufferSnapshot(sessionId) + legacy.emitExit(sessionId, 0) + expect(provider.hasPty(sessionId)).toBe(true) + provider.ackColdRestore(sessionId) + + expect(legacy.ackColdRestore).toHaveBeenCalledExactlyOnceWith(sessionId) + expect(current.ackColdRestore).not.toHaveBeenCalled() + }) + it('forwards replay output from fallback and daemon providers', () => { const current = createDaemonAdapter('daemon') const fallback = createProvider('fallback') diff --git a/src/main/daemon/degraded-daemon-pty-provider.ts b/src/main/daemon/degraded-daemon-pty-provider.ts index a607495d3b8..7b1add71043 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.ts @@ -13,6 +13,7 @@ import type { } from '../providers/types' import { findDaemonAdapter, listProviderSessionIds } from './degraded-daemon-session-routing' import { probePtyOwners } from './daemon-pty-liveness-probe' +import { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowledgement-routes' export class DegradedDaemonPtyProvider implements IPtyProvider { readonly routesFreshSpawnsToLocalProvider = true @@ -23,6 +24,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { private legacy: DaemonPtyAdapter[] private fallback: IPtyProvider private sessionProviders = new Map() + private readonly snapshotAcks = new DaemonSnapshotAcknowledgementRoutes() private unsubscribers: (() => void)[] = [] private dataListeners: ((payload: PtyDataEvent) => void)[] = [] private exitListeners: ((payload: { id: string; code: number }) => void)[] = [] @@ -143,8 +145,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { id: string, opts?: { scrollbackRows?: number } ): Promise { - // Why: recovery must reach the legacy adapter that owns the thinned session model. - return (await this.providerFor(id).getBufferSnapshot?.(id, opts)) ?? null + return await this.snapshotAcks.capture(id, opts, this.providerFor(id), this.allDaemonAdapters()) } async clearBuffer(id: string): Promise { @@ -252,9 +253,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { } ackColdRestore(sessionId: string): void { - findDaemonAdapter(this.sessionProviders, this.allDaemonAdapters(), sessionId)?.ackColdRestore( - sessionId - ) + this.snapshotAcks.acknowledge(sessionId) } clearTombstone(sessionId: string): void { diff --git a/src/main/ipc/pty-management.test.ts b/src/main/ipc/pty-management.test.ts index 4e30669fb9e..954ba42f276 100644 --- a/src/main/ipc/pty-management.test.ts +++ b/src/main/ipc/pty-management.test.ts @@ -313,6 +313,26 @@ describe('pty:management IPC handlers', () => { expect(current.shutdown).toHaveBeenCalledTimes(1) }) + it('keeps sessions remaining when their owner becomes unavailable while polling', async () => { + const current = makeAdapter(5, [makeSession('uncertain-session')]) + vi.mocked(current.listSessions) + .mockResolvedValueOnce([ + makeSession('uncertain-session') as Omit + ]) + .mockRejectedValue(new Error('listing unavailable')) + const { registerDaemonManagementHandlers } = await importFresh() + getDaemonProviderMock.mockReturnValue(await makeRouter(current)) + registerDaemonManagementHandlers() + + const result = await runKillAllWithPolls(buildHandlerMap()['pty:management:killAll']) + + expect(result).toEqual({ + killedCount: 0, + remainingCount: 1, + killedSessionIds: [] + }) + }) + it('does not count respawned sessions with fresh IDs against remainingCount', async () => { // Why: mounted panes may re-call pty:spawn with brand-new session IDs // while killAll is polling (tab remount, navigate-back). Those fresh @@ -385,6 +405,44 @@ describe('pty:management IPC handlers', () => { killedSessionIds: ['b'] }) }) + + it('kills and counts colliding ids through their exact owning adapters', async () => { + const sessionId = 'colliding-session' + const currentSessions = [makeSession(sessionId, { incarnationId: 'current-incarnation' })] + const legacySessions = [ + makeSession(sessionId, { + protocolVersion: 3, + incarnationId: 'legacy-incarnation' + }) + ] + const current = makeAdapter(5, []) + const legacy = makeAdapter(3, []) + current.listSessions = vi.fn(async () => + currentSessions.map(({ protocolVersion: _pv, ...session }) => session) + ) + legacy.listSessions = vi.fn(async () => + legacySessions.map(({ protocolVersion: _pv, ...session }) => session) + ) + current.shutdown = vi.fn(async () => { + currentSessions.splice(0) + }) + legacy.shutdown = vi.fn(async () => { + legacySessions.splice(0) + }) + const { registerDaemonManagementHandlers } = await importFresh() + getDaemonProviderMock.mockReturnValue(await makeRouter(current, [legacy])) + registerDaemonManagementHandlers() + + const result = await runKillAllWithPolls(buildHandlerMap()['pty:management:killAll']) + + expect(result).toEqual({ + killedCount: 2, + remainingCount: 0, + killedSessionIds: [sessionId, sessionId] + }) + expect(current.shutdown).toHaveBeenCalledExactlyOnceWith(sessionId, { immediate: true }) + expect(legacy.shutdown).toHaveBeenCalledExactlyOnceWith(sessionId, { immediate: true }) + }) }) describe('killOne', () => { @@ -420,6 +478,55 @@ describe('pty:management IPC handlers', () => { expect(current.shutdown).not.toHaveBeenCalled() }) + it('fails a bare colliding id closed and routes an incarnation-qualified kill', async () => { + const sessionId = 'colliding-session' + const current = makeAdapter(5, [ + makeSession(sessionId, { incarnationId: 'current-incarnation' }) + ]) + const legacy = makeAdapter(3, [ + makeSession(sessionId, { + protocolVersion: 3, + incarnationId: 'legacy-incarnation' + }) + ]) + const { registerDaemonManagementHandlers } = await importFresh() + getDaemonProviderMock.mockReturnValue(await makeRouter(current, [legacy])) + registerDaemonManagementHandlers() + const handler = buildHandlerMap()['pty:management:killOne'] + + await expect(handler({}, { sessionId })).resolves.toEqual({ success: false }) + await expect( + handler( + {}, + { + sessionId, + protocolVersion: 3, + incarnationId: 'legacy-incarnation' + } + ) + ).resolves.toEqual({ success: true }) + + expect(current.shutdown).not.toHaveBeenCalled() + expect(legacy.shutdown).toHaveBeenCalledExactlyOnceWith(sessionId, { immediate: true }) + }) + + it('fails closed when another adapter cannot rule out a colliding id', async () => { + const current = makeAdapter(5, [makeSession('current-session')]) + const legacy = makeAdapter(3, []) + legacy.listSessions.mockRejectedValue(new Error('listing unavailable')) + const { registerDaemonManagementHandlers } = await importFresh() + getDaemonProviderMock.mockReturnValue(await makeRouter(current, [legacy])) + registerDaemonManagementHandlers() + + const result = await buildHandlerMap()['pty:management:killOne']( + {}, + { sessionId: 'current-session', protocolVersion: 5 } + ) + + expect(result).toEqual({ success: false }) + expect(current.shutdown).not.toHaveBeenCalled() + }) + it('rejects empty/missing sessionId without hitting the adapter', async () => { const current = makeAdapter(5, [makeSession('new-1')]) const { registerDaemonManagementHandlers } = await importFresh() diff --git a/src/main/ipc/pty-management.ts b/src/main/ipc/pty-management.ts index f188fb3aad5..29b9cbebe57 100644 --- a/src/main/ipc/pty-management.ts +++ b/src/main/ipc/pty-management.ts @@ -29,17 +29,46 @@ function isDaemonDegraded(): boolean { return getDaemonProvider() instanceof DegradedDaemonPtyProvider } -async function collectSessions(adapters: DaemonPtyAdapter[]): Promise { +type OwnedDaemonSession = { + owner: DaemonPtyAdapter + session: DaemonSessionInfo +} + +type DaemonSessionCollection = { + sessions: OwnedDaemonSession[] + unavailableOwners: Set +} + +async function collectSessions(adapters: DaemonPtyAdapter[]): Promise { const results = await Promise.allSettled( adapters.map(async (adapter) => { const sessions = await adapter.listSessions() - return sessions.map((s) => ({ - ...s, - protocolVersion: adapter.protocolVersion + return sessions.map((session) => ({ + owner: adapter, + session: { ...session, protocolVersion: adapter.protocolVersion } })) }) ) - return results.flatMap((r) => (r.status === 'fulfilled' ? r.value : [])) + const sessions: OwnedDaemonSession[] = [] + const unavailableOwners = new Set() + for (const [index, result] of results.entries()) { + if (result.status === 'fulfilled') { + sessions.push(...result.value) + } else { + unavailableOwners.add(adapters[index]) + } + } + return { sessions, unavailableOwners } +} + +function sameSessionIncarnation(left: OwnedDaemonSession, right: OwnedDaemonSession): boolean { + return ( + left.owner === right.owner && + left.session.sessionId === right.session.sessionId && + (left.session.incarnationId === undefined || + right.session.incarnationId === undefined || + left.session.incarnationId === right.session.incarnationId) + ) } export function registerDaemonManagementHandlers(): void { @@ -51,7 +80,9 @@ export function registerDaemonManagementHandlers(): void { ipcMain.handle( 'pty:management:listSessions', async (): Promise<{ sessions: DaemonSessionInfo[]; degraded: boolean }> => { - const sessions = await collectSessions(getDaemonAdapters()) + const sessions = (await collectSessions(getDaemonAdapters())).sessions.map( + ({ session }) => session + ) return { sessions, degraded: isDaemonDegraded() } } ) @@ -65,9 +96,7 @@ export function registerDaemonManagementHandlers(): void { killedSessionIds: string[] }> => { const adapters = getDaemonAdapters() - // Why: snapshot session IDs up front so mid-kill respawns aren't counted as "remaining". - const initial = await collectSessions(adapters) - const initialIds = new Set(initial.map((s) => s.sessionId)) + const initial = (await collectSessions(adapters)).sessions const initialCount = initial.length if (initialCount === 0) { @@ -76,61 +105,64 @@ export function registerDaemonManagementHandlers(): void { // Why: no retry — session.kill() is idempotent and runs its own kill ladder; allSettled so one rejection doesn't abort the rest. await Promise.allSettled( - initial.map(async (session) => { - // Why: assumes PROTOCOL_VERSION stays distinct from PREVIOUS_DAEMON_PROTOCOL_VERSIONS (types.ts), else legacy sessions misroute here. - const owner = adapters.find((a) => a.protocolVersion === session.protocolVersion) - if (!owner) { - return - } + initial.map(async ({ owner, session }) => { // Why: immediate=true only matters to legacy/future adapters; swallow rejections since remainingCount reports stuck sessions. await owner.shutdown(session.sessionId, { immediate: true }).catch(() => {}) }) ) // Why: count only the initial-snapshot intersection so renderer respawns mid-kill aren't counted as remaining. - let remainingOriginalCount = initialCount - let remainingOriginalIds = initialIds + let remainingOriginal = initial for (let attempt = 0; attempt < MAX_POLL_ATTEMPTS; attempt += 1) { await sleep(POLL_INTERVAL_MS) const current = await collectSessions(adapters) - remainingOriginalIds = new Set( - current - .filter((session) => initialIds.has(session.sessionId)) - .map((session) => session.sessionId) + remainingOriginal = initial.filter( + (original) => + current.unavailableOwners.has(original.owner) || + current.sessions.some((candidate) => sameSessionIncarnation(original, candidate)) ) - remainingOriginalCount = remainingOriginalIds.size - if (remainingOriginalCount === 0) { + if (remainingOriginal.length === 0) { break } } - const killedCount = initialCount - remainingOriginalCount + const killed = initial.filter( + (original) => + !remainingOriginal.some((candidate) => sameSessionIncarnation(original, candidate)) + ) return { - killedCount, - remainingCount: remainingOriginalCount, - killedSessionIds: [...initialIds].filter( - (sessionId) => !remainingOriginalIds.has(sessionId) - ) + killedCount: killed.length, + remainingCount: remainingOriginal.length, + killedSessionIds: killed.map(({ session }) => session.sessionId) } } ) ipcMain.handle( 'pty:management:killOne', - async (_event, args: { sessionId: string }): Promise<{ success: boolean }> => { + async ( + _event, + args: { sessionId: string; protocolVersion?: number; incarnationId?: string } + ): Promise<{ success: boolean }> => { if (typeof args?.sessionId !== 'string' || args.sessionId.length === 0) { return { success: false } } const adapters = getDaemonAdapters() - const sessions = await collectSessions(adapters) - const match = sessions.find((s) => s.sessionId === args.sessionId) - if (!match) { + const collection = await collectSessions(adapters) + if (collection.unavailableOwners.size > 0) { return { success: false } } - const owner = adapters.find((a) => a.protocolVersion === match.protocolVersion) - if (!owner) { + const matches = collection.sessions.filter( + ({ session }) => + session.sessionId === args.sessionId && + (args.protocolVersion === undefined || + session.protocolVersion === args.protocolVersion) && + (args.incarnationId === undefined || session.incarnationId === args.incarnationId) + ) + if (matches.length !== 1) { return { success: false } } + const [{ owner }] = matches try { await owner.shutdown(args.sessionId, { immediate: true }) return { success: true } diff --git a/src/preload/api-types.ts b/src/preload/api-types.ts index 81796d17eb3..da2fa16380c 100644 --- a/src/preload/api-types.ts +++ b/src/preload/api-types.ts @@ -723,6 +723,7 @@ export type PreflightApi = { // Mirror of daemon's `DaemonSessionInfo` (src/main/daemon/types.ts); not imported — preload can't depend on main-only protocol types. export type PtyManagementSession = { sessionId: string + incarnationId?: string state: 'created' | 'spawning' | 'running' | 'exiting' | 'exited' shellState: 'pending' | 'ready' | 'timed_out' | 'unsupported' isAlive: boolean @@ -742,7 +743,11 @@ export type PtyManagementApi = { remainingCount: number killedSessionIds?: string[] }> - killOne: (args: { sessionId: string }) => Promise<{ success: boolean }> + killOne: (args: { + sessionId: string + protocolVersion: number + incarnationId?: string + }) => Promise<{ success: boolean }> restart: () => Promise<{ success: boolean }> } diff --git a/src/preload/index.ts b/src/preload/index.ts index 7c63fd9756d..b6421a0c25b 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -1233,7 +1233,8 @@ const api = { management: { listSessions: () => ipcRenderer.invoke('pty:management:listSessions'), killAll: () => ipcRenderer.invoke('pty:management:killAll'), - killOne: (args: { sessionId: string }) => ipcRenderer.invoke('pty:management:killOne', args), + killOne: (args: { sessionId: string; protocolVersion: number; incarnationId?: string }) => + ipcRenderer.invoke('pty:management:killOne', args), restart: () => ipcRenderer.invoke('pty:management:restart') } }, diff --git a/src/renderer/src/components/settings/ManageSessionsSection.tsx b/src/renderer/src/components/settings/ManageSessionsSection.tsx index de48b535e3d..f9092eabc15 100644 --- a/src/renderer/src/components/settings/ManageSessionsSection.tsx +++ b/src/renderer/src/components/settings/ManageSessionsSection.tsx @@ -132,7 +132,9 @@ export function ManageSessionsSection(): React.JSX.Element { mutationInFlight.current = true try { const { success } = await window.api.pty.management.killOne({ - sessionId: session.sessionId + sessionId: session.sessionId, + protocolVersion: session.protocolVersion, + ...(session.incarnationId ? { incarnationId: session.incarnationId } : {}) }) if (success) { toast.success( diff --git a/src/renderer/src/components/terminal-pane/pty-connection.test.ts b/src/renderer/src/components/terminal-pane/pty-connection.test.ts index db6030d00fd..64d9cd51723 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.test.ts @@ -11104,6 +11104,30 @@ describe('connectPanePty', () => { expect(deps.updateTabPtyId).not.toHaveBeenCalled() }) + it('retains restored layout and rendered state when daemon routing is unavailable', async () => { + const { connectPanePty } = await import('./pty-connection') + const transport = createMockTransport('tab-pty') + transport.connect.mockImplementation(async ({ sessionId }: { sessionId?: string }) => + sessionId ? { id: sessionId, routingUnavailable: true as const } : 'unexpected-fresh-pty' + ) + transportFactoryQueue.push(transport) + const pane = createPane(1) + const deps = createDeps({ + restoredLeafId: LEAF_1, + restoredPtyIdByLeafId: { [LEAF_1]: 'tab-pty' } + }) + + connectPanePty(pane as never, createManager(1) as never, deps as never) + await flushAsyncTicks(20) + + expect(transport.connect).toHaveBeenCalledTimes(1) + expect(deps.clearExitedPanePtyLayoutBinding).not.toHaveBeenCalled() + expect(deps.syncPanePtyLayoutBinding).not.toHaveBeenCalledWith(1, null) + expect(deps.clearTabPtyId).not.toHaveBeenCalled() + expect(deps.updateTabPtyId).not.toHaveBeenCalled() + expect(pane.terminal.clear).not.toHaveBeenCalled() + }) + // Why: Phase 6 deleted the hidden-skip grammar — every hidden chunk rides the background scheduler, none content-scanned. it('queues hidden PTY bytes on the background scheduler without per-chunk scanning', async () => { const { connectPanePty } = await import('./pty-connection') diff --git a/src/renderer/src/components/terminal-pane/pty-connection.ts b/src/renderer/src/components/terminal-pane/pty-connection.ts index 188fdf11c55..58d6151a645 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.ts @@ -7569,6 +7569,9 @@ export function connectPanePty( // Why: the transport already delivered the dead session's final frame + exit; treat as terminal state, not a failed reattach. return true } + if (connectResult?.routingUnavailable) { + return true + } const retryPtyId = connectResult?.id ?? diff --git a/src/renderer/src/components/terminal-pane/pty-transport-types.ts b/src/renderer/src/components/terminal-pane/pty-transport-types.ts index ef7d9974464..363ce4e97a8 100644 --- a/src/renderer/src/components/terminal-pane/pty-transport-types.ts +++ b/src/renderer/src/components/terminal-pane/pty-transport-types.ts @@ -50,6 +50,8 @@ export type PtyConnectResult = { /** The requested session exited while it had no primary pane handler. Its * buffered final data/exit were delivered, so callers must not fresh-spawn. */ exitedBeforeAttach?: boolean + /** Ownership could not be proven, so the caller must retain its rendered frame. */ + routingUnavailable?: true /** The provider adopted an existing session rather than creating a fresh one. * Startup commands may be ignored; recovery still requires separate ownership evidence. */ isReattach?: boolean diff --git a/src/renderer/src/components/terminal-pane/pty-transport.test.ts b/src/renderer/src/components/terminal-pane/pty-transport.test.ts index ea7d9d043e1..d0704d9f2fc 100644 --- a/src/renderer/src/components/terminal-pane/pty-transport.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-transport.test.ts @@ -189,6 +189,29 @@ describe('createIpcPtyTransport', () => { expect(write).not.toHaveBeenCalled() }) + it('returns an explicit retained-frame result when daemon routing is unavailable', async () => { + const { createIpcPtyTransport } = await import('./pty-transport') + const spawn = window.api.pty.spawn as unknown as ReturnType + spawn.mockRejectedValueOnce( + new Error( + "Error invoking remote method 'pty:spawn': daemon_session_routing_unavailable: owner unknown" + ) + ) + const onError = vi.fn() + + const result = await createIpcPtyTransport({}).connect({ + url: '', + sessionId: 'preserved-session', + callbacks: { onError } + }) + + expect(result).toEqual({ + id: 'preserved-session', + routingUnavailable: true + }) + expect(onError).toHaveBeenCalledWith('daemon_session_routing_unavailable: owner unknown') + }) + it('ignores a stale exit for a previous PTY after reconnecting the same transport', async () => { const { createIpcPtyTransport } = await import('./pty-transport') const spawn = window.api.pty.spawn as unknown as ReturnType diff --git a/src/renderer/src/components/terminal-pane/pty-transport.ts b/src/renderer/src/components/terminal-pane/pty-transport.ts index 6b777cbb21e..22da26d2f70 100644 --- a/src/renderer/src/components/terminal-pane/pty-transport.ts +++ b/src/renderer/src/components/terminal-pane/pty-transport.ts @@ -44,6 +44,7 @@ import { } from '../../../../shared/agent-status-osc' import { extractIpcErrorMessage } from '@/lib/ipc-error' import { isTuiAgent } from '../../../../shared/tui-agent-config' +import { isDaemonSessionRoutingUnavailable } from '../../../../shared/daemon-session-routing-error' // Re-export public API so existing consumers keep working. export { @@ -881,6 +882,13 @@ export function createIpcPtyTransport(opts: IpcPtyTransportOptions = {}): PtyTra return spawnResult.id } catch (err) { const msg = extractIpcErrorMessage(err, err instanceof Error ? err.message : String(err)) + if (options.sessionId && isDaemonSessionRoutingUnavailable(msg)) { + storedCallbacks.onError?.(msg) + return { + id: options.sessionId, + routingUnavailable: true + } satisfies PtyConnectResult + } if ( connectionId && options.sessionId && diff --git a/src/shared/daemon-session-routing-error.ts b/src/shared/daemon-session-routing-error.ts new file mode 100644 index 00000000000..900e3def247 --- /dev/null +++ b/src/shared/daemon-session-routing-error.ts @@ -0,0 +1,5 @@ +export const DAEMON_SESSION_ROUTING_UNAVAILABLE_MARKER = 'daemon_session_routing_unavailable' + +export function isDaemonSessionRoutingUnavailable(message: string): boolean { + return message.includes(DAEMON_SESSION_ROUTING_UNAVAILABLE_MARKER) +}