fix(daemon): close round 14 ownership races

This commit is contained in:
OrcaWin
2026-07-20 08:41:35 -07:00
parent d85acefc41
commit 2f36bf47df
13 changed files with 527 additions and 260 deletions
+15 -1
View File
@@ -281,18 +281,29 @@ describe('DaemonPtyRouter', () => {
expect(current.spawn).toHaveBeenCalledWith({ sessionId: 'legacy-session', cols: 80, rows: 24 })
})
it('uses mapped adapter liveness instead of routing-cache presence for hasPty', async () => {
it('retains a sleeping legacy owner without reporting a logical exit', async () => {
const current = createAdapter('current')
const legacy = createAdapter('legacy', ['legacy-session'])
const router = new DaemonPtyRouter({ current, legacy: [legacy] })
const exitSpy = vi.fn()
router.onExit(exitSpy)
await router.discoverLegacySessions()
expect(router.hasPty('legacy-session')).toBe(true)
await router.shutdown('legacy-session', { keepHistory: true })
legacy.emitExit('legacy-session', 0)
await router.listProcesses()
expect(router.hasPty('legacy-session')).toBe(false)
expect(current.hasPty).not.toHaveBeenCalledWith('legacy-session')
expect(exitSpy).not.toHaveBeenCalled()
await router.spawn({ sessionId: 'legacy-session', cols: 80, rows: 24 })
expect(legacy.spawn).toHaveBeenCalledWith({
sessionId: 'legacy-session',
cols: 80,
rows: 24
})
})
it('fails listProcesses closed when any routed adapter cannot list sessions', async () => {
@@ -397,10 +408,13 @@ describe('DaemonPtyRouter', () => {
const firstLegacy = createAdapter('legacy-a', ['duplicate-session'])
const secondLegacy = createAdapter('legacy-b', ['duplicate-session'])
const router = new DaemonPtyRouter({ current, legacy: [firstLegacy, secondLegacy] })
const exitSpy = vi.fn()
router.onExit(exitSpy)
await expect(router.listProcesses()).rejects.toThrow('Ambiguous PTY session ownership')
current.emitExit('duplicate-session', 0)
expect(exitSpy).not.toHaveBeenCalled()
await expect(
router.shutdown('duplicate-session', {
immediate: true
+17 -9
View File
@@ -35,9 +35,10 @@ export class DaemonPtyRouter implements IPtyProvider {
}
}),
adapter.onExit((payload) => {
this.sessionRouting.remove(payload.id, adapter)
for (const listener of this.exitListeners) {
listener(payload)
if (this.sessionRouting.handleExit(payload.id, adapter)) {
for (const listener of this.exitListeners) {
listener(payload)
}
}
})
)
@@ -49,13 +50,10 @@ export class DaemonPtyRouter implements IPtyProvider {
const refresh = this.sessionRouting.beginInventoryRefresh()
try {
const sessions = await adapter.listProcesses()
for (const session of sessions) {
this.sessionRouting.addInventorySession(session.id, adapter, refresh)
}
this.sessionRouting.refreshLive([adapter], [sessions], refresh)
} catch (error) {
console.warn('[daemon] Failed to discover legacy daemon sessions', error)
} finally {
this.sessionRouting.finishInventoryRefresh(refresh)
console.warn('[daemon] Failed to discover legacy daemon sessions', error)
}
}
}
@@ -113,13 +111,23 @@ export class DaemonPtyRouter implements IPtyProvider {
throw this.ambiguousOwnershipError([id])
}
const adapter = this.adapterFor(id)
await adapter.shutdown(id, opts)
// Why: sleep passes keepHistory=true and re-spawns against the same
// sessionId on wake. If we delete the routing entry here, adapterFor()
// falls back to `this.current` on wake — for a session that originally
// lived on a legacy adapter (different protocolVersion), the wake-side
// createOrAttach lands on the wrong adapter and creates a fresh session,
// losing the cold-restore from the legacy adapter's history dir.
if (opts.keepHistory) {
this.sessionRouting.beginSleep(id, adapter)
}
try {
await adapter.shutdown(id, opts)
} catch (error) {
if (opts.keepHistory) {
this.sessionRouting.cancelSleep(id, adapter)
}
throw error
}
if (!opts.keepHistory) {
this.sessionRouting.remove(id, adapter)
}
@@ -0,0 +1,83 @@
import { describe, expect, it } from 'vitest'
import { DaemonPtySessionRouting } from './daemon-pty-session-routing'
type Owner = { label: string }
const processInfo = (id: string) => ({ id, cwd: '', title: id })
describe('DaemonPtySessionRouting', () => {
it('applies a daemon snapshot beside a newer fallback spawn', () => {
const fallback: Owner = { label: 'fallback' }
const daemon: Owner = { label: 'daemon' }
const routing = new DaemonPtySessionRouting<Owner>()
const refresh = routing.beginInventoryRefresh()
routing.add('duplicate', fallback)
const ambiguous = routing.refreshLive(
[fallback, daemon],
[[], [processInfo('duplicate')]],
refresh
)
expect(ambiguous).toEqual(['duplicate'])
expect(routing.isAmbiguous('duplicate')).toBe(true)
})
it('supersedes an older provider snapshot even when the newer snapshot is empty', () => {
const owner: Owner = { label: 'daemon' }
const routing = new DaemonPtySessionRouting<Owner>()
const older = routing.beginInventoryRefresh()
const newer = routing.beginInventoryRefresh()
routing.refreshLive([owner], [[]], newer)
routing.refreshLive([owner], [[processInfo('stale')]], older)
expect(routing.get('stale')).toBeUndefined()
})
it('reports logical exit only after the final physical owner exits', () => {
const first: Owner = { label: 'first' }
const second: Owner = { label: 'second' }
const third: Owner = { label: 'third' }
const routing = new DaemonPtySessionRouting<Owner>()
routing.add('duplicate', first)
routing.add('duplicate', second)
routing.add('duplicate', third)
expect(routing.handleExit('duplicate', first)).toBe(false)
expect(routing.handleExit('duplicate', second)).toBe(false)
expect(routing.handleExit('duplicate', third)).toBe(true)
})
it('prunes absent live routes but preserves an intentionally sleeping owner', () => {
const owner: Owner = { label: 'legacy' }
const routing = new DaemonPtySessionRouting<Owner>()
routing.add('stale-live', owner)
let refresh = routing.beginInventoryRefresh()
routing.refreshLive([owner], [[]], refresh)
expect(routing.get('stale-live')).toBeUndefined()
routing.add('sleeping', owner)
routing.beginSleep('sleeping', owner)
expect(routing.handleExit('sleeping', owner)).toBe(false)
refresh = routing.beginInventoryRefresh()
routing.refreshLive([owner], [[]], refresh)
expect(routing.get('sleeping')).toBe(owner)
routing.add('sleeping', owner)
expect(routing.handleExit('sleeping', owner)).toBe(true)
})
it('does not let an in-progress sleep inventory expose its expected exit', () => {
const owner: Owner = { label: 'legacy' }
const routing = new DaemonPtySessionRouting<Owner>()
routing.add('sleeping', owner)
routing.beginSleep('sleeping', owner)
const refresh = routing.beginInventoryRefresh()
routing.refreshLive([owner], [[processInfo('sleeping')]], refresh)
expect(routing.handleExit('sleeping', owner)).toBe(false)
expect(routing.get('sleeping')).toBe(owner)
})
})
+158 -111
View File
@@ -10,24 +10,27 @@ type SessionMutation = {
refresh?: DaemonPtyInventoryRefresh
}
export class DaemonPtySessionRouting {
private adapters = new Map<string, DaemonPtyAdapter>()
private ambiguousAdapters = new Map<string, Set<DaemonPtyAdapter>>()
/** Tracks every physical owner of a logical PTY id across inventory races. */
export class DaemonPtySessionRouting<Owner extends object = DaemonPtyAdapter> {
private owners = new Map<string, Owner>()
private ambiguousOwners = new Map<string, Set<Owner>>()
private sleepingOwners = new Map<string, Set<Owner>>()
private revision = 0
private sessionMutations = new Map<string, SessionMutation>()
private sessionMutations = new Map<string, Map<Owner, SessionMutation>>()
private activeInventoryRefreshes = new Set<DaemonPtyInventoryRefresh>()
private ownerInventoryRefreshes = new Map<Owner, DaemonPtyInventoryRefresh>()
get(sessionId: string): DaemonPtyAdapter | undefined {
return this.adapters.get(sessionId)
get(sessionId: string): Owner | undefined {
return this.owners.get(sessionId)
}
isAmbiguous(sessionId: string): boolean {
return this.ambiguousAdapters.has(sessionId)
return this.ambiguousOwners.has(sessionId)
}
beginInventoryRefresh(): DaemonPtyInventoryRefresh {
// Why: a later overlapping snapshot may supersede an earlier refresh,
// while an exit/add event must supersede every snapshot already in flight.
// while an exit/add event must supersede only that provider's in-flight result.
this.revision += 1
const refresh = { revision: this.revision }
this.activeInventoryRefreshes.add(refresh)
@@ -43,132 +46,169 @@ export class DaemonPtySessionRouting {
const oldestRevision = Math.min(
...[...this.activeInventoryRefreshes].map((active) => active.revision)
)
for (const [id, mutation] of this.sessionMutations) {
if (mutation.revision <= oldestRevision) {
for (const [id, ownerMutations] of this.sessionMutations) {
for (const [owner, mutation] of ownerMutations) {
if (mutation.revision <= oldestRevision) {
ownerMutations.delete(owner)
}
}
if (ownerMutations.size === 0) {
this.sessionMutations.delete(id)
}
}
}
applyInventoryMutation(
sessionId: string,
refresh: DaemonPtyInventoryRefresh,
mutation: () => void
): boolean {
if (!this.isInventoryResultCurrent(sessionId, refresh)) {
return false
}
mutation()
this.recordMutation(sessionId, refresh)
return true
}
addInventorySession(
sessionId: string,
adapter: DaemonPtyAdapter,
refresh: DaemonPtyInventoryRefresh
): void {
this.applyInventoryMutation(sessionId, refresh, () => this.addRoute(sessionId, adapter))
}
recordExternalMutation(sessionId: string): void {
this.recordMutation(sessionId)
}
idsFor(adapter: DaemonPtyAdapter): string[] {
const ids = [...this.adapters].filter(([, owner]) => owner === adapter).map(([id]) => id)
for (const [id, owners] of this.ambiguousAdapters) {
if (owners.has(adapter)) {
idsFor(owner: Owner): string[] {
const ids = [...this.owners].filter(([, candidate]) => candidate === owner).map(([id]) => id)
for (const [id, candidates] of this.ambiguousOwners) {
if (candidates.has(owner)) {
ids.push(id)
}
}
return ids.sort()
}
add(sessionId: string, adapter: DaemonPtyAdapter): void {
this.recordMutation(sessionId)
this.addRoute(sessionId, adapter)
add(sessionId: string, owner: Owner): void {
this.recordMutation(sessionId, owner)
this.removeSleepingOwner(sessionId, owner)
this.addRoute(sessionId, owner)
}
private addRoute(sessionId: string, adapter: DaemonPtyAdapter): void {
const ambiguous = this.ambiguousAdapters.get(sessionId)
if (ambiguous) {
ambiguous.add(adapter)
return
}
const existing = this.adapters.get(sessionId)
if (existing && existing !== adapter) {
this.adapters.delete(sessionId)
this.ambiguousAdapters.set(sessionId, new Set([existing, adapter]))
return
}
this.adapters.set(sessionId, adapter)
remove(sessionId: string, owner: Owner): boolean {
this.recordMutation(sessionId, owner)
this.removeSleepingOwner(sessionId, owner)
this.removeRoute(sessionId, owner)
return !this.hasOwner(sessionId)
}
remove(sessionId: string, adapter: DaemonPtyAdapter): void {
this.recordMutation(sessionId)
const ambiguous = this.ambiguousAdapters.get(sessionId)
if (ambiguous) {
ambiguous.delete(adapter)
if (ambiguous.size === 1) {
this.ambiguousAdapters.delete(sessionId)
this.adapters.set(sessionId, ambiguous.values().next().value as DaemonPtyAdapter)
} else if (ambiguous.size === 0) {
this.ambiguousAdapters.delete(sessionId)
this.adapters.delete(sessionId)
}
return
}
if (this.adapters.get(sessionId) === adapter) {
this.adapters.delete(sessionId)
handleExit(sessionId: string, owner: Owner): boolean {
this.recordMutation(sessionId, owner)
// Why: keepHistory intentionally exits the physical process while retaining
// the provider that owns its cold-restore state for the later wake.
if (this.isSleeping(sessionId, owner)) {
return false
}
this.removeRoute(sessionId, owner)
return !this.hasOwner(sessionId)
}
beginSleep(sessionId: string, owner: Owner): void {
this.recordMutation(sessionId, owner)
this.addRoute(sessionId, owner)
const sleeping = this.sleepingOwners.get(sessionId) ?? new Set<Owner>()
sleeping.add(owner)
this.sleepingOwners.set(sessionId, sleeping)
}
cancelSleep(sessionId: string, owner: Owner): void {
this.recordMutation(sessionId, owner)
this.removeSleepingOwner(sessionId, owner)
}
refreshLive(
adapters: DaemonPtyAdapter[],
inventoryOwners: Owner[],
results: PtyProcessInfo[][],
refresh: DaemonPtyInventoryRefresh
): string[] {
const owners = new Map<string, Set<DaemonPtyAdapter>>()
for (let index = 0; index < adapters.length; index += 1) {
for (const session of results[index]) {
const sessionOwners = owners.get(session.id) ?? new Set<DaemonPtyAdapter>()
sessionOwners.add(adapters[index])
owners.set(session.id, sessionOwners)
}
}
// Why: unique absent sessions may be sleeping on legacy history, but a
// successful all-daemon inventory can retire stale live ambiguity.
for (const id of this.ambiguousAdapters.keys()) {
if (!owners.has(id) && this.isInventoryResultCurrent(id, refresh)) {
this.applyInventoryMutation(id, refresh, () => this.ambiguousAdapters.delete(id))
}
}
const ambiguousIds: string[] = []
for (const [id, sessionOwners] of owners) {
if (!this.isInventoryResultCurrent(id, refresh)) {
continue
}
if (sessionOwners.size === 1) {
this.applyInventoryMutation(id, refresh, () => {
this.ambiguousAdapters.delete(id)
this.adapters.set(id, sessionOwners.values().next().value as DaemonPtyAdapter)
})
} else {
this.applyInventoryMutation(id, refresh, () => {
this.adapters.delete(id)
this.ambiguousAdapters.set(id, sessionOwners)
})
ambiguousIds.push(id)
}
for (let index = 0; index < inventoryOwners.length; index += 1) {
this.refreshOwner(inventoryOwners[index], results[index], refresh)
}
this.finishInventoryRefresh(refresh)
return ambiguousIds.sort()
return [...this.ambiguousOwners.keys()].sort()
}
private isInventoryResultCurrent(sessionId: string, refresh: DaemonPtyInventoryRefresh): boolean {
const mutation = this.sessionMutations.get(sessionId)
private refreshOwner(
owner: Owner,
sessions: PtyProcessInfo[],
refresh: DaemonPtyInventoryRefresh
): void {
const latestRefresh = this.ownerInventoryRefreshes.get(owner)
if (latestRefresh && latestRefresh.revision > refresh.revision) {
return
}
this.ownerInventoryRefreshes.set(owner, refresh)
const reportedIds = new Set(sessions.map((session) => session.id))
const relevantIds = new Set([...this.idsFor(owner), ...reportedIds])
for (const sessionId of relevantIds) {
this.applyInventoryMutation(sessionId, owner, refresh, () => {
if (reportedIds.has(sessionId)) {
this.addRoute(sessionId, owner)
} else if (!this.isSleeping(sessionId, owner)) {
this.removeRoute(sessionId, owner)
}
})
}
}
private applyInventoryMutation(
sessionId: string,
owner: Owner,
refresh: DaemonPtyInventoryRefresh,
mutation: () => void
): boolean {
if (!this.isInventoryResultCurrent(sessionId, owner, refresh)) {
return false
}
mutation()
this.recordMutation(sessionId, owner, refresh)
return true
}
private addRoute(sessionId: string, owner: Owner): void {
const ambiguous = this.ambiguousOwners.get(sessionId)
if (ambiguous) {
ambiguous.add(owner)
return
}
const existing = this.owners.get(sessionId)
if (existing && existing !== owner) {
this.owners.delete(sessionId)
this.ambiguousOwners.set(sessionId, new Set([existing, owner]))
return
}
this.owners.set(sessionId, owner)
}
private removeRoute(sessionId: string, owner: Owner): void {
const ambiguous = this.ambiguousOwners.get(sessionId)
if (ambiguous) {
ambiguous.delete(owner)
if (ambiguous.size === 1) {
this.ambiguousOwners.delete(sessionId)
this.owners.set(sessionId, ambiguous.values().next().value as Owner)
} else if (ambiguous.size === 0) {
this.ambiguousOwners.delete(sessionId)
this.owners.delete(sessionId)
}
return
}
if (this.owners.get(sessionId) === owner) {
this.owners.delete(sessionId)
}
}
private hasOwner(sessionId: string): boolean {
return this.owners.has(sessionId) || this.ambiguousOwners.has(sessionId)
}
private isSleeping(sessionId: string, owner: Owner): boolean {
return this.sleepingOwners.get(sessionId)?.has(owner) === true
}
private removeSleepingOwner(sessionId: string, owner: Owner): void {
const sleeping = this.sleepingOwners.get(sessionId)
sleeping?.delete(owner)
if (sleeping?.size === 0) {
this.sleepingOwners.delete(sessionId)
}
}
private isInventoryResultCurrent(
sessionId: string,
owner: Owner,
refresh: DaemonPtyInventoryRefresh
): boolean {
const mutation = this.sessionMutations.get(sessionId)?.get(owner)
return (
mutation === undefined ||
mutation.revision <= refresh.revision ||
@@ -177,10 +217,17 @@ export class DaemonPtySessionRouting {
)
}
private recordMutation(sessionId: string, refresh?: DaemonPtyInventoryRefresh): void {
private recordMutation(
sessionId: string,
owner: Owner,
refresh?: DaemonPtyInventoryRefresh
): void {
this.revision += 1
if (this.activeInventoryRefreshes.size > 0) {
this.sessionMutations.set(sessionId, { revision: this.revision, refresh })
if (this.activeInventoryRefreshes.size === 0) {
return
}
const ownerMutations = this.sessionMutations.get(sessionId) ?? new Map()
ownerMutations.set(owner, { revision: this.revision, refresh })
this.sessionMutations.set(sessionId, ownerMutations)
}
}
@@ -2,7 +2,8 @@ import type { IPtyProvider } from '../providers/types'
export async function shutdownDegradedFallbackSessions<T extends IPtyProvider>(
sessionProviders: Map<string, T>,
fallback: T
fallback: T,
onShutdown?: (sessionId: string) => void
): Promise<number> {
const ids = [...sessionProviders]
.filter(([, provider]) => provider === fallback)
@@ -11,6 +12,7 @@ export async function shutdownDegradedFallbackSessions<T extends IPtyProvider>(
ids.map(async (id) => {
await fallback.shutdown(id, { immediate: true })
sessionProviders.delete(id)
onShutdown?.(id)
})
)
// Why: fallback cleanup must not abort the user's daemon-restart recovery path.
@@ -273,6 +273,71 @@ describe('DegradedDaemonPtyProvider', () => {
warn.mockRestore()
})
it('discovers an already-live fallback collision before direct shutdown', async () => {
const current = createDaemonAdapter('daemon', ['duplicate-session'])
const fallback = createProvider('fallback', ['duplicate-session'])
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
await provider.discoverDaemonSessions()
await expect(provider.shutdown('duplicate-session', { immediate: true })).rejects.toThrow(
'Ambiguous PTY session ownership across degraded providers: duplicate-session'
)
expect(current.shutdown).not.toHaveBeenCalled()
expect(fallback.shutdown).not.toHaveBeenCalled()
})
it('probes every provider before destructive shutdown without startup inventory', async () => {
const current = createDaemonAdapter('daemon', ['duplicate-session'])
const fallback = createProvider('fallback', ['duplicate-session'])
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
await expect(provider.shutdown('duplicate-session', { immediate: true })).rejects.toThrow(
'Ambiguous PTY session ownership'
)
expect(current.shutdown).not.toHaveBeenCalled()
expect(fallback.shutdown).not.toHaveBeenCalled()
})
it('keeps a daemon inventory result when fallback spawns during the listing', async () => {
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
let resolveDaemon!: (sessions: PtyProcessInfo[]) => void
vi.mocked(current.listProcesses).mockReturnValueOnce(
new Promise((resolve) => {
resolveDaemon = resolve
})
)
const listing = provider.listProcesses()
await provider.spawn({ sessionId: 'duplicate-session', cols: 80, rows: 24 })
resolveDaemon([{ id: 'duplicate-session', cwd: '', title: 'daemon' }])
await expect(listing).rejects.toThrow(
'Ambiguous PTY session ownership across degraded providers: duplicate-session'
)
await expect(provider.shutdown('duplicate-session', { immediate: true })).rejects.toThrow(
'Ambiguous PTY session ownership'
)
})
it('waits for the final fallback owner before forwarding a duplicate exit', async () => {
const current = createDaemonAdapter('daemon', ['duplicate-session'])
const fallback = createProvider('fallback', ['duplicate-session'])
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
const exitSpy = vi.fn()
provider.onExit(exitSpy)
await provider.discoverDaemonSessions()
current.emitExit('duplicate-session', 0)
expect(exitSpy).not.toHaveBeenCalled()
fallback.emitExit('duplicate-session', 0)
expect(exitSpy).toHaveBeenCalledOnce()
expect(exitSpy).toHaveBeenCalledWith({ id: 'duplicate-session', code: 0 })
})
it('routes authoritative recovery snapshots to the owning daemon', async () => {
const current = createDaemonAdapter('daemon', ['daemon-session'])
const fallback = createProvider('fallback')
@@ -408,4 +473,19 @@ describe('DegradedDaemonPtyProvider', () => {
expect(provider.getCurrentDaemonSessionIds()).toEqual([])
expect(provider.hasPty('legacy-session')).toBe(true)
})
it('suppresses a synthetic daemon exit while a fallback duplicate survives', async () => {
const current = createDaemonAdapter('daemon', ['duplicate-session'])
const fallback = createProvider('fallback', ['duplicate-session'])
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
const exitSpy = vi.fn()
provider.onExit(exitSpy)
await provider.discoverDaemonSessions()
provider.fanoutCurrentDaemonSyntheticExits(-1)
expect(exitSpy).not.toHaveBeenCalled()
expect(provider.hasPty('duplicate-session')).toBe(true)
expect(provider.getCurrentDaemonSessionIds()).toEqual([])
})
})
+38 -37
View File
@@ -24,6 +24,8 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
private current: DaemonPtyAdapter
private legacy: DaemonPtyAdapter[]
private fallback: DegradedManagedPtyProvider
private daemonAdapters: DaemonPtyAdapter[]
private providers: DegradedManagedPtyProvider[]
private sessionRouting: DegradedDaemonSessionRouting
private unsubscribers: (() => void)[] = []
private dataListeners: ((payload: PtyDataEvent) => void)[] = []
@@ -37,9 +39,11 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
this.current = opts.current
this.legacy = opts.legacy
this.fallback = opts.fallback
this.sessionRouting = new DegradedDaemonSessionRouting(this.fallback, this.allDaemonAdapters())
this.daemonAdapters = [this.current, ...this.legacy]
this.providers = [this.fallback, ...this.daemonAdapters]
this.sessionRouting = new DegradedDaemonSessionRouting(this.fallback, this.daemonAdapters)
for (const provider of this.allProviders()) {
for (const provider of this.providers) {
this.unsubscribers.push(
provider.onData((payload) => {
for (const listener of this.dataListeners) {
@@ -47,9 +51,10 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}
}),
provider.onExit((payload) => {
this.sessionRouting.remove(payload.id, provider)
for (const listener of this.exitListeners) {
listener(payload)
if (this.sessionRouting.handleExit(payload.id, provider)) {
for (const listener of this.exitListeners) {
listener(payload)
}
}
})
)
@@ -57,17 +62,16 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}
async discoverDaemonSessions(): Promise<void> {
for (const adapter of this.allDaemonAdapters()) {
// Why: fallback sessions can already exist after startup fail-open; include
// them so a duplicate preserved daemon owner fails closed before shutdown.
for (const provider of this.providers) {
const refresh = this.sessionRouting.beginInventoryRefresh()
try {
const sessions = await adapter.listProcesses()
for (const session of sessions) {
this.sessionRouting.addDaemonInventorySession(session.id, adapter, refresh)
}
const sessions = await provider.listProcesses()
this.sessionRouting.refreshProvider(provider, sessions, refresh)
} catch (error) {
console.warn('[daemon] Failed to discover degraded daemon sessions', error)
} finally {
this.sessionRouting.finishInventoryRefresh(refresh)
console.warn('[daemon] Failed to discover degraded PTY sessions', error)
}
}
}
@@ -116,10 +120,10 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
throw this.ambiguousOwnershipError([id])
}
const target = this.providerFor(id)
await target.shutdown(id, opts)
if (!opts.keepHistory) {
this.sessionRouting.remove(id, target)
if (this.sessionRouting.isAmbiguous(id)) {
throw this.ambiguousOwnershipError([id])
}
await this.sessionRouting.shutdown(id, target, opts)
}
async sendSignal(id: string, signal: string): Promise<void> {
@@ -180,7 +184,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}
async listProcesses(opts?: { deadlineMs?: number }): Promise<PtyProcessInfo[]> {
const providers = this.allProviders()
const providers = this.providers
const refresh = this.sessionRouting.beginInventoryRefresh()
let results: PtyProcessInfo[][]
try {
@@ -215,7 +219,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}
onBackgroundStreamEvent(callback: (payload: PtyBackgroundStreamEvent) => void): () => void {
const unsubscribes = this.allProviders().flatMap(
const unsubscribes = this.providers.flatMap(
(provider) => provider.onBackgroundStreamEvent?.(callback) ?? []
)
return () => {
@@ -226,7 +230,7 @@ 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.providers.map((provider) => provider.onReplay(callback))
let active = true
const trackedUnsubscribe = (): void => {
if (!active) {
@@ -269,7 +273,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}> {
const alive: string[] = []
const killed: string[] = []
for (const adapter of this.allDaemonAdapters()) {
for (const adapter of this.daemonAdapters) {
const result = await adapter.reconcileOnStartup(validWorktreeIds)
for (const id of result.alive) {
alive.push(id)
@@ -285,7 +289,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
dispose(): void {
this.disposeProviderOnly()
for (const adapter of this.allDaemonAdapters()) {
for (const adapter of this.daemonAdapters) {
adapter.dispose()
}
}
@@ -297,7 +301,11 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}
async shutdownFallbackSessions(): Promise<number> {
return shutdownDegradedFallbackSessions(this.sessionRouting.providerMap(), this.fallback)
return shutdownDegradedFallbackSessions(
this.sessionRouting.providerMapFor(this.fallback),
this.fallback,
(id) => this.sessionRouting.remove(id, this.fallback)
)
}
getCurrentDaemonSessionIds(): string[] {
@@ -306,19 +314,20 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
fanoutCurrentDaemonSyntheticExits(code: number): void {
for (const id of this.getCurrentDaemonSessionIds()) {
this.sessionRouting.remove(id, this.current)
// Why: sessions discovered from listProcesses may not exist in the
// adapter's active-session set, but restart still kills that daemon.
// oxlint-disable-next-line unicorn/no-useless-spread -- copy-safe: listeners may unsubscribe during iteration
for (const listener of [...this.exitListeners]) {
listener({ id, code })
if (this.sessionRouting.handleExit(id, this.current)) {
// Why: sessions discovered from listProcesses may not exist in the
// adapter's active-session set, but restart still kills that daemon.
// oxlint-disable-next-line unicorn/no-useless-spread -- copy-safe: listeners may unsubscribe during iteration
for (const listener of [...this.exitListeners]) {
listener({ id, code })
}
}
}
}
async disconnectOnly(): Promise<void> {
this.disposeProviderOnly()
await Promise.all(this.allDaemonAdapters().map((adapter) => adapter.disconnectOnly()))
await Promise.all(this.daemonAdapters.map((adapter) => adapter.disconnectOnly()))
}
getCurrentAdapter(): DaemonPtyAdapter {
@@ -330,7 +339,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}
getAllAdapters(): readonly DaemonPtyAdapter[] {
return this.allDaemonAdapters()
return this.daemonAdapters
}
private providerFor(sessionId: string): DegradedManagedPtyProvider {
@@ -342,12 +351,4 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
`Ambiguous PTY session ownership across degraded providers: ${sessionIds.join(', ')}`
)
}
private allProviders(): DegradedManagedPtyProvider[] {
return [this.fallback, ...this.allDaemonAdapters()]
}
private allDaemonAdapters(): DaemonPtyAdapter[] {
return [this.current, ...this.legacy]
}
}
@@ -11,8 +11,7 @@ export type DegradedManagedPtyProvider = IPtyProvider & {
}
export class DegradedDaemonSessionRouting {
private providers = new Map<string, DegradedManagedPtyProvider>()
private daemonRouting = new DaemonPtySessionRouting()
private routing = new DaemonPtySessionRouting<DegradedManagedPtyProvider>()
constructor(
private fallback: DegradedManagedPtyProvider,
@@ -20,52 +19,59 @@ export class DegradedDaemonSessionRouting {
) {}
get(sessionId: string): DegradedManagedPtyProvider | undefined {
if (this.isAmbiguous(sessionId)) {
return undefined
}
return this.daemonRouting.get(sessionId) ?? this.providers.get(sessionId)
return this.routing.get(sessionId)
}
isAmbiguous(sessionId: string): boolean {
return (
this.daemonRouting.isAmbiguous(sessionId) ||
(this.providers.has(sessionId) && this.daemonRouting.get(sessionId) !== undefined)
)
return this.routing.isAmbiguous(sessionId)
}
beginInventoryRefresh(): DaemonPtyInventoryRefresh {
return this.daemonRouting.beginInventoryRefresh()
return this.routing.beginInventoryRefresh()
}
finishInventoryRefresh(refresh: DaemonPtyInventoryRefresh): void {
this.daemonRouting.finishInventoryRefresh(refresh)
}
addDaemonInventorySession(
sessionId: string,
adapter: DaemonPtyAdapter,
refresh: DaemonPtyInventoryRefresh
): void {
this.daemonRouting.addInventorySession(sessionId, adapter, refresh)
this.routing.finishInventoryRefresh(refresh)
}
add(sessionId: string, provider: DegradedManagedPtyProvider): void {
if (this.isDaemonAdapter(provider)) {
this.daemonRouting.add(sessionId, provider)
return
}
this.daemonRouting.recordExternalMutation(sessionId)
this.providers.set(sessionId, provider)
this.routing.add(sessionId, provider)
}
remove(sessionId: string, provider: DegradedManagedPtyProvider): void {
if (this.isDaemonAdapter(provider)) {
this.daemonRouting.remove(sessionId, provider)
return
remove(sessionId: string, provider: DegradedManagedPtyProvider): boolean {
return this.routing.remove(sessionId, provider)
}
handleExit(sessionId: string, provider: DegradedManagedPtyProvider): boolean {
return this.routing.handleExit(sessionId, provider)
}
beginSleep(sessionId: string, provider: DegradedManagedPtyProvider): void {
this.routing.beginSleep(sessionId, provider)
}
cancelSleep(sessionId: string, provider: DegradedManagedPtyProvider): void {
this.routing.cancelSleep(sessionId, provider)
}
async shutdown(
sessionId: string,
provider: DegradedManagedPtyProvider,
opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number }
): Promise<void> {
if (opts.keepHistory) {
this.beginSleep(sessionId, provider)
}
this.daemonRouting.recordExternalMutation(sessionId)
if (this.providers.get(sessionId) === provider) {
this.providers.delete(sessionId)
try {
await provider.shutdown(sessionId, opts)
} catch (error) {
if (opts.keepHistory) {
this.cancelSleep(sessionId, provider)
}
throw error
}
if (!opts.keepHistory) {
this.remove(sessionId, provider)
}
}
@@ -74,31 +80,28 @@ export class DegradedDaemonSessionRouting {
daemonResults: PtyProcessInfo[][],
refresh: DaemonPtyInventoryRefresh
): string[] {
for (const session of fallbackSessions) {
this.daemonRouting.applyInventoryMutation(session.id, refresh, () => {
this.providers.set(session.id, this.fallback)
})
}
const ambiguousIds = new Set(
this.daemonRouting.refreshLive(this.daemonAdapters, daemonResults, refresh)
return this.routing.refreshLive(
this.allProviders(),
[fallbackSessions, ...daemonResults],
refresh
)
for (const sessionId of this.providers.keys()) {
if (
this.daemonRouting.get(sessionId) !== undefined ||
this.daemonRouting.isAmbiguous(sessionId)
) {
ambiguousIds.add(sessionId)
}
}
return [...ambiguousIds].sort()
}
refreshProvider(
provider: DegradedManagedPtyProvider,
sessions: PtyProcessInfo[],
refresh: DaemonPtyInventoryRefresh
): string[] {
return this.routing.refreshLive([provider], [sessions], refresh)
}
idsForDaemon(adapter: DaemonPtyAdapter): string[] {
return this.daemonRouting.idsFor(adapter)
return this.routing.idsFor(adapter)
}
daemonFor(sessionId: string): DaemonPtyAdapter | null {
return this.isAmbiguous(sessionId) ? null : (this.daemonRouting.get(sessionId) ?? null)
const provider = this.routing.get(sessionId)
return provider && this.isDaemonAdapter(provider) ? provider : null
}
hasPty(sessionId: string): boolean {
@@ -109,8 +112,7 @@ export class DegradedDaemonSessionRouting {
if (routed) {
return routed.hasPty?.(sessionId) ?? true
}
const discovered = this.discoverExistingProvider(sessionId)
return this.isAmbiguous(sessionId) || discovered !== undefined
return this.discoverExistingProvider(sessionId) !== undefined || this.isAmbiguous(sessionId)
}
providerFor(
@@ -120,39 +122,30 @@ export class DegradedDaemonSessionRouting {
if (this.isAmbiguous(sessionId)) {
return ambiguousProvider
}
const routed = this.get(sessionId)
if (routed) {
return routed
}
const discovered = this.discoverExistingProvider(sessionId)
return this.isAmbiguous(sessionId) ? ambiguousProvider : (discovered ?? this.fallback)
const provider = this.get(sessionId) ?? this.discoverExistingProvider(sessionId)
return this.isAmbiguous(sessionId) ? ambiguousProvider : (provider ?? this.fallback)
}
providerMap(): Map<string, DegradedManagedPtyProvider> {
return this.providers
providerMapFor(provider: DegradedManagedPtyProvider): Map<string, DegradedManagedPtyProvider> {
return new Map(this.routing.idsFor(provider).map((id) => [id, provider]))
}
private discoverExistingProvider(sessionId: string): DegradedManagedPtyProvider | undefined {
for (const adapter of this.daemonAdapters) {
if (adapter.hasPty(sessionId)) {
this.add(sessionId, adapter)
// Why: probing every provider before selecting one prevents an unlisted
// fallback/daemon collision from being mistaken for unique ownership.
for (const provider of this.allProviders()) {
if (provider.hasPty?.(sessionId) === true) {
this.add(sessionId, provider)
}
}
if (this.isAmbiguous(sessionId)) {
return undefined
}
const daemon = this.get(sessionId)
if (daemon) {
return daemon
}
if (this.fallback.hasPty?.(sessionId) === true) {
this.add(sessionId, this.fallback)
return this.fallback
}
return undefined
return this.isAmbiguous(sessionId) ? undefined : this.get(sessionId)
}
private isDaemonAdapter(provider: DegradedManagedPtyProvider): provider is DaemonPtyAdapter {
return this.daemonAdapters.includes(provider as DaemonPtyAdapter)
}
private allProviders(): DegradedManagedPtyProvider[] {
return [this.fallback, ...this.daemonAdapters]
}
}
@@ -55,7 +55,7 @@ describe('TerminalSessionTeardown Windows descendant termination', () => {
}
})
it('terminates an agent Job before force-killing its root', async () => {
it('signals an agent root before terminating its Job', async () => {
const session = createSession({ agent: true, nativeWindows: true })
sessions.set('agent', session)
@@ -64,8 +64,8 @@ describe('TerminalSessionTeardown Windows descendant termination', () => {
expect(session.beginTermination).toHaveBeenCalledOnce()
expect(killWithDescendantSweepMock).not.toHaveBeenCalled()
expect(session.terminateJobTree).toHaveBeenCalledOnce()
expect(session.terminateJobTree.mock.invocationCallOrder[0]).toBeLessThan(
session.forceKillAndWaitForExit.mock.invocationCallOrder[0]!
expect(session.forceKillAndWaitForExit.mock.invocationCallOrder[0]).toBeLessThan(
session.terminateJobTree.mock.invocationCallOrder[0]!
)
})
+4 -11
View File
@@ -145,21 +145,14 @@ export class TerminalSessionTeardown {
}
private async terminateWindowsTree(session: Session, killRoot: () => void): Promise<void> {
let nativeCompletion: Promise<boolean> | undefined
try {
nativeCompletion = session.terminateJobTree(WINDOWS_PTY_JOB_DRAIN_TIMEOUT_MS)
} catch (error) {
// Why: even a synchronous native bridge failure must not strand the
// still-owned ConPTY root while destructive teardown fails closed.
killRoot()
throw error
}
// Why: Job termination can synchronously reap the root while node-pty waits
// for descendants, so send any numeric root signal before PID reuse is possible.
killRoot()
const nativeCompletion = session.terminateJobTree(WINDOWS_PTY_JOB_DRAIN_TIMEOUT_MS)
if (!nativeCompletion) {
killRoot()
throw new Error(`Windows PTY Job ownership unavailable for process ${session.pid}`)
}
killRoot()
if (!(await nativeCompletion)) {
throw new Error(`Windows PTY Job did not drain for process ${session.pid}`)
}