fix(daemon): fence session routing by owner

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