fix(daemon): harden generation route fences

This commit is contained in:
Brennan Benson
2026-07-30 22:37:03 -07:00
parent a8ed9dce24
commit 86d3331aa0
25 changed files with 2162 additions and 382 deletions
@@ -0,0 +1,46 @@
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import {
DaemonSessionOwnerUnknownError,
type DaemonSessionRoute,
DaemonSessionUnavailableError,
sameDaemonIncarnation
} from './daemon-session-route'
export function ownerForFreshDaemonSession(
sessionId: string | undefined,
route: DaemonSessionRoute | undefined,
current: DaemonPtyAdapter,
clearRoute: () => void
): DaemonPtyAdapter {
if (!sessionId || !route) {
return current
}
if (route.state === 'owned' && route.owner === current) {
return current
}
assertFreshDaemonSessionAvailable(sessionId, route, clearRoute)
return current
}
export function assertFreshDaemonSessionAvailable(
sessionId: string | undefined,
route: DaemonSessionRoute | undefined,
clearRoute: () => void
): void {
if (!sessionId || !route) {
return
}
if (route.state === 'unavailable') {
const ownerIncarnation = route.owner.getLastAuthenticatedDaemonIdentity()
if (
route.incarnation &&
ownerIncarnation &&
!sameDaemonIncarnation(route.incarnation, ownerIncarnation)
) {
clearRoute()
return
}
throw new DaemonSessionUnavailableError(sessionId)
}
throw new DaemonSessionOwnerUnknownError(sessionId)
}
+61
View File
@@ -510,4 +510,65 @@ describe('killStaleDaemon pid identity guards', () => {
await closeServer(replacementSocket)
}
})
it('preserves a replacement pidfile installed as the targeted process exits', async () => {
if (process.platform === 'win32') {
return
}
const replacementSocket = createServer((socket) => socket.end())
await new Promise<void>((resolve, reject) => {
replacementSocket.once('error', reject)
replacementSocket.listen(socketPath, () => {
replacementSocket.off('error', reject)
resolve()
})
})
const pidPath = getDaemonPidPath(dir)
const replacementPidfile = serializeDaemonPidFile({
pid: process.pid,
startedAtMs: null,
launchNonce: 'replacement'
})
const child = spawn(
process.execPath,
[
'-e',
"const fs=require('node:fs');const [pidPath,pidfile]=process.argv.slice(1);process.on('SIGTERM',()=>{fs.writeFileSync(pidPath,pidfile);process.exit(0)});process.send?.('ready');setInterval(()=>{},1000)",
pidPath,
replacementPidfile,
'daemon-entry',
socketPath,
tokenPath
],
{ stdio: ['ignore', 'ignore', 'ignore', 'ipc'] }
)
await new Promise<void>((resolve, reject) => {
child.once('error', reject)
child.on('message', (message) => {
if (message === 'ready') {
resolve()
}
})
})
writeFileSync(
pidPath,
serializeDaemonPidFile({
pid: child.pid!,
startedAtMs: null,
launchNonce: 'target'
}),
{ mode: 0o600 }
)
try {
await expect(killStaleDaemon(dir, socketPath, tokenPath)).resolves.toBe(true)
expect(readFileSync(pidPath, 'utf8')).toBe(replacementPidfile)
} finally {
if (child.exitCode === null) {
child.kill('SIGKILL')
}
await closeServer(replacementSocket)
}
})
})
+7 -7
View File
@@ -10,7 +10,7 @@ import {
} from '../../shared/process-output-field-scanner'
import { isStartupDiagnosticsEnabled, logStartupDiagnostic } from '../startup/startup-diagnostics'
import { encodeNdjson } from './ndjson'
import { getDaemonPidPath } from './daemon-spawner'
import { getDaemonPidPath, unlinkOwnedDaemonPidFile } from './daemon-spawner'
import {
PROTOCOL_VERSION,
type HelloMessage,
@@ -666,6 +666,8 @@ export async function killStaleDaemon(
const pidPath = getDaemonPidPath(runtimeDir, protocolVersion)
let killedDaemon = false
let identifiedDaemon = false
let identifiedPid: number | null = null
let identifiedLaunchNonce: string | null = null
try {
const parsedPid = parseDaemonPidFile(readFileSync(pidPath, 'utf8'))
if (
@@ -673,6 +675,8 @@ export async function killStaleDaemon(
(await isDaemonProcess(parsedPid.pid, socketPath, tokenPath, parsedPid.startedAtMs))
) {
identifiedDaemon = true
identifiedPid = parsedPid.pid
identifiedLaunchNonce = parsedPid.launchNonce
const { pid, startedAtMs } = parsedPid
process.kill(pid, 'SIGTERM')
const deadline = Date.now() + KILL_WAIT_MS
@@ -711,12 +715,8 @@ export async function killStaleDaemon(
// PID file missing or process already dead
}
if (identifiedDaemon) {
try {
unlinkSync(pidPath)
} catch {
// Best-effort
}
if (identifiedDaemon && identifiedPid !== null && identifiedLaunchNonce !== null) {
unlinkOwnedDaemonPidFile(pidPath, identifiedPid, identifiedLaunchNonce)
}
const socketIsLive = await canConnectSocket(socketPath)
@@ -1527,6 +1527,7 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => {
await adapter.listProcesses()
const firstIdentity = adapter.getLastAuthenticatedDaemonIdentity()
expect(firstIdentity).not.toBeNull()
expect(adapter.matchesLastAuthenticatedDaemonIdentity(firstIdentity)).toBe(true)
const identityChanges: {
previous: NonNullable<typeof firstIdentity>
current: NonNullable<typeof firstIdentity>
@@ -1568,6 +1569,10 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => {
}
])
expect(adapter.getLastAuthenticatedDaemonIdentity()).toEqual(identityChanges[0]?.current)
expect(adapter.matchesLastAuthenticatedDaemonIdentity(firstIdentity)).toBe(false)
expect(
adapter.matchesLastAuthenticatedDaemonIdentity(identityChanges[0]?.current ?? null)
).toBe(true)
})
it('isolates audit observation listeners from inventory and later listeners', async () => {
+7
View File
@@ -299,6 +299,13 @@ export class DaemonPtyAdapter implements IPtyProvider {
return this.lastAuthenticatedIdentity ? { ...this.lastAuthenticatedIdentity } : null
}
matchesLastAuthenticatedDaemonIdentity(identity: DaemonEndpointIdentity | null): boolean {
if (!identity || !this.lastAuthenticatedIdentity) {
return identity === this.lastAuthenticatedIdentity
}
return sameEndpointIdentity(identity, this.lastAuthenticatedIdentity)
}
getLastAuditObservation(): DaemonAuditObservation | null {
return this.lastAuditObservation
}
@@ -1,20 +0,0 @@
import type { IPtyProvider } from '../providers/types'
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
export async function probePtyOwners(
id: string,
routed: IPtyProvider | undefined,
possibleOwners: readonly DaemonPtyAdapter[]
): Promise<boolean | null> {
if (routed) {
return routed.probePtyLiveness
? await routed.probePtyLiveness(id)
: (routed.hasPty?.(id) ?? null)
}
const results = await Promise.all(possibleOwners.map((provider) => provider.probePtyLiveness(id)))
return results.some((result) => result === true)
? true
: results.every((result) => result === false)
? false
: null
}
@@ -6,7 +6,9 @@ import type { PtyProcessInfo, PtySpawnOptions } from '../providers/types'
type AdapterHarness = {
adapter: DaemonPtyAdapter
emitData: (sessionId: string, data: string) => void
emitIdentityChange: (previous: DaemonEndpointIdentity, current: DaemonEndpointIdentity) => void
emitExit: (sessionId: string, code: number) => void
setIdentity: (identity: DaemonEndpointIdentity) => void
}
@@ -28,9 +30,18 @@ function createAdapter(
previous: DaemonEndpointIdentity
current: DaemonEndpointIdentity
}) => void)[] = []
const dataListeners: ((payload: { id: string; data: string }) => void)[] = []
const exitListeners: ((payload: { id: string; code: number }) => void)[] = []
const adapter = {
protocolVersion: label === 'current' ? 30 : 29,
getLastAuthenticatedDaemonIdentity: vi.fn(() => ({ ...daemonIdentity })),
matchesLastAuthenticatedDaemonIdentity: vi.fn(
(candidate: DaemonEndpointIdentity | null) =>
candidate !== null &&
candidate.pid === daemonIdentity.pid &&
candidate.startedAtMs === daemonIdentity.startedAtMs &&
candidate.launchNonce === daemonIdentity.launchNonce
),
onDaemonIdentityChanged: vi.fn(
(
listener: (event: {
@@ -55,8 +66,14 @@ function createAdapter(
getBufferSnapshot: vi.fn(async () => null),
ackColdRestore: vi.fn(),
reconcileOnStartup: vi.fn(async () => reconcileResult),
onData: vi.fn(() => () => {}),
onExit: vi.fn(() => () => {}),
onData: vi.fn((listener: (payload: { id: string; data: string }) => void) => {
dataListeners.push(listener)
return () => {}
}),
onExit: vi.fn((listener: (payload: { id: string; code: number }) => void) => {
exitListeners.push(listener)
return () => {}
}),
onBackgroundStreamEvent: vi.fn(() => () => {}),
onWriteUnavailable: vi.fn(() => () => {}),
dispose: vi.fn(),
@@ -65,6 +82,11 @@ function createAdapter(
return {
adapter,
emitData: (sessionId, data) => {
for (const listener of dataListeners) {
listener({ id: sessionId, data })
}
},
setIdentity: (next) => {
daemonIdentity = next
},
@@ -72,6 +94,11 @@ function createAdapter(
for (const listener of identityListeners) {
listener({ previous, current })
}
},
emitExit: (sessionId, code) => {
for (const listener of exitListeners) {
listener({ id: sessionId, code })
}
}
}
}
@@ -81,6 +108,25 @@ describe('DaemonPtyRouter routing safety', () => {
vi.restoreAllMocks()
})
it('does not clone daemon identity for established stream chunks', () => {
const current = createAdapter('current')
const data = vi.fn()
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: []
})
router.onData(data)
current.emitData('stream-session', 'first')
vi.mocked(current.adapter.getLastAuthenticatedDaemonIdentity).mockClear()
for (let index = 0; index < 100; index += 1) {
current.emitData('stream-session', `chunk-${index}`)
}
expect(data).toHaveBeenCalledTimes(101)
expect(current.adapter.getLastAuthenticatedDaemonIdentity).not.toHaveBeenCalled()
})
it('uses conclusive probes while fencing absence after a failed discovery', async () => {
const current = createAdapter('current')
const legacy = createAdapter('legacy', ['legacy-session'])
@@ -151,6 +197,36 @@ describe('DaemonPtyRouter routing safety', () => {
expect(undiscovered.adapter.sendSignal).not.toHaveBeenCalled()
})
it('probes every adapter before collapsing an ambiguous route', async () => {
const sessionId = 'late-collision'
const firstSessions = [sessionId]
const lateSessions: string[] = []
const current = createAdapter('current', [sessionId], {
alive: [sessionId],
killed: []
})
const first = createAdapter('legacy-first', firstSessions, {
alive: [sessionId],
killed: []
})
const late = createAdapter('legacy-late', lateSessions)
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: [first.adapter, late.adapter]
})
await router.reconcileOnStartup(new Set())
firstSessions.splice(0)
lateSessions.push(sessionId)
await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow(
'daemon_session_routing_unavailable'
)
expect(current.adapter.sendSignal).not.toHaveBeenCalled()
expect(first.adapter.sendSignal).not.toHaveBeenCalled()
expect(late.adapter.sendSignal).not.toHaveBeenCalled()
})
it('fails existing operations unknown when an owner probe is inconclusive', async () => {
const current = createAdapter('current')
const legacy = createAdapter('legacy')
@@ -175,7 +251,41 @@ describe('DaemonPtyRouter routing safety', () => {
expect(current.adapter.sendSignal).not.toHaveBeenCalled()
})
it('probes every adapter once and memoizes the unique owner', async () => {
it('keeps inconclusive synchronous liveness non-authoritative', async () => {
const sessionId = 'uncertain-liveness'
const current = createAdapter('current')
const legacy = createAdapter('legacy')
vi.mocked(legacy.adapter.listProcesses).mockRejectedValueOnce(new Error('listing unavailable'))
vi.spyOn(console, 'warn').mockImplementation(() => {})
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: [legacy.adapter]
})
await router.discoverLegacySessions()
expect(() => router.hasPty(sessionId)).toThrow('daemon_session_routing_unavailable')
legacy.emitExit(sessionId, 0)
expect(() => router.hasPty(sessionId)).toThrow('daemon_session_routing_unavailable')
})
it('keeps replacement-owned synchronous liveness non-authoritative', async () => {
const sessionId = 'replacement-liveness'
const legacy = createAdapter('legacy', [sessionId])
const router = new DaemonPtyRouter({
current: createAdapter('current').adapter,
legacy: [legacy.adapter]
})
await router.discoverLegacySessions()
legacy.setIdentity(identity('legacy', 11))
expect(() => router.hasPty(sessionId)).toThrow('daemon_session_routing_unavailable')
expect(legacy.adapter.hasPty).not.toHaveBeenCalled()
})
it('deduplicates concurrent owner probes and memoizes the unique owner', async () => {
const current = createAdapter('current')
const legacy = createAdapter('legacy', ['surviving-session'])
const router = new DaemonPtyRouter({
@@ -183,8 +293,10 @@ describe('DaemonPtyRouter routing safety', () => {
legacy: [legacy.adapter]
})
await router.sendSignal('surviving-session', 'SIGINT')
await router.shutdown('surviving-session', { immediate: true })
await Promise.all([
router.sendSignal('surviving-session', 'SIGINT'),
router.shutdown('surviving-session', { immediate: true })
])
expect(current.adapter.probePtyLiveness).toHaveBeenCalledExactlyOnceWith('surviving-session')
expect(legacy.adapter.probePtyLiveness).toHaveBeenCalledExactlyOnceWith('surviving-session')
@@ -249,6 +361,193 @@ describe('DaemonPtyRouter routing safety', () => {
expect(legacy.adapter.sendSignal).not.toHaveBeenCalled()
})
it('does not collapse an ambiguous route onto a replacement incarnation', async () => {
const sessionId = 'replacement-collision'
const currentSessions = [sessionId]
const current = createAdapter('current', currentSessions, {
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())
currentSessions.splice(0)
legacy.setIdentity(identity('legacy', 11))
await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow(
'daemon_session_routing_unavailable'
)
expect(current.adapter.sendSignal).not.toHaveBeenCalled()
expect(legacy.adapter.sendSignal).not.toHaveBeenCalled()
})
it('does not admit a replacement ambiguous candidate after the other owner exits', async () => {
const sessionId = 'stale-identity-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)
await router.discoverLegacySessions()
legacy.emitIdentityChange(previous, replacement)
current.emitExit(sessionId, 0)
expect(router.getSessionRouteState(sessionId)).toBe('unavailable')
await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow(
'daemon_session_routing_unavailable'
)
expect(current.adapter.sendSignal).not.toHaveBeenCalled()
expect(legacy.adapter.sendSignal).not.toHaveBeenCalled()
})
it('does not transfer an owned route from replacement output', async () => {
const sessionId = 'replacement-output'
const current = createAdapter('current')
const legacy = createAdapter('legacy', [sessionId])
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: [legacy.adapter]
})
const data = vi.fn()
router.onData(data)
await router.discoverLegacySessions()
legacy.setIdentity(identity('legacy', 11))
legacy.emitData(sessionId, 'replacement frame')
expect(data).not.toHaveBeenCalled()
expect(router.getSessionRouteState(sessionId)).toBe('unavailable')
await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow(
'daemon_session_routing_unavailable'
)
})
it('rejects an owner probe whose daemon incarnation changes in flight', async () => {
const sessionId = 'probe-replacement-race'
const current = createAdapter('current')
const legacy = createAdapter('legacy', [sessionId])
let finishProbe: ((result: boolean) => void) | undefined
vi.mocked(legacy.adapter.probePtyLiveness).mockReturnValue(
new Promise((resolve) => {
finishProbe = resolve
})
)
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: [legacy.adapter]
})
const previous = identity('legacy', 10)
const replacement = identity('legacy', 11)
const signaling = router.sendSignal(sessionId, 'SIGTERM')
await vi.waitFor(() => expect(finishProbe).toBeDefined())
legacy.setIdentity(replacement)
legacy.emitIdentityChange(previous, replacement)
finishProbe?.(true)
await expect(signaling).rejects.toThrow('daemon_session_routing_unavailable')
expect(current.adapter.sendSignal).not.toHaveBeenCalled()
expect(legacy.adapter.sendSignal).not.toHaveBeenCalled()
})
it('does not apply reconciliation after an earlier adapter is replaced', async () => {
const sessionId = 'stale-reconciliation'
const current = createAdapter('current', [], {
alive: [sessionId],
killed: []
})
const legacy = createAdapter('legacy')
let finishLegacyReconciliation:
| ((result: { alive: string[]; killed: string[] }) => void)
| undefined
vi.mocked(legacy.adapter.reconcileOnStartup).mockReturnValue(
new Promise((resolve) => {
finishLegacyReconciliation = resolve
})
)
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: [legacy.adapter]
})
const previous = identity('current', 20)
const replacement = identity('current', 21)
const reconciling = router.reconcileOnStartup(new Set())
await vi.waitFor(() => expect(finishLegacyReconciliation).toBeDefined())
current.setIdentity(replacement)
current.emitIdentityChange(previous, replacement)
await router.spawn({
sessionId,
isNewSession: true,
cols: 80,
rows: 24
})
finishLegacyReconciliation?.({ alive: [], killed: [] })
await expect(reconciling).rejects.toThrow(
'daemon incarnation changed during startup reconciliation'
)
expect(router.getSessionRouteState(sessionId)).toBe('owned')
await router.sendSignal(sessionId, 'SIGTERM')
expect(current.adapter.sendSignal).toHaveBeenCalledExactlyOnceWith(sessionId, 'SIGTERM')
})
it('rejects a conclusive-absence result when discovery becomes incomplete in flight', async () => {
const sessionId = 'stale-absence-probe'
const current = createAdapter('current')
const legacy = createAdapter('legacy')
let finishCurrentProbe: ((result: boolean) => void) | undefined
let finishLegacyProbe: ((result: boolean) => void) | undefined
vi.mocked(current.adapter.probePtyLiveness).mockReturnValue(
new Promise((resolve) => {
finishCurrentProbe = resolve
})
)
vi.mocked(legacy.adapter.probePtyLiveness).mockReturnValue(
new Promise((resolve) => {
finishLegacyProbe = resolve
})
)
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: [legacy.adapter]
})
const signaling = router.sendSignal(sessionId, 'SIGTERM')
await vi.waitFor(() => {
expect(finishCurrentProbe).toBeDefined()
expect(finishLegacyProbe).toBeDefined()
})
vi.mocked(legacy.adapter.listProcesses).mockRejectedValueOnce(new Error('listing unavailable'))
vi.spyOn(console, 'warn').mockImplementation(() => {})
await router.discoverLegacySessions()
finishCurrentProbe?.(false)
finishLegacyProbe?.(false)
await expect(signaling).rejects.toThrow('daemon_session_routing_unavailable')
expect(current.adapter.sendSignal).not.toHaveBeenCalled()
expect(legacy.adapter.sendSignal).not.toHaveBeenCalled()
})
it('keeps a canonical spawn collision ambiguous', async () => {
const current = createAdapter('current')
const legacy = createAdapter('legacy', ['canonical-session'])
@@ -279,7 +578,7 @@ describe('DaemonPtyRouter routing safety', () => {
expect(legacy.adapter.sendSignal).not.toHaveBeenCalled()
})
it('scopes unavailable tombstones to the authenticated daemon incarnation', async () => {
it('keeps an existing id tombstoned when a replacement reports the same id', async () => {
const sessionId = 'legacy-session'
const current = createAdapter('current')
const legacy = createAdapter('legacy', [sessionId])
@@ -301,9 +600,31 @@ describe('DaemonPtyRouter routing safety', () => {
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')
expect(router.getSessionRouteState(sessionId)).toBe('unavailable')
await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow(
'daemon_session_routing_unavailable'
)
expect(legacy.adapter.sendSignal).not.toHaveBeenCalled()
})
it('does not transfer an owned route from replacement discovery', async () => {
const sessionId = 'replacement-discovery'
const current = createAdapter('current')
const legacy = createAdapter('legacy', [sessionId])
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: [legacy.adapter]
})
await router.discoverLegacySessions()
legacy.setIdentity(identity('legacy', 11))
await router.discoverLegacySessions()
expect(router.getSessionRouteState(sessionId)).toBe('unavailable')
await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow(
'daemon_session_routing_unavailable'
)
expect(legacy.adapter.sendSignal).not.toHaveBeenCalled()
})
it('allows explicit fresh reuse after the tombstone owner is replaced', async () => {
@@ -389,6 +710,166 @@ describe('DaemonPtyRouter routing safety', () => {
expect(legacy.adapter.ackColdRestore).not.toHaveBeenCalled()
})
it('keeps the newest snapshot producer when captures finish out of order', async () => {
const sessionId = 'concurrent-snapshot'
const current = createAdapter('current')
const legacy = createAdapter('legacy', [sessionId])
let finishFirstCapture: ((snapshot: null) => void) | undefined
vi.mocked(legacy.adapter.getBufferSnapshot)
.mockImplementationOnce(
() =>
new Promise((resolve) => {
finishFirstCapture = resolve
})
)
.mockResolvedValueOnce({
data: 'newest frame',
cols: 80,
rows: 24,
seq: 2,
source: 'headless'
})
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: [legacy.adapter]
})
await router.discoverLegacySessions()
const firstCapture = router.getBufferSnapshot(sessionId)
await vi.waitFor(() => expect(finishFirstCapture).toBeDefined())
await router.getBufferSnapshot(sessionId)
finishFirstCapture?.(null)
await firstCapture
router.ackColdRestore(sessionId)
expect(legacy.adapter.ackColdRestore).toHaveBeenCalledExactlyOnceWith(sessionId)
})
it('does not acknowledge an old snapshot against a replacement daemon', async () => {
const sessionId = 'replacement-snapshot'
const current = createAdapter('current')
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]
})
const previous = identity('legacy', 10)
const replacement = identity('legacy', 11)
await router.discoverLegacySessions()
await router.getBufferSnapshot(sessionId)
legacy.setIdentity(replacement)
legacy.emitIdentityChange(previous, replacement)
router.ackColdRestore(sessionId)
expect(legacy.adapter.ackColdRestore).not.toHaveBeenCalled()
})
it('does not acknowledge a snapshot whose producer changes incarnation in flight', async () => {
const sessionId = 'in-flight-replacement-snapshot'
const current = createAdapter('current')
const legacy = createAdapter('legacy', [sessionId])
let finishSnapshot:
| ((snapshot: {
data: string
cols: number
rows: number
seq: number
source: 'headless'
}) => void)
| undefined
vi.mocked(legacy.adapter.getBufferSnapshot).mockImplementation(
() =>
new Promise((resolve) => {
finishSnapshot = resolve
})
)
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: [legacy.adapter]
})
const previous = identity('legacy', 10)
const replacement = identity('legacy', 11)
await router.discoverLegacySessions()
const capturing = router.getBufferSnapshot(sessionId)
await vi.waitFor(() => expect(finishSnapshot).toBeDefined())
legacy.setIdentity(replacement)
legacy.emitIdentityChange(previous, replacement)
finishSnapshot?.({
data: 'legacy frame',
cols: 80,
rows: 24,
seq: 1,
source: 'headless'
})
await capturing
router.ackColdRestore(sessionId)
expect(legacy.adapter.ackColdRestore).not.toHaveBeenCalled()
})
it('drops pending snapshot acknowledgements only for the exiting producer', async () => {
const sessionId = 'exiting-snapshot'
const current = createAdapter('current')
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)
current.emitExit(sessionId, 0)
router.ackColdRestore(sessionId)
expect(legacy.adapter.ackColdRestore).toHaveBeenCalledExactlyOnceWith(sessionId)
await router.getBufferSnapshot(sessionId)
legacy.emitExit(sessionId, 0)
router.ackColdRestore(sessionId)
expect(legacy.adapter.ackColdRestore).toHaveBeenCalledOnce()
})
it('bounds unavailable route retention', async () => {
const killed = Array.from({ length: 1001 }, (_, index) => `closed-${index}`)
const currentSessions: string[] = []
const current = createAdapter('current', currentSessions, { alive: [], killed })
const router = new DaemonPtyRouter({
current: current.adapter,
legacy: []
})
const exit = vi.fn()
router.onExit(exit)
await router.reconcileOnStartup(new Set())
expect(router.getSessionRouteState('closed-0')).toBeNull()
expect(router.getSessionRouteState('closed-1')).toBe('unavailable')
expect(router.getSessionRouteState('closed-1000')).toBe('unavailable')
currentSessions.push('closed-0')
await expect(router.sendSignal('closed-0', 'SIGTERM')).rejects.toThrow(
'daemon_session_routing_unavailable'
)
expect(() => current.emitExit('closed-0', 0)).not.toThrow()
expect(exit).not.toHaveBeenCalled()
expect(current.adapter.sendSignal).not.toHaveBeenCalled()
})
it('passes folder and git-worktree keys through the same reconciliation path', async () => {
const current = createAdapter('current')
const legacy = createAdapter('legacy')
@@ -1,11 +1,16 @@
import type { PtyProviderBufferSnapshot, PtySpawnOptions, PtySpawnResult } from '../providers/types'
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import type { DaemonEndpointIdentity } from './daemon-hello-protocol'
import { DaemonSessionRouteTable } from './daemon-session-route-table'
import { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowledgement-routes'
export class DaemonPtyRouterSessionRouting {
private readonly routes: DaemonSessionRouteTable
private readonly snapshotAcks = new DaemonSnapshotAcknowledgementRoutes()
private readonly freshSpawnsInFlight = new Map<
string,
{ owner: DaemonPtyAdapter; count: number }
>()
private readonly identityUnsubscribes: (() => void)[] = []
constructor(
@@ -16,16 +21,22 @@ export class DaemonPtyRouterSessionRouting {
for (const adapter of adapters) {
this.identityUnsubscribes.push(
adapter.onDaemonIdentityChanged(({ previous }) => {
this.snapshotAcks.dropAdapterIncarnation(adapter, previous)
this.routes.markOwnerUnavailable(adapter, previous)
})
)
}
}
recordDiscovery(adapter: DaemonPtyAdapter, sessions: readonly { id: string }[]): void {
recordDiscovery(
adapter: DaemonPtyAdapter,
sessions: readonly { id: string }[],
incarnation?: DaemonEndpointIdentity | null
): void {
this.routes.recordCompleteDiscovery(
adapter,
sessions.map((session) => session.id)
sessions.map((session) => session.id),
incarnation
)
}
@@ -51,6 +62,32 @@ export class DaemonPtyRouterSessionRouting {
this.routes.recordOwned(result.id, target)
}
beginSpawn(opts: PtySpawnOptions, target: DaemonPtyAdapter): void {
if (!opts.isNewSession || !opts.sessionId) {
return
}
const inFlight = this.freshSpawnsInFlight.get(opts.sessionId)
this.freshSpawnsInFlight.set(opts.sessionId, {
owner: target,
count: (inFlight?.count ?? 0) + 1
})
}
endSpawn(opts: PtySpawnOptions): void {
if (!opts.isNewSession || !opts.sessionId) {
return
}
const inFlight = this.freshSpawnsInFlight.get(opts.sessionId)
if (!inFlight || inFlight.count <= 1) {
this.freshSpawnsInFlight.delete(opts.sessionId)
return
}
this.freshSpawnsInFlight.set(opts.sessionId, {
owner: inFlight.owner,
count: inFlight.count - 1
})
}
async owner(sessionId: string): Promise<DaemonPtyAdapter> {
return await this.routes.resolveOwner(sessionId)
}
@@ -72,14 +109,7 @@ export class DaemonPtyRouterSessionRouting {
}
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))
return this.routes.hasPty(sessionId)
}
async probePtyLiveness(sessionId: string): Promise<boolean | null> {
@@ -99,9 +129,7 @@ export class DaemonPtyRouterSessionRouting {
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
return await this.snapshotAcks.capture(sessionId, opts, owner, this.adapters)
}
acknowledgeBufferSnapshot(sessionId: string): void {
@@ -120,17 +148,37 @@ export class DaemonPtyRouterSessionRouting {
}
if (!keepHistory) {
this.routes.markUnavailable(sessionId, owner)
this.snapshotAcks.drop(sessionId)
this.snapshotAcks.dropForProducer(sessionId, owner)
}
}
recordReconciledAlive(sessionId: string, owner: DaemonPtyAdapter): void {
this.routes.recordOwned(sessionId, owner)
recordReconciledAlive(
sessionId: string,
owner: DaemonPtyAdapter,
incarnation?: DaemonEndpointIdentity | null
): void {
this.routes.recordOwnedIncarnation(
sessionId,
owner,
incarnation ?? owner.getLastAuthenticatedDaemonIdentity()
)
}
recordReconciledKilled(sessionId: string, owner: DaemonPtyAdapter): void {
this.routes.markUnavailable(sessionId, owner)
this.snapshotAcks.drop(sessionId)
recordReconciledKilled(
sessionId: string,
owner: DaemonPtyAdapter,
incarnation?: DaemonEndpointIdentity | null
): void {
this.routes.markUnavailable(
sessionId,
owner,
incarnation ?? owner.getLastAuthenticatedDaemonIdentity()
)
this.snapshotAcks.dropForProducer(
sessionId,
owner,
incarnation ?? owner.getLastAuthenticatedDaemonIdentity()
)
}
shouldForwardEvent(sessionId: string, owner: DaemonPtyAdapter): boolean {
@@ -138,18 +186,36 @@ export class DaemonPtyRouterSessionRouting {
}
shouldForwardStreamEvent(sessionId: string, owner: DaemonPtyAdapter): boolean {
const freshSpawn = this.freshSpawnsInFlight.get(sessionId)
if (freshSpawn && freshSpawn.owner !== owner) {
this.routes.recordFreshOwned(sessionId, freshSpawn.owner)
this.routes.recordOwned(sessionId, owner)
return false
}
return this.routes.recordDataOwner(sessionId, owner)
}
recordExit(sessionId: string, owner: DaemonPtyAdapter): boolean {
const freshSpawn = this.freshSpawnsInFlight.get(sessionId)
if (freshSpawn && freshSpawn.owner !== owner) {
if (this.routes.get(sessionId)) {
this.routes.markUnavailable(sessionId, owner)
}
this.snapshotAcks.dropForProducer(sessionId, owner)
return false
}
const shouldForward = this.shouldForwardEvent(sessionId, owner)
this.routes.markUnavailable(sessionId, owner)
this.snapshotAcks.dropForProducer(sessionId, owner)
return shouldForward
}
markAdapterUnavailable(adapter: DaemonPtyAdapter): void {
markAdapterUnavailable(
adapter: DaemonPtyAdapter,
incarnation = adapter.getLastAuthenticatedDaemonIdentity()
): void {
if (this.adapters.includes(adapter)) {
this.routes.markOwnerUnavailable(adapter)
this.routes.markOwnerUnavailable(adapter, incarnation)
}
}
@@ -165,5 +231,7 @@ export class DaemonPtyRouterSessionRouting {
for (const unsubscribe of this.identityUnsubscribes.splice(0)) {
unsubscribe()
}
this.freshSpawnsInFlight.clear()
this.snapshotAcks.clear()
}
}
+35
View File
@@ -533,6 +533,41 @@ describe('DaemonPtyRouter', () => {
)
})
it('fences foreign stream events received before a fresh spawn reply', async () => {
const sessionId = 'fresh-stale-events'
const current = createAdapter('current')
const legacy = createAdapter('legacy')
let resolveSpawn: ((result: PtySpawnResult) => void) | undefined
vi.mocked(current.spawn).mockReturnValue(
new Promise((resolve) => {
resolveSpawn = resolve
})
)
const router = new DaemonPtyRouter({ current, legacy: [legacy] })
const data = vi.fn()
const exit = vi.fn()
router.onData(data)
router.onExit(exit)
const spawning = router.spawn({
sessionId,
isNewSession: true,
cols: 80,
rows: 24
})
legacy.emitData(sessionId, 'stale legacy frame')
legacy.emitExit(sessionId, 0)
resolveSpawn?.({ id: sessionId })
await spawning
expect(data).not.toHaveBeenCalled()
expect(exit).not.toHaveBeenCalled()
expect(router.getSessionRouteState(sessionId)).toBe('owned')
await router.sendSignal(sessionId, 'SIGTERM')
expect(current.sendSignal).toHaveBeenCalledExactlyOnceWith(sessionId, 'SIGTERM')
expect(legacy.sendSignal).not.toHaveBeenCalled()
})
it('forwards gap events and explicit sequence accounting from every adapter', () => {
const current = createAdapter('current')
const legacy = createAdapter('legacy')
+42 -8
View File
@@ -12,6 +12,7 @@ import type { PtyProcessInspection } from '../providers/pty-process-inspection'
import { shouldHandoffDaemonHistory } from './daemon-history-handoff'
import type { DaemonPtyRouterDataEvent, DaemonPtyRouterExitEvent } from './daemon-pty-router-events'
import { DaemonPtyRouterSessionRouting } from './daemon-pty-router-session-routing'
import { sameDaemonIncarnation } from './daemon-session-route'
export class DaemonPtyRouter implements IPtyProvider {
private current: DaemonPtyAdapter
@@ -45,8 +46,13 @@ export class DaemonPtyRouter implements IPtyProvider {
async discoverLegacySessions(): Promise<void> {
for (const adapter of this.legacy) {
try {
const before = adapter.getLastAuthenticatedDaemonIdentity()
const sessions = await adapter.listProcesses()
this.routing.recordDiscovery(adapter, sessions)
const incarnation = adapter.getLastAuthenticatedDaemonIdentity()
if (before && !sameDaemonIncarnation(before, incarnation)) {
throw new Error('daemon incarnation changed during session discovery')
}
this.routing.recordDiscovery(adapter, sessions, incarnation)
} catch (error) {
this.routing.recordDiscoveryFailure(adapter)
console.warn('[daemon] Failed to discover legacy daemon sessions', error)
@@ -57,9 +63,14 @@ export class DaemonPtyRouter implements IPtyProvider {
async spawn(opts: PtySpawnOptions): Promise<PtySpawnResult> {
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, opts)
return result
this.routing.beginSpawn(opts, target)
try {
const result = await target.spawn(opts)
this.routing.recordSpawn(result, target, opts)
return result
} finally {
this.routing.endSpawn(opts)
}
}
supportsGitCredentialGuardHost(sessionId?: string): boolean {
@@ -241,8 +252,21 @@ export class DaemonPtyRouter implements IPtyProvider {
}> {
const alive: string[] = []
const killed: string[] = []
const reconciled: {
adapter: DaemonPtyAdapter
incarnation: ReturnType<DaemonPtyAdapter['getLastAuthenticatedDaemonIdentity']>
alive: string[]
killed: string[]
}[] = []
for (const adapter of this.allAdapters()) {
const before = adapter.getLastAuthenticatedDaemonIdentity()
const result = await adapter.reconcileOnStartup(validWorktreeIds)
const incarnation = adapter.getLastAuthenticatedDaemonIdentity()
if (before && !sameDaemonIncarnation(before, incarnation)) {
this.routing.markAdapterUnavailable(adapter, before)
throw new Error('daemon incarnation changed during startup reconciliation')
}
reconciled.push({ adapter, incarnation, ...result })
// Why: daemon startup can reconcile many restored sessions; spreading
// those arrays into push can exceed JavaScript's argument limit.
for (const id of result.alive) {
@@ -251,11 +275,21 @@ export class DaemonPtyRouter implements IPtyProvider {
for (const id of result.killed) {
killed.push(id)
}
for (const id of result.alive) {
this.routing.recordReconciledAlive(id, adapter)
}
for (const { adapter, incarnation } of reconciled) {
if (!sameDaemonIncarnation(incarnation, adapter.getLastAuthenticatedDaemonIdentity())) {
this.routing.markAdapterUnavailable(adapter, incarnation)
throw new Error('daemon incarnation changed during startup reconciliation')
}
for (const id of result.killed) {
this.routing.recordReconciledKilled(id, adapter)
}
for (const { adapter, incarnation, alive: sessionIds } of reconciled) {
for (const id of sessionIds) {
this.routing.recordReconciledAlive(id, adapter, incarnation)
}
}
for (const { adapter, incarnation, killed: sessionIds } of reconciled) {
for (const id of sessionIds) {
this.routing.recordReconciledKilled(id, adapter, incarnation)
}
}
return { alive, killed }
@@ -0,0 +1,49 @@
import type { DaemonEndpointIdentity } from './daemon-hello-protocol'
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import { sameDaemonIncarnation } from './daemon-session-route'
export type DaemonSessionOwnerProbe = {
owner: DaemonPtyAdapter
result: boolean | null
incarnation: DaemonEndpointIdentity | null
}
export async function probeDaemonSessionOwner(
owner: DaemonPtyAdapter,
sessionId: string
): Promise<DaemonSessionOwnerProbe> {
const before = owner.getLastAuthenticatedDaemonIdentity()
try {
const result = await owner.probePtyLiveness(sessionId)
const incarnation = owner.getLastAuthenticatedDaemonIdentity()
return {
owner,
result: before && !sameDaemonIncarnation(before, incarnation) ? null : result,
incarnation
}
} catch {
return {
owner,
result: null,
incarnation: owner.getLastAuthenticatedDaemonIdentity()
}
}
}
export function probeDaemonSessionOwnerSync(
owner: DaemonPtyAdapter,
sessionId: string
): DaemonSessionOwnerProbe {
const before = owner.getLastAuthenticatedDaemonIdentity()
try {
const result = owner.hasPty(sessionId)
const incarnation = owner.getLastAuthenticatedDaemonIdentity()
return {
owner,
result: before && !sameDaemonIncarnation(before, incarnation) ? null : result,
incarnation
}
} catch {
return { owner, result: null, incarnation: owner.getLastAuthenticatedDaemonIdentity() }
}
}
@@ -0,0 +1,24 @@
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
export class DaemonSessionOwnerResolutionCache {
private readonly pending = new Map<string, Promise<DaemonPtyAdapter>>()
async resolve(
sessionId: string,
start: () => Promise<DaemonPtyAdapter>
): Promise<DaemonPtyAdapter> {
const existing = this.pending.get(sessionId)
if (existing) {
return await existing
}
const resolution = start()
this.pending.set(sessionId, resolution)
try {
return await resolution
} finally {
if (this.pending.get(sessionId) === resolution) {
this.pending.delete(sessionId)
}
}
}
}
@@ -0,0 +1,51 @@
import type { DaemonEndpointIdentity } from './daemon-hello-protocol'
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import { probeDaemonSessionOwner } from './daemon-session-owner-probes'
import {
DaemonSessionGoneError,
DaemonSessionOwnerUnknownError,
sameDaemonIncarnation
} from './daemon-session-route'
type ResolveDaemonSessionOwnerOptions = {
sessionId: string
candidates: readonly DaemonPtyAdapter[]
expectedIncarnations?: ReadonlyMap<DaemonPtyAdapter, DaemonEndpointIdentity | null>
discoveryComplete: boolean
transfer: (owner: DaemonPtyAdapter, incarnation: DaemonEndpointIdentity | null) => void
recordAmbiguous: (
candidates: ReadonlyMap<DaemonPtyAdapter, DaemonEndpointIdentity | null>
) => void
}
export async function resolveDaemonSessionOwner({
sessionId,
candidates,
expectedIncarnations,
discoveryComplete,
transfer,
recordAmbiguous
}: ResolveDaemonSessionOwnerOptions): Promise<DaemonPtyAdapter> {
const probed = await Promise.all(
candidates.map((owner) => probeDaemonSessionOwner(owner, sessionId))
)
const results = probed.map((result) =>
sameDaemonIncarnation(result.incarnation, result.owner.getLastAuthenticatedDaemonIdentity()) &&
(!expectedIncarnations?.has(result.owner) ||
sameDaemonIncarnation(expectedIncarnations.get(result.owner) ?? null, result.incarnation))
? result
: { ...result, result: null }
)
const owners = results.filter(({ result }) => result === true)
if (owners.length === 1 && results.every(({ result }) => result !== null)) {
transfer(owners[0].owner, owners[0].incarnation)
return owners[0].owner
}
if (owners.length > 1) {
recordAmbiguous(new Map(owners.map(({ owner, incarnation }) => [owner, incarnation])))
}
if (owners.length === 0 && results.every(({ result }) => result === false) && discoveryComplete) {
throw new DaemonSessionGoneError(sessionId)
}
throw new DaemonSessionOwnerUnknownError(sessionId)
}
@@ -0,0 +1,56 @@
import type { DaemonEndpointIdentity } from './daemon-hello-protocol'
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import { probeDaemonSessionOwnerSync } from './daemon-session-owner-probes'
import {
DaemonSessionGoneError,
DaemonSessionOwnerUnknownError,
type DaemonSessionRoute,
DaemonSessionUnavailableError
} from './daemon-session-route'
type ResolveDaemonSessionOwnerSyncOptions = {
sessionId: string
route: DaemonSessionRoute | undefined
discoveryIncomplete: boolean
adapters: readonly DaemonPtyAdapter[]
transfer: (owner: DaemonPtyAdapter, incarnation: DaemonEndpointIdentity | null) => void
recordAmbiguous: (
candidates: ReadonlyMap<DaemonPtyAdapter, DaemonEndpointIdentity | null>
) => void
}
export function resolveDaemonSessionOwnerSync({
sessionId,
route,
discoveryIncomplete,
adapters,
transfer,
recordAmbiguous
}: ResolveDaemonSessionOwnerSyncOptions): DaemonPtyAdapter {
if (route?.state === 'owned') {
return route.owner
}
if (route?.state === 'unavailable') {
throw new DaemonSessionUnavailableError(sessionId)
}
if (route?.state === 'ambiguous') {
throw new DaemonSessionOwnerUnknownError(sessionId)
}
const results = adapters.map((owner) => probeDaemonSessionOwnerSync(owner, sessionId))
const owners = results.filter(({ result }) => result === true)
if (results.some(({ result }) => result === null)) {
throw new DaemonSessionOwnerUnknownError(sessionId)
}
if (owners.length === 1) {
transfer(owners[0].owner, owners[0].incarnation)
return owners[0].owner
}
if (owners.length > 1) {
recordAmbiguous(new Map(owners.map(({ owner, incarnation }) => [owner, incarnation])))
throw new DaemonSessionOwnerUnknownError(sessionId)
}
if (discoveryIncomplete) {
throw new DaemonSessionOwnerUnknownError(sessionId)
}
throw new DaemonSessionGoneError(sessionId)
}
@@ -0,0 +1,83 @@
import type { DaemonEndpointIdentity } from './daemon-hello-protocol'
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import { createOwnedDaemonSessionRoute, sameDaemonIncarnation } from './daemon-session-route'
import type { DaemonSessionRouteStore } from './daemon-session-route-store'
export function markDaemonSessionUnavailable(
routes: DaemonSessionRouteStore,
sessionId: string,
owner: DaemonPtyAdapter,
incarnation: DaemonEndpointIdentity | null
): void {
const route = routes.get(sessionId)
if (!route) {
routes.set(sessionId, {
state: 'unavailable',
owner,
incarnation
})
return
}
if (
route.state === 'owned' &&
route.owner === owner &&
sameDaemonIncarnation(route.incarnation, incarnation)
) {
routes.set(sessionId, {
state: 'unavailable',
owner,
incarnation: route.incarnation
})
return
}
if (
route.state !== 'ambiguous' ||
!route.candidates.has(owner) ||
!sameDaemonIncarnation(route.candidates.get(owner) ?? null, incarnation)
) {
return
}
const candidates = new Map(route.candidates)
candidates.delete(owner)
if (candidates.size === 1) {
for (const [candidate, candidateIncarnation] of candidates) {
routes.set(sessionId, createOwnedDaemonSessionRoute(candidate, candidateIncarnation))
}
} else if (candidates.size > 1) {
routes.set(sessionId, { state: 'ambiguous', candidates })
} else {
routes.set(sessionId, {
state: 'unavailable',
owner,
incarnation
})
}
}
export function markDaemonSessionOwnerUnavailable(
routes: DaemonSessionRouteStore,
owner: DaemonPtyAdapter,
incarnation: DaemonEndpointIdentity | null
): void {
for (const [sessionId, route] of routes) {
if (
route.state === 'ambiguous' &&
route.candidates.has(owner) &&
sameDaemonIncarnation(route.candidates.get(owner) ?? null, incarnation)
) {
markDaemonSessionUnavailable(routes, sessionId, owner, incarnation)
continue
}
if (
route.state === 'owned' &&
route.owner === owner &&
sameDaemonIncarnation(route.incarnation, incarnation)
) {
routes.set(sessionId, {
state: 'unavailable',
owner,
incarnation: route.incarnation
})
}
}
}
@@ -0,0 +1,57 @@
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import type { DaemonSessionRoute } from './daemon-session-route'
const MAX_UNAVAILABLE_ROUTES = 1000
export class DaemonSessionRouteStore implements Iterable<[string, DaemonSessionRoute]> {
private readonly routes = new Map<string, DaemonSessionRoute>()
private readonly unavailableRouteIds = new Set<string>()
private unavailableRouteEvicted = false
get(sessionId: string): DaemonSessionRoute | undefined {
return this.routes.get(sessionId)
}
set(sessionId: string, route: DaemonSessionRoute): void {
this.routes.set(sessionId, route)
this.unavailableRouteIds.delete(sessionId)
if (route.state !== 'unavailable') {
return
}
this.unavailableRouteIds.add(sessionId)
while (this.unavailableRouteIds.size > MAX_UNAVAILABLE_ROUTES) {
const oldest = this.unavailableRouteIds.values().next().value
if (!oldest) {
return
}
this.unavailableRouteIds.delete(oldest)
if (this.routes.get(oldest)?.state === 'unavailable') {
this.routes.delete(oldest)
this.unavailableRouteEvicted = true
}
}
}
delete(sessionId: string): void {
this.routes.delete(sessionId)
this.unavailableRouteIds.delete(sessionId)
}
ownedSessionIds(owner: DaemonPtyAdapter): string[] {
const sessionIds: string[] = []
for (const [sessionId, route] of this.routes) {
if (route.state === 'owned' && route.owner === owner) {
sessionIds.push(sessionId)
}
}
return sessionIds
}
hasUnavailableRouteEviction(): boolean {
return this.unavailableRouteEvicted
}
[Symbol.iterator](): MapIterator<[string, DaemonSessionRoute]> {
return this.routes[Symbol.iterator]()
}
}
+202 -162
View File
@@ -3,34 +3,56 @@ import {
createOwnedDaemonSessionRoute,
DaemonSessionGoneError,
DaemonSessionOwnerUnknownError,
type DaemonSessionRoute,
DaemonSessionUnavailableError,
isCurrentOwnedDaemonSessionRoute,
sameDaemonIncarnation
} from './daemon-session-route'
import { probeDaemonSessionOwner } from './daemon-session-owner-probes'
import { DaemonSessionRouteStore } from './daemon-session-route-store'
import {
assertFreshDaemonSessionAvailable,
ownerForFreshDaemonSession
} from './daemon-fresh-session-routing'
import { resolveDaemonSessionOwnerSync } from './daemon-session-owner-sync-resolution'
import { resolveDaemonSessionOwner } from './daemon-session-owner-resolution'
import {
markDaemonSessionOwnerUnavailable,
markDaemonSessionUnavailable
} from './daemon-session-route-availability'
import { DaemonSessionOwnerResolutionCache } from './daemon-session-owner-resolution-cache'
export class DaemonSessionRouteTable {
private readonly routes = new Map<string, DaemonSessionRoute>()
private readonly routes = new DaemonSessionRouteStore()
private readonly incompleteDiscoveries = new Set<DaemonPtyAdapter>()
private readonly ownerResolutions = new DaemonSessionOwnerResolutionCache()
constructor(private readonly adapters: readonly DaemonPtyAdapter[]) {}
get(sessionId: string): DaemonSessionRoute | undefined {
return this.routes.get(sessionId)
}
readonly get = this.routes.get.bind(this.routes)
getOwned(sessionId: string): DaemonPtyAdapter | undefined {
const route = this.routes.get(sessionId)
return route?.state === 'owned' ? route.owner : undefined
return route?.state === 'owned' && isCurrentOwnedDaemonSessionRoute(route)
? route.owner
: undefined
}
shouldForwardEvent(sessionId: string, owner: DaemonPtyAdapter): boolean {
const route = this.routes.get(sessionId)
return !route || (route.state === 'owned' && route.owner === owner)
if (!route) {
return !this.routes.hasUnavailableRouteEviction()
}
return (
route.state === 'owned' && route.owner === owner && isCurrentOwnedDaemonSessionRoute(route)
)
}
recordDataOwner(sessionId: string, owner: DaemonPtyAdapter): boolean {
const route = this.routes.get(sessionId)
if (!route) {
if (this.routes.hasUnavailableRouteEviction()) {
return false
}
this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner))
return true
}
@@ -38,43 +60,78 @@ export class DaemonSessionRouteTable {
this.recordOwned(sessionId, owner)
return this.shouldForwardEvent(sessionId, owner)
}
if (route.state === 'ambiguous' || route.owner === owner) {
return route.state === 'owned'
if (route.state === 'ambiguous') {
return false
}
if (route.owner === owner) {
if (isCurrentOwnedDaemonSessionRoute(route)) {
return true
}
this.markUnavailable(sessionId, owner, route.incarnation)
return false
}
this.recordOwned(sessionId, owner)
return false
}
recordOwned(sessionId: string, owner: DaemonPtyAdapter): void {
this.recordOwnedIncarnation(sessionId, owner, owner.getLastAuthenticatedDaemonIdentity())
}
recordOwnedIncarnation(
sessionId: string,
owner: DaemonPtyAdapter,
incarnation: ReturnType<DaemonPtyAdapter['getLastAuthenticatedDaemonIdentity']>
): void {
const route = this.routes.get(sessionId)
if (!route) {
this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner))
if (this.routes.hasUnavailableRouteEviction()) {
return
}
this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner, incarnation))
return
}
if (route.state === 'owned' && route.owner === owner) {
this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner))
if (!sameDaemonIncarnation(route.incarnation, incarnation)) {
this.markUnavailable(sessionId, owner, route.incarnation)
return
}
this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner, incarnation))
return
}
if (route.state === 'unavailable') {
const incarnation = owner.getLastAuthenticatedDaemonIdentity()
if (route.owner !== owner || !sameDaemonIncarnation(route.incarnation, incarnation)) {
this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner))
}
return
}
if (route.state === 'ambiguous' && route.candidates.has(owner)) {
const candidateIncarnation = route.candidates.get(owner) ?? null
if (!sameDaemonIncarnation(candidateIncarnation, incarnation)) {
this.markUnavailable(sessionId, owner, candidateIncarnation)
return
}
}
const candidates =
route.state === 'ambiguous' ? new Set(route.candidates) : new Set([route.owner])
candidates.add(owner)
route.state === 'ambiguous'
? new Map(route.candidates)
: new Map([[route.owner, route.incarnation]])
candidates.set(owner, incarnation)
this.routes.set(sessionId, { state: 'ambiguous', candidates })
}
transfer(sessionId: string, owner: DaemonPtyAdapter): void {
this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner))
transfer(
sessionId: string,
owner: DaemonPtyAdapter,
incarnation = owner.getLastAuthenticatedDaemonIdentity()
): void {
this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner, incarnation))
}
recordFreshOwned(sessionId: string, owner: DaemonPtyAdapter): void {
const route = this.routes.get(sessionId)
if (route?.state === 'unavailable' && route.owner === owner) {
if (
!route ||
route.state === 'unavailable' ||
(route.state === 'owned' && route.owner === owner && !isCurrentOwnedDaemonSessionRoute(route))
) {
this.transfer(sessionId, owner)
return
}
@@ -85,142 +142,149 @@ export class DaemonSessionRouteTable {
this.incompleteDiscoveries.add(owner)
}
recordCompleteDiscovery(owner: DaemonPtyAdapter, sessionIds: readonly string[]): void {
recordCompleteDiscovery(
owner: DaemonPtyAdapter,
sessionIds: readonly string[],
incarnation = owner.getLastAuthenticatedDaemonIdentity()
): void {
for (const sessionId of sessionIds) {
this.recordOwned(sessionId, owner)
this.recordOwnedIncarnation(sessionId, owner, incarnation)
}
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)) {
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()
})
}
}
markUnavailable(
sessionId: string,
owner: DaemonPtyAdapter,
incarnation = owner.getLastAuthenticatedDaemonIdentity()
): void {
markDaemonSessionUnavailable(this.routes, sessionId, owner, incarnation)
}
markOwnerUnavailable(
owner: DaemonPtyAdapter,
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 &&
sameDaemonIncarnation(route.incarnation, incarnation)
) {
this.routes.set(sessionId, {
state: 'unavailable',
owner,
incarnation: route.incarnation
})
}
}
markDaemonSessionOwnerUnavailable(this.routes, owner, 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') {
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)
return ownerForFreshDaemonSession(
sessionId,
sessionId ? this.routes.get(sessionId) : undefined,
current,
() => sessionId && this.routes.delete(sessionId)
)
}
async resolveOwner(sessionId: string): Promise<DaemonPtyAdapter> {
assertFreshSpawnAvailable(sessionId: string | undefined): void {
assertFreshDaemonSessionAvailable(
sessionId,
sessionId ? this.routes.get(sessionId) : undefined,
() => sessionId && this.routes.delete(sessionId)
)
}
hasPty(sessionId: string): boolean {
const route = this.routes.get(sessionId)
if (route?.state === 'owned') {
return route.owner
if (!route && this.routes.hasUnavailableRouteEviction()) {
throw new DaemonSessionOwnerUnknownError(sessionId)
}
if (route?.state === 'unavailable') {
throw new DaemonSessionUnavailableError(sessionId)
}
const candidates =
route?.state === 'ambiguous' && this.incompleteDiscoveries.size === 0
? [...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 (route?.state === 'ambiguous') {
throw new DaemonSessionOwnerUnknownError(sessionId)
}
if (owners.length > 1) {
this.routes.set(sessionId, { state: 'ambiguous', candidates: new Set(owners) })
if (route?.state === 'owned') {
if (!isCurrentOwnedDaemonSessionRoute(route)) {
this.markUnavailable(sessionId, route.owner, route.incarnation)
throw new DaemonSessionUnavailableError(sessionId)
}
return route.owner.hasPty(sessionId)
}
if (
owners.length === 0 &&
results.every(({ result }) => result === false) &&
this.incompleteDiscoveries.size === 0
) {
throw new DaemonSessionGoneError(sessionId)
try {
this.resolveOwnerSync(sessionId)
return true
} catch (error) {
if (error instanceof DaemonSessionGoneError) {
return false
}
throw error
}
throw new DaemonSessionOwnerUnknownError(sessionId)
}
async resolveOwner(sessionId: string): Promise<DaemonPtyAdapter> {
const route = this.routes.get(sessionId)
if (!route && this.routes.hasUnavailableRouteEviction()) {
throw new DaemonSessionOwnerUnknownError(sessionId)
}
if (route?.state === 'owned') {
if (isCurrentOwnedDaemonSessionRoute(route)) {
return route.owner
}
this.markUnavailable(sessionId, route.owner, route.incarnation)
throw new DaemonSessionUnavailableError(sessionId)
}
if (route?.state === 'unavailable') {
throw new DaemonSessionUnavailableError(sessionId)
}
return await this.ownerResolutions.resolve(sessionId, async () => {
const discoveryComplete = this.incompleteDiscoveries.size === 0
const accept = (): void => {
if (this.routes.get(sessionId) !== route) {
throw new DaemonSessionOwnerUnknownError(sessionId)
}
}
try {
return await resolveDaemonSessionOwner({
sessionId,
candidates: this.adapters,
expectedIncarnations: route?.state === 'ambiguous' ? route.candidates : undefined,
discoveryComplete,
transfer: (owner, incarnation) => {
accept()
this.transfer(sessionId, owner, incarnation)
},
recordAmbiguous: (ambiguousCandidates) => {
accept()
this.routes.set(sessionId, {
state: 'ambiguous',
candidates: ambiguousCandidates
})
}
})
} catch (error) {
if (
error instanceof DaemonSessionGoneError &&
(this.routes.get(sessionId) !== route ||
discoveryComplete !== (this.incompleteDiscoveries.size === 0))
) {
throw new DaemonSessionOwnerUnknownError(sessionId)
}
throw error
}
})
}
async probeLiveness(sessionId: string): Promise<boolean | null> {
const route = this.routes.get(sessionId)
if (!route && this.routes.hasUnavailableRouteEviction()) {
return null
}
if (route?.state === 'unavailable' || route?.state === 'ambiguous') {
return null
}
if (route?.state === 'owned') {
return await probeOwner(route.owner, sessionId)
if (!isCurrentOwnedDaemonSessionRoute(route)) {
this.markUnavailable(sessionId, route.owner, route.incarnation)
return null
}
const result = await probeDaemonSessionOwner(route.owner, sessionId)
return sameDaemonIncarnation(route.incarnation, result.incarnation) &&
sameDaemonIncarnation(result.incarnation, result.owner.getLastAuthenticatedDaemonIdentity())
? result.result
: null
}
try {
await this.resolveOwner(sessionId)
@@ -232,47 +296,23 @@ export class DaemonSessionRouteTable {
resolveOwnerSync(sessionId: string): DaemonPtyAdapter {
const route = this.routes.get(sessionId)
if (route?.state === 'owned') {
return route.owner
if (!route && this.routes.hasUnavailableRouteEviction()) {
throw new DaemonSessionOwnerUnknownError(sessionId)
}
if (route?.state === 'unavailable') {
if (route?.state === 'owned' && !isCurrentOwnedDaemonSessionRoute(route)) {
this.markUnavailable(sessionId, route.owner, route.incarnation)
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)
return resolveDaemonSessionOwnerSync({
sessionId,
route,
discoveryIncomplete: this.incompleteDiscoveries.size > 0,
adapters: this.adapters,
transfer: (owner, incarnation) => this.transfer(sessionId, owner, incarnation),
recordAmbiguous: (candidates) =>
this.routes.set(sessionId, { state: 'ambiguous', candidates })
})
}
}
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
}
readonly getOwnedSessionIds = this.routes.ownedSessionIds.bind(this.routes)
}
+13 -3
View File
@@ -15,7 +15,7 @@ export type DaemonSessionRoute =
}
| {
state: 'ambiguous'
candidates: ReadonlySet<DaemonPtyAdapter>
candidates: ReadonlyMap<DaemonPtyAdapter, DaemonEndpointIdentity | null>
}
export class DaemonSessionOwnerUnknownError extends Error {
@@ -40,12 +40,13 @@ export class DaemonSessionGoneError extends Error {
}
export function createOwnedDaemonSessionRoute(
owner: DaemonPtyAdapter
owner: DaemonPtyAdapter,
incarnation = owner.getLastAuthenticatedDaemonIdentity()
): Extract<DaemonSessionRoute, { state: 'owned' }> {
return {
state: 'owned',
owner,
incarnation: owner.getLastAuthenticatedDaemonIdentity()
incarnation
}
}
@@ -62,3 +63,12 @@ export function sameDaemonIncarnation(
left.launchNonce === right.launchNonce
)
}
export function isCurrentOwnedDaemonSessionRoute(
route: Extract<DaemonSessionRoute, { state: 'owned' }>
): boolean {
if (typeof route.owner.matchesLastAuthenticatedDaemonIdentity === 'function') {
return route.owner.matchesLastAuthenticatedDaemonIdentity(route.incarnation)
}
return sameDaemonIncarnation(route.incarnation, route.owner.getLastAuthenticatedDaemonIdentity())
}
@@ -1,8 +1,21 @@
import type { IPtyProvider, PtyProviderBufferSnapshot } from '../providers/types'
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import type { DaemonEndpointIdentity } from './daemon-hello-protocol'
import { sameDaemonIncarnation } from './daemon-session-route'
export class DaemonSnapshotAcknowledgementRoutes {
private readonly producers = new Map<string, DaemonPtyAdapter>()
private readonly producers = new Map<
string,
{ adapter: DaemonPtyAdapter; incarnation: DaemonEndpointIdentity | null }
>()
private readonly pendingCaptures = new Map<
string,
{
token: symbol
adapter: DaemonPtyAdapter | undefined
incarnation: DaemonEndpointIdentity | null
}
>()
async capture(
sessionId: string,
@@ -10,44 +23,95 @@ export class DaemonSnapshotAcknowledgementRoutes {
provider: IPtyProvider,
adapters: readonly DaemonPtyAdapter[]
): Promise<PtyProviderBufferSnapshot | null> {
const snapshot = (await provider.getBufferSnapshot?.(sessionId, opts)) ?? null
this.recordAdapter(sessionId, snapshot, provider, adapters)
return snapshot
const producer = adapters.find((adapter) => adapter === provider)
const before = producer?.getLastAuthenticatedDaemonIdentity() ?? null
const token = Symbol(sessionId)
this.pendingCaptures.set(sessionId, { token, adapter: producer, incarnation: before })
try {
const snapshot = (await provider.getBufferSnapshot?.(sessionId, opts)) ?? null
if (this.pendingCaptures.get(sessionId)?.token !== token) {
return snapshot
}
const incarnation = producer?.getLastAuthenticatedDaemonIdentity() ?? null
if (before && !sameDaemonIncarnation(before, incarnation)) {
this.producers.delete(sessionId)
return snapshot
}
this.record(sessionId, snapshot, producer, incarnation)
return snapshot
} finally {
if (this.pendingCaptures.get(sessionId)?.token === token) {
this.pendingCaptures.delete(sessionId)
}
}
}
record(
private record(
sessionId: string,
snapshot: PtyProviderBufferSnapshot | null,
producer: DaemonPtyAdapter
producer: DaemonPtyAdapter | undefined,
incarnation: DaemonEndpointIdentity | null
): void {
if (snapshot) {
this.producers.set(sessionId, producer)
if (snapshot && producer) {
this.producers.set(sessionId, {
adapter: producer,
incarnation
})
} else {
this.producers.delete(sessionId)
}
}
recordAdapter(
dropForProducer(
sessionId: string,
snapshot: PtyProviderBufferSnapshot | null,
provider: IPtyProvider,
adapters: readonly DaemonPtyAdapter[]
adapter: DaemonPtyAdapter,
incarnation = adapter.getLastAuthenticatedDaemonIdentity()
): void {
const producer = adapters.find((adapter) => adapter === provider)
if (producer) {
this.record(sessionId, snapshot, producer)
} else {
const producer = this.producers.get(sessionId)
if (producer?.adapter === adapter && sameDaemonIncarnation(producer.incarnation, incarnation)) {
this.producers.delete(sessionId)
}
const pending = this.pendingCaptures.get(sessionId)
if (pending?.adapter === adapter && sameDaemonIncarnation(pending.incarnation, incarnation)) {
this.pendingCaptures.delete(sessionId)
}
}
dropAdapterIncarnation(
adapter: DaemonPtyAdapter,
incarnation: DaemonEndpointIdentity | null
): void {
for (const [sessionId, producer] of this.producers) {
if (
producer.adapter === adapter &&
sameDaemonIncarnation(producer.incarnation, incarnation)
) {
this.producers.delete(sessionId)
}
}
for (const [sessionId, pending] of this.pendingCaptures) {
if (pending.adapter === adapter && sameDaemonIncarnation(pending.incarnation, incarnation)) {
this.pendingCaptures.delete(sessionId)
}
}
}
acknowledge(sessionId: string): void {
const producer = this.producers.get(sessionId)
this.producers.delete(sessionId)
producer?.ackColdRestore(sessionId)
if (
producer &&
sameDaemonIncarnation(
producer.incarnation,
producer.adapter.getLastAuthenticatedDaemonIdentity()
)
) {
producer.adapter.ackColdRestore(sessionId)
}
}
drop(sessionId: string): void {
this.producers.delete(sessionId)
clear(): void {
this.producers.clear()
this.pendingCaptures.clear()
}
}
@@ -36,6 +36,9 @@ function createProvider(
providesAgentSessionOwnerListings: vi.fn(() => authoritativeOwnerListings),
write: vi.fn(),
resize: vi.fn(),
pauseProducer: vi.fn(),
resumeProducer: vi.fn(),
setPtyBackgrounded: vi.fn(),
shutdown: vi.fn(async (id: string) => {
const idx = sessions.indexOf(id)
if (idx !== -1) {
@@ -124,6 +127,8 @@ function createDaemonAdapter(
return {
...createProvider(label, sessions, true),
protocolVersion: 13,
getLastAuthenticatedDaemonIdentity: vi.fn(() => null),
onDaemonIdentityChanged: vi.fn(() => () => {}),
listSessions: vi.fn(async () => []),
ackColdRestore: vi.fn(),
clearTombstone: vi.fn(),
@@ -228,20 +233,71 @@ describe('DegradedDaemonPtyProvider', () => {
expect(fallback.write).toHaveBeenCalledWith(fresh.id, 'new\n')
})
it('routes a previously daemon-backed id to fallback after daemon exit removes the mapping', async () => {
it('fences existing ids after incomplete daemon discovery while allowing explicit fresh ids', async () => {
const sessionId = 'uncertain-session'
const current = createDaemonAdapter('daemon')
const legacy = createDaemonAdapter('legacy')
const fallback = createProvider('fallback')
vi.mocked(legacy.listProcesses).mockRejectedValueOnce(new Error('listing unavailable'))
vi.mocked(legacy.probePtyLiveness).mockResolvedValue(null)
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback })
await provider.discoverDaemonSessions()
await expect(provider.spawn({ sessionId, cols: 80, rows: 24 })).rejects.toThrow(
'daemon_session_routing_unavailable'
)
await expect(
provider.spawn({ sessionId: 'fresh-session', isNewSession: true, cols: 80, rows: 24 })
).resolves.toMatchObject({ id: 'fresh-session' })
expect(fallback.spawn).toHaveBeenCalledExactlyOnceWith({
sessionId: 'fresh-session',
isNewSession: true,
cols: 80,
rows: 24
})
warn.mockRestore()
})
it('keeps cross-generation daemon collisions away from the fallback', async () => {
const sessionId = 'daemon-collision'
const current = createDaemonAdapter('current', [sessionId])
const legacy = createDaemonAdapter('legacy', [sessionId])
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback })
const data = vi.fn()
provider.onData(data)
await provider.discoverDaemonSessions()
current.emitData(sessionId, 'current output')
legacy.emitData(sessionId, 'legacy output')
await expect(provider.spawn({ sessionId, cols: 80, rows: 24 })).rejects.toThrow(
'daemon_session_routing_unavailable'
)
expect(() => provider.pauseProducer(sessionId)).not.toThrow()
expect(() => provider.resumeProducer(sessionId)).not.toThrow()
expect(() => provider.setPtyBackgrounded(sessionId, true)).not.toThrow()
expect(data).not.toHaveBeenCalled()
expect(fallback.spawn).not.toHaveBeenCalled()
expect(fallback.pauseProducer).not.toHaveBeenCalled()
expect(fallback.resumeProducer).not.toHaveBeenCalled()
expect(fallback.setPtyBackgrounded).not.toHaveBeenCalled()
})
it('tombstones a previously daemon-backed id after its daemon exits', async () => {
const current = createDaemonAdapter('daemon', ['daemon-session'])
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
await provider.discoverDaemonSessions()
current.emitExit('daemon-session', 0)
await provider.spawn({ sessionId: 'daemon-session', cols: 80, rows: 24 })
await expect(
provider.spawn({ sessionId: 'daemon-session', cols: 80, rows: 24 })
).rejects.toThrow('daemon_session_routing_unavailable')
expect(fallback.spawn).toHaveBeenCalledWith({
sessionId: 'daemon-session',
cols: 80,
rows: 24
})
expect(fallback.spawn).not.toHaveBeenCalled()
})
it('caches a provider discovered by hasPty before routing later operations', () => {
@@ -254,6 +310,23 @@ describe('DegradedDaemonPtyProvider', () => {
expect(current.write).toHaveBeenCalledWith('daemon-session', 'kept-on-daemon\n')
expect(fallback.write).not.toHaveBeenCalled()
expect(current.hasPty).toHaveBeenCalledExactlyOnceWith('daemon-session')
})
it('probes each daemon once before memoizing a unique owner', () => {
const current = createDaemonAdapter('current')
const legacy = createDaemonAdapter('legacy', ['legacy-session'])
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback })
provider.write('legacy-session', 'first\n')
provider.resize('legacy-session', 100, 30)
expect(current.hasPty).toHaveBeenCalledExactlyOnceWith('legacy-session')
expect(legacy.hasPty).toHaveBeenCalledExactlyOnceWith('legacy-session')
expect(legacy.write).toHaveBeenCalledExactlyOnceWith('legacy-session', 'first\n')
expect(legacy.resize).toHaveBeenCalledExactlyOnceWith('legacy-session', 100, 30)
expect(fallback.hasPty).not.toHaveBeenCalled()
})
it('probes daemon owners without borrowing fallback liveness', async () => {
@@ -267,6 +340,9 @@ describe('DegradedDaemonPtyProvider', () => {
expect(fallback.probePtyLiveness).not.toHaveBeenCalled()
vi.mocked(current.probePtyLiveness).mockResolvedValue(true)
await expect(provider.probePtyLiveness('unknown-session')).resolves.toBeNull()
vi.mocked(legacy.probePtyLiveness).mockResolvedValue(false)
await expect(provider.probePtyLiveness('unknown-session')).resolves.toBe(true)
})
@@ -294,9 +370,9 @@ describe('DegradedDaemonPtyProvider', () => {
})
})
it('acknowledges a colliding id through the daemon that produced its snapshot', async () => {
it('keeps colliding liveness unknown and acknowledges through the snapshot producer', async () => {
const sessionId = 'colliding-session'
const current = createDaemonAdapter('current', [sessionId])
const current = createDaemonAdapter('current')
const legacy = createDaemonAdapter('legacy', [sessionId])
const fallback = createProvider('fallback')
legacy.getBufferSnapshot = vi.fn(async () => ({
@@ -314,21 +390,27 @@ describe('DegradedDaemonPtyProvider', () => {
await provider.discoverDaemonSessions()
await provider.getBufferSnapshot(sessionId)
legacy.emitExit(sessionId, 0)
expect(provider.hasPty(sessionId)).toBe(true)
current.emitData(sessionId, 'colliding current output')
expect(() => provider.hasPty(sessionId)).toThrow('daemon_session_routing_unavailable')
provider.ackColdRestore(sessionId)
expect(legacy.ackColdRestore).toHaveBeenCalledExactlyOnceWith(sessionId)
expect(current.ackColdRestore).not.toHaveBeenCalled()
})
it('forwards replay output from fallback and daemon providers', () => {
it('forwards replay output from owned fallback and daemon providers', async () => {
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
const replaySpy = vi.fn()
const unsubscribe = provider.onReplay(replaySpy)
await provider.spawn({
sessionId: 'fallback-session',
isNewSession: true,
cols: 80,
rows: 24
})
current.emitReplay('daemon-session', 'daemon replay')
fallback.emitReplay('fallback-session', 'fallback replay')
unsubscribe()
@@ -345,6 +427,154 @@ describe('DegradedDaemonPtyProvider', () => {
})
})
it('withholds replay output when daemon and fallback ids collide', async () => {
const sessionId = 'replay-collision'
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
const replay = vi.fn()
provider.onReplay(replay)
await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 })
current.emitReplay(sessionId, 'daemon replay')
fallback.emitReplay(sessionId, 'fallback replay')
expect(replay).not.toHaveBeenCalled()
})
it('keeps a daemon survivor live when a colliding fallback exits first', async () => {
const sessionId = 'fallback-exits-first'
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
const data = vi.fn()
const exit = vi.fn()
provider.onData(data)
provider.onExit(exit)
await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 })
current.emitData(sessionId, 'withheld collision')
fallback.emitExit(sessionId, 0)
current.emitData(sessionId, 'daemon survivor')
expect(exit).not.toHaveBeenCalled()
expect(data).toHaveBeenCalledExactlyOnceWith({
id: sessionId,
data: 'daemon survivor'
})
})
it('keeps a fallback survivor live when a colliding daemon exits first', async () => {
const sessionId = 'daemon-exits-first'
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
const data = vi.fn()
const exit = vi.fn()
provider.onData(data)
provider.onExit(exit)
await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 })
current.emitData(sessionId, 'withheld collision')
current.emitExit(sessionId, 0)
fallback.emitData(sessionId, 'fallback survivor')
fallback.emitExit(sessionId, 0)
expect(data).toHaveBeenCalledExactlyOnceWith({
id: sessionId,
data: 'fallback survivor'
})
expect(exit).toHaveBeenCalledExactlyOnceWith({ id: sessionId, code: 0 })
})
it('keeps a fallback collision fenced while another daemon owner survives', async () => {
const sessionId = 'two-daemon-fallback-collision'
const current = createDaemonAdapter('current')
const legacy = createDaemonAdapter('legacy')
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback })
const data = vi.fn()
provider.onData(data)
await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 })
current.emitData(sessionId, 'current collision')
legacy.emitData(sessionId, 'legacy collision')
current.emitExit(sessionId, 0)
fallback.emitData(sessionId, 'still colliding')
expect(() => provider.write(sessionId, 'blocked')).toThrow('daemon_session_routing_unavailable')
expect(data).not.toHaveBeenCalled()
legacy.emitExit(sessionId, 0)
fallback.emitData(sessionId, 'fallback survivor')
provider.write(sessionId, 'allowed')
expect(data).toHaveBeenCalledExactlyOnceWith({
id: sessionId,
data: 'fallback survivor'
})
expect(fallback.write).toHaveBeenCalledExactlyOnceWith(sessionId, 'allowed')
})
it('records fallback ownership from output emitted before spawn resolves', async () => {
const sessionId = 'in-flight-fallback-spawn'
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
const data = vi.fn()
provider.onData(data)
vi.mocked(fallback.spawn).mockImplementationOnce(async () => {
fallback.emitData(sessionId, 'fallback first')
return { id: sessionId }
})
await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 })
current.emitData(sessionId, 'daemon collision')
fallback.emitData(sessionId, 'withheld fallback collision')
expect(data).toHaveBeenCalledExactlyOnceWith({
id: sessionId,
data: 'fallback first'
})
})
it('rejects unknown fallback output after unavailable tombstones are evicted', async () => {
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
for (let index = 0; index <= 1_000; index += 1) {
const sessionId = `evicted-${index}`
current.emitData(sessionId, 'frame')
current.emitExit(sessionId, 0)
}
const data = vi.fn()
provider.onData(data)
fallback.emitData('unknown-fallback', 'unsafe frame')
expect(data).not.toHaveBeenCalled()
expect(() => provider.write('unknown-fallback', 'unsafe input')).toThrow(
'daemon_session_routing_unavailable'
)
vi.mocked(fallback.spawn).mockImplementationOnce(async () => {
fallback.emitData('explicit-fallback', 'safe frame')
return { id: 'explicit-fallback' }
})
await provider.spawn({
sessionId: 'explicit-fallback',
isNewSession: true,
cols: 80,
rows: 24
})
expect(data).toHaveBeenCalledExactlyOnceWith({
id: 'explicit-fallback',
data: 'safe frame'
})
})
it('preserves explicit sequence accounting on daemon data events', () => {
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
@@ -396,8 +626,13 @@ describe('DegradedDaemonPtyProvider', () => {
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
const stuck = await provider.spawn({ sessionId: 'stuck', cols: 80, rows: 24 })
await provider.spawn({ sessionId: 'ok', cols: 80, rows: 24 })
const stuck = await provider.spawn({
sessionId: 'stuck',
isNewSession: true,
cols: 80,
rows: 24
})
await provider.spawn({ sessionId: 'ok', isNewSession: true, cols: 80, rows: 24 })
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
vi.mocked(fallback.shutdown).mockImplementation(async (id: string) => {
if (id === stuck.id) {
@@ -434,6 +669,29 @@ describe('DegradedDaemonPtyProvider', () => {
expect(provider.hasPty('legacy-session')).toBe(true)
})
it('reconciles live owners before same-id tombstones regardless of adapter order', async () => {
const sessionId = 'reconciled-owner'
const current = createDaemonAdapter('current')
const legacy = createDaemonAdapter('legacy', [sessionId])
const fallback = createProvider('fallback')
vi.mocked(current.reconcileOnStartup).mockResolvedValue({
alive: [],
killed: [sessionId]
})
vi.mocked(legacy.reconcileOnStartup).mockResolvedValue({
alive: [sessionId],
killed: []
})
const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback })
await provider.reconcileOnStartup(new Set())
await provider.sendSignal(sessionId, 'SIGTERM')
expect(legacy.sendSignal).toHaveBeenCalledExactlyOnceWith(sessionId, 'SIGTERM')
expect(current.sendSignal).not.toHaveBeenCalled()
expect(fallback.sendSignal).not.toHaveBeenCalled()
})
it('keeps an exited legacy daemon poisoning listProcesses after construction', async () => {
const current = createDaemonAdapter('daemon', ['current-session'])
const legacy = createDaemonAdapter('legacy', ['legacy-session'])
+71 -90
View File
@@ -1,6 +1,5 @@
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import { combineUnsubscribes } from './combine-unsubscribes'
import { shutdownDegradedFallbackSessions } from './degraded-daemon-fallback-shutdown'
import { inspectPtyProviderProcess } from '../providers/pty-process-inspection'
import type {
IPtyProvider,
@@ -11,9 +10,8 @@ import type {
PtySpawnOptions,
PtySpawnResult
} 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'
import { DegradedDaemonSessionRouting } from './degraded-daemon-session-routing'
export class DegradedDaemonPtyProvider implements IPtyProvider {
readonly routesFreshSpawnsToLocalProvider = true
@@ -23,7 +21,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
private current: DaemonPtyAdapter
private legacy: DaemonPtyAdapter[]
private fallback: IPtyProvider
private sessionProviders = new Map<string, IPtyProvider>()
private readonly routing: DegradedDaemonSessionRouting
private readonly snapshotAcks = new DaemonSnapshotAcknowledgementRoutes()
private unsubscribers: (() => void)[] = []
private dataListeners: ((payload: PtyDataEvent) => void)[] = []
@@ -37,16 +35,27 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
this.current = opts.current
this.legacy = opts.legacy
this.fallback = opts.fallback
this.routing = new DegradedDaemonSessionRouting(
this.current,
this.legacy,
this.fallback,
this.snapshotAcks
)
for (const provider of this.allProviders()) {
this.unsubscribers.push(
provider.onData((payload) => {
if (!this.routing.shouldForwardStreamEvent(provider, payload.id)) {
return
}
for (const listener of this.dataListeners) {
listener(payload)
}
}),
provider.onExit((payload) => {
this.sessionProviders.delete(payload.id)
if (!this.routing.recordExit(provider, payload.id)) {
return
}
for (const listener of this.exitListeners) {
listener(payload)
}
@@ -56,124 +65,114 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}
async discoverDaemonSessions(): Promise<void> {
for (const adapter of this.allDaemonAdapters()) {
try {
const sessions = await adapter.listProcesses()
for (const session of sessions) {
this.sessionProviders.set(session.id, adapter)
}
} catch (error) {
console.warn('[daemon] Failed to discover degraded daemon sessions', error)
}
}
await this.routing.discover()
}
async spawn(opts: PtySpawnOptions): Promise<PtySpawnResult> {
const mapped = opts.sessionId ? this.sessionProviders.get(opts.sessionId) : undefined
const target = mapped ?? this.fallback
const result = await target.spawn(opts)
this.sessionProviders.set(result.id, target)
return result
return await this.routing.spawn(opts)
}
async attach(id: string): Promise<void> {
await this.providerFor(id).attach(id)
await this.routing.providerFor(id).attach(id)
}
hasPty(id: string): boolean {
const mapped = this.sessionProviders.get(id)
return mapped ? (mapped.hasPty?.(id) ?? true) : this.findProviderForExistingSession(id) !== null
return this.routing.hasPty(id)
}
async probePtyLiveness(id: string): Promise<boolean | null> {
return await probePtyOwners(id, this.sessionProviders.get(id), this.allDaemonAdapters())
return await this.routing.probePtyLiveness(id)
}
// Why: an unknown id cannot borrow listing authority from the fresh-spawn provider.
providesAgentSessionOwnerListings = (ptyId: string): boolean =>
(
this.sessionProviders.get(ptyId) ?? this.findProviderForExistingSession(ptyId)
)?.providesAgentSessionOwnerListings?.(ptyId) === true
this.routing.ownerForHint(ptyId)?.providesAgentSessionOwnerListings?.(ptyId) === true
write(id: string, data: string): void {
this.providerFor(id).write(id, data)
this.routing.providerFor(id).write(id, data)
}
resize(id: string, cols: number, rows: number): void {
this.providerFor(id).resize(id, cols, rows)
this.routing.providerFor(id).resize(id, cols, rows)
}
pauseProducer(id: string): void {
this.providerFor(id).pauseProducer?.(id)
this.routing.ownerForHint(id)?.pauseProducer?.(id)
}
resumeProducer(id: string): void {
this.providerFor(id).resumeProducer?.(id)
this.routing.ownerForHint(id)?.resumeProducer?.(id)
}
setPtyBackgrounded(id: string, background: boolean): void {
this.providerFor(id).setPtyBackgrounded?.(id, background)
this.routing.ownerForHint(id)?.setPtyBackgrounded?.(id, background)
}
async shutdown(
id: string,
opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number }
): Promise<void> {
await this.providerFor(id).shutdown(id, opts)
const provider = this.routing.providerFor(id)
await provider.shutdown(id, opts)
if (!opts.keepHistory) {
this.sessionProviders.delete(id)
this.routing.recordShutdown(id, provider)
}
}
async sendSignal(id: string, signal: string): Promise<void> {
await this.providerFor(id).sendSignal(id, signal)
await this.routing.providerFor(id).sendSignal(id, signal)
}
async getCwd(id: string): Promise<string> {
return this.providerFor(id).getCwd(id)
return this.routing.providerFor(id).getCwd(id)
}
async getInitialCwd(id: string): Promise<string> {
return this.providerFor(id).getInitialCwd(id)
return this.routing.providerFor(id).getInitialCwd(id)
}
async getAppliedSize(id: string): Promise<{ cols: number; rows: number } | null> {
return (await this.providerFor(id).getAppliedSize?.(id)) ?? null
return (await this.routing.providerFor(id).getAppliedSize?.(id)) ?? null
}
async getBufferSnapshot(
id: string,
opts?: { scrollbackRows?: number }
): Promise<PtyProviderBufferSnapshot | null> {
return await this.snapshotAcks.capture(id, opts, this.providerFor(id), this.allDaemonAdapters())
return await this.snapshotAcks.capture(
id,
opts,
this.routing.providerFor(id),
this.allDaemonAdapters()
)
}
async clearBuffer(id: string): Promise<void> {
await this.providerFor(id).clearBuffer(id)
await this.routing.providerFor(id).clearBuffer(id)
}
async closeStartupQueryAuthority(id: string): Promise<number> {
return (await this.providerFor(id).closeStartupQueryAuthority?.(id)) ?? 0
return (await this.routing.providerFor(id).closeStartupQueryAuthority?.(id)) ?? 0
}
acknowledgeDataEvent(id: string, charCount: number): void {
this.providerFor(id).acknowledgeDataEvent(id, charCount)
this.routing.providerFor(id).acknowledgeDataEvent(id, charCount)
}
async hasChildProcesses(id: string): Promise<boolean> {
return this.providerFor(id).hasChildProcesses(id)
return this.routing.providerFor(id).hasChildProcesses(id)
}
async getForegroundProcess(id: string): Promise<string | null> {
return this.providerFor(id).getForegroundProcess(id)
return this.routing.providerFor(id).getForegroundProcess(id)
}
inspectProcess(id: string) {
return this.hasPty(id)
? inspectPtyProviderProcess(this.providerFor(id), id)
? inspectPtyProviderProcess(this.routing.providerFor(id), id)
: Promise.reject(new Error('terminal_gone'))
}
async confirmForegroundProcess(id: string): Promise<string | null> {
return this.providerFor(id).confirmForegroundProcess?.(id) ?? null
return this.routing.providerFor(id).confirmForegroundProcess?.(id) ?? null
}
async serialize(ids: string[]): Promise<string> {
@@ -186,7 +185,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
async listProcesses(opts?: { deadlineMs?: number }): Promise<PtyProcessInfo[]> {
const results = await Promise.all(
this.allProviders().map((provider) => provider.listProcesses(opts))
this.routing.allProviders().map((provider) => provider.listProcesses(opts))
)
return results.flat()
}
@@ -211,7 +210,15 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
onBackgroundStreamEvent(callback: (payload: PtyBackgroundStreamEvent) => void): () => void {
return combineUnsubscribes(
this.allProviders().flatMap((provider) => provider.onBackgroundStreamEvent?.(callback) ?? [])
this.allProviders().flatMap((provider) => {
return (
provider.onBackgroundStreamEvent?.((payload) => {
if (this.routing.shouldForwardStreamEvent(provider, payload.id)) {
callback(payload)
}
}) ?? []
)
})
)
}
@@ -225,7 +232,13 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}
onReplay(callback: (payload: { id: string; data: string }) => void): () => void {
const unsubscribes = this.allProviders().map((provider) => provider.onReplay(callback))
const unsubscribes = this.allProviders().map((provider) =>
provider.onReplay((payload) => {
if (this.routing.shouldForwardStreamEvent(provider, payload.id)) {
callback(payload)
}
})
)
let active = true
const trackedUnsubscribe = (): void => {
if (!active) {
@@ -257,29 +270,15 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}
clearTombstone(sessionId: string): void {
findDaemonAdapter(this.sessionProviders, this.allDaemonAdapters(), sessionId)?.clearTombstone(
sessionId
)
this.routing.clearTombstone(sessionId)
}
async reconcileOnStartup(validWorktreeIds: Set<string>): Promise<{
alive: string[]
killed: string[]
}> {
const alive: string[] = []
const killed: string[] = []
for (const adapter of this.allDaemonAdapters()) {
const result = await adapter.reconcileOnStartup(validWorktreeIds)
for (const id of result.alive) {
alive.push(id)
this.sessionProviders.set(id, adapter)
}
for (const id of result.killed) {
killed.push(id)
this.sessionProviders.delete(id)
}
}
return { alive, killed }
const result = await this.routing.reconcile(validWorktreeIds)
return result
}
dispose(): void {
@@ -291,19 +290,19 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
disposeProviderOnly(): void {
combineUnsubscribes(this.unsubscribers.splice(0))()
this.routing.dispose()
}
async shutdownFallbackSessions(): Promise<number> {
return shutdownDegradedFallbackSessions(this.sessionProviders, this.fallback)
return await this.routing.shutdownFallbackSessions()
}
getCurrentDaemonSessionIds(): string[] {
return listProviderSessionIds(this.sessionProviders, this.current)
return this.routing.currentSessionIds()
}
fanoutCurrentDaemonSyntheticExits(code: number): void {
for (const id of this.getCurrentDaemonSessionIds()) {
this.sessionProviders.delete(id)
for (const id of this.routing.recordCurrentSyntheticExits()) {
// Why: restart kills listed sessions even when the adapter did not track them active.
// oxlint-disable-next-line unicorn/no-useless-spread -- copy-safe: listeners may unsubscribe during iteration
for (const listener of [...this.exitListeners]) {
@@ -329,29 +328,11 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
return this.allDaemonAdapters()
}
private providerFor(sessionId: string): IPtyProvider {
return (
this.sessionProviders.get(sessionId) ??
this.findProviderForExistingSession(sessionId) ??
this.fallback
)
}
private findProviderForExistingSession(sessionId: string): IPtyProvider | null {
for (const provider of this.allProviders()) {
if (provider.hasPty?.(sessionId) === true) {
this.sessionProviders.set(sessionId, provider)
return provider
}
}
return null
}
private allProviders(): IPtyProvider[] {
return [this.fallback, ...this.allDaemonAdapters()]
return this.routing.allProviders()
}
private allDaemonAdapters(): DaemonPtyAdapter[] {
return [this.current, ...this.legacy]
return [...this.routing.allAdapters()]
}
}
@@ -1,22 +1,312 @@
import type { IPtyProvider } from '../providers/types'
import type { IPtyProvider, PtySpawnOptions, PtySpawnResult } from '../providers/types'
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import { shutdownDegradedFallbackSessions } from './degraded-daemon-fallback-shutdown'
import { DaemonSessionRouteTable } from './daemon-session-route-table'
import { DaemonSessionOwnerUnknownError, sameDaemonIncarnation } from './daemon-session-route'
import type { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowledgement-routes'
import { DegradedFallbackSpawnRoutes } from './degraded-fallback-spawn-routes'
export function listProviderSessionIds(
sessionProviders: ReadonlyMap<string, IPtyProvider>,
provider: IPtyProvider
): string[] {
return [...sessionProviders]
.filter(([, mappedProvider]) => mappedProvider === provider)
.map(([id]) => id)
}
export class DegradedDaemonSessionRouting {
private readonly adapters: DaemonPtyAdapter[]
private readonly adapterSet: ReadonlySet<DaemonPtyAdapter>
private readonly routes: DaemonSessionRouteTable
private readonly fallbackSessions = new Map<string, IPtyProvider>()
private readonly fallbackCollisions = new Set<string>()
private readonly fallbackSpawns = new DegradedFallbackSpawnRoutes()
private readonly identityUnsubscribes: (() => void)[] = []
export function findDaemonAdapter(
sessionProviders: ReadonlyMap<string, IPtyProvider>,
daemonAdapters: readonly DaemonPtyAdapter[],
sessionId: string
): DaemonPtyAdapter | null {
const provider = sessionProviders.get(sessionId)
return provider && daemonAdapters.includes(provider as DaemonPtyAdapter)
? (provider as DaemonPtyAdapter)
: null
constructor(
private readonly current: DaemonPtyAdapter,
legacy: readonly DaemonPtyAdapter[],
private readonly fallback: IPtyProvider,
private readonly snapshotAcks: DaemonSnapshotAcknowledgementRoutes
) {
this.adapters = [current, ...legacy]
this.adapterSet = new Set(this.adapters)
this.routes = new DaemonSessionRouteTable(this.adapters)
for (const adapter of this.adapters) {
this.identityUnsubscribes.push(
adapter.onDaemonIdentityChanged(({ previous }) => {
this.snapshotAcks.dropAdapterIncarnation(adapter, previous)
this.routes.markOwnerUnavailable(adapter, previous)
})
)
}
}
async discover(): Promise<void> {
for (const adapter of this.adapters) {
try {
const before = adapter.getLastAuthenticatedDaemonIdentity()
const sessions = await adapter.listProcesses()
const incarnation = adapter.getLastAuthenticatedDaemonIdentity()
if (before && !sameDaemonIncarnation(before, incarnation)) {
throw new Error('daemon incarnation changed during session discovery')
}
this.routes.recordCompleteDiscovery(
adapter,
sessions.map((session) => session.id),
incarnation
)
for (const session of sessions) {
if (this.fallbackSessions.has(session.id)) {
this.fallbackCollisions.add(session.id)
}
}
} catch (error) {
this.routes.recordDiscoveryFailure(adapter)
console.warn('[daemon] Failed to discover degraded daemon sessions', error)
}
}
}
async spawn(opts: PtySpawnOptions): Promise<PtySpawnResult> {
const target = await this.spawnTarget(opts)
const inFlightFallbackId = target === this.fallback ? opts.sessionId : undefined
if (inFlightFallbackId) {
this.fallbackSpawns.begin(inFlightFallbackId)
}
try {
const result = await target.spawn(opts)
if (target === this.fallback) {
if (!this.fallbackSpawns.hasExited(result.id)) {
this.fallbackSessions.set(result.id, target)
}
if (this.routes.get(result.id)) {
this.fallbackCollisions.add(result.id)
}
} else {
this.routes.recordOwned(result.id, target as DaemonPtyAdapter)
}
return result
} finally {
if (inFlightFallbackId) {
this.fallbackSpawns.end(inFlightFallbackId)
}
}
}
providerFor(sessionId: string): IPtyProvider {
if (this.fallbackCollisions.has(sessionId)) {
throw new DaemonSessionOwnerUnknownError(sessionId)
}
return this.fallbackSessions.get(sessionId) ?? this.routes.resolveOwnerSync(sessionId)
}
ownerForHint(sessionId: string): IPtyProvider | null {
try {
return this.providerFor(sessionId)
} catch {
return null
}
}
hasPty(sessionId: string): boolean {
if (this.fallbackCollisions.has(sessionId)) {
return true
}
const fallback = this.fallbackSessions.get(sessionId)
return fallback ? (fallback.hasPty?.(sessionId) ?? true) : this.routes.hasPty(sessionId)
}
async probePtyLiveness(sessionId: string): Promise<boolean | null> {
if (this.fallbackCollisions.has(sessionId)) {
return null
}
const fallback = this.fallbackSessions.get(sessionId)
if (!fallback) {
return await this.routes.probeLiveness(sessionId)
}
return fallback.probePtyLiveness
? await fallback.probePtyLiveness(sessionId)
: (fallback.hasPty?.(sessionId) ?? null)
}
shouldForwardStreamEvent(provider: IPtyProvider, sessionId: string): boolean {
const adapter = this.daemonAdapter(provider)
if (adapter) {
const shouldForward = this.routes.recordDataOwner(sessionId, adapter)
if (this.fallbackSessions.has(sessionId)) {
this.fallbackCollisions.add(sessionId)
return false
}
return shouldForward
}
const route = this.routes.get(sessionId)
if (!route) {
return (
!this.fallbackSpawns.hasExited(sessionId) &&
(this.fallbackSessions.get(sessionId) === provider ||
this.fallbackSpawns.isInFlight(sessionId))
)
}
if (
this.fallbackSessions.get(sessionId) === provider &&
route.state === 'unavailable' &&
!this.fallbackCollisions.has(sessionId)
) {
this.fallbackCollisions.delete(sessionId)
return true
}
if (route) {
this.fallbackCollisions.add(sessionId)
return false
}
return true
}
recordExit(provider: IPtyProvider, sessionId: string): boolean {
const adapter = this.daemonAdapter(provider)
if (!adapter) {
const isOwned =
this.fallbackSessions.get(sessionId) === provider ||
this.fallbackSpawns.isInFlight(sessionId)
if (!isOwned) {
return false
}
this.fallbackSpawns.recordExit(sessionId)
this.fallbackSessions.delete(sessionId)
this.fallbackCollisions.delete(sessionId)
const route = this.routes.get(sessionId)
return !route || route.state === 'unavailable'
}
this.snapshotAcks.dropForProducer(sessionId, adapter)
if (this.fallbackSessions.has(sessionId)) {
this.routes.markUnavailable(sessionId, adapter)
if (this.routes.get(sessionId)?.state === 'unavailable') {
this.fallbackCollisions.delete(sessionId)
}
return false
}
const shouldForward = this.routes.shouldForwardEvent(sessionId, adapter)
this.routes.markUnavailable(sessionId, adapter)
return shouldForward
}
recordShutdown(sessionId: string, provider: IPtyProvider, keepHistory?: boolean): void {
if (keepHistory) {
return
}
if (provider === this.fallback) {
this.fallbackSessions.delete(sessionId)
this.fallbackCollisions.delete(sessionId)
} else {
const adapter = provider as DaemonPtyAdapter
this.routes.markUnavailable(sessionId, adapter)
this.snapshotAcks.dropForProducer(sessionId, adapter)
}
}
async reconcile(validWorktreeIds: Set<string>): Promise<{ alive: string[]; killed: string[] }> {
const alive: string[] = []
const killed: string[] = []
const reconciled: {
adapter: DaemonPtyAdapter
incarnation: ReturnType<DaemonPtyAdapter['getLastAuthenticatedDaemonIdentity']>
alive: string[]
killed: string[]
}[] = []
for (const adapter of this.adapters) {
const before = adapter.getLastAuthenticatedDaemonIdentity()
const result = await adapter.reconcileOnStartup(validWorktreeIds)
const incarnation = adapter.getLastAuthenticatedDaemonIdentity()
if (before && !sameDaemonIncarnation(before, incarnation)) {
this.routes.markOwnerUnavailable(adapter, before)
throw new Error('daemon incarnation changed during startup reconciliation')
}
reconciled.push({ adapter, incarnation, ...result })
for (const id of result.alive) {
alive.push(id)
}
for (const id of result.killed) {
killed.push(id)
}
}
for (const { adapter, incarnation } of reconciled) {
if (!sameDaemonIncarnation(incarnation, adapter.getLastAuthenticatedDaemonIdentity())) {
this.routes.markOwnerUnavailable(adapter, incarnation)
throw new Error('daemon incarnation changed during startup reconciliation')
}
}
for (const { adapter, incarnation, alive: sessionIds } of reconciled) {
for (const id of sessionIds) {
this.routes.recordOwnedIncarnation(id, adapter, incarnation)
if (this.fallbackSessions.has(id)) {
this.fallbackCollisions.add(id)
}
}
}
for (const { adapter, incarnation, killed: sessionIds } of reconciled) {
for (const id of sessionIds) {
this.routes.markUnavailable(id, adapter, incarnation)
this.snapshotAcks.dropForProducer(id, adapter, incarnation)
if (this.routes.get(id)?.state === 'unavailable') {
this.fallbackCollisions.delete(id)
}
}
}
return { alive, killed }
}
clearTombstone(sessionId: string): void {
this.routes.getOwned(sessionId)?.clearTombstone(sessionId)
}
currentSessionIds(): string[] {
return this.routes.getOwnedSessionIds(this.current)
}
recordCurrentSyntheticExits(): string[] {
const forwarded: string[] = []
for (const id of this.currentSessionIds()) {
if (this.routes.shouldForwardEvent(id, this.current)) {
forwarded.push(id)
}
this.routes.markUnavailable(id, this.current)
this.snapshotAcks.dropForProducer(id, this.current)
if (this.routes.get(id)?.state === 'unavailable') {
this.fallbackCollisions.delete(id)
}
}
return forwarded
}
async shutdownFallbackSessions(): Promise<number> {
return await shutdownDegradedFallbackSessions(this.fallbackSessions, this.fallback)
}
allProviders(): IPtyProvider[] {
return [this.fallback, ...this.adapters]
}
allAdapters(): readonly DaemonPtyAdapter[] {
return this.adapters
}
dispose(): void {
for (const unsubscribe of this.identityUnsubscribes.splice(0)) {
unsubscribe()
}
this.fallbackSessions.clear()
this.fallbackCollisions.clear()
this.fallbackSpawns.clear()
this.snapshotAcks.clear()
}
private async spawnTarget(opts: PtySpawnOptions): Promise<IPtyProvider> {
if (opts.sessionId && this.fallbackCollisions.has(opts.sessionId)) {
throw new DaemonSessionOwnerUnknownError(opts.sessionId)
}
if (!opts.sessionId || opts.isNewSession) {
this.routes.assertFreshSpawnAvailable(opts.sessionId)
return this.fallback
}
return (
this.fallbackSessions.get(opts.sessionId) ?? (await this.routes.resolveOwner(opts.sessionId))
)
}
private daemonAdapter(provider: IPtyProvider): DaemonPtyAdapter | null {
return this.adapterSet.has(provider as DaemonPtyAdapter) ? (provider as DaemonPtyAdapter) : null
}
}
@@ -0,0 +1,41 @@
export class DegradedFallbackSpawnRoutes {
private readonly counts = new Map<string, number>()
private readonly exited = new Set<string>()
begin(sessionId: string): void {
const count = this.counts.get(sessionId) ?? 0
if (count === 0) {
this.exited.delete(sessionId)
}
this.counts.set(sessionId, count + 1)
}
end(sessionId: string): void {
const remaining = (this.counts.get(sessionId) ?? 1) - 1
if (remaining > 0) {
this.counts.set(sessionId, remaining)
return
}
this.counts.delete(sessionId)
this.exited.delete(sessionId)
}
isInFlight(sessionId: string): boolean {
return this.counts.has(sessionId)
}
hasExited(sessionId: string): boolean {
return this.exited.has(sessionId)
}
recordExit(sessionId: string): void {
if (this.isInFlight(sessionId)) {
this.exited.add(sessionId)
}
}
clear(): void {
this.counts.clear()
this.exited.clear()
}
}
+28
View File
@@ -383,6 +383,34 @@ describe('pty:management IPC handlers', () => {
})
})
it('does not count a replacement incarnation of the same id as remaining', async () => {
const sessionId = 'reused-id'
const current = makeAdapter(5, [])
vi.mocked(current.listSessions)
.mockResolvedValueOnce([
makeSession(sessionId, {
incarnationId: 'original'
}) as Omit<DaemonSessionInfo, 'protocolVersion'>
])
.mockResolvedValue([
makeSession(sessionId, {
incarnationId: 'replacement'
}) as Omit<DaemonSessionInfo, 'protocolVersion'>
])
const { registerDaemonManagementHandlers } = await importFresh()
getDaemonProviderMock.mockReturnValue(await makeRouter(current))
registerDaemonManagementHandlers()
const result = await runKillAllWithPolls(buildHandlerMap()['pty:management:killAll'])
expect(result).toEqual({
killedCount: 1,
remainingCount: 0,
killedSessionIds: [sessionId]
})
expect(current.shutdown).toHaveBeenCalledExactlyOnceWith(sessionId, { immediate: true })
})
it('swallows per-session shutdown rejections without stopping the batch', async () => {
const sessionsList = [makeSession('a'), makeSession('b')]
const current = makeAdapter(5, [])
+42 -11
View File
@@ -61,13 +61,45 @@ async function collectSessions(adapters: DaemonPtyAdapter[]): Promise<DaemonSess
return { sessions, unavailableOwners }
}
function sameSessionIncarnation(left: OwnedDaemonSession, right: OwnedDaemonSession): boolean {
type DaemonSessionIncarnationBucket = {
hasUnqualifiedSession: boolean
incarnationIds: Set<string>
}
function indexSessionIncarnations(
sessions: readonly OwnedDaemonSession[]
): Map<DaemonPtyAdapter, Map<string, DaemonSessionIncarnationBucket>> {
const byOwner = new Map<DaemonPtyAdapter, Map<string, DaemonSessionIncarnationBucket>>()
for (const { owner, session } of sessions) {
let bySessionId = byOwner.get(owner)
if (!bySessionId) {
bySessionId = new Map()
byOwner.set(owner, bySessionId)
}
let bucket = bySessionId.get(session.sessionId)
if (!bucket) {
bucket = { hasUnqualifiedSession: false, incarnationIds: new Set() }
bySessionId.set(session.sessionId, bucket)
}
if (session.incarnationId === undefined) {
bucket.hasUnqualifiedSession = true
} else {
bucket.incarnationIds.add(session.incarnationId)
}
}
return byOwner
}
function hasSessionIncarnation(
index: ReadonlyMap<DaemonPtyAdapter, ReadonlyMap<string, DaemonSessionIncarnationBucket>>,
original: OwnedDaemonSession
): boolean {
const bucket = index.get(original.owner)?.get(original.session.sessionId)
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)
bucket !== undefined &&
(original.session.incarnationId === undefined ||
bucket.hasUnqualifiedSession ||
bucket.incarnationIds.has(original.session.incarnationId))
)
}
@@ -120,20 +152,19 @@ export function registerDaemonManagementHandlers(): void {
for (let attempt = 0; attempt < MAX_POLL_ATTEMPTS; attempt += 1) {
await sleep(POLL_INTERVAL_MS)
const current = await collectSessions(adapters)
const currentIndex = indexSessionIncarnations(current.sessions)
remainingOriginal = initial.filter(
(original) =>
current.unavailableOwners.has(original.owner) ||
current.sessions.some((candidate) => sameSessionIncarnation(original, candidate))
hasSessionIncarnation(currentIndex, original)
)
if (remainingOriginal.length === 0) {
break
}
}
const killed = initial.filter(
(original) =>
!remainingOriginal.some((candidate) => sameSessionIncarnation(original, candidate))
)
const remainingSet = new Set(remainingOriginal)
const killed = initial.filter((original) => !remainingSet.has(original))
return {
killedCount: killed.length,
remainingCount: remainingOriginal.length,