diff --git a/src/main/daemon/daemon-health.test.ts b/src/main/daemon/daemon-health.test.ts index 8d266caae9b..f1128e7e370 100644 --- a/src/main/daemon/daemon-health.test.ts +++ b/src/main/daemon/daemon-health.test.ts @@ -1,8 +1,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { existsSync, 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' +import { spawn } from 'node:child_process' import { DaemonServer } from './daemon-server' import { getDaemonPidPath, serializeDaemonPidFile } from './daemon-spawner' import { @@ -436,9 +437,77 @@ describe('killStaleDaemon pid identity guards', () => { ([, sig]) => sig === 'SIGTERM' || sig === 'SIGKILL' ) expect(terminationSignals).toEqual([]) - expect(readFileSync(pidPath, 'utf8')).toContain(`"pid":${process.pid}`) + expect(parseDaemonPidFile(readFileSync(pidPath, 'utf8'))?.pid).toBe(process.pid) } finally { killSpy.mockRestore() } }) + + it('preserves replacement identity files when the pid changes ownership during escalation', async () => { + if (process.platform === 'win32') { + return + } + + const replacementSocket = createServer((socket) => socket.end()) + await new Promise((resolve, reject) => { + replacementSocket.once('error', reject) + replacementSocket.listen(socketPath, () => { + replacementSocket.off('error', reject) + resolve() + }) + }) + const child = spawn( + process.execPath, + [ + '-e', + "process.on('SIGTERM',()=>{process.title='replacement-process';process.send?.('replaced')});process.send?.('ready');setInterval(()=>{},1000)", + 'daemon-entry', + socketPath, + tokenPath + ], + { stdio: ['ignore', 'ignore', 'ignore', 'ipc'] } + ) + await new Promise((resolve, reject) => { + child.once('error', reject) + child.on('message', (message) => { + if (message === 'ready') { + resolve() + } + }) + }) + const pid = child.pid + expect(pid).toEqual(expect.any(Number)) + const pidPath = getDaemonPidPath(dir) + writeFileSync(pidPath, serializeDaemonPidFile({ pid: pid!, startedAtMs: null }), { + mode: 0o600 + }) + const replacementObserved = new Promise((resolve) => { + child.on('message', (message) => { + if (message === 'replaced') { + resolve() + } + }) + }) + + vi.useFakeTimers() + try { + const killing = killStaleDaemon(dir, socketPath, tokenPath) + await replacementObserved + await vi.advanceTimersByTimeAsync(3_100) + await expect(killing).resolves.toBe(true) + expect(parseDaemonPidFile(readFileSync(pidPath, 'utf8'))?.pid).toBe(pid) + expect(existsSync(socketPath)).toBe(true) + } finally { + vi.useRealTimers() + const exited = + child.exitCode === null + ? new Promise((resolve) => child.once('exit', () => resolve())) + : Promise.resolve() + if (child.exitCode === null) { + child.kill('SIGKILL') + } + await exited + await closeServer(replacementSocket) + } + }) }) diff --git a/src/main/daemon/daemon-health.ts b/src/main/daemon/daemon-health.ts index 06e940134a1..384c7e4ecb4 100644 --- a/src/main/daemon/daemon-health.ts +++ b/src/main/daemon/daemon-health.ts @@ -693,6 +693,7 @@ export async function killStaleDaemon( // process that happens to now own the same pid. if (!(await isDaemonProcess(pid, socketPath, tokenPath, startedAtMs))) { console.warn('[daemon] Skipping SIGKILL for stale daemon: reason=pid_recycled') + identifiedDaemon = false exited = true killedDaemon = true } else { @@ -719,7 +720,7 @@ export async function killStaleDaemon( } const socketIsLive = await canConnectSocket(socketPath) - if (process.platform !== 'win32' && existsSync(socketPath) && (killedDaemon || !socketIsLive)) { + if (process.platform !== 'win32' && existsSync(socketPath) && !socketIsLive) { try { unlinkSync(socketPath) } catch { diff --git a/src/main/daemon/daemon-init.test.ts b/src/main/daemon/daemon-init.test.ts index 444627f59e6..4e4e3e4155f 100644 --- a/src/main/daemon/daemon-init.test.ts +++ b/src/main/daemon/daemon-init.test.ts @@ -216,6 +216,8 @@ type MockAdapter = { shutdown: ReturnType dispose: ReturnType disconnectOnly: ReturnType + getLastAuthenticatedDaemonIdentity: ReturnType + onDaemonIdentityChanged: ReturnType onData: ReturnType onExit: ReturnType // Why: the router calls onData/onExit on each adapter; the stub returns a no-op unsubscribe so router subscription doesn't explode. @@ -336,6 +338,8 @@ vi.mock('./daemon-pty-adapter', () => ({ readonly shutdown: ReturnType readonly dispose: ReturnType readonly disconnectOnly: ReturnType + readonly getLastAuthenticatedDaemonIdentity: ReturnType + readonly onDaemonIdentityChanged: ReturnType readonly onData: ReturnType readonly onExit: ReturnType readonly callOrder: string[] @@ -365,6 +369,8 @@ vi.mock('./daemon-pty-adapter', () => ({ throw disconnectOnlyError } }) + this.getLastAuthenticatedDaemonIdentity = vi.fn(() => null) + this.onDaemonIdentityChanged = vi.fn(() => () => {}) this.onData = vi.fn(() => { if (routerSubscriptionError.current) { const error = routerSubscriptionError.current diff --git a/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts b/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts index 19e82a5faef..cac0a809f2e 100644 --- a/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts +++ b/src/main/daemon/daemon-pty-adapter-subscription-fanout.ts @@ -10,17 +10,23 @@ export class DaemonPtyAdapterSubscriptionFanout { constructor( private readonly adapters: readonly DaemonPtyAdapter[], - onAdapterExit: (adapter: DaemonPtyAdapter, id: string) => void + private readonly shouldForwardStreamEvent: (adapter: DaemonPtyAdapter, id: string) => boolean, + onAdapterExit: (adapter: DaemonPtyAdapter, id: string) => boolean ) { for (const adapter of adapters) { this.unsubscribers.push( adapter.onData((payload) => { + if (!shouldForwardStreamEvent(adapter, payload.id)) { + return + } for (const listener of this.dataListeners) { listener(payload) } }), adapter.onExit((payload) => { - onAdapterExit(adapter, payload.id) + if (!onAdapterExit(adapter, payload.id)) { + return + } for (const listener of this.exitListeners) { listener(payload) } @@ -41,7 +47,13 @@ export class DaemonPtyAdapterSubscriptionFanout { onBackgroundStreamEvent(callback: (payload: PtyBackgroundStreamEvent) => void): () => void { return combineUnsubscribes( - this.adapters.map((adapter) => adapter.onBackgroundStreamEvent(callback)) + this.adapters.map((adapter) => + adapter.onBackgroundStreamEvent((payload) => { + if (this.shouldForwardStreamEvent(adapter, payload.id)) { + callback(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 index acab8477c26..c38f5367468 100644 --- a/src/main/daemon/daemon-pty-router-routing-safety.test.ts +++ b/src/main/daemon/daemon-pty-router-routing-safety.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it, vi } from 'vitest' +import { afterEach, 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' @@ -77,11 +77,15 @@ function createAdapter( } describe('DaemonPtyRouter routing safety', () => { - it('fences unknown existing ids until a later complete legacy listing', async () => { + afterEach(() => { + vi.restoreAllMocks() + }) + + it('uses conclusive probes while fencing absence after a failed discovery', 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(() => {}) + vi.spyOn(console, 'warn').mockImplementation(() => {}) const router = new DaemonPtyRouter({ current: current.adapter, legacy: [legacy.adapter] @@ -96,18 +100,17 @@ describe('DaemonPtyRouter routing safety', () => { 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( + await expect( + router.spawn({ sessionId: 'legacy-session', cols: 80, rows: 24 }) + ).resolves.toMatchObject({ id: 'legacy-session' }) + await expect(router.shutdown('missing-session', { immediate: true })).rejects.toThrow( 'daemon_session_routing_unavailable' ) await router.discoverLegacySessions() - await router.spawn({ sessionId: 'legacy-session', cols: 80, rows: 24 }) + await expect(router.shutdown('missing-session', { immediate: true })).rejects.toThrow( + 'terminal_gone' + ) expect(current.adapter.spawn).toHaveBeenCalledExactlyOnceWith({ sessionId: 'fresh-session', @@ -120,7 +123,32 @@ describe('DaemonPtyRouter routing safety', () => { cols: 80, rows: 24 }) - warn.mockRestore() + }) + + it('probes adapters outside an ambiguous route while discovery is incomplete', async () => { + const sessionId = 'hidden-collision' + const firstSessions = [sessionId] + const current = createAdapter('current') + const first = createAdapter('legacy-first', firstSessions) + const second = createAdapter('legacy-second', [sessionId]) + const undiscovered = createAdapter('legacy-undiscovered', [sessionId]) + vi.mocked(undiscovered.adapter.listProcesses).mockRejectedValueOnce( + new Error('listing unavailable') + ) + vi.spyOn(console, 'warn').mockImplementation(() => {}) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [first.adapter, second.adapter, undiscovered.adapter] + }) + + await router.discoverLegacySessions() + firstSessions.splice(0) + + await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + expect(second.adapter.sendSignal).not.toHaveBeenCalled() + expect(undiscovered.adapter.sendSignal).not.toHaveBeenCalled() }) it('fails existing operations unknown when an owner probe is inconclusive', async () => { @@ -194,6 +222,63 @@ describe('DaemonPtyRouter routing safety', () => { expect(current.adapter.shutdown).not.toHaveBeenCalled() }) + it('collapses an ambiguous route after one daemon incarnation is replaced', async () => { + const sessionId = 'replaced-collision' + 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] + }) + const previous = identity('legacy', 10) + const replacement = identity('legacy', 11) + + await router.reconcileOnStartup(new Set()) + legacy.setIdentity(replacement) + legacy.emitIdentityChange(previous, replacement) + + expect(router.getSessionRouteState(sessionId)).toBe('owned') + await router.sendSignal(sessionId, 'SIGTERM') + expect(current.adapter.sendSignal).toHaveBeenCalledExactlyOnceWith(sessionId, 'SIGTERM') + expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() + }) + + it('keeps a canonical spawn collision ambiguous', async () => { + const current = createAdapter('current') + const legacy = createAdapter('legacy', ['canonical-session']) + vi.mocked(current.adapter.spawn).mockResolvedValue({ + id: 'canonical-session', + incarnationId: 'current-incarnation' + }) + vi.mocked(current.adapter.probePtyLiveness).mockResolvedValue(true) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + await router.discoverLegacySessions() + + await router.spawn({ + sessionId: 'requested-session', + isNewSession: true, + cols: 80, + rows: 24, + agentSessionEnsure: {} as never + }) + + expect(router.getSessionRouteState('canonical-session')).toBe('ambiguous') + await expect(router.sendSignal('canonical-session', 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + expect(current.adapter.sendSignal).not.toHaveBeenCalled() + expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() + }) + it('scopes unavailable tombstones to the authenticated daemon incarnation', async () => { const sessionId = 'legacy-session' const current = createAdapter('current') @@ -221,6 +306,35 @@ describe('DaemonPtyRouter routing safety', () => { expect(legacy.adapter.sendSignal).toHaveBeenCalledWith(sessionId, 'SIGTERM') }) + it('allows explicit fresh reuse after the tombstone owner is replaced', async () => { + const sessionId = 'legacy-session' + const legacySessions = [sessionId] + const current = createAdapter('current') + const legacy = createAdapter('legacy', legacySessions) + 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) + legacySessions.splice(0) + await router.discoverLegacySessions() + + await expect( + router.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + ).resolves.toMatchObject({ id: sessionId }) + expect(current.adapter.spawn).toHaveBeenCalledExactlyOnceWith({ + sessionId, + isNewSession: true, + cols: 80, + rows: 24 + }) + }) + it('acknowledges a colliding id through the adapter that produced its snapshot', async () => { const sessionId = 'colliding-snapshot' const current = createAdapter('current', [sessionId], { @@ -249,6 +363,32 @@ describe('DaemonPtyRouter routing safety', () => { expect(current.adapter.ackColdRestore).not.toHaveBeenCalled() }) + it('clears a stale snapshot producer after a later null capture', async () => { + const sessionId = 'legacy-snapshot' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + vi.mocked(legacy.adapter.getBufferSnapshot) + .mockResolvedValueOnce({ + data: 'legacy frame', + cols: 80, + rows: 24, + seq: 1, + source: 'headless' + }) + .mockResolvedValueOnce(null) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.discoverLegacySessions() + await router.getBufferSnapshot(sessionId) + await router.getBufferSnapshot(sessionId) + router.ackColdRestore(sessionId) + + expect(legacy.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') diff --git a/src/main/daemon/daemon-pty-router-session-routing.ts b/src/main/daemon/daemon-pty-router-session-routing.ts index cc90ada1961..a8043213adb 100644 --- a/src/main/daemon/daemon-pty-router-session-routing.ts +++ b/src/main/daemon/daemon-pty-router-session-routing.ts @@ -40,11 +40,15 @@ export class DaemonPtyRouterSessionRouting { return this.routes.resolveOwner(opts.sessionId) } - recordSpawn(result: PtySpawnResult, target: DaemonPtyAdapter): void { + recordSpawn(result: PtySpawnResult, target: DaemonPtyAdapter, opts: PtySpawnOptions): void { if (result.exitedBeforeSpawnReply) { return } - this.routes.transfer(result.id, target) + if (opts.isNewSession && opts.sessionId === result.id) { + this.routes.recordFreshOwned(result.id, target) + return + } + this.routes.recordOwned(result.id, target) } async owner(sessionId: string): Promise { @@ -55,6 +59,14 @@ export class DaemonPtyRouterSessionRouting { return this.routes.resolveOwnerSync(sessionId) } + ownerForHint(sessionId: string): DaemonPtyAdapter | null { + try { + return this.ownerSync(sessionId) + } catch { + return null + } + } + ownerOrCurrent(sessionId?: string): DaemonPtyAdapter { return sessionId ? this.ownerSync(sessionId) : this.current } @@ -121,8 +133,18 @@ export class DaemonPtyRouterSessionRouting { this.snapshotAcks.drop(sessionId) } - recordExit(sessionId: string, owner: DaemonPtyAdapter): void { + shouldForwardEvent(sessionId: string, owner: DaemonPtyAdapter): boolean { + return this.routes.shouldForwardEvent(sessionId, owner) + } + + shouldForwardStreamEvent(sessionId: string, owner: DaemonPtyAdapter): boolean { + return this.routes.recordDataOwner(sessionId, owner) + } + + recordExit(sessionId: string, owner: DaemonPtyAdapter): boolean { + const shouldForward = this.shouldForwardEvent(sessionId, owner) this.routes.markUnavailable(sessionId, owner) + return shouldForward } markAdapterUnavailable(adapter: DaemonPtyAdapter): void { diff --git a/src/main/daemon/daemon-pty-router.test.ts b/src/main/daemon/daemon-pty-router.test.ts index 5cafdc03877..ab49db4cb21 100644 --- a/src/main/daemon/daemon-pty-router.test.ts +++ b/src/main/daemon/daemon-pty-router.test.ts @@ -83,6 +83,8 @@ function createAdapter( writes.push({ id, data }) }), resize: vi.fn(), + pauseProducer: vi.fn(), + resumeProducer: vi.fn(), setPtyBackgrounded: vi.fn(), getBufferSnapshot: vi.fn(async () => null), shutdown: vi.fn(async (id: string) => { @@ -440,6 +442,97 @@ describe('DaemonPtyRouter', () => { }) }) + it('silently drops best-effort hints for an unavailable route', async () => { + const current = createAdapter('current') + const legacy = createAdapter('legacy', ['legacy-session']) + const router = new DaemonPtyRouter({ current, legacy: [legacy] }) + await router.discoverLegacySessions() + legacy.emitExit('legacy-session', 0) + + expect(() => router.pauseProducer('legacy-session')).not.toThrow() + expect(() => router.resumeProducer('legacy-session')).not.toThrow() + expect(() => router.setPtyBackgrounded('legacy-session', true)).not.toThrow() + expect(current.pauseProducer).not.toHaveBeenCalled() + expect(current.resumeProducer).not.toHaveBeenCalled() + expect(current.setPtyBackgrounded).not.toHaveBeenCalled() + }) + + it('fences colliding output and keeps a surviving owner after the other exits', 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, legacy: [legacy] }) + const data = vi.fn() + const exit = vi.fn() + router.onData(data) + router.onExit(exit) + await router.reconcileOnStartup(new Set()) + + current.emitData(sessionId, 'current') + legacy.emitData(sessionId, 'legacy') + current.emitExit(sessionId, 0) + + expect(data).not.toHaveBeenCalled() + expect(exit).not.toHaveBeenCalled() + expect(router.getSessionRouteState(sessionId)).toBe('owned') + + legacy.emitData(sessionId, 'survivor') + expect(data).toHaveBeenCalledExactlyOnceWith({ id: sessionId, data: 'survivor' }) + }) + + it('records data ownership before discovery and fences later collision evidence', () => { + const sessionId = 'undiscovered-collision' + const current = createAdapter('current', [sessionId]) + const legacy = createAdapter('legacy', [sessionId]) + const router = new DaemonPtyRouter({ current, legacy: [legacy] }) + const data = vi.fn() + router.onData(data) + + current.emitData(sessionId, 'current') + legacy.emitData(sessionId, 'legacy') + current.emitData(sessionId, 'current-after-collision') + + expect(data).toHaveBeenCalledExactlyOnceWith({ id: sessionId, data: 'current' }) + expect(router.getSessionRouteState(sessionId)).toBe('ambiguous') + }) + + it('preserves collision evidence received before a fresh spawn reply', async () => { + const sessionId = 'fresh-collision' + const currentSessions = [sessionId] + const legacySessions = [sessionId] + const current = createAdapter('current', currentSessions) + const legacy = createAdapter('legacy', legacySessions) + let resolveSpawn: ((result: PtySpawnResult) => void) | undefined + vi.mocked(current.spawn).mockReturnValue( + new Promise((resolve) => { + resolveSpawn = resolve + }) + ) + const router = new DaemonPtyRouter({ current, legacy: [legacy] }) + + const spawning = router.spawn({ + sessionId, + isNewSession: true, + cols: 80, + rows: 24 + }) + current.emitData(sessionId, 'current') + legacy.emitData(sessionId, 'legacy') + resolveSpawn?.({ id: sessionId }) + await spawning + + expect(router.getSessionRouteState(sessionId)).toBe('ambiguous') + await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + }) + it('forwards gap events and explicit sequence accounting from every adapter', () => { const current = createAdapter('current') const legacy = createAdapter('legacy') @@ -470,6 +563,22 @@ describe('DaemonPtyRouter', () => { }) }) + it('fences background stream events from a colliding adapter', async () => { + const sessionId = 'background-collision' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + const router = new DaemonPtyRouter({ current, legacy: [legacy] }) + const backgroundSpy = vi.fn() + router.onBackgroundStreamEvent(backgroundSpy) + await router.discoverLegacySessions() + + current.emitBackground({ id: sessionId, kind: 'dataGap', droppedChars: 512 }) + legacy.emitBackground({ id: sessionId, kind: 'transientFact', fact: { kind: 'bell' } }) + + expect(backgroundSpy).not.toHaveBeenCalled() + expect(router.getSessionRouteState(sessionId)).toBe('ambiguous') + }) + it('tombstones a legacy route after the routed session exits', async () => { const current = createAdapter('current') const legacy = createAdapter('legacy', ['legacy-session']) @@ -630,6 +739,20 @@ describe('DaemonPtyRouter', () => { expect(current.write).toHaveBeenCalledWith('current-alive', 'new\n') }) + it('accepts a live owner after another adapter tombstones the same id', async () => { + const sessionId = 'reconciled-owner' + const current = createAdapter('current', [], { alive: [], killed: [sessionId] }) + const legacy = createAdapter('legacy', [], { alive: [sessionId], killed: [] }) + const router = new DaemonPtyRouter({ current, legacy: [legacy] }) + + await router.reconcileOnStartup(new Set()) + await router.sendSignal(sessionId, 'SIGTERM') + + expect(router.getSessionRouteState(sessionId)).toBe('owned') + expect(legacy.sendSignal).toHaveBeenCalledExactlyOnceWith(sessionId, 'SIGTERM') + expect(current.sendSignal).not.toHaveBeenCalled() + }) + it('merges large startup reconciliation results', async () => { const alive = buildSessionIds('alive', LARGE_RECONCILE_SESSION_COUNT) const killed = buildSessionIds('killed', LARGE_RECONCILE_SESSION_COUNT) diff --git a/src/main/daemon/daemon-pty-router.ts b/src/main/daemon/daemon-pty-router.ts index b3130451af1..ca31abcaa5a 100644 --- a/src/main/daemon/daemon-pty-router.ts +++ b/src/main/daemon/daemon-pty-router.ts @@ -25,9 +25,8 @@ export class DaemonPtyRouter implements IPtyProvider { this.routing = new DaemonPtyRouterSessionRouting(this.current, this.allAdapters()) this.subscriptions = new DaemonPtyAdapterSubscriptionFanout( this.allAdapters(), - (adapter, id) => { - this.routing.recordExit(id, adapter) - } + (adapter, id) => this.routing.shouldForwardStreamEvent(id, adapter), + (adapter, id) => this.routing.recordExit(id, adapter) ) } @@ -59,7 +58,7 @@ export class DaemonPtyRouter implements IPtyProvider { const targetResult = this.routing.spawnTarget(opts) const target = targetResult instanceof Promise ? await targetResult : targetResult const result = await target.spawn(opts) - this.routing.recordSpawn(result, target) + this.routing.recordSpawn(result, target, opts) return result } @@ -104,15 +103,15 @@ export class DaemonPtyRouter implements IPtyProvider { } pauseProducer(id: string): void { - this.routing.ownerSync(id).pauseProducer(id) + this.routing.ownerForHint(id)?.pauseProducer(id) } resumeProducer(id: string): void { - this.routing.ownerSync(id).resumeProducer(id) + this.routing.ownerForHint(id)?.resumeProducer(id) } setPtyBackgrounded(id: string, background: boolean): void { - this.routing.ownerSync(id).setPtyBackgrounded(id, background) + this.routing.ownerForHint(id)?.setPtyBackgrounded(id, background) } async shutdown( diff --git a/src/main/daemon/daemon-session-route-table.ts b/src/main/daemon/daemon-session-route-table.ts index fb5bc6c2dda..a8462e246a4 100644 --- a/src/main/daemon/daemon-session-route-table.ts +++ b/src/main/daemon/daemon-session-route-table.ts @@ -1,43 +1,12 @@ -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' - } -} +import { + createOwnedDaemonSessionRoute, + DaemonSessionGoneError, + DaemonSessionOwnerUnknownError, + type DaemonSessionRoute, + DaemonSessionUnavailableError, + sameDaemonIncarnation +} from './daemon-session-route' export class DaemonSessionRouteTable { private readonly routes = new Map() @@ -54,20 +23,42 @@ export class DaemonSessionRouteTable { return route?.state === 'owned' ? route.owner : undefined } + shouldForwardEvent(sessionId: string, owner: DaemonPtyAdapter): boolean { + const route = this.routes.get(sessionId) + return !route || (route.state === 'owned' && route.owner === owner) + } + + recordDataOwner(sessionId: string, owner: DaemonPtyAdapter): boolean { + const route = this.routes.get(sessionId) + if (!route) { + this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner)) + return true + } + if (route.state === 'unavailable') { + this.recordOwned(sessionId, owner) + return this.shouldForwardEvent(sessionId, owner) + } + if (route.state === 'ambiguous' || route.owner === owner) { + return route.state === 'owned' + } + this.recordOwned(sessionId, owner) + return false + } + recordOwned(sessionId: string, owner: DaemonPtyAdapter): void { const route = this.routes.get(sessionId) if (!route) { - this.routes.set(sessionId, ownedRoute(owner)) + this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner)) return } if (route.state === 'owned' && route.owner === owner) { - this.routes.set(sessionId, ownedRoute(owner)) + this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner)) return } if (route.state === 'unavailable') { const incarnation = owner.getLastAuthenticatedDaemonIdentity() - if (route.owner === owner && !sameIncarnation(route.incarnation, incarnation)) { - this.routes.set(sessionId, ownedRoute(owner)) + if (route.owner !== owner || !sameDaemonIncarnation(route.incarnation, incarnation)) { + this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner)) } return } @@ -78,7 +69,16 @@ export class DaemonSessionRouteTable { } transfer(sessionId: string, owner: DaemonPtyAdapter): void { - this.routes.set(sessionId, ownedRoute(owner)) + this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner)) + } + + recordFreshOwned(sessionId: string, owner: DaemonPtyAdapter): void { + const route = this.routes.get(sessionId) + if (route?.state === 'unavailable' && route.owner === owner) { + this.transfer(sessionId, owner) + return + } + this.recordOwned(sessionId, owner) } recordDiscoveryFailure(owner: DaemonPtyAdapter): void { @@ -111,11 +111,21 @@ export class DaemonSessionRouteTable { return } if (route.state === 'ambiguous' && route.candidates.has(owner)) { - this.routes.set(sessionId, { - state: 'unavailable', - owner, - incarnation: owner.getLastAuthenticatedDaemonIdentity() - }) + const candidates = new Set(route.candidates) + candidates.delete(owner) + if (candidates.size === 1) { + for (const candidate of candidates) { + this.routes.set(sessionId, createOwnedDaemonSessionRoute(candidate)) + } + } else if (candidates.size > 1) { + this.routes.set(sessionId, { state: 'ambiguous', candidates }) + } else { + this.routes.set(sessionId, { + state: 'unavailable', + owner, + incarnation: owner.getLastAuthenticatedDaemonIdentity() + }) + } } } @@ -124,10 +134,14 @@ export class DaemonSessionRouteTable { incarnation = owner.getLastAuthenticatedDaemonIdentity() ): void { for (const [sessionId, route] of this.routes) { + if (route.state === 'ambiguous' && route.candidates.has(owner)) { + this.markUnavailable(sessionId, owner) + continue + } if ( route.state === 'owned' && route.owner === owner && - sameIncarnation(route.incarnation, incarnation) + sameDaemonIncarnation(route.incarnation, incarnation) ) { this.routes.set(sessionId, { state: 'unavailable', @@ -150,6 +164,15 @@ export class DaemonSessionRouteTable { return current } if (route.state === 'unavailable') { + const ownerIncarnation = route.owner.getLastAuthenticatedDaemonIdentity() + if ( + route.incarnation && + ownerIncarnation && + !sameDaemonIncarnation(route.incarnation, ownerIncarnation) + ) { + this.routes.delete(sessionId) + return current + } throw new DaemonSessionUnavailableError(sessionId) } throw new DaemonSessionOwnerUnknownError(sessionId) @@ -163,10 +186,10 @@ export class DaemonSessionRouteTable { 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 candidates = + route?.state === 'ambiguous' && this.incompleteDiscoveries.size === 0 + ? [...route.candidates] + : this.adapters const results = await Promise.all( candidates.map(async (owner) => ({ owner, @@ -238,14 +261,6 @@ export class DaemonSessionRouteTable { } } -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) @@ -261,17 +276,3 @@ function hasPty(owner: DaemonPtyAdapter, sessionId: string): boolean | null { 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-session-route.ts b/src/main/daemon/daemon-session-route.ts new file mode 100644 index 00000000000..18882170c25 --- /dev/null +++ b/src/main/daemon/daemon-session-route.ts @@ -0,0 +1,64 @@ +import { DAEMON_SESSION_ROUTING_UNAVAILABLE_MARKER } from '../../shared/daemon-session-routing-error' +import type { DaemonEndpointIdentity } from './daemon-hello-protocol' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' + +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 function createOwnedDaemonSessionRoute( + owner: DaemonPtyAdapter +): Extract { + return { + state: 'owned', + owner, + incarnation: owner.getLastAuthenticatedDaemonIdentity() + } +} + +export function sameDaemonIncarnation( + 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 index 576212b9aeb..ad0b1f7b812 100644 --- a/src/main/daemon/daemon-snapshot-acknowledgement-routes.ts +++ b/src/main/daemon/daemon-snapshot-acknowledgement-routes.ts @@ -22,6 +22,8 @@ export class DaemonSnapshotAcknowledgementRoutes { ): void { if (snapshot) { this.producers.set(sessionId, producer) + } else { + this.producers.delete(sessionId) } } @@ -34,6 +36,8 @@ export class DaemonSnapshotAcknowledgementRoutes { const producer = adapters.find((adapter) => adapter === provider) if (producer) { this.record(sessionId, snapshot, producer) + } else { + this.producers.delete(sessionId) } } diff --git a/src/main/ipc/pty-management.test.ts b/src/main/ipc/pty-management.test.ts index 954ba42f276..ca5678ef462 100644 --- a/src/main/ipc/pty-management.test.ts +++ b/src/main/ipc/pty-management.test.ts @@ -289,6 +289,21 @@ describe('pty:management IPC handlers', () => { expect(legacy.shutdown).toHaveBeenCalledWith('old-1', { immediate: true }) }) + it('rejects before killing when the initial owner inventory is incomplete', 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() + + await expect(buildHandlerMap()['pty:management:killAll']({})).rejects.toThrow( + 'session inventory is unavailable' + ) + expect(current.shutdown).not.toHaveBeenCalled() + expect(legacy.shutdown).not.toHaveBeenCalled() + }) + it('reports remainingCount when sessions refuse to die after the poll window', async () => { const sessions = [makeSession('stuck')] const current = makeAdapter(5, []) diff --git a/src/main/ipc/pty-management.ts b/src/main/ipc/pty-management.ts index 29b9cbebe57..f53ff13646f 100644 --- a/src/main/ipc/pty-management.ts +++ b/src/main/ipc/pty-management.ts @@ -96,7 +96,11 @@ export function registerDaemonManagementHandlers(): void { killedSessionIds: string[] }> => { const adapters = getDaemonAdapters() - const initial = (await collectSessions(adapters)).sessions + const initialCollection = await collectSessions(adapters) + if (initialCollection.unavailableOwners.size > 0) { + throw new Error('Cannot kill daemon sessions while session inventory is unavailable') + } + const initial = initialCollection.sessions const initialCount = initial.length if (initialCount === 0) { diff --git a/src/main/ipc/pty.test.ts b/src/main/ipc/pty.test.ts index c0798fe39fa..2d376aa05c6 100644 --- a/src/main/ipc/pty.test.ts +++ b/src/main/ipc/pty.test.ts @@ -2839,8 +2839,10 @@ describe('registerPtyHandlers', () => { describe('daemon-active provider (parity with LocalPtyProvider)', () => { // Why: under the daemon, LocalPtyProvider.buildSpawnEnv never runs, so host-local env injection must happen in the pty:spawn handler instead. + type GitCredentialGuardSupport = boolean | ((sessionId?: string) => boolean) + function setupDaemonAdapter( - supportsGitCredentialGuardHost = true, + supportsGitCredentialGuardHost: GitCredentialGuardSupport = true, reportedWslDistro?: string | null, supportsAgentSessionClaims = true, supportsAgentSessionCreateOperations = supportsAgentSessionClaims @@ -2859,7 +2861,10 @@ describe('registerPtyHandlers', () => { ) setLocalPtyProvider({ spawn: daemonSpawn, - supportsGitCredentialGuardHost: () => supportsGitCredentialGuardHost, + supportsGitCredentialGuardHost: (sessionId?: string) => + typeof supportsGitCredentialGuardHost === 'function' + ? supportsGitCredentialGuardHost(sessionId) + : supportsGitCredentialGuardHost, supportsAgentSessionClaims: () => supportsAgentSessionClaims, supportsAgentSessionCreateOperations: () => supportsAgentSessionCreateOperations, write: vi.fn(), @@ -2943,7 +2948,7 @@ describe('registerPtyHandlers', () => { launchAgent?: TuiAgent envToDelete?: string[] }, - supportsGitCredentialGuardHost = true + supportsGitCredentialGuardHost: GitCredentialGuardSupport = true ): Promise { const daemonSpawn = setupDaemonAdapter(supportsGitCredentialGuardHost) const savedEnv: Record = {} @@ -3002,7 +3007,7 @@ describe('registerPtyHandlers', () => { command?: string launchAgent?: TuiAgent }, - supportsGitCredentialGuardHost = true + supportsGitCredentialGuardHost: GitCredentialGuardSupport = true ): Promise> { return ( await daemonSpawnAndGetOptions( @@ -4071,6 +4076,25 @@ describe('registerPtyHandlers', () => { }) }) + it('queries fresh daemon capability without resolving the minted session id', async () => { + const supportsGuard = vi.fn((sessionId?: string) => { + if (sessionId) { + throw new Error('fresh id was resolved as an existing route') + } + return true + }) + const daemonSpawn = setupDaemonAdapter(supportsGuard) + handlers.clear() + registerPtyHandlers(mainWindow as never) + + await expect( + handlers.get('pty:spawn')!(null, { cols: 80, rows: 24, env: {} }) + ).resolves.toBeDefined() + + expect(supportsGuard).toHaveBeenCalledExactlyOnceWith(undefined) + expect(daemonSpawn).toHaveBeenCalledOnce() + }) + it('respects a caller-provided sessionId instead of minting a new one', async () => { const daemonSpawn = setupDaemonAdapter() handlers.clear() @@ -6233,6 +6257,23 @@ describe('registerPtyHandlers', () => { ]) }) + it('reports unknown snapshot capability when session routing is unavailable', () => { + registerPtyHandlers(mainWindow as never) + setLocalPtyProvider({ + canProvideAuthoritativeBufferSnapshot() { + throw new Error('daemon_session_routing_unavailable') + } + } as never) + const listener = onMock.mock.calls.find( + ([channel]) => channel === 'pty:getAuthoritativeBufferSnapshotCapabilitiesSync' + )?.[1] as ((event: { returnValue?: unknown }, args: { ids: unknown[] }) => void) | undefined + const event: { returnValue?: unknown } = {} + + listener?.(event, { ids: ['ambiguous-pty'] }) + + expect(event.returnValue).toEqual([{ id: 'ambiguous-pty', authoritative: null }]) + }) + it('checks single-PTY liveness without listing every session', async () => { const hasPty = vi.fn((id: string) => id === 'live-pty') const listProcesses = vi.fn(async () => { diff --git a/src/main/ipc/pty.ts b/src/main/ipc/pty.ts index 260f0d4d77d..75fc48112a7 100644 --- a/src/main/ipc/pty.ts +++ b/src/main/ipc/pty.ts @@ -1298,6 +1298,23 @@ function routesFreshSpawnsToLocalProvider( return (provider as FreshLocalFallbackProvider).routesFreshSpawnsToLocalProvider === true } +function getAuthoritativeBufferSnapshotCapability( + provider: IPtyProvider | undefined, + ptyId: string +): boolean | null { + if (!provider) { + return null + } + if (provider.canProvideAuthoritativeBufferSnapshot) { + try { + return provider.canProvideAuthoritativeBufferSnapshot(ptyId) + } catch { + return null + } + } + return routesFreshSpawnsToLocalProvider(provider) ? false : null +} + function beginPtySpawnForWorktree( worktreeId: string | undefined, cwd: string | undefined, @@ -3813,7 +3830,9 @@ export function registerPtyHandlers( wslDistro: codexSelectionTarget.runtime === 'wsl' ? expectedWslDistro : null, agentStatusHooksEnabled: isAgentStatusHooksEnabled(getSettings?.()), networkProxySettings: getSettings?.(), - deferGitConfigGuardToDaemon: provider.supportsGitCredentialGuardHost?.(sessionId) === true + deferGitConfigGuardToDaemon: + provider.supportsGitCredentialGuardHost?.(isMintedSessionId ? undefined : sessionId) === + true }) stampWslOrchestrationCompatibilityHost( env, @@ -5001,7 +5020,9 @@ export function registerPtyHandlers( agentStatusHooksEnabled: isAgentStatusHooksEnabled(getSettings?.()), networkProxySettings: getSettings?.(), deferGitConfigGuardToDaemon: - provider.supportsGitCredentialGuardHost?.(effectiveSessionId) === true + provider.supportsGitCredentialGuardHost?.( + isMintedSessionId ? undefined : effectiveSessionId + ) === true }) stampWslOrchestrationCompatibilityHost( env, @@ -6128,11 +6149,7 @@ export function registerPtyHandlers( // Why: degraded routing mixes preserved daemons with an in-process fallback; keep all panes mounted rather than guess ownership. capabilities.push({ id: value, - authoritative: provider?.canProvideAuthoritativeBufferSnapshot - ? provider.canProvideAuthoritativeBufferSnapshot(value) - : provider && routesFreshSpawnsToLocalProvider(provider) - ? false - : null + authoritative: getAuthoritativeBufferSnapshotCapability(provider, value) }) } // Why: cold deferral runs during render before hidden panes mount; this in-memory route lookup lets legacy PTYs mount in that pass. 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 64d9cd51723..9261ded6a52 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.test.ts @@ -11126,6 +11126,8 @@ describe('connectPanePty', () => { expect(deps.clearTabPtyId).not.toHaveBeenCalled() expect(deps.updateTabPtyId).not.toHaveBeenCalled() expect(pane.terminal.clear).not.toHaveBeenCalled() + expect(window.api.pty.settlePaneSerializer).not.toHaveBeenCalled() + expect(window.api.pty.clearPendingPaneSerializer).toHaveBeenCalled() }) // Why: Phase 6 deleted the hidden-skip grammar — every hidden chunk rides the background scheduler, none content-scanned. diff --git a/src/renderer/src/components/terminal-pane/pty-connection.ts b/src/renderer/src/components/terminal-pane/pty-connection.ts index 58d6151a645..0842ea8858d 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.ts @@ -8253,7 +8253,12 @@ export function connectPanePty( finishReattachLiveDataDeferral(accepted, outputCallbacks.generation) const gen = await preSignalPromise if (typeof gen === 'number') { - if (!accepted) { + const routingUnavailable = + result && + typeof result === 'object' && + 'routingUnavailable' in result && + result.routingUnavailable === true + if (!accepted || routingUnavailable) { await window.api.pty.clearPendingPaneSerializer(cacheKey, gen).catch(() => {}) } else if (!isRemoteRuntimePtyId(pendingSessionId)) { const settledPtyId = @@ -8489,7 +8494,12 @@ export function connectPanePty( finishReattachLiveDataDeferral(accepted, outputCallbacks.generation) const gen = await preSignalPromise if (typeof gen === 'number') { - if (!accepted) { + const routingUnavailable = + result && + typeof result === 'object' && + 'routingUnavailable' in result && + result.routingUnavailable === true + if (!accepted || routingUnavailable) { await window.api.pty.clearPendingPaneSerializer(cacheKey, gen).catch(() => {}) } else if (!isRemoteRuntimePtyId(deferredReattachSessionId)) { const settledPtyId =