fix(daemon): order PTY inventory routing updates

This commit is contained in:
OrcaWin
2026-07-20 07:54:53 -07:00
parent df92cd16b4
commit d85acefc41
6 changed files with 290 additions and 37 deletions
+88 -6
View File
@@ -1,7 +1,12 @@
import { describe, expect, it, vi } from 'vitest'
import { DaemonPtyRouter } from './daemon-pty-router'
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import type { PtyBackgroundStreamEvent, PtySpawnOptions, PtySpawnResult } from '../providers/types'
import type {
PtyBackgroundStreamEvent,
PtyProcessInfo,
PtySpawnOptions,
PtySpawnResult
} from '../providers/types'
import { GIT_CREDENTIAL_GUARD_HOST_PROTOCOL_VERSION } from './types'
type AdapterMock = DaemonPtyAdapter & {
@@ -182,6 +187,31 @@ describe('DaemonPtyRouter', () => {
expect(current.write).toHaveBeenCalledWith(fresh.id, 'new\n')
})
it('does not discover a session from an inventory response older than exit', async () => {
const current = createAdapter('current')
const legacy = createAdapter('legacy', ['exited-session'])
let resolveLegacy!: (sessions: PtyProcessInfo[]) => void
vi.mocked(legacy.listProcesses).mockReturnValueOnce(
new Promise((resolve) => {
resolveLegacy = resolve
})
)
const router = new DaemonPtyRouter({ current, legacy: [legacy] })
const discovery = router.discoverLegacySessions()
legacy.emitExit('exited-session', 0)
resolveLegacy([{ id: 'exited-session', cwd: '', title: 'stale legacy' }])
await discovery
await router.spawn({ sessionId: 'exited-session', cols: 80, rows: 24 })
expect(current.spawn).toHaveBeenCalledWith({
sessionId: 'exited-session',
cols: 80,
rows: 24
})
expect(legacy.spawn).not.toHaveBeenCalled()
})
it('routes background hints and authoritative snapshots to the session owner', async () => {
const current = createAdapter('current')
const legacy = createAdapter('legacy', ['legacy-session'])
@@ -289,7 +319,61 @@ describe('DaemonPtyRouter', () => {
expect(legacy.write).toHaveBeenCalledWith('legacy-session', 'after recovery\n')
})
it('rejects destructive teardown when recovered daemons report a duplicate session id', async () => {
it('does not restore a route from an inventory response older than exit', async () => {
const current = createAdapter('current')
const legacy = createAdapter('legacy', ['exited-session'])
const router = new DaemonPtyRouter({ current, legacy: [legacy] })
await router.discoverLegacySessions()
let resolveLegacy!: (sessions: PtyProcessInfo[]) => void
vi.mocked(legacy.listProcesses).mockReturnValueOnce(
new Promise((resolve) => {
resolveLegacy = resolve
})
)
const listing = router.listProcesses()
legacy.emitExit('exited-session', 0)
resolveLegacy([{ id: 'exited-session', cwd: '', title: 'stale legacy' }])
await listing
await router.spawn({ sessionId: 'exited-session', cols: 80, rows: 24 })
expect(current.spawn).toHaveBeenCalledWith({
sessionId: 'exited-session',
cols: 80,
rows: 24
})
expect(legacy.spawn).not.toHaveBeenCalled()
})
it('keeps a later overlapping inventory result over an earlier response', async () => {
const current = createAdapter('current', ['migrated-session'])
const legacy = createAdapter('legacy')
let resolveOldCurrent!: (sessions: PtyProcessInfo[]) => void
let resolveOldLegacy!: (sessions: PtyProcessInfo[]) => void
vi.mocked(current.listProcesses).mockReturnValueOnce(
new Promise((resolve) => {
resolveOldCurrent = resolve
})
)
vi.mocked(legacy.listProcesses).mockReturnValueOnce(
new Promise((resolve) => {
resolveOldLegacy = resolve
})
)
const router = new DaemonPtyRouter({ current, legacy: [legacy] })
const olderListing = router.listProcesses()
await router.listProcesses()
resolveOldCurrent([])
resolveOldLegacy([{ id: 'migrated-session', cwd: '', title: 'stale legacy' }])
await olderListing
router.write('migrated-session', 'new owner\n')
expect(current.write).toHaveBeenCalledWith('migrated-session', 'new owner\n')
expect(legacy.write).not.toHaveBeenCalled()
})
it('rejects shutdown when recovered daemons report a duplicate session id', async () => {
const current = createAdapter('current', ['duplicate-session'])
const legacy = createAdapter('legacy', ['duplicate-session'])
vi.mocked(legacy.listProcesses).mockRejectedValueOnce(new Error('legacy unavailable'))
@@ -301,8 +385,7 @@ describe('DaemonPtyRouter', () => {
)
await expect(
router.shutdown('duplicate-session', {
immediate: true,
deadlineMs: Date.now() + 5_000
immediate: true
})
).rejects.toThrow('Ambiguous PTY session ownership')
expect(current.shutdown).not.toHaveBeenCalled()
@@ -320,8 +403,7 @@ describe('DaemonPtyRouter', () => {
await expect(
router.shutdown('duplicate-session', {
immediate: true,
deadlineMs: Date.now() + 5_000
immediate: true
})
).rejects.toThrow('Ambiguous PTY session ownership')
expect(firstLegacy.shutdown).not.toHaveBeenCalled()
+14 -4
View File
@@ -46,13 +46,16 @@ export class DaemonPtyRouter implements IPtyProvider {
async discoverLegacySessions(): Promise<void> {
for (const adapter of this.legacy) {
const refresh = this.sessionRouting.beginInventoryRefresh()
try {
const sessions = await adapter.listProcesses()
for (const session of sessions) {
this.sessionRouting.add(session.id, adapter)
this.sessionRouting.addInventorySession(session.id, adapter, refresh)
}
} catch (error) {
console.warn('[daemon] Failed to discover legacy daemon sessions', error)
} finally {
this.sessionRouting.finishInventoryRefresh(refresh)
}
}
}
@@ -106,7 +109,7 @@ export class DaemonPtyRouter implements IPtyProvider {
id: string,
opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number }
): Promise<void> {
if (opts.deadlineMs !== undefined && this.sessionRouting.isAmbiguous(id)) {
if (this.sessionRouting.isAmbiguous(id)) {
throw this.ambiguousOwnershipError([id])
}
const adapter = this.adapterFor(id)
@@ -185,8 +188,15 @@ export class DaemonPtyRouter implements IPtyProvider {
// Why: runtime exact-stop/liveness flows must fail closed if any adapter
// cannot provide a trustworthy process list.
const adapters = this.allAdapters()
const results = await Promise.all(adapters.map((adapter) => adapter.listProcesses(opts)))
const ambiguousIds = this.sessionRouting.refreshLive(adapters, results)
const refresh = this.sessionRouting.beginInventoryRefresh()
let results: PtyProcessInfo[][]
try {
results = await Promise.all(adapters.map((adapter) => adapter.listProcesses(opts)))
} catch (error) {
this.sessionRouting.finishInventoryRefresh(refresh)
throw error
}
const ambiguousIds = this.sessionRouting.refreshLive(adapters, results, refresh)
if (ambiguousIds.length > 0) {
throw this.ambiguousOwnershipError(ambiguousIds)
}
+104 -7
View File
@@ -1,9 +1,21 @@
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import type { PtyProcessInfo } from '../providers/types'
export type DaemonPtyInventoryRefresh = {
readonly revision: number
}
type SessionMutation = {
revision: number
refresh?: DaemonPtyInventoryRefresh
}
export class DaemonPtySessionRouting {
private adapters = new Map<string, DaemonPtyAdapter>()
private ambiguousAdapters = new Map<string, Set<DaemonPtyAdapter>>()
private revision = 0
private sessionMutations = new Map<string, SessionMutation>()
private activeInventoryRefreshes = new Set<DaemonPtyInventoryRefresh>()
get(sessionId: string): DaemonPtyAdapter | undefined {
return this.adapters.get(sessionId)
@@ -13,6 +25,56 @@ export class DaemonPtySessionRouting {
return this.ambiguousAdapters.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.
this.revision += 1
const refresh = { revision: this.revision }
this.activeInventoryRefreshes.add(refresh)
return refresh
}
finishInventoryRefresh(refresh: DaemonPtyInventoryRefresh): void {
this.activeInventoryRefreshes.delete(refresh)
if (this.activeInventoryRefreshes.size === 0) {
this.sessionMutations.clear()
return
}
const oldestRevision = Math.min(
...[...this.activeInventoryRefreshes].map((active) => active.revision)
)
for (const [id, mutation] of this.sessionMutations) {
if (mutation.revision <= oldestRevision) {
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) {
@@ -24,6 +86,11 @@ export class DaemonPtySessionRouting {
}
add(sessionId: string, adapter: DaemonPtyAdapter): void {
this.recordMutation(sessionId)
this.addRoute(sessionId, adapter)
}
private addRoute(sessionId: string, adapter: DaemonPtyAdapter): void {
const ambiguous = this.ambiguousAdapters.get(sessionId)
if (ambiguous) {
ambiguous.add(adapter)
@@ -39,6 +106,7 @@ export class DaemonPtySessionRouting {
}
remove(sessionId: string, adapter: DaemonPtyAdapter): void {
this.recordMutation(sessionId)
const ambiguous = this.ambiguousAdapters.get(sessionId)
if (ambiguous) {
ambiguous.delete(adapter)
@@ -56,7 +124,11 @@ export class DaemonPtySessionRouting {
}
}
refreshLive(adapters: DaemonPtyAdapter[], results: PtyProcessInfo[][]): string[] {
refreshLive(
adapters: DaemonPtyAdapter[],
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]) {
@@ -69,21 +141,46 @@ export class DaemonPtySessionRouting {
// 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.ambiguousAdapters.delete(id)
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.ambiguousAdapters.delete(id)
this.adapters.set(id, sessionOwners.values().next().value as DaemonPtyAdapter)
this.applyInventoryMutation(id, refresh, () => {
this.ambiguousAdapters.delete(id)
this.adapters.set(id, sessionOwners.values().next().value as DaemonPtyAdapter)
})
} else {
this.adapters.delete(id)
this.ambiguousAdapters.set(id, sessionOwners)
this.applyInventoryMutation(id, refresh, () => {
this.adapters.delete(id)
this.ambiguousAdapters.set(id, sessionOwners)
})
ambiguousIds.push(id)
}
}
this.finishInventoryRefresh(refresh)
return ambiguousIds.sort()
}
private isInventoryResultCurrent(sessionId: string, refresh: DaemonPtyInventoryRefresh): boolean {
const mutation = this.sessionMutations.get(sessionId)
return (
mutation === undefined ||
mutation.revision <= refresh.revision ||
mutation.refresh === refresh ||
(mutation.refresh !== undefined && mutation.refresh.revision < refresh.revision)
)
}
private recordMutation(sessionId: string, refresh?: DaemonPtyInventoryRefresh): void {
this.revision += 1
if (this.activeInventoryRefreshes.size > 0) {
this.sessionMutations.set(sessionId, { revision: this.revision, refresh })
}
}
}
@@ -1,7 +1,12 @@
import { describe, expect, it, vi } from 'vitest'
import { DegradedDaemonPtyProvider } from './degraded-daemon-pty-provider'
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import type { IPtyProvider, PtySpawnOptions, PtySpawnResult } from '../providers/types'
import type {
IPtyProvider,
PtyProcessInfo,
PtySpawnOptions,
PtySpawnResult
} from '../providers/types'
type ProviderMock = IPtyProvider & {
emitData: (id: string, data: string, sequenceChars?: number) => void
@@ -190,7 +195,33 @@ describe('DegradedDaemonPtyProvider', () => {
warn.mockRestore()
})
it('fails deadline teardown closed for duplicate ownership across degraded daemons', async () => {
it('does not restore a daemon route from an inventory response older than exit', async () => {
const current = createDaemonAdapter('daemon', ['exited-session'])
const fallback = createProvider('fallback')
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
await provider.discoverDaemonSessions()
let resolveDaemon!: (sessions: PtyProcessInfo[]) => void
vi.mocked(current.listProcesses).mockReturnValueOnce(
new Promise((resolve) => {
resolveDaemon = resolve
})
)
const listing = provider.listProcesses()
current.emitExit('exited-session', 0)
resolveDaemon([{ id: 'exited-session', cwd: '', title: 'stale daemon' }])
await listing
await provider.spawn({ sessionId: 'exited-session', cols: 80, rows: 24 })
expect(fallback.spawn).toHaveBeenCalledWith({
sessionId: 'exited-session',
cols: 80,
rows: 24
})
expect(current.spawn).not.toHaveBeenCalled()
})
it('fails shutdown closed for duplicate ownership across degraded daemons', async () => {
const current = createDaemonAdapter('current', ['duplicate-session'])
const legacy = createDaemonAdapter('legacy', ['duplicate-session'])
const fallback = createProvider('fallback')
@@ -200,8 +231,7 @@ describe('DegradedDaemonPtyProvider', () => {
await expect(
provider.shutdown('duplicate-session', {
immediate: true,
deadlineMs: Date.now() + 1000
immediate: true
})
).rejects.toThrow(
'Ambiguous PTY session ownership across degraded providers: duplicate-session'
@@ -229,8 +259,7 @@ describe('DegradedDaemonPtyProvider', () => {
)
await expect(
provider.shutdown('duplicate-session', {
immediate: true,
deadlineMs: Date.now() + 1000
immediate: true
})
).rejects.toThrow(
'Ambiguous PTY session ownership across degraded providers: duplicate-session'
+16 -10
View File
@@ -58,13 +58,16 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
async discoverDaemonSessions(): Promise<void> {
for (const adapter of this.allDaemonAdapters()) {
const refresh = this.sessionRouting.beginInventoryRefresh()
try {
const sessions = await adapter.listProcesses()
for (const session of sessions) {
this.sessionRouting.add(session.id, adapter)
this.sessionRouting.addDaemonInventorySession(session.id, adapter, refresh)
}
} catch (error) {
console.warn('[daemon] Failed to discover degraded daemon sessions', error)
} finally {
this.sessionRouting.finishInventoryRefresh(refresh)
}
}
}
@@ -109,7 +112,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
id: string,
opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number }
): Promise<void> {
if (opts.deadlineMs !== undefined && this.sessionRouting.isAmbiguous(id)) {
if (this.sessionRouting.isAmbiguous(id)) {
throw this.ambiguousOwnershipError([id])
}
const target = this.providerFor(id)
@@ -178,8 +181,15 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
async listProcesses(opts?: { deadlineMs?: number }): Promise<PtyProcessInfo[]> {
const providers = this.allProviders()
const results = await Promise.all(providers.map((provider) => provider.listProcesses(opts)))
const ambiguousIds = this.sessionRouting.refreshSessions(results[0], results.slice(1))
const refresh = this.sessionRouting.beginInventoryRefresh()
let results: PtyProcessInfo[][]
try {
results = await Promise.all(providers.map((provider) => provider.listProcesses(opts)))
} catch (error) {
this.sessionRouting.finishInventoryRefresh(refresh)
throw error
}
const ambiguousIds = this.sessionRouting.refreshSessions(results[0], results.slice(1), refresh)
if (ambiguousIds.length > 0) {
throw this.ambiguousOwnershipError(ambiguousIds)
}
@@ -246,11 +256,11 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
}
ackColdRestore(sessionId: string): void {
this.daemonAdapterFor(sessionId)?.ackColdRestore(sessionId)
this.sessionRouting.daemonFor(sessionId)?.ackColdRestore(sessionId)
}
clearTombstone(sessionId: string): void {
this.daemonAdapterFor(sessionId)?.clearTombstone(sessionId)
this.sessionRouting.daemonFor(sessionId)?.clearTombstone(sessionId)
}
async reconcileOnStartup(validWorktreeIds: Set<string>): Promise<{
@@ -327,10 +337,6 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
return this.sessionRouting.providerFor(sessionId, this.current)
}
private daemonAdapterFor(sessionId: string): DaemonPtyAdapter | null {
return this.sessionRouting.daemonFor(sessionId)
}
private ambiguousOwnershipError(sessionIds: string[]): Error {
return new Error(
`Ambiguous PTY session ownership across degraded providers: ${sessionIds.join(', ')}`
@@ -1,5 +1,8 @@
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
import { DaemonPtySessionRouting } from './daemon-pty-session-routing'
import {
DaemonPtySessionRouting,
type DaemonPtyInventoryRefresh
} from './daemon-pty-session-routing'
import type { IPtyProvider, PtyProcessInfo } from '../providers/types'
export type DegradedManagedPtyProvider = IPtyProvider & {
@@ -30,11 +33,28 @@ export class DegradedDaemonSessionRouting {
)
}
beginInventoryRefresh(): DaemonPtyInventoryRefresh {
return this.daemonRouting.beginInventoryRefresh()
}
finishInventoryRefresh(refresh: DaemonPtyInventoryRefresh): void {
this.daemonRouting.finishInventoryRefresh(refresh)
}
addDaemonInventorySession(
sessionId: string,
adapter: DaemonPtyAdapter,
refresh: DaemonPtyInventoryRefresh
): void {
this.daemonRouting.addInventorySession(sessionId, adapter, 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)
}
@@ -43,16 +63,25 @@ export class DegradedDaemonSessionRouting {
this.daemonRouting.remove(sessionId, provider)
return
}
this.daemonRouting.recordExternalMutation(sessionId)
if (this.providers.get(sessionId) === provider) {
this.providers.delete(sessionId)
}
}
refreshSessions(fallbackSessions: PtyProcessInfo[], daemonResults: PtyProcessInfo[][]): string[] {
refreshSessions(
fallbackSessions: PtyProcessInfo[],
daemonResults: PtyProcessInfo[][],
refresh: DaemonPtyInventoryRefresh
): string[] {
for (const session of fallbackSessions) {
this.providers.set(session.id, this.fallback)
this.daemonRouting.applyInventoryMutation(session.id, refresh, () => {
this.providers.set(session.id, this.fallback)
})
}
const ambiguousIds = new Set(this.daemonRouting.refreshLive(this.daemonAdapters, daemonResults))
const ambiguousIds = new Set(
this.daemonRouting.refreshLive(this.daemonAdapters, daemonResults, refresh)
)
for (const sessionId of this.providers.keys()) {
if (
this.daemonRouting.get(sessionId) !== undefined ||