fix(daemon): close routing fence gaps

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