diff --git a/src/main/daemon/daemon-pty-router.test.ts b/src/main/daemon/daemon-pty-router.test.ts index 5b2ba936966..763fca2c012 100644 --- a/src/main/daemon/daemon-pty-router.test.ts +++ b/src/main/daemon/daemon-pty-router.test.ts @@ -1,7 +1,12 @@ import { describe, expect, it, vi } from 'vitest' import { DaemonPtyRouter } from './daemon-pty-router' import type { DaemonPtyAdapter } from './daemon-pty-adapter' -import type { PtyBackgroundStreamEvent, PtySpawnOptions, PtySpawnResult } from '../providers/types' +import type { + PtyBackgroundStreamEvent, + PtyProcessInfo, + PtySpawnOptions, + PtySpawnResult +} from '../providers/types' import { GIT_CREDENTIAL_GUARD_HOST_PROTOCOL_VERSION } from './types' type AdapterMock = DaemonPtyAdapter & { @@ -182,6 +187,31 @@ describe('DaemonPtyRouter', () => { expect(current.write).toHaveBeenCalledWith(fresh.id, 'new\n') }) + it('does not discover a session from an inventory response older than exit', async () => { + const current = createAdapter('current') + const legacy = createAdapter('legacy', ['exited-session']) + let resolveLegacy!: (sessions: PtyProcessInfo[]) => void + vi.mocked(legacy.listProcesses).mockReturnValueOnce( + new Promise((resolve) => { + resolveLegacy = resolve + }) + ) + const router = new DaemonPtyRouter({ current, legacy: [legacy] }) + + const discovery = router.discoverLegacySessions() + legacy.emitExit('exited-session', 0) + resolveLegacy([{ id: 'exited-session', cwd: '', title: 'stale legacy' }]) + await discovery + await router.spawn({ sessionId: 'exited-session', cols: 80, rows: 24 }) + + expect(current.spawn).toHaveBeenCalledWith({ + sessionId: 'exited-session', + cols: 80, + rows: 24 + }) + expect(legacy.spawn).not.toHaveBeenCalled() + }) + it('routes background hints and authoritative snapshots to the session owner', async () => { const current = createAdapter('current') const legacy = createAdapter('legacy', ['legacy-session']) @@ -289,7 +319,61 @@ describe('DaemonPtyRouter', () => { expect(legacy.write).toHaveBeenCalledWith('legacy-session', 'after recovery\n') }) - it('rejects destructive teardown when recovered daemons report a duplicate session id', async () => { + it('does not restore a route from an inventory response older than exit', async () => { + const current = createAdapter('current') + const legacy = createAdapter('legacy', ['exited-session']) + const router = new DaemonPtyRouter({ current, legacy: [legacy] }) + await router.discoverLegacySessions() + let resolveLegacy!: (sessions: PtyProcessInfo[]) => void + vi.mocked(legacy.listProcesses).mockReturnValueOnce( + new Promise((resolve) => { + resolveLegacy = resolve + }) + ) + + const listing = router.listProcesses() + legacy.emitExit('exited-session', 0) + resolveLegacy([{ id: 'exited-session', cwd: '', title: 'stale legacy' }]) + await listing + await router.spawn({ sessionId: 'exited-session', cols: 80, rows: 24 }) + + expect(current.spawn).toHaveBeenCalledWith({ + sessionId: 'exited-session', + cols: 80, + rows: 24 + }) + expect(legacy.spawn).not.toHaveBeenCalled() + }) + + it('keeps a later overlapping inventory result over an earlier response', async () => { + const current = createAdapter('current', ['migrated-session']) + const legacy = createAdapter('legacy') + let resolveOldCurrent!: (sessions: PtyProcessInfo[]) => void + let resolveOldLegacy!: (sessions: PtyProcessInfo[]) => void + vi.mocked(current.listProcesses).mockReturnValueOnce( + new Promise((resolve) => { + resolveOldCurrent = resolve + }) + ) + vi.mocked(legacy.listProcesses).mockReturnValueOnce( + new Promise((resolve) => { + resolveOldLegacy = resolve + }) + ) + const router = new DaemonPtyRouter({ current, legacy: [legacy] }) + + const olderListing = router.listProcesses() + await router.listProcesses() + resolveOldCurrent([]) + resolveOldLegacy([{ id: 'migrated-session', cwd: '', title: 'stale legacy' }]) + await olderListing + router.write('migrated-session', 'new owner\n') + + expect(current.write).toHaveBeenCalledWith('migrated-session', 'new owner\n') + expect(legacy.write).not.toHaveBeenCalled() + }) + + it('rejects shutdown when recovered daemons report a duplicate session id', async () => { const current = createAdapter('current', ['duplicate-session']) const legacy = createAdapter('legacy', ['duplicate-session']) vi.mocked(legacy.listProcesses).mockRejectedValueOnce(new Error('legacy unavailable')) @@ -301,8 +385,7 @@ describe('DaemonPtyRouter', () => { ) await expect( router.shutdown('duplicate-session', { - immediate: true, - deadlineMs: Date.now() + 5_000 + immediate: true }) ).rejects.toThrow('Ambiguous PTY session ownership') expect(current.shutdown).not.toHaveBeenCalled() @@ -320,8 +403,7 @@ describe('DaemonPtyRouter', () => { await expect( router.shutdown('duplicate-session', { - immediate: true, - deadlineMs: Date.now() + 5_000 + immediate: true }) ).rejects.toThrow('Ambiguous PTY session ownership') expect(firstLegacy.shutdown).not.toHaveBeenCalled() diff --git a/src/main/daemon/daemon-pty-router.ts b/src/main/daemon/daemon-pty-router.ts index b6797b39a51..b3fe9307927 100644 --- a/src/main/daemon/daemon-pty-router.ts +++ b/src/main/daemon/daemon-pty-router.ts @@ -46,13 +46,16 @@ export class DaemonPtyRouter implements IPtyProvider { async discoverLegacySessions(): Promise { for (const adapter of this.legacy) { + const refresh = this.sessionRouting.beginInventoryRefresh() try { const sessions = await adapter.listProcesses() for (const session of sessions) { - this.sessionRouting.add(session.id, adapter) + this.sessionRouting.addInventorySession(session.id, adapter, refresh) } } catch (error) { console.warn('[daemon] Failed to discover legacy daemon sessions', error) + } finally { + this.sessionRouting.finishInventoryRefresh(refresh) } } } @@ -106,7 +109,7 @@ export class DaemonPtyRouter implements IPtyProvider { id: string, opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number } ): Promise { - if (opts.deadlineMs !== undefined && this.sessionRouting.isAmbiguous(id)) { + if (this.sessionRouting.isAmbiguous(id)) { throw this.ambiguousOwnershipError([id]) } const adapter = this.adapterFor(id) @@ -185,8 +188,15 @@ export class DaemonPtyRouter implements IPtyProvider { // Why: runtime exact-stop/liveness flows must fail closed if any adapter // cannot provide a trustworthy process list. const adapters = this.allAdapters() - const results = await Promise.all(adapters.map((adapter) => adapter.listProcesses(opts))) - const ambiguousIds = this.sessionRouting.refreshLive(adapters, results) + const refresh = this.sessionRouting.beginInventoryRefresh() + let results: PtyProcessInfo[][] + try { + results = await Promise.all(adapters.map((adapter) => adapter.listProcesses(opts))) + } catch (error) { + this.sessionRouting.finishInventoryRefresh(refresh) + throw error + } + const ambiguousIds = this.sessionRouting.refreshLive(adapters, results, refresh) if (ambiguousIds.length > 0) { throw this.ambiguousOwnershipError(ambiguousIds) } diff --git a/src/main/daemon/daemon-pty-session-routing.ts b/src/main/daemon/daemon-pty-session-routing.ts index c9b082578a1..a95526813c2 100644 --- a/src/main/daemon/daemon-pty-session-routing.ts +++ b/src/main/daemon/daemon-pty-session-routing.ts @@ -1,9 +1,21 @@ import type { DaemonPtyAdapter } from './daemon-pty-adapter' import type { PtyProcessInfo } from '../providers/types' +export type DaemonPtyInventoryRefresh = { + readonly revision: number +} + +type SessionMutation = { + revision: number + refresh?: DaemonPtyInventoryRefresh +} + export class DaemonPtySessionRouting { private adapters = new Map() private ambiguousAdapters = new Map>() + private revision = 0 + private sessionMutations = new Map() + private activeInventoryRefreshes = new Set() get(sessionId: string): DaemonPtyAdapter | undefined { return this.adapters.get(sessionId) @@ -13,6 +25,56 @@ export class DaemonPtySessionRouting { return this.ambiguousAdapters.has(sessionId) } + beginInventoryRefresh(): DaemonPtyInventoryRefresh { + // Why: a later overlapping snapshot may supersede an earlier refresh, + // while an exit/add event must supersede every snapshot already in flight. + this.revision += 1 + const refresh = { revision: this.revision } + this.activeInventoryRefreshes.add(refresh) + return refresh + } + + finishInventoryRefresh(refresh: DaemonPtyInventoryRefresh): void { + this.activeInventoryRefreshes.delete(refresh) + if (this.activeInventoryRefreshes.size === 0) { + this.sessionMutations.clear() + return + } + const oldestRevision = Math.min( + ...[...this.activeInventoryRefreshes].map((active) => active.revision) + ) + for (const [id, mutation] of this.sessionMutations) { + if (mutation.revision <= oldestRevision) { + this.sessionMutations.delete(id) + } + } + } + + applyInventoryMutation( + sessionId: string, + refresh: DaemonPtyInventoryRefresh, + mutation: () => void + ): boolean { + if (!this.isInventoryResultCurrent(sessionId, refresh)) { + return false + } + mutation() + this.recordMutation(sessionId, refresh) + return true + } + + addInventorySession( + sessionId: string, + adapter: DaemonPtyAdapter, + refresh: DaemonPtyInventoryRefresh + ): void { + this.applyInventoryMutation(sessionId, refresh, () => this.addRoute(sessionId, adapter)) + } + + recordExternalMutation(sessionId: string): void { + this.recordMutation(sessionId) + } + idsFor(adapter: DaemonPtyAdapter): string[] { const ids = [...this.adapters].filter(([, owner]) => owner === adapter).map(([id]) => id) for (const [id, owners] of this.ambiguousAdapters) { @@ -24,6 +86,11 @@ export class DaemonPtySessionRouting { } add(sessionId: string, adapter: DaemonPtyAdapter): void { + this.recordMutation(sessionId) + this.addRoute(sessionId, adapter) + } + + private addRoute(sessionId: string, adapter: DaemonPtyAdapter): void { const ambiguous = this.ambiguousAdapters.get(sessionId) if (ambiguous) { ambiguous.add(adapter) @@ -39,6 +106,7 @@ export class DaemonPtySessionRouting { } remove(sessionId: string, adapter: DaemonPtyAdapter): void { + this.recordMutation(sessionId) const ambiguous = this.ambiguousAdapters.get(sessionId) if (ambiguous) { ambiguous.delete(adapter) @@ -56,7 +124,11 @@ export class DaemonPtySessionRouting { } } - refreshLive(adapters: DaemonPtyAdapter[], results: PtyProcessInfo[][]): string[] { + refreshLive( + adapters: DaemonPtyAdapter[], + results: PtyProcessInfo[][], + refresh: DaemonPtyInventoryRefresh + ): string[] { const owners = new Map>() for (let index = 0; index < adapters.length; index += 1) { for (const session of results[index]) { @@ -69,21 +141,46 @@ export class DaemonPtySessionRouting { // Why: unique absent sessions may be sleeping on legacy history, but a // successful all-daemon inventory can retire stale live ambiguity. for (const id of this.ambiguousAdapters.keys()) { - if (!owners.has(id)) { - this.ambiguousAdapters.delete(id) + if (!owners.has(id) && this.isInventoryResultCurrent(id, refresh)) { + this.applyInventoryMutation(id, refresh, () => this.ambiguousAdapters.delete(id)) } } const ambiguousIds: string[] = [] for (const [id, sessionOwners] of owners) { + if (!this.isInventoryResultCurrent(id, refresh)) { + continue + } if (sessionOwners.size === 1) { - this.ambiguousAdapters.delete(id) - this.adapters.set(id, sessionOwners.values().next().value as DaemonPtyAdapter) + this.applyInventoryMutation(id, refresh, () => { + this.ambiguousAdapters.delete(id) + this.adapters.set(id, sessionOwners.values().next().value as DaemonPtyAdapter) + }) } else { - this.adapters.delete(id) - this.ambiguousAdapters.set(id, sessionOwners) + this.applyInventoryMutation(id, refresh, () => { + this.adapters.delete(id) + this.ambiguousAdapters.set(id, sessionOwners) + }) ambiguousIds.push(id) } } + this.finishInventoryRefresh(refresh) return ambiguousIds.sort() } + + private isInventoryResultCurrent(sessionId: string, refresh: DaemonPtyInventoryRefresh): boolean { + const mutation = this.sessionMutations.get(sessionId) + return ( + mutation === undefined || + mutation.revision <= refresh.revision || + mutation.refresh === refresh || + (mutation.refresh !== undefined && mutation.refresh.revision < refresh.revision) + ) + } + + private recordMutation(sessionId: string, refresh?: DaemonPtyInventoryRefresh): void { + this.revision += 1 + if (this.activeInventoryRefreshes.size > 0) { + this.sessionMutations.set(sessionId, { revision: this.revision, refresh }) + } + } } diff --git a/src/main/daemon/degraded-daemon-pty-provider.test.ts b/src/main/daemon/degraded-daemon-pty-provider.test.ts index e443ea5b5e4..aed861e5855 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.test.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.test.ts @@ -1,7 +1,12 @@ import { describe, expect, it, vi } from 'vitest' import { DegradedDaemonPtyProvider } from './degraded-daemon-pty-provider' import type { DaemonPtyAdapter } from './daemon-pty-adapter' -import type { IPtyProvider, PtySpawnOptions, PtySpawnResult } from '../providers/types' +import type { + IPtyProvider, + PtyProcessInfo, + PtySpawnOptions, + PtySpawnResult +} from '../providers/types' type ProviderMock = IPtyProvider & { emitData: (id: string, data: string, sequenceChars?: number) => void @@ -190,7 +195,33 @@ describe('DegradedDaemonPtyProvider', () => { warn.mockRestore() }) - it('fails deadline teardown closed for duplicate ownership across degraded daemons', async () => { + it('does not restore a daemon route from an inventory response older than exit', async () => { + const current = createDaemonAdapter('daemon', ['exited-session']) + const fallback = createProvider('fallback') + const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) + await provider.discoverDaemonSessions() + let resolveDaemon!: (sessions: PtyProcessInfo[]) => void + vi.mocked(current.listProcesses).mockReturnValueOnce( + new Promise((resolve) => { + resolveDaemon = resolve + }) + ) + + const listing = provider.listProcesses() + current.emitExit('exited-session', 0) + resolveDaemon([{ id: 'exited-session', cwd: '', title: 'stale daemon' }]) + await listing + await provider.spawn({ sessionId: 'exited-session', cols: 80, rows: 24 }) + + expect(fallback.spawn).toHaveBeenCalledWith({ + sessionId: 'exited-session', + cols: 80, + rows: 24 + }) + expect(current.spawn).not.toHaveBeenCalled() + }) + + it('fails shutdown closed for duplicate ownership across degraded daemons', async () => { const current = createDaemonAdapter('current', ['duplicate-session']) const legacy = createDaemonAdapter('legacy', ['duplicate-session']) const fallback = createProvider('fallback') @@ -200,8 +231,7 @@ describe('DegradedDaemonPtyProvider', () => { await expect( provider.shutdown('duplicate-session', { - immediate: true, - deadlineMs: Date.now() + 1000 + immediate: true }) ).rejects.toThrow( 'Ambiguous PTY session ownership across degraded providers: duplicate-session' @@ -229,8 +259,7 @@ describe('DegradedDaemonPtyProvider', () => { ) await expect( provider.shutdown('duplicate-session', { - immediate: true, - deadlineMs: Date.now() + 1000 + immediate: true }) ).rejects.toThrow( 'Ambiguous PTY session ownership across degraded providers: duplicate-session' diff --git a/src/main/daemon/degraded-daemon-pty-provider.ts b/src/main/daemon/degraded-daemon-pty-provider.ts index 8a8ae448f03..5c0e74d0e1e 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.ts @@ -58,13 +58,16 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { async discoverDaemonSessions(): Promise { for (const adapter of this.allDaemonAdapters()) { + const refresh = this.sessionRouting.beginInventoryRefresh() try { const sessions = await adapter.listProcesses() for (const session of sessions) { - this.sessionRouting.add(session.id, adapter) + this.sessionRouting.addDaemonInventorySession(session.id, adapter, refresh) } } catch (error) { console.warn('[daemon] Failed to discover degraded daemon sessions', error) + } finally { + this.sessionRouting.finishInventoryRefresh(refresh) } } } @@ -109,7 +112,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { id: string, opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number } ): Promise { - if (opts.deadlineMs !== undefined && this.sessionRouting.isAmbiguous(id)) { + if (this.sessionRouting.isAmbiguous(id)) { throw this.ambiguousOwnershipError([id]) } const target = this.providerFor(id) @@ -178,8 +181,15 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { async listProcesses(opts?: { deadlineMs?: number }): Promise { const providers = this.allProviders() - const results = await Promise.all(providers.map((provider) => provider.listProcesses(opts))) - const ambiguousIds = this.sessionRouting.refreshSessions(results[0], results.slice(1)) + const refresh = this.sessionRouting.beginInventoryRefresh() + let results: PtyProcessInfo[][] + try { + results = await Promise.all(providers.map((provider) => provider.listProcesses(opts))) + } catch (error) { + this.sessionRouting.finishInventoryRefresh(refresh) + throw error + } + const ambiguousIds = this.sessionRouting.refreshSessions(results[0], results.slice(1), refresh) if (ambiguousIds.length > 0) { throw this.ambiguousOwnershipError(ambiguousIds) } @@ -246,11 +256,11 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { } ackColdRestore(sessionId: string): void { - this.daemonAdapterFor(sessionId)?.ackColdRestore(sessionId) + this.sessionRouting.daemonFor(sessionId)?.ackColdRestore(sessionId) } clearTombstone(sessionId: string): void { - this.daemonAdapterFor(sessionId)?.clearTombstone(sessionId) + this.sessionRouting.daemonFor(sessionId)?.clearTombstone(sessionId) } async reconcileOnStartup(validWorktreeIds: Set): Promise<{ @@ -327,10 +337,6 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { return this.sessionRouting.providerFor(sessionId, this.current) } - private daemonAdapterFor(sessionId: string): DaemonPtyAdapter | null { - return this.sessionRouting.daemonFor(sessionId) - } - private ambiguousOwnershipError(sessionIds: string[]): Error { return new Error( `Ambiguous PTY session ownership across degraded providers: ${sessionIds.join(', ')}` diff --git a/src/main/daemon/degraded-daemon-session-routing.ts b/src/main/daemon/degraded-daemon-session-routing.ts index 6e4669eac7f..d6d2b26c9fc 100644 --- a/src/main/daemon/degraded-daemon-session-routing.ts +++ b/src/main/daemon/degraded-daemon-session-routing.ts @@ -1,5 +1,8 @@ import type { DaemonPtyAdapter } from './daemon-pty-adapter' -import { DaemonPtySessionRouting } from './daemon-pty-session-routing' +import { + DaemonPtySessionRouting, + type DaemonPtyInventoryRefresh +} from './daemon-pty-session-routing' import type { IPtyProvider, PtyProcessInfo } from '../providers/types' export type DegradedManagedPtyProvider = IPtyProvider & { @@ -30,11 +33,28 @@ export class DegradedDaemonSessionRouting { ) } + beginInventoryRefresh(): DaemonPtyInventoryRefresh { + return this.daemonRouting.beginInventoryRefresh() + } + + finishInventoryRefresh(refresh: DaemonPtyInventoryRefresh): void { + this.daemonRouting.finishInventoryRefresh(refresh) + } + + addDaemonInventorySession( + sessionId: string, + adapter: DaemonPtyAdapter, + refresh: DaemonPtyInventoryRefresh + ): void { + this.daemonRouting.addInventorySession(sessionId, adapter, refresh) + } + add(sessionId: string, provider: DegradedManagedPtyProvider): void { if (this.isDaemonAdapter(provider)) { this.daemonRouting.add(sessionId, provider) return } + this.daemonRouting.recordExternalMutation(sessionId) this.providers.set(sessionId, provider) } @@ -43,16 +63,25 @@ export class DegradedDaemonSessionRouting { this.daemonRouting.remove(sessionId, provider) return } + this.daemonRouting.recordExternalMutation(sessionId) if (this.providers.get(sessionId) === provider) { this.providers.delete(sessionId) } } - refreshSessions(fallbackSessions: PtyProcessInfo[], daemonResults: PtyProcessInfo[][]): string[] { + refreshSessions( + fallbackSessions: PtyProcessInfo[], + daemonResults: PtyProcessInfo[][], + refresh: DaemonPtyInventoryRefresh + ): string[] { for (const session of fallbackSessions) { - this.providers.set(session.id, this.fallback) + this.daemonRouting.applyInventoryMutation(session.id, refresh, () => { + this.providers.set(session.id, this.fallback) + }) } - const ambiguousIds = new Set(this.daemonRouting.refreshLive(this.daemonAdapters, daemonResults)) + const ambiguousIds = new Set( + this.daemonRouting.refreshLive(this.daemonAdapters, daemonResults, refresh) + ) for (const sessionId of this.providers.keys()) { if ( this.daemonRouting.get(sessionId) !== undefined ||