From 15d604d8599f6da924fbf39b7eae64ce46343a4a Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 31 Jul 2026 00:20:47 -0700 Subject: [PATCH] fix(daemon): validate recovery routing evidence --- src/main/daemon/daemon-health.ts | 8 +- .../daemon-pty-adapter-subscription-fanout.ts | 14 +- ...aemon-pty-router-routing-safety-fixture.ts | 17 +- .../daemon-pty-router-session-routing.ts | 18 +- src/main/daemon/daemon-pty-router.test.ts | 22 --- src/main/daemon/daemon-pty-router.ts | 1 + ...emon-pty-write-unavailable-routing.test.ts | 50 +++++ .../daemon-spawn-route-validation.test.ts | 77 ++++++++ .../daemon/daemon-spawn-route-validation.ts | 18 ++ .../daemon-stale-pidfile-cleanup.test.ts | 38 ++++ .../degraded-daemon-pty-provider-fixture.ts | 140 ++++++++++++++ .../degraded-daemon-pty-provider.test.ts | 179 ++++-------------- .../daemon/degraded-daemon-pty-provider.ts | 8 +- .../daemon/degraded-daemon-session-routing.ts | 15 +- .../degraded-fallback-session-routes.ts | 22 ++- 15 files changed, 445 insertions(+), 182 deletions(-) create mode 100644 src/main/daemon/daemon-pty-write-unavailable-routing.test.ts create mode 100644 src/main/daemon/daemon-spawn-route-validation.test.ts create mode 100644 src/main/daemon/daemon-spawn-route-validation.ts create mode 100644 src/main/daemon/degraded-daemon-pty-provider-fixture.ts diff --git a/src/main/daemon/daemon-health.ts b/src/main/daemon/daemon-health.ts index 5ff89c8102b..5bafa91d2fa 100644 --- a/src/main/daemon/daemon-health.ts +++ b/src/main/daemon/daemon-health.ts @@ -568,10 +568,10 @@ async function matchDaemonProcess( return probeProcessExistence(pid) === 'absent' ? 'stale' : 'unknown' } } - return commandLineMatchesDaemon(commandLine, socketPath, tokenPath) && - startTimeMatches(pid, startedAtMs) - ? 'match' - : 'stale' + if (!commandLineMatchesDaemon(commandLine, socketPath, tokenPath)) { + return 'stale' + } + return startTimeMatches(pid, startedAtMs) ? 'match' : 'unknown' } async function isDaemonProcess( diff --git a/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts b/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts index cac0a809f2e..4845b5f8046 100644 --- a/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts +++ b/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts @@ -11,6 +11,10 @@ export class DaemonPtyAdapterSubscriptionFanout { constructor( private readonly adapters: readonly DaemonPtyAdapter[], private readonly shouldForwardStreamEvent: (adapter: DaemonPtyAdapter, id: string) => boolean, + private readonly shouldForwardWriteUnavailable: ( + adapter: DaemonPtyAdapter, + id: string + ) => boolean, onAdapterExit: (adapter: DaemonPtyAdapter, id: string) => boolean ) { for (const adapter of adapters) { @@ -60,7 +64,15 @@ export class DaemonPtyAdapterSubscriptionFanout { // Why: main subscribes on the routed provider, so without this the dead-endpoint // fan-out never reaches the renderer and only the written pane recovers (STA-2373). onWriteUnavailable(callback: (payload: { id: string }) => void): () => void { - return combineUnsubscribes(this.adapters.map((adapter) => adapter.onWriteUnavailable(callback))) + return combineUnsubscribes( + this.adapters.map((adapter) => + adapter.onWriteUnavailable((payload) => { + if (this.shouldForwardWriteUnavailable(adapter, payload.id)) { + callback(payload) + } + }) + ) + ) } onReplay(_callback: (payload: { id: string; data: string }) => void): () => void { diff --git a/src/main/daemon/daemon-pty-router-routing-safety-fixture.ts b/src/main/daemon/daemon-pty-router-routing-safety-fixture.ts index 382f88b7026..23f605b415f 100644 --- a/src/main/daemon/daemon-pty-router-routing-safety-fixture.ts +++ b/src/main/daemon/daemon-pty-router-routing-safety-fixture.ts @@ -8,6 +8,7 @@ export type AdapterHarness = { emitData: (sessionId: string, data: string) => void emitIdentityChange: (previous: DaemonEndpointIdentity, current: DaemonEndpointIdentity) => void emitExit: (sessionId: string, code: number) => void + emitWriteUnavailable: (sessionId: string) => void setIdentity: (identity: DaemonEndpointIdentity) => void } @@ -31,6 +32,7 @@ export function createAdapter( }) => void)[] = [] const dataListeners: ((payload: { id: string; data: string }) => void)[] = [] const exitListeners: ((payload: { id: string; code: number }) => void)[] = [] + const writeUnavailableListeners: ((payload: { id: string }) => void)[] = [] const adapter = { protocolVersion: label === 'current' ? 30 : 29, getLastAuthenticatedDaemonIdentity: vi.fn(() => ({ ...daemonIdentity })), @@ -74,7 +76,15 @@ export function createAdapter( return () => {} }), onBackgroundStreamEvent: vi.fn(() => () => {}), - onWriteUnavailable: vi.fn(() => () => {}), + onWriteUnavailable: vi.fn((listener: (payload: { id: string }) => void) => { + writeUnavailableListeners.push(listener) + return () => { + const index = writeUnavailableListeners.indexOf(listener) + if (index !== -1) { + writeUnavailableListeners.splice(index, 1) + } + } + }), dispose: vi.fn(), disconnectOnly: vi.fn(async () => {}) } as unknown as DaemonPtyAdapter @@ -98,6 +108,11 @@ export function createAdapter( for (const listener of exitListeners) { listener({ id: sessionId, code }) } + }, + emitWriteUnavailable: (sessionId) => { + for (const listener of writeUnavailableListeners) { + listener({ id: sessionId }) + } } } } diff --git a/src/main/daemon/daemon-pty-router-session-routing.ts b/src/main/daemon/daemon-pty-router-session-routing.ts index 93acee652b0..01d8b1ad127 100644 --- a/src/main/daemon/daemon-pty-router-session-routing.ts +++ b/src/main/daemon/daemon-pty-router-session-routing.ts @@ -3,6 +3,7 @@ import type { DaemonPtyAdapter } from './daemon-pty-adapter' import type { DaemonEndpointIdentity } from './daemon-hello-protocol' import { DaemonSessionRouteTable } from './daemon-session-route-table' import { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowledgement-routes' +import { recordValidatedDaemonSpawn } from './daemon-spawn-route-validation' export class DaemonPtyRouterSessionRouting { private readonly routes: DaemonSessionRouteTable @@ -64,11 +65,15 @@ export class DaemonPtyRouterSessionRouting { if (result.exitedBeforeSpawnReply) { return } - if (authoritativeIntent && opts.sessionId === result.id) { - this.routes.recordFreshOwned(result.id, target) + if (authoritativeIntent) { + if (opts.sessionId === result.id) { + this.routes.recordFreshOwned(result.id, target) + } else { + this.routes.recordOwned(result.id, target) + } return } - this.routes.recordOwned(result.id, target) + recordValidatedDaemonSpawn(this.routes, result.id, opts.sessionId, target) } beginSpawn(opts: PtySpawnOptions, target: DaemonPtyAdapter): boolean { @@ -212,6 +217,13 @@ export class DaemonPtyRouterSessionRouting { return this.routes.recordDataOwner(sessionId, owner) } + shouldForwardWriteUnavailable(sessionId: string, owner: DaemonPtyAdapter): boolean { + const authoritativeSpawn = this.authoritativeSpawnsInFlight.get(sessionId) + return authoritativeSpawn + ? authoritativeSpawn.owner === owner + : this.routes.getOwned(sessionId) === owner + } + recordExit(sessionId: string, owner: DaemonPtyAdapter): boolean { const authoritativeSpawn = this.authoritativeSpawnsInFlight.get(sessionId) if (authoritativeSpawn && authoritativeSpawn.owner !== owner) { diff --git a/src/main/daemon/daemon-pty-router.test.ts b/src/main/daemon/daemon-pty-router.test.ts index 936eaf9278d..307f4f95e3d 100644 --- a/src/main/daemon/daemon-pty-router.test.ts +++ b/src/main/daemon/daemon-pty-router.test.ts @@ -184,28 +184,6 @@ function createAdapter( } as unknown as AdapterMock } -it('forwards dead-endpoint write-unavailable signals from every routed adapter', () => { - // Why revert-sensitive: main subscribes on the ROUTED provider, so if the router - // does not forward this the STA-2373 fan-out never reaches the renderer and only - // the written pane recovers — siblings stay frozen. The router is the live - // localProvider whenever a legacy daemon socket exists (protocol bump mid-session). - const current = createAdapter('current') - const legacy = createAdapter('legacy') - const router = new DaemonPtyRouter({ current, legacy: [legacy] }) - const recovered: string[] = [] - - const unsubscribe = router.onWriteUnavailable(({ id }) => recovered.push(id)) - current.triggerWriteUnavailable('current-pane') - legacy.triggerWriteUnavailable('legacy-pane') - - expect(recovered).toEqual(['current-pane', 'legacy-pane']) - - unsubscribe() - current.triggerWriteUnavailable('after-unsubscribe') - legacy.triggerWriteUnavailable('after-unsubscribe') - expect(recovered).toEqual(['current-pane', 'legacy-pane']) -}) - it('rejects completion inspection when no daemon owns the session', async () => { const router = new DaemonPtyRouter({ current: createAdapter('current'), diff --git a/src/main/daemon/daemon-pty-router.ts b/src/main/daemon/daemon-pty-router.ts index 46d12268218..3d0428886b0 100644 --- a/src/main/daemon/daemon-pty-router.ts +++ b/src/main/daemon/daemon-pty-router.ts @@ -27,6 +27,7 @@ export class DaemonPtyRouter implements IPtyProvider { this.subscriptions = new DaemonPtyAdapterSubscriptionFanout( this.allAdapters(), (adapter, id) => this.routing.shouldForwardStreamEvent(id, adapter), + (adapter, id) => this.routing.shouldForwardWriteUnavailable(id, adapter), (adapter, id) => this.routing.recordExit(id, adapter) ) } diff --git a/src/main/daemon/daemon-pty-write-unavailable-routing.test.ts b/src/main/daemon/daemon-pty-write-unavailable-routing.test.ts new file mode 100644 index 00000000000..6105c4c24c5 --- /dev/null +++ b/src/main/daemon/daemon-pty-write-unavailable-routing.test.ts @@ -0,0 +1,50 @@ +import { expect, it } from 'vitest' +import { DaemonPtyRouter } from './daemon-pty-router' +import { createAdapter, identity } from './daemon-pty-router-routing-safety-fixture' + +it('forwards write-unavailable signals from each exact routed owner', () => { + const current = createAdapter('current') + const legacy = createAdapter('legacy') + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + const recovered: string[] = [] + current.emitData('current-pane', 'route evidence') + legacy.emitData('legacy-pane', 'route evidence') + const unsubscribe = router.onWriteUnavailable(({ id }) => recovered.push(id)) + + current.emitWriteUnavailable('current-pane') + legacy.emitWriteUnavailable('legacy-pane') + + expect(recovered).toEqual(['current-pane', 'legacy-pane']) + unsubscribe() + current.emitWriteUnavailable('current-pane') + expect(recovered).toEqual(['current-pane', 'legacy-pane']) +}) + +it('withholds write-unavailable signals from foreign and stale adapter routes', () => { + const current = createAdapter('current') + const legacy = createAdapter('legacy') + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + const recovered: string[] = [] + const previous = identity('current', 20) + const replacement = identity('current', 21) + current.emitData('shared-pane', 'route evidence') + router.onWriteUnavailable(({ id }) => recovered.push(id)) + + legacy.emitWriteUnavailable('shared-pane') + current.emitWriteUnavailable('shared-pane') + current.emitData('ambiguous-pane', 'current collision evidence') + legacy.emitData('ambiguous-pane', 'legacy collision evidence') + current.emitWriteUnavailable('ambiguous-pane') + legacy.emitWriteUnavailable('ambiguous-pane') + current.setIdentity(replacement) + current.emitIdentityChange(previous, replacement) + current.emitWriteUnavailable('shared-pane') + + expect(recovered).toEqual(['shared-pane']) +}) diff --git a/src/main/daemon/daemon-spawn-route-validation.test.ts b/src/main/daemon/daemon-spawn-route-validation.test.ts new file mode 100644 index 00000000000..a597e409af4 --- /dev/null +++ b/src/main/daemon/daemon-spawn-route-validation.test.ts @@ -0,0 +1,77 @@ +import { expect, it, vi } from 'vitest' +import type { PtySpawnResult } from '../providers/types' +import type { DaemonIdentityChangeEvent } from './daemon-pty-adapter' +import { DaemonPtyRouter } from './daemon-pty-router' +import { createAdapter, identity } from './daemon-pty-router-routing-safety-fixture' +import { DegradedDaemonPtyProvider } from './degraded-daemon-pty-provider' +import { createDaemonAdapter, createProvider } from './degraded-daemon-pty-provider-fixture' + +it('rejects an existing-session spawn after its selected daemon is replaced', async () => { + const sessionId = 'replaced-during-spawn' + const legacy = createAdapter('legacy', [sessionId]) + const previous = identity('legacy', 10) + const replacement = identity('legacy', 11) + let resolveSpawn: ((result: { id: string }) => void) | undefined + let markSpawnStarted: (() => void) | undefined + const spawnStarted = new Promise((resolve) => { + markSpawnStarted = resolve + }) + vi.mocked(legacy.adapter.spawn).mockImplementationOnce( + () => + new Promise((resolve) => { + markSpawnStarted?.() + resolveSpawn = resolve + }) + ) + const router = new DaemonPtyRouter({ + current: createAdapter('current').adapter, + legacy: [legacy.adapter] + }) + await router.discoverLegacySessions() + + const spawning = router.spawn({ sessionId, cols: 80, rows: 24 }) + await spawnStarted + legacy.setIdentity(replacement) + legacy.emitIdentityChange(previous, replacement) + resolveSpawn?.({ id: sessionId }) + + await expect(spawning).rejects.toThrow('daemon_session_routing_unavailable') + expect(router.getSessionRouteState(sessionId)).toBe('unavailable') +}) + +it('rejects a degraded existing spawn after its selected daemon is replaced', async () => { + const sessionId = 'replaced-daemon-spawn' + const previous = { pid: 1, startedAtMs: 1, launchNonce: 'previous' } + const replacement = { pid: 2, startedAtMs: 2, launchNonce: 'replacement' } + const current = createDaemonAdapter('daemon', [sessionId]) + const fallback = createProvider('fallback') + let daemonIdentity = previous + let notifyIdentityChange!: (event: DaemonIdentityChangeEvent) => void + vi.mocked(current.getLastAuthenticatedDaemonIdentity).mockImplementation(() => daemonIdentity) + vi.mocked(current.onDaemonIdentityChanged).mockImplementation((listener) => { + notifyIdentityChange = listener + return () => {} + }) + let resolveSpawn!: (result: PtySpawnResult) => void + let markSpawnStarted!: () => void + const spawnStarted = new Promise((resolve) => { + markSpawnStarted = resolve + }) + vi.mocked(current.spawn).mockImplementationOnce( + () => + new Promise((resolve) => { + markSpawnStarted() + resolveSpawn = resolve + }) + ) + const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) + await provider.discoverDaemonSessions() + + const spawning = provider.spawn({ sessionId, cols: 80, rows: 24 }) + await spawnStarted + daemonIdentity = replacement + notifyIdentityChange({ previous, current: replacement }) + resolveSpawn({ id: sessionId }) + + await expect(spawning).rejects.toThrow('daemon_session_routing_unavailable') +}) diff --git a/src/main/daemon/daemon-spawn-route-validation.ts b/src/main/daemon/daemon-spawn-route-validation.ts new file mode 100644 index 00000000000..27109268dd0 --- /dev/null +++ b/src/main/daemon/daemon-spawn-route-validation.ts @@ -0,0 +1,18 @@ +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { DaemonSessionUnavailableError } from './daemon-session-route' +import type { DaemonSessionRouteTable } from './daemon-session-route-table' + +export function recordValidatedDaemonSpawn( + routes: DaemonSessionRouteTable, + resultSessionId: string, + requestedSessionId: string | undefined, + owner: DaemonPtyAdapter +): void { + routes.recordOwned(resultSessionId, owner) + if ( + routes.getOwned(resultSessionId) !== owner || + (requestedSessionId !== undefined && routes.getOwned(requestedSessionId) !== owner) + ) { + throw new DaemonSessionUnavailableError(requestedSessionId ?? resultSessionId) + } +} diff --git a/src/main/daemon/daemon-stale-pidfile-cleanup.test.ts b/src/main/daemon/daemon-stale-pidfile-cleanup.test.ts index 87ef7a7e168..2f223543783 100644 --- a/src/main/daemon/daemon-stale-pidfile-cleanup.test.ts +++ b/src/main/daemon/daemon-stale-pidfile-cleanup.test.ts @@ -109,6 +109,44 @@ describe('stale daemon pidfile cleanup', () => { } }) + it('preserves matching POSIX command evidence when the start time is inconclusive', async () => { + if (process.platform === 'win32') { + return + } + const child = spawn( + process.execPath, + [ + '-e', + "process.send?.('ready');setInterval(()=>{},1000)", + 'daemon-entry', + socketPath, + tokenPath + ], + { stdio: ['ignore', 'ignore', 'ignore', 'ipc'] } + ) + await new Promise((resolve) => { + child.on('message', (message) => { + if (message === 'ready') { + resolve() + } + }) + }) + const record = serializeDaemonPidFile({ + pid: child.pid!, + startedAtMs: Date.now() - 5_000, + launchNonce: 'slow-bootstrap' + }) + writeFileSync(pidPath, record, { mode: 0o600 }) + + try { + await expect(killStaleDaemon(dir, socketPath, tokenPath)).resolves.toBe(false) + expect(readFileSync(pidPath, 'utf8')).toBe(record) + expect(child.exitCode).toBeNull() + } finally { + child.kill('SIGKILL') + } + }) + it('preserves the exact record when process access is rejected', async () => { const record = serializeDaemonPidFile({ pid: process.pid, diff --git a/src/main/daemon/degraded-daemon-pty-provider-fixture.ts b/src/main/daemon/degraded-daemon-pty-provider-fixture.ts new file mode 100644 index 00000000000..fb3676fa72a --- /dev/null +++ b/src/main/daemon/degraded-daemon-pty-provider-fixture.ts @@ -0,0 +1,140 @@ +import { vi } from 'vitest' +import type { PtyProcessInspection } from '../providers/pty-process-inspection' +import type { IPtyProvider, PtySpawnOptions, PtySpawnResult } from '../providers/types' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' + +export type ProviderMock = IPtyProvider & { + probePtyLiveness: (id: string) => Promise + inspectProcess: (id: string) => Promise + emitData: (id: string, data: string, sequenceChars?: number) => void + emitReplay: (id: string, data: string) => void + emitExit: (id: string, code: number) => void + triggerWriteUnavailable: (id: string) => void + onWriteUnavailable: (callback: (payload: { id: string }) => void) => () => void +} + +export function createProvider( + label: string, + sessions: string[] = [], + authoritativeOwnerListings = false +): ProviderMock { + const dataListeners: ((payload: { id: string; data: string; sequenceChars?: number }) => void)[] = + [] + const replayListeners: ((payload: { id: string; data: string }) => void)[] = [] + const exitListeners: ((payload: { id: string; code: number }) => void)[] = [] + const writeUnavailableListeners: ((payload: { id: string }) => void)[] = [] + return { + spawn: vi.fn(async (opts: PtySpawnOptions): Promise => { + const id = opts.sessionId ?? `${label}-new` + sessions.push(id) + return { id } + }), + attach: vi.fn(async () => {}), + hasPty: vi.fn((id: string) => sessions.includes(id)), + probePtyLiveness: vi.fn(async (id: string) => sessions.includes(id)), + providesAgentSessionOwnerListings: vi.fn(() => authoritativeOwnerListings), + write: vi.fn(), + resize: vi.fn(), + pauseProducer: vi.fn(), + resumeProducer: vi.fn(), + setPtyBackgrounded: vi.fn(), + shutdown: vi.fn(async (id: string) => { + const idx = sessions.indexOf(id) + if (idx !== -1) { + sessions.splice(idx, 1) + } + }), + sendSignal: vi.fn(async () => {}), + getCwd: vi.fn(async () => ''), + getInitialCwd: vi.fn(async () => ''), + clearBuffer: vi.fn(async () => {}), + acknowledgeDataEvent: vi.fn(), + hasChildProcesses: vi.fn(async () => false), + getForegroundProcess: vi.fn(async () => null), + inspectProcess: vi.fn(async () => ({ foregroundProcess: null, hasChildProcesses: false })), + confirmForegroundProcess: vi.fn(async () => `${label}-confirmed`), + serialize: vi.fn(async () => '{}'), + revive: vi.fn(async () => {}), + listProcesses: vi.fn(async () => sessions.map((id) => ({ id, cwd: '', title: label }))), + getDefaultShell: vi.fn(async () => '/bin/zsh'), + getProfiles: vi.fn(async () => []), + onData: vi.fn( + (callback: (payload: { id: string; data: string; sequenceChars?: number }) => void) => { + dataListeners.push(callback) + return () => { + const idx = dataListeners.indexOf(callback) + if (idx !== -1) { + dataListeners.splice(idx, 1) + } + } + } + ), + onReplay: vi.fn((callback: (payload: { id: string; data: string }) => void) => { + replayListeners.push(callback) + return () => { + const idx = replayListeners.indexOf(callback) + if (idx !== -1) { + replayListeners.splice(idx, 1) + } + } + }), + onExit: vi.fn((callback: (payload: { id: string; code: number }) => void) => { + exitListeners.push(callback) + return () => { + const idx = exitListeners.indexOf(callback) + if (idx !== -1) { + exitListeners.splice(idx, 1) + } + } + }), + emitData: (id: string, data: string, sequenceChars?: number) => { + for (const listener of dataListeners) { + listener({ id, data, ...(sequenceChars === undefined ? {} : { sequenceChars }) }) + } + }, + emitReplay: (id: string, data: string) => { + for (const listener of replayListeners) { + listener({ id, data }) + } + }, + emitExit: (id: string, code: number) => { + for (const listener of exitListeners) { + listener({ id, code }) + } + }, + onWriteUnavailable: vi.fn((callback: (payload: { id: string }) => void) => { + writeUnavailableListeners.push(callback) + return () => { + const idx = writeUnavailableListeners.indexOf(callback) + if (idx !== -1) { + writeUnavailableListeners.splice(idx, 1) + } + } + }), + triggerWriteUnavailable: (id: string) => { + for (const listener of writeUnavailableListeners) { + listener({ id }) + } + } + } +} + +export function createDaemonAdapter( + label: string, + sessions: string[] = [] +): DaemonPtyAdapter & ProviderMock { + return { + ...createProvider(label, sessions, true), + protocolVersion: 13, + getLastAuthenticatedDaemonIdentity: vi.fn(() => null), + onDaemonIdentityChanged: vi.fn(() => () => {}), + listSessions: vi.fn(async () => []), + ackColdRestore: vi.fn(), + clearTombstone: vi.fn(), + reconcileOnStartup: vi.fn(async () => ({ alive: sessions, killed: [] })), + dispose: vi.fn(), + disconnectOnly: vi.fn(async () => {}), + getActiveSessionIds: vi.fn(() => []), + fanoutSyntheticExits: vi.fn() + } as unknown as DaemonPtyAdapter & ProviderMock +} diff --git a/src/main/daemon/degraded-daemon-pty-provider.test.ts b/src/main/daemon/degraded-daemon-pty-provider.test.ts index efca6b6e7ed..2e46f262148 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.test.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.test.ts @@ -1,154 +1,19 @@ import { describe, expect, it, vi } from 'vitest' import { DegradedDaemonPtyProvider } from './degraded-daemon-pty-provider' -import type { DaemonIdentityChangeEvent, DaemonPtyAdapter } from './daemon-pty-adapter' -import type { IPtyProvider, PtySpawnOptions, PtySpawnResult } from '../providers/types' -import type { PtyProcessInspection } from '../providers/pty-process-inspection' +import type { DaemonIdentityChangeEvent } from './daemon-pty-adapter' +import type { PtySpawnResult } from '../providers/types' +import { createDaemonAdapter, createProvider } from './degraded-daemon-pty-provider-fixture' -type ProviderMock = IPtyProvider & { - probePtyLiveness: (id: string) => Promise - inspectProcess: (id: string) => Promise - emitData: (id: string, data: string, sequenceChars?: number) => void - emitReplay: (id: string, data: string) => void - emitExit: (id: string, code: number) => void - triggerWriteUnavailable: (id: string) => void - onWriteUnavailable: (callback: (payload: { id: string }) => void) => () => void -} - -function createProvider( - label: string, - sessions: string[] = [], - authoritativeOwnerListings = false -): ProviderMock { - const dataListeners: ((payload: { id: string; data: string; sequenceChars?: number }) => void)[] = - [] - const replayListeners: ((payload: { id: string; data: string }) => void)[] = [] - const exitListeners: ((payload: { id: string; code: number }) => void)[] = [] - const writeUnavailableListeners: ((payload: { id: string }) => void)[] = [] - return { - spawn: vi.fn(async (opts: PtySpawnOptions): Promise => { - const id = opts.sessionId ?? `${label}-new` - sessions.push(id) - return { id } - }), - attach: vi.fn(async () => {}), - hasPty: vi.fn((id: string) => sessions.includes(id)), - probePtyLiveness: vi.fn(async (id: string) => sessions.includes(id)), - providesAgentSessionOwnerListings: vi.fn(() => authoritativeOwnerListings), - write: vi.fn(), - resize: vi.fn(), - pauseProducer: vi.fn(), - resumeProducer: vi.fn(), - setPtyBackgrounded: vi.fn(), - shutdown: vi.fn(async (id: string) => { - const idx = sessions.indexOf(id) - if (idx !== -1) { - sessions.splice(idx, 1) - } - }), - sendSignal: vi.fn(async () => {}), - getCwd: vi.fn(async () => ''), - getInitialCwd: vi.fn(async () => ''), - clearBuffer: vi.fn(async () => {}), - acknowledgeDataEvent: vi.fn(), - hasChildProcesses: vi.fn(async () => false), - getForegroundProcess: vi.fn(async () => null), - inspectProcess: vi.fn(async () => ({ foregroundProcess: null, hasChildProcesses: false })), - confirmForegroundProcess: vi.fn(async () => `${label}-confirmed`), - serialize: vi.fn(async () => '{}'), - revive: vi.fn(async () => {}), - listProcesses: vi.fn(async () => sessions.map((id) => ({ id, cwd: '', title: label }))), - getDefaultShell: vi.fn(async () => '/bin/zsh'), - getProfiles: vi.fn(async () => []), - onData: vi.fn( - (callback: (payload: { id: string; data: string; sequenceChars?: number }) => void) => { - dataListeners.push(callback) - return () => { - const idx = dataListeners.indexOf(callback) - if (idx !== -1) { - dataListeners.splice(idx, 1) - } - } - } - ), - onReplay: vi.fn((callback: (payload: { id: string; data: string }) => void) => { - replayListeners.push(callback) - return () => { - const idx = replayListeners.indexOf(callback) - if (idx !== -1) { - replayListeners.splice(idx, 1) - } - } - }), - onExit: vi.fn((callback: (payload: { id: string; code: number }) => void) => { - exitListeners.push(callback) - return () => { - const idx = exitListeners.indexOf(callback) - if (idx !== -1) { - exitListeners.splice(idx, 1) - } - } - }), - emitData: (id: string, data: string, sequenceChars?: number) => { - for (const listener of dataListeners) { - listener({ id, data, ...(sequenceChars === undefined ? {} : { sequenceChars }) }) - } - }, - emitReplay: (id: string, data: string) => { - for (const listener of replayListeners) { - listener({ id, data }) - } - }, - emitExit: (id: string, code: number) => { - for (const listener of exitListeners) { - listener({ id, code }) - } - }, - onWriteUnavailable: vi.fn((callback: (payload: { id: string }) => void) => { - writeUnavailableListeners.push(callback) - return () => { - const idx = writeUnavailableListeners.indexOf(callback) - if (idx !== -1) { - writeUnavailableListeners.splice(idx, 1) - } - } - }), - triggerWriteUnavailable: (id: string) => { - for (const listener of writeUnavailableListeners) { - listener({ id }) - } - } - } -} - -function createDaemonAdapter( - label: string, - sessions: string[] = [] -): DaemonPtyAdapter & ProviderMock { - return { - ...createProvider(label, sessions, true), - protocolVersion: 13, - getLastAuthenticatedDaemonIdentity: vi.fn(() => null), - onDaemonIdentityChanged: vi.fn(() => () => {}), - listSessions: vi.fn(async () => []), - ackColdRestore: vi.fn(), - clearTombstone: vi.fn(), - reconcileOnStartup: vi.fn(async () => ({ alive: sessions, killed: [] })), - dispose: vi.fn(), - disconnectOnly: vi.fn(async () => {}), - getActiveSessionIds: vi.fn(() => []), - fanoutSyntheticExits: vi.fn() - } as unknown as DaemonPtyAdapter & ProviderMock -} - -it('forwards dead-endpoint write-unavailable signals from the daemon adapters', () => { +it('forwards dead-endpoint write-unavailable signals from exact daemon owners', async () => { // Why revert-sensitive: this provider is the live localProvider in degraded launch // mode and main subscribes on it, so without forwarding the STA-2373 fan-out reaches // no listener and sibling panes stay frozen. - const current = createDaemonAdapter('daemon') - const legacy = createDaemonAdapter('legacy') + const current = createDaemonAdapter('daemon', ['daemon-pane']) + const legacy = createDaemonAdapter('legacy', ['legacy-pane']) const fallback = createProvider('fallback') const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback }) const recovered: string[] = [] + await provider.discoverDaemonSessions() const unsubscribe = provider.onWriteUnavailable(({ id }) => recovered.push(id)) current.triggerWriteUnavailable('daemon-pane') @@ -160,6 +25,36 @@ it('forwards dead-endpoint write-unavailable signals from the daemon adapters', expect(recovered).toEqual(['daemon-pane', 'legacy-pane']) }) +it('withholds degraded write-unavailable signals from a foreign daemon route', async () => { + const current = createDaemonAdapter('daemon', ['shared-pane']) + const legacy = createDaemonAdapter('legacy') + const fallback = createProvider('fallback') + const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback }) + const recovered: string[] = [] + await provider.discoverDaemonSessions() + provider.onWriteUnavailable(({ id }) => recovered.push(id)) + + legacy.triggerWriteUnavailable('shared-pane') + current.triggerWriteUnavailable('shared-pane') + + expect(recovered).toEqual(['shared-pane']) +}) + +it('withholds degraded write-unavailable signals for fallback collisions', async () => { + const sessionId = 'fallback-collision' + const current = createDaemonAdapter('daemon', [sessionId]) + const fallback = createProvider('fallback') + const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) + await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + await provider.discoverDaemonSessions() + const recovered: string[] = [] + provider.onWriteUnavailable(({ id }) => recovered.push(id)) + + current.triggerWriteUnavailable(sessionId) + + expect(recovered).toEqual([]) +}) + it('rejects completion inspection instead of borrowing the fallback provider', async () => { const provider = new DegradedDaemonPtyProvider({ current: createDaemonAdapter('daemon'), diff --git a/src/main/daemon/degraded-daemon-pty-provider.ts b/src/main/daemon/degraded-daemon-pty-provider.ts index 93a87d98212..ed90a41f0bb 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.ts @@ -227,7 +227,13 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { // adapters only — the local fallback has no dead-socket problem. onWriteUnavailable(callback: (payload: { id: string }) => void): () => void { return combineUnsubscribes( - this.allDaemonAdapters().map((adapter) => adapter.onWriteUnavailable(callback)) + this.allDaemonAdapters().map((adapter) => + adapter.onWriteUnavailable((payload) => { + if (this.routing.shouldForwardWriteUnavailable(adapter, payload.id)) { + callback(payload) + } + }) + ) ) } diff --git a/src/main/daemon/degraded-daemon-session-routing.ts b/src/main/daemon/degraded-daemon-session-routing.ts index 452f4806a0f..ee17e3f0835 100644 --- a/src/main/daemon/degraded-daemon-session-routing.ts +++ b/src/main/daemon/degraded-daemon-session-routing.ts @@ -25,7 +25,7 @@ export class DegradedDaemonSessionRouting { this.adapters = [current, ...legacy] this.adapterSet = new Set(this.adapters) this.routes = new DaemonSessionRouteTable(this.adapters) - this.fallbackRoutes = new FallbackRoutes(this.fallbackOwners, this.routes) + this.fallbackRoutes = new FallbackRoutes(this.fallbackOwners, this.routes, this.fallbackSpawns) for (const adapter of this.adapters) { this.identityUnsubscribes.push( adapter.onDaemonIdentityChanged(({ previous }) => { @@ -76,7 +76,7 @@ export class DegradedDaemonSessionRouting { this.fallbackRoutes.recordSpawnedSession(result.id, target) } } else { - this.routes.recordOwned(result.id, target as DaemonPtyAdapter) + this.fallbackRoutes.recordDaemonSpawn(result.id, opts.sessionId, target as DaemonPtyAdapter) } return result } finally { @@ -158,11 +158,12 @@ export class DegradedDaemonSessionRouting { this.fallbackRoutes.clearCollision(sessionId) return true } - if (route) { - this.fallbackRoutes.recordCollision(sessionId) - return false - } - return true + this.fallbackRoutes.recordCollision(sessionId) + return false + } + + shouldForwardWriteUnavailable(adapter: DaemonPtyAdapter, sessionId: string): boolean { + return this.fallbackRoutes.shouldForwardWriteUnavailable(sessionId, adapter) } recordExit(provider: IPtyProvider, sessionId: string): boolean { diff --git a/src/main/daemon/degraded-fallback-session-routes.ts b/src/main/daemon/degraded-fallback-session-routes.ts index d679d2cddc4..a3758e08fda 100644 --- a/src/main/daemon/degraded-fallback-session-routes.ts +++ b/src/main/daemon/degraded-fallback-session-routes.ts @@ -2,6 +2,8 @@ import type { IPtyProvider } from '../providers/types' import type { DaemonPtyAdapter } from './daemon-pty-adapter' import { sameDaemonIncarnation } from './daemon-session-route' import type { DaemonSessionRouteTable } from './daemon-session-route-table' +import { recordValidatedDaemonSpawn } from './daemon-spawn-route-validation' +import type { DegradedFallbackSpawnRoutes } from './degraded-fallback-spawn-routes' type FallbackLivenessProbe = { provider: IPtyProvider @@ -15,7 +17,8 @@ export class DegradedFallbackSessionRoutes { constructor( private readonly sessions: Map, - private readonly daemonRoutes: DaemonSessionRouteTable + private readonly daemonRoutes: DaemonSessionRouteTable, + private readonly spawns: DegradedFallbackSpawnRoutes ) {} hasCollision(sessionId: string): boolean { @@ -41,12 +44,29 @@ export class DegradedFallbackSessionRoutes { } } + recordDaemonSpawn( + resultSessionId: string, + requestedSessionId: string | undefined, + owner: DaemonPtyAdapter + ): void { + recordValidatedDaemonSpawn(this.daemonRoutes, resultSessionId, requestedSessionId, owner) + } + clearUnownedCollision(sessionId: string): void { if (!this.sessions.has(sessionId)) { this.clearCollision(sessionId) } } + shouldForwardWriteUnavailable(sessionId: string, owner: DaemonPtyAdapter): boolean { + return ( + !this.spawns.hasLiveCandidate(sessionId) && + !this.sessions.has(sessionId) && + !this.collisionIds.has(sessionId) && + this.daemonRoutes.getOwned(sessionId) === owner + ) + } + deleteSession(sessionId: string): void { this.sessions.delete(sessionId) this.epochs.delete(sessionId)