mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 00:03:45 +00:00
fix(daemon): validate recovery routing evidence
This commit is contained in:
@@ -568,10 +568,10 @@ async function matchDaemonProcess(
|
||||
return probeProcessExistence(pid) === 'absent' ? 'stale' : 'unknown'
|
||||
}
|
||||
}
|
||||
return commandLineMatchesDaemon(commandLine, socketPath, tokenPath) &&
|
||||
startTimeMatches(pid, startedAtMs)
|
||||
? 'match'
|
||||
: 'stale'
|
||||
if (!commandLineMatchesDaemon(commandLine, socketPath, tokenPath)) {
|
||||
return 'stale'
|
||||
}
|
||||
return startTimeMatches(pid, startedAtMs) ? 'match' : 'unknown'
|
||||
}
|
||||
|
||||
async function isDaemonProcess(
|
||||
|
||||
@@ -11,6 +11,10 @@ export class DaemonPtyAdapterSubscriptionFanout {
|
||||
constructor(
|
||||
private readonly adapters: readonly DaemonPtyAdapter[],
|
||||
private readonly shouldForwardStreamEvent: (adapter: DaemonPtyAdapter, id: string) => boolean,
|
||||
private readonly shouldForwardWriteUnavailable: (
|
||||
adapter: DaemonPtyAdapter,
|
||||
id: string
|
||||
) => boolean,
|
||||
onAdapterExit: (adapter: DaemonPtyAdapter, id: string) => boolean
|
||||
) {
|
||||
for (const adapter of adapters) {
|
||||
@@ -60,7 +64,15 @@ export class DaemonPtyAdapterSubscriptionFanout {
|
||||
// Why: main subscribes on the routed provider, so without this the dead-endpoint
|
||||
// fan-out never reaches the renderer and only the written pane recovers (STA-2373).
|
||||
onWriteUnavailable(callback: (payload: { id: string }) => void): () => void {
|
||||
return combineUnsubscribes(this.adapters.map((adapter) => adapter.onWriteUnavailable(callback)))
|
||||
return combineUnsubscribes(
|
||||
this.adapters.map((adapter) =>
|
||||
adapter.onWriteUnavailable((payload) => {
|
||||
if (this.shouldForwardWriteUnavailable(adapter, payload.id)) {
|
||||
callback(payload)
|
||||
}
|
||||
})
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
onReplay(_callback: (payload: { id: string; data: string }) => void): () => void {
|
||||
|
||||
@@ -8,6 +8,7 @@ export type AdapterHarness = {
|
||||
emitData: (sessionId: string, data: string) => void
|
||||
emitIdentityChange: (previous: DaemonEndpointIdentity, current: DaemonEndpointIdentity) => void
|
||||
emitExit: (sessionId: string, code: number) => void
|
||||
emitWriteUnavailable: (sessionId: string) => void
|
||||
setIdentity: (identity: DaemonEndpointIdentity) => void
|
||||
}
|
||||
|
||||
@@ -31,6 +32,7 @@ export function createAdapter(
|
||||
}) => void)[] = []
|
||||
const dataListeners: ((payload: { id: string; data: string }) => void)[] = []
|
||||
const exitListeners: ((payload: { id: string; code: number }) => void)[] = []
|
||||
const writeUnavailableListeners: ((payload: { id: string }) => void)[] = []
|
||||
const adapter = {
|
||||
protocolVersion: label === 'current' ? 30 : 29,
|
||||
getLastAuthenticatedDaemonIdentity: vi.fn(() => ({ ...daemonIdentity })),
|
||||
@@ -74,7 +76,15 @@ export function createAdapter(
|
||||
return () => {}
|
||||
}),
|
||||
onBackgroundStreamEvent: vi.fn(() => () => {}),
|
||||
onWriteUnavailable: vi.fn(() => () => {}),
|
||||
onWriteUnavailable: vi.fn((listener: (payload: { id: string }) => void) => {
|
||||
writeUnavailableListeners.push(listener)
|
||||
return () => {
|
||||
const index = writeUnavailableListeners.indexOf(listener)
|
||||
if (index !== -1) {
|
||||
writeUnavailableListeners.splice(index, 1)
|
||||
}
|
||||
}
|
||||
}),
|
||||
dispose: vi.fn(),
|
||||
disconnectOnly: vi.fn(async () => {})
|
||||
} as unknown as DaemonPtyAdapter
|
||||
@@ -98,6 +108,11 @@ export function createAdapter(
|
||||
for (const listener of exitListeners) {
|
||||
listener({ id: sessionId, code })
|
||||
}
|
||||
},
|
||||
emitWriteUnavailable: (sessionId) => {
|
||||
for (const listener of writeUnavailableListeners) {
|
||||
listener({ id: sessionId })
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ import type { DaemonPtyAdapter } from './daemon-pty-adapter'
|
||||
import type { DaemonEndpointIdentity } from './daemon-hello-protocol'
|
||||
import { DaemonSessionRouteTable } from './daemon-session-route-table'
|
||||
import { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowledgement-routes'
|
||||
import { recordValidatedDaemonSpawn } from './daemon-spawn-route-validation'
|
||||
|
||||
export class DaemonPtyRouterSessionRouting {
|
||||
private readonly routes: DaemonSessionRouteTable
|
||||
@@ -64,11 +65,15 @@ export class DaemonPtyRouterSessionRouting {
|
||||
if (result.exitedBeforeSpawnReply) {
|
||||
return
|
||||
}
|
||||
if (authoritativeIntent && opts.sessionId === result.id) {
|
||||
this.routes.recordFreshOwned(result.id, target)
|
||||
if (authoritativeIntent) {
|
||||
if (opts.sessionId === result.id) {
|
||||
this.routes.recordFreshOwned(result.id, target)
|
||||
} else {
|
||||
this.routes.recordOwned(result.id, target)
|
||||
}
|
||||
return
|
||||
}
|
||||
this.routes.recordOwned(result.id, target)
|
||||
recordValidatedDaemonSpawn(this.routes, result.id, opts.sessionId, target)
|
||||
}
|
||||
|
||||
beginSpawn(opts: PtySpawnOptions, target: DaemonPtyAdapter): boolean {
|
||||
@@ -212,6 +217,13 @@ export class DaemonPtyRouterSessionRouting {
|
||||
return this.routes.recordDataOwner(sessionId, owner)
|
||||
}
|
||||
|
||||
shouldForwardWriteUnavailable(sessionId: string, owner: DaemonPtyAdapter): boolean {
|
||||
const authoritativeSpawn = this.authoritativeSpawnsInFlight.get(sessionId)
|
||||
return authoritativeSpawn
|
||||
? authoritativeSpawn.owner === owner
|
||||
: this.routes.getOwned(sessionId) === owner
|
||||
}
|
||||
|
||||
recordExit(sessionId: string, owner: DaemonPtyAdapter): boolean {
|
||||
const authoritativeSpawn = this.authoritativeSpawnsInFlight.get(sessionId)
|
||||
if (authoritativeSpawn && authoritativeSpawn.owner !== owner) {
|
||||
|
||||
@@ -184,28 +184,6 @@ function createAdapter(
|
||||
} as unknown as AdapterMock
|
||||
}
|
||||
|
||||
it('forwards dead-endpoint write-unavailable signals from every routed adapter', () => {
|
||||
// Why revert-sensitive: main subscribes on the ROUTED provider, so if the router
|
||||
// does not forward this the STA-2373 fan-out never reaches the renderer and only
|
||||
// the written pane recovers — siblings stay frozen. The router is the live
|
||||
// localProvider whenever a legacy daemon socket exists (protocol bump mid-session).
|
||||
const current = createAdapter('current')
|
||||
const legacy = createAdapter('legacy')
|
||||
const router = new DaemonPtyRouter({ current, legacy: [legacy] })
|
||||
const recovered: string[] = []
|
||||
|
||||
const unsubscribe = router.onWriteUnavailable(({ id }) => recovered.push(id))
|
||||
current.triggerWriteUnavailable('current-pane')
|
||||
legacy.triggerWriteUnavailable('legacy-pane')
|
||||
|
||||
expect(recovered).toEqual(['current-pane', 'legacy-pane'])
|
||||
|
||||
unsubscribe()
|
||||
current.triggerWriteUnavailable('after-unsubscribe')
|
||||
legacy.triggerWriteUnavailable('after-unsubscribe')
|
||||
expect(recovered).toEqual(['current-pane', 'legacy-pane'])
|
||||
})
|
||||
|
||||
it('rejects completion inspection when no daemon owns the session', async () => {
|
||||
const router = new DaemonPtyRouter({
|
||||
current: createAdapter('current'),
|
||||
|
||||
@@ -27,6 +27,7 @@ export class DaemonPtyRouter implements IPtyProvider {
|
||||
this.subscriptions = new DaemonPtyAdapterSubscriptionFanout(
|
||||
this.allAdapters(),
|
||||
(adapter, id) => this.routing.shouldForwardStreamEvent(id, adapter),
|
||||
(adapter, id) => this.routing.shouldForwardWriteUnavailable(id, adapter),
|
||||
(adapter, id) => this.routing.recordExit(id, adapter)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
import { expect, it } from 'vitest'
|
||||
import { DaemonPtyRouter } from './daemon-pty-router'
|
||||
import { createAdapter, identity } from './daemon-pty-router-routing-safety-fixture'
|
||||
|
||||
it('forwards write-unavailable signals from each exact routed owner', () => {
|
||||
const current = createAdapter('current')
|
||||
const legacy = createAdapter('legacy')
|
||||
const router = new DaemonPtyRouter({
|
||||
current: current.adapter,
|
||||
legacy: [legacy.adapter]
|
||||
})
|
||||
const recovered: string[] = []
|
||||
current.emitData('current-pane', 'route evidence')
|
||||
legacy.emitData('legacy-pane', 'route evidence')
|
||||
const unsubscribe = router.onWriteUnavailable(({ id }) => recovered.push(id))
|
||||
|
||||
current.emitWriteUnavailable('current-pane')
|
||||
legacy.emitWriteUnavailable('legacy-pane')
|
||||
|
||||
expect(recovered).toEqual(['current-pane', 'legacy-pane'])
|
||||
unsubscribe()
|
||||
current.emitWriteUnavailable('current-pane')
|
||||
expect(recovered).toEqual(['current-pane', 'legacy-pane'])
|
||||
})
|
||||
|
||||
it('withholds write-unavailable signals from foreign and stale adapter routes', () => {
|
||||
const current = createAdapter('current')
|
||||
const legacy = createAdapter('legacy')
|
||||
const router = new DaemonPtyRouter({
|
||||
current: current.adapter,
|
||||
legacy: [legacy.adapter]
|
||||
})
|
||||
const recovered: string[] = []
|
||||
const previous = identity('current', 20)
|
||||
const replacement = identity('current', 21)
|
||||
current.emitData('shared-pane', 'route evidence')
|
||||
router.onWriteUnavailable(({ id }) => recovered.push(id))
|
||||
|
||||
legacy.emitWriteUnavailable('shared-pane')
|
||||
current.emitWriteUnavailable('shared-pane')
|
||||
current.emitData('ambiguous-pane', 'current collision evidence')
|
||||
legacy.emitData('ambiguous-pane', 'legacy collision evidence')
|
||||
current.emitWriteUnavailable('ambiguous-pane')
|
||||
legacy.emitWriteUnavailable('ambiguous-pane')
|
||||
current.setIdentity(replacement)
|
||||
current.emitIdentityChange(previous, replacement)
|
||||
current.emitWriteUnavailable('shared-pane')
|
||||
|
||||
expect(recovered).toEqual(['shared-pane'])
|
||||
})
|
||||
@@ -0,0 +1,77 @@
|
||||
import { expect, it, vi } from 'vitest'
|
||||
import type { PtySpawnResult } from '../providers/types'
|
||||
import type { DaemonIdentityChangeEvent } from './daemon-pty-adapter'
|
||||
import { DaemonPtyRouter } from './daemon-pty-router'
|
||||
import { createAdapter, identity } from './daemon-pty-router-routing-safety-fixture'
|
||||
import { DegradedDaemonPtyProvider } from './degraded-daemon-pty-provider'
|
||||
import { createDaemonAdapter, createProvider } from './degraded-daemon-pty-provider-fixture'
|
||||
|
||||
it('rejects an existing-session spawn after its selected daemon is replaced', async () => {
|
||||
const sessionId = 'replaced-during-spawn'
|
||||
const legacy = createAdapter('legacy', [sessionId])
|
||||
const previous = identity('legacy', 10)
|
||||
const replacement = identity('legacy', 11)
|
||||
let resolveSpawn: ((result: { id: string }) => void) | undefined
|
||||
let markSpawnStarted: (() => void) | undefined
|
||||
const spawnStarted = new Promise<void>((resolve) => {
|
||||
markSpawnStarted = resolve
|
||||
})
|
||||
vi.mocked(legacy.adapter.spawn).mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
markSpawnStarted?.()
|
||||
resolveSpawn = resolve
|
||||
})
|
||||
)
|
||||
const router = new DaemonPtyRouter({
|
||||
current: createAdapter('current').adapter,
|
||||
legacy: [legacy.adapter]
|
||||
})
|
||||
await router.discoverLegacySessions()
|
||||
|
||||
const spawning = router.spawn({ sessionId, cols: 80, rows: 24 })
|
||||
await spawnStarted
|
||||
legacy.setIdentity(replacement)
|
||||
legacy.emitIdentityChange(previous, replacement)
|
||||
resolveSpawn?.({ id: sessionId })
|
||||
|
||||
await expect(spawning).rejects.toThrow('daemon_session_routing_unavailable')
|
||||
expect(router.getSessionRouteState(sessionId)).toBe('unavailable')
|
||||
})
|
||||
|
||||
it('rejects a degraded existing spawn after its selected daemon is replaced', async () => {
|
||||
const sessionId = 'replaced-daemon-spawn'
|
||||
const previous = { pid: 1, startedAtMs: 1, launchNonce: 'previous' }
|
||||
const replacement = { pid: 2, startedAtMs: 2, launchNonce: 'replacement' }
|
||||
const current = createDaemonAdapter('daemon', [sessionId])
|
||||
const fallback = createProvider('fallback')
|
||||
let daemonIdentity = previous
|
||||
let notifyIdentityChange!: (event: DaemonIdentityChangeEvent) => void
|
||||
vi.mocked(current.getLastAuthenticatedDaemonIdentity).mockImplementation(() => daemonIdentity)
|
||||
vi.mocked(current.onDaemonIdentityChanged).mockImplementation((listener) => {
|
||||
notifyIdentityChange = listener
|
||||
return () => {}
|
||||
})
|
||||
let resolveSpawn!: (result: PtySpawnResult) => void
|
||||
let markSpawnStarted!: () => void
|
||||
const spawnStarted = new Promise<void>((resolve) => {
|
||||
markSpawnStarted = resolve
|
||||
})
|
||||
vi.mocked(current.spawn).mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
markSpawnStarted()
|
||||
resolveSpawn = resolve
|
||||
})
|
||||
)
|
||||
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
|
||||
await provider.discoverDaemonSessions()
|
||||
|
||||
const spawning = provider.spawn({ sessionId, cols: 80, rows: 24 })
|
||||
await spawnStarted
|
||||
daemonIdentity = replacement
|
||||
notifyIdentityChange({ previous, current: replacement })
|
||||
resolveSpawn({ id: sessionId })
|
||||
|
||||
await expect(spawning).rejects.toThrow('daemon_session_routing_unavailable')
|
||||
})
|
||||
@@ -0,0 +1,18 @@
|
||||
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
|
||||
import { DaemonSessionUnavailableError } from './daemon-session-route'
|
||||
import type { DaemonSessionRouteTable } from './daemon-session-route-table'
|
||||
|
||||
export function recordValidatedDaemonSpawn(
|
||||
routes: DaemonSessionRouteTable,
|
||||
resultSessionId: string,
|
||||
requestedSessionId: string | undefined,
|
||||
owner: DaemonPtyAdapter
|
||||
): void {
|
||||
routes.recordOwned(resultSessionId, owner)
|
||||
if (
|
||||
routes.getOwned(resultSessionId) !== owner ||
|
||||
(requestedSessionId !== undefined && routes.getOwned(requestedSessionId) !== owner)
|
||||
) {
|
||||
throw new DaemonSessionUnavailableError(requestedSessionId ?? resultSessionId)
|
||||
}
|
||||
}
|
||||
@@ -109,6 +109,44 @@ describe('stale daemon pidfile cleanup', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('preserves matching POSIX command evidence when the start time is inconclusive', async () => {
|
||||
if (process.platform === 'win32') {
|
||||
return
|
||||
}
|
||||
const child = spawn(
|
||||
process.execPath,
|
||||
[
|
||||
'-e',
|
||||
"process.send?.('ready');setInterval(()=>{},1000)",
|
||||
'daemon-entry',
|
||||
socketPath,
|
||||
tokenPath
|
||||
],
|
||||
{ stdio: ['ignore', 'ignore', 'ignore', 'ipc'] }
|
||||
)
|
||||
await new Promise<void>((resolve) => {
|
||||
child.on('message', (message) => {
|
||||
if (message === 'ready') {
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
})
|
||||
const record = serializeDaemonPidFile({
|
||||
pid: child.pid!,
|
||||
startedAtMs: Date.now() - 5_000,
|
||||
launchNonce: 'slow-bootstrap'
|
||||
})
|
||||
writeFileSync(pidPath, record, { mode: 0o600 })
|
||||
|
||||
try {
|
||||
await expect(killStaleDaemon(dir, socketPath, tokenPath)).resolves.toBe(false)
|
||||
expect(readFileSync(pidPath, 'utf8')).toBe(record)
|
||||
expect(child.exitCode).toBeNull()
|
||||
} finally {
|
||||
child.kill('SIGKILL')
|
||||
}
|
||||
})
|
||||
|
||||
it('preserves the exact record when process access is rejected', async () => {
|
||||
const record = serializeDaemonPidFile({
|
||||
pid: process.pid,
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
import { vi } from 'vitest'
|
||||
import type { PtyProcessInspection } from '../providers/pty-process-inspection'
|
||||
import type { IPtyProvider, PtySpawnOptions, PtySpawnResult } from '../providers/types'
|
||||
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
|
||||
|
||||
export type ProviderMock = IPtyProvider & {
|
||||
probePtyLiveness: (id: string) => Promise<boolean | null>
|
||||
inspectProcess: (id: string) => Promise<PtyProcessInspection>
|
||||
emitData: (id: string, data: string, sequenceChars?: number) => void
|
||||
emitReplay: (id: string, data: string) => void
|
||||
emitExit: (id: string, code: number) => void
|
||||
triggerWriteUnavailable: (id: string) => void
|
||||
onWriteUnavailable: (callback: (payload: { id: string }) => void) => () => void
|
||||
}
|
||||
|
||||
export function createProvider(
|
||||
label: string,
|
||||
sessions: string[] = [],
|
||||
authoritativeOwnerListings = false
|
||||
): ProviderMock {
|
||||
const dataListeners: ((payload: { id: string; data: string; sequenceChars?: number }) => void)[] =
|
||||
[]
|
||||
const replayListeners: ((payload: { id: string; data: string }) => void)[] = []
|
||||
const exitListeners: ((payload: { id: string; code: number }) => void)[] = []
|
||||
const writeUnavailableListeners: ((payload: { id: string }) => void)[] = []
|
||||
return {
|
||||
spawn: vi.fn(async (opts: PtySpawnOptions): Promise<PtySpawnResult> => {
|
||||
const id = opts.sessionId ?? `${label}-new`
|
||||
sessions.push(id)
|
||||
return { id }
|
||||
}),
|
||||
attach: vi.fn(async () => {}),
|
||||
hasPty: vi.fn((id: string) => sessions.includes(id)),
|
||||
probePtyLiveness: vi.fn(async (id: string) => sessions.includes(id)),
|
||||
providesAgentSessionOwnerListings: vi.fn(() => authoritativeOwnerListings),
|
||||
write: vi.fn(),
|
||||
resize: vi.fn(),
|
||||
pauseProducer: vi.fn(),
|
||||
resumeProducer: vi.fn(),
|
||||
setPtyBackgrounded: vi.fn(),
|
||||
shutdown: vi.fn(async (id: string) => {
|
||||
const idx = sessions.indexOf(id)
|
||||
if (idx !== -1) {
|
||||
sessions.splice(idx, 1)
|
||||
}
|
||||
}),
|
||||
sendSignal: vi.fn(async () => {}),
|
||||
getCwd: vi.fn(async () => ''),
|
||||
getInitialCwd: vi.fn(async () => ''),
|
||||
clearBuffer: vi.fn(async () => {}),
|
||||
acknowledgeDataEvent: vi.fn(),
|
||||
hasChildProcesses: vi.fn(async () => false),
|
||||
getForegroundProcess: vi.fn(async () => null),
|
||||
inspectProcess: vi.fn(async () => ({ foregroundProcess: null, hasChildProcesses: false })),
|
||||
confirmForegroundProcess: vi.fn(async () => `${label}-confirmed`),
|
||||
serialize: vi.fn(async () => '{}'),
|
||||
revive: vi.fn(async () => {}),
|
||||
listProcesses: vi.fn(async () => sessions.map((id) => ({ id, cwd: '', title: label }))),
|
||||
getDefaultShell: vi.fn(async () => '/bin/zsh'),
|
||||
getProfiles: vi.fn(async () => []),
|
||||
onData: vi.fn(
|
||||
(callback: (payload: { id: string; data: string; sequenceChars?: number }) => void) => {
|
||||
dataListeners.push(callback)
|
||||
return () => {
|
||||
const idx = dataListeners.indexOf(callback)
|
||||
if (idx !== -1) {
|
||||
dataListeners.splice(idx, 1)
|
||||
}
|
||||
}
|
||||
}
|
||||
),
|
||||
onReplay: vi.fn((callback: (payload: { id: string; data: string }) => void) => {
|
||||
replayListeners.push(callback)
|
||||
return () => {
|
||||
const idx = replayListeners.indexOf(callback)
|
||||
if (idx !== -1) {
|
||||
replayListeners.splice(idx, 1)
|
||||
}
|
||||
}
|
||||
}),
|
||||
onExit: vi.fn((callback: (payload: { id: string; code: number }) => void) => {
|
||||
exitListeners.push(callback)
|
||||
return () => {
|
||||
const idx = exitListeners.indexOf(callback)
|
||||
if (idx !== -1) {
|
||||
exitListeners.splice(idx, 1)
|
||||
}
|
||||
}
|
||||
}),
|
||||
emitData: (id: string, data: string, sequenceChars?: number) => {
|
||||
for (const listener of dataListeners) {
|
||||
listener({ id, data, ...(sequenceChars === undefined ? {} : { sequenceChars }) })
|
||||
}
|
||||
},
|
||||
emitReplay: (id: string, data: string) => {
|
||||
for (const listener of replayListeners) {
|
||||
listener({ id, data })
|
||||
}
|
||||
},
|
||||
emitExit: (id: string, code: number) => {
|
||||
for (const listener of exitListeners) {
|
||||
listener({ id, code })
|
||||
}
|
||||
},
|
||||
onWriteUnavailable: vi.fn((callback: (payload: { id: string }) => void) => {
|
||||
writeUnavailableListeners.push(callback)
|
||||
return () => {
|
||||
const idx = writeUnavailableListeners.indexOf(callback)
|
||||
if (idx !== -1) {
|
||||
writeUnavailableListeners.splice(idx, 1)
|
||||
}
|
||||
}
|
||||
}),
|
||||
triggerWriteUnavailable: (id: string) => {
|
||||
for (const listener of writeUnavailableListeners) {
|
||||
listener({ id })
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function createDaemonAdapter(
|
||||
label: string,
|
||||
sessions: string[] = []
|
||||
): DaemonPtyAdapter & ProviderMock {
|
||||
return {
|
||||
...createProvider(label, sessions, true),
|
||||
protocolVersion: 13,
|
||||
getLastAuthenticatedDaemonIdentity: vi.fn(() => null),
|
||||
onDaemonIdentityChanged: vi.fn(() => () => {}),
|
||||
listSessions: vi.fn(async () => []),
|
||||
ackColdRestore: vi.fn(),
|
||||
clearTombstone: vi.fn(),
|
||||
reconcileOnStartup: vi.fn(async () => ({ alive: sessions, killed: [] })),
|
||||
dispose: vi.fn(),
|
||||
disconnectOnly: vi.fn(async () => {}),
|
||||
getActiveSessionIds: vi.fn(() => []),
|
||||
fanoutSyntheticExits: vi.fn()
|
||||
} as unknown as DaemonPtyAdapter & ProviderMock
|
||||
}
|
||||
@@ -1,154 +1,19 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { DegradedDaemonPtyProvider } from './degraded-daemon-pty-provider'
|
||||
import type { DaemonIdentityChangeEvent, DaemonPtyAdapter } from './daemon-pty-adapter'
|
||||
import type { IPtyProvider, PtySpawnOptions, PtySpawnResult } from '../providers/types'
|
||||
import type { PtyProcessInspection } from '../providers/pty-process-inspection'
|
||||
import type { DaemonIdentityChangeEvent } from './daemon-pty-adapter'
|
||||
import type { PtySpawnResult } from '../providers/types'
|
||||
import { createDaemonAdapter, createProvider } from './degraded-daemon-pty-provider-fixture'
|
||||
|
||||
type ProviderMock = IPtyProvider & {
|
||||
probePtyLiveness: (id: string) => Promise<boolean | null>
|
||||
inspectProcess: (id: string) => Promise<PtyProcessInspection>
|
||||
emitData: (id: string, data: string, sequenceChars?: number) => void
|
||||
emitReplay: (id: string, data: string) => void
|
||||
emitExit: (id: string, code: number) => void
|
||||
triggerWriteUnavailable: (id: string) => void
|
||||
onWriteUnavailable: (callback: (payload: { id: string }) => void) => () => void
|
||||
}
|
||||
|
||||
function createProvider(
|
||||
label: string,
|
||||
sessions: string[] = [],
|
||||
authoritativeOwnerListings = false
|
||||
): ProviderMock {
|
||||
const dataListeners: ((payload: { id: string; data: string; sequenceChars?: number }) => void)[] =
|
||||
[]
|
||||
const replayListeners: ((payload: { id: string; data: string }) => void)[] = []
|
||||
const exitListeners: ((payload: { id: string; code: number }) => void)[] = []
|
||||
const writeUnavailableListeners: ((payload: { id: string }) => void)[] = []
|
||||
return {
|
||||
spawn: vi.fn(async (opts: PtySpawnOptions): Promise<PtySpawnResult> => {
|
||||
const id = opts.sessionId ?? `${label}-new`
|
||||
sessions.push(id)
|
||||
return { id }
|
||||
}),
|
||||
attach: vi.fn(async () => {}),
|
||||
hasPty: vi.fn((id: string) => sessions.includes(id)),
|
||||
probePtyLiveness: vi.fn(async (id: string) => sessions.includes(id)),
|
||||
providesAgentSessionOwnerListings: vi.fn(() => authoritativeOwnerListings),
|
||||
write: vi.fn(),
|
||||
resize: vi.fn(),
|
||||
pauseProducer: vi.fn(),
|
||||
resumeProducer: vi.fn(),
|
||||
setPtyBackgrounded: vi.fn(),
|
||||
shutdown: vi.fn(async (id: string) => {
|
||||
const idx = sessions.indexOf(id)
|
||||
if (idx !== -1) {
|
||||
sessions.splice(idx, 1)
|
||||
}
|
||||
}),
|
||||
sendSignal: vi.fn(async () => {}),
|
||||
getCwd: vi.fn(async () => ''),
|
||||
getInitialCwd: vi.fn(async () => ''),
|
||||
clearBuffer: vi.fn(async () => {}),
|
||||
acknowledgeDataEvent: vi.fn(),
|
||||
hasChildProcesses: vi.fn(async () => false),
|
||||
getForegroundProcess: vi.fn(async () => null),
|
||||
inspectProcess: vi.fn(async () => ({ foregroundProcess: null, hasChildProcesses: false })),
|
||||
confirmForegroundProcess: vi.fn(async () => `${label}-confirmed`),
|
||||
serialize: vi.fn(async () => '{}'),
|
||||
revive: vi.fn(async () => {}),
|
||||
listProcesses: vi.fn(async () => sessions.map((id) => ({ id, cwd: '', title: label }))),
|
||||
getDefaultShell: vi.fn(async () => '/bin/zsh'),
|
||||
getProfiles: vi.fn(async () => []),
|
||||
onData: vi.fn(
|
||||
(callback: (payload: { id: string; data: string; sequenceChars?: number }) => void) => {
|
||||
dataListeners.push(callback)
|
||||
return () => {
|
||||
const idx = dataListeners.indexOf(callback)
|
||||
if (idx !== -1) {
|
||||
dataListeners.splice(idx, 1)
|
||||
}
|
||||
}
|
||||
}
|
||||
),
|
||||
onReplay: vi.fn((callback: (payload: { id: string; data: string }) => void) => {
|
||||
replayListeners.push(callback)
|
||||
return () => {
|
||||
const idx = replayListeners.indexOf(callback)
|
||||
if (idx !== -1) {
|
||||
replayListeners.splice(idx, 1)
|
||||
}
|
||||
}
|
||||
}),
|
||||
onExit: vi.fn((callback: (payload: { id: string; code: number }) => void) => {
|
||||
exitListeners.push(callback)
|
||||
return () => {
|
||||
const idx = exitListeners.indexOf(callback)
|
||||
if (idx !== -1) {
|
||||
exitListeners.splice(idx, 1)
|
||||
}
|
||||
}
|
||||
}),
|
||||
emitData: (id: string, data: string, sequenceChars?: number) => {
|
||||
for (const listener of dataListeners) {
|
||||
listener({ id, data, ...(sequenceChars === undefined ? {} : { sequenceChars }) })
|
||||
}
|
||||
},
|
||||
emitReplay: (id: string, data: string) => {
|
||||
for (const listener of replayListeners) {
|
||||
listener({ id, data })
|
||||
}
|
||||
},
|
||||
emitExit: (id: string, code: number) => {
|
||||
for (const listener of exitListeners) {
|
||||
listener({ id, code })
|
||||
}
|
||||
},
|
||||
onWriteUnavailable: vi.fn((callback: (payload: { id: string }) => void) => {
|
||||
writeUnavailableListeners.push(callback)
|
||||
return () => {
|
||||
const idx = writeUnavailableListeners.indexOf(callback)
|
||||
if (idx !== -1) {
|
||||
writeUnavailableListeners.splice(idx, 1)
|
||||
}
|
||||
}
|
||||
}),
|
||||
triggerWriteUnavailable: (id: string) => {
|
||||
for (const listener of writeUnavailableListeners) {
|
||||
listener({ id })
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function createDaemonAdapter(
|
||||
label: string,
|
||||
sessions: string[] = []
|
||||
): DaemonPtyAdapter & ProviderMock {
|
||||
return {
|
||||
...createProvider(label, sessions, true),
|
||||
protocolVersion: 13,
|
||||
getLastAuthenticatedDaemonIdentity: vi.fn(() => null),
|
||||
onDaemonIdentityChanged: vi.fn(() => () => {}),
|
||||
listSessions: vi.fn(async () => []),
|
||||
ackColdRestore: vi.fn(),
|
||||
clearTombstone: vi.fn(),
|
||||
reconcileOnStartup: vi.fn(async () => ({ alive: sessions, killed: [] })),
|
||||
dispose: vi.fn(),
|
||||
disconnectOnly: vi.fn(async () => {}),
|
||||
getActiveSessionIds: vi.fn(() => []),
|
||||
fanoutSyntheticExits: vi.fn()
|
||||
} as unknown as DaemonPtyAdapter & ProviderMock
|
||||
}
|
||||
|
||||
it('forwards dead-endpoint write-unavailable signals from the daemon adapters', () => {
|
||||
it('forwards dead-endpoint write-unavailable signals from exact daemon owners', async () => {
|
||||
// Why revert-sensitive: this provider is the live localProvider in degraded launch
|
||||
// mode and main subscribes on it, so without forwarding the STA-2373 fan-out reaches
|
||||
// no listener and sibling panes stay frozen.
|
||||
const current = createDaemonAdapter('daemon')
|
||||
const legacy = createDaemonAdapter('legacy')
|
||||
const current = createDaemonAdapter('daemon', ['daemon-pane'])
|
||||
const legacy = createDaemonAdapter('legacy', ['legacy-pane'])
|
||||
const fallback = createProvider('fallback')
|
||||
const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback })
|
||||
const recovered: string[] = []
|
||||
await provider.discoverDaemonSessions()
|
||||
|
||||
const unsubscribe = provider.onWriteUnavailable(({ id }) => recovered.push(id))
|
||||
current.triggerWriteUnavailable('daemon-pane')
|
||||
@@ -160,6 +25,36 @@ it('forwards dead-endpoint write-unavailable signals from the daemon adapters',
|
||||
expect(recovered).toEqual(['daemon-pane', 'legacy-pane'])
|
||||
})
|
||||
|
||||
it('withholds degraded write-unavailable signals from a foreign daemon route', async () => {
|
||||
const current = createDaemonAdapter('daemon', ['shared-pane'])
|
||||
const legacy = createDaemonAdapter('legacy')
|
||||
const fallback = createProvider('fallback')
|
||||
const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback })
|
||||
const recovered: string[] = []
|
||||
await provider.discoverDaemonSessions()
|
||||
provider.onWriteUnavailable(({ id }) => recovered.push(id))
|
||||
|
||||
legacy.triggerWriteUnavailable('shared-pane')
|
||||
current.triggerWriteUnavailable('shared-pane')
|
||||
|
||||
expect(recovered).toEqual(['shared-pane'])
|
||||
})
|
||||
|
||||
it('withholds degraded write-unavailable signals for fallback collisions', async () => {
|
||||
const sessionId = 'fallback-collision'
|
||||
const current = createDaemonAdapter('daemon', [sessionId])
|
||||
const fallback = createProvider('fallback')
|
||||
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
|
||||
await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 })
|
||||
await provider.discoverDaemonSessions()
|
||||
const recovered: string[] = []
|
||||
provider.onWriteUnavailable(({ id }) => recovered.push(id))
|
||||
|
||||
current.triggerWriteUnavailable(sessionId)
|
||||
|
||||
expect(recovered).toEqual([])
|
||||
})
|
||||
|
||||
it('rejects completion inspection instead of borrowing the fallback provider', async () => {
|
||||
const provider = new DegradedDaemonPtyProvider({
|
||||
current: createDaemonAdapter('daemon'),
|
||||
|
||||
@@ -227,7 +227,13 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
|
||||
// adapters only — the local fallback has no dead-socket problem.
|
||||
onWriteUnavailable(callback: (payload: { id: string }) => void): () => void {
|
||||
return combineUnsubscribes(
|
||||
this.allDaemonAdapters().map((adapter) => adapter.onWriteUnavailable(callback))
|
||||
this.allDaemonAdapters().map((adapter) =>
|
||||
adapter.onWriteUnavailable((payload) => {
|
||||
if (this.routing.shouldForwardWriteUnavailable(adapter, payload.id)) {
|
||||
callback(payload)
|
||||
}
|
||||
})
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -25,7 +25,7 @@ export class DegradedDaemonSessionRouting {
|
||||
this.adapters = [current, ...legacy]
|
||||
this.adapterSet = new Set(this.adapters)
|
||||
this.routes = new DaemonSessionRouteTable(this.adapters)
|
||||
this.fallbackRoutes = new FallbackRoutes(this.fallbackOwners, this.routes)
|
||||
this.fallbackRoutes = new FallbackRoutes(this.fallbackOwners, this.routes, this.fallbackSpawns)
|
||||
for (const adapter of this.adapters) {
|
||||
this.identityUnsubscribes.push(
|
||||
adapter.onDaemonIdentityChanged(({ previous }) => {
|
||||
@@ -76,7 +76,7 @@ export class DegradedDaemonSessionRouting {
|
||||
this.fallbackRoutes.recordSpawnedSession(result.id, target)
|
||||
}
|
||||
} else {
|
||||
this.routes.recordOwned(result.id, target as DaemonPtyAdapter)
|
||||
this.fallbackRoutes.recordDaemonSpawn(result.id, opts.sessionId, target as DaemonPtyAdapter)
|
||||
}
|
||||
return result
|
||||
} finally {
|
||||
@@ -158,11 +158,12 @@ export class DegradedDaemonSessionRouting {
|
||||
this.fallbackRoutes.clearCollision(sessionId)
|
||||
return true
|
||||
}
|
||||
if (route) {
|
||||
this.fallbackRoutes.recordCollision(sessionId)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
this.fallbackRoutes.recordCollision(sessionId)
|
||||
return false
|
||||
}
|
||||
|
||||
shouldForwardWriteUnavailable(adapter: DaemonPtyAdapter, sessionId: string): boolean {
|
||||
return this.fallbackRoutes.shouldForwardWriteUnavailable(sessionId, adapter)
|
||||
}
|
||||
|
||||
recordExit(provider: IPtyProvider, sessionId: string): boolean {
|
||||
|
||||
@@ -2,6 +2,8 @@ import type { IPtyProvider } from '../providers/types'
|
||||
import type { DaemonPtyAdapter } from './daemon-pty-adapter'
|
||||
import { sameDaemonIncarnation } from './daemon-session-route'
|
||||
import type { DaemonSessionRouteTable } from './daemon-session-route-table'
|
||||
import { recordValidatedDaemonSpawn } from './daemon-spawn-route-validation'
|
||||
import type { DegradedFallbackSpawnRoutes } from './degraded-fallback-spawn-routes'
|
||||
|
||||
type FallbackLivenessProbe = {
|
||||
provider: IPtyProvider
|
||||
@@ -15,7 +17,8 @@ export class DegradedFallbackSessionRoutes {
|
||||
|
||||
constructor(
|
||||
private readonly sessions: Map<string, IPtyProvider>,
|
||||
private readonly daemonRoutes: DaemonSessionRouteTable
|
||||
private readonly daemonRoutes: DaemonSessionRouteTable,
|
||||
private readonly spawns: DegradedFallbackSpawnRoutes
|
||||
) {}
|
||||
|
||||
hasCollision(sessionId: string): boolean {
|
||||
@@ -41,12 +44,29 @@ export class DegradedFallbackSessionRoutes {
|
||||
}
|
||||
}
|
||||
|
||||
recordDaemonSpawn(
|
||||
resultSessionId: string,
|
||||
requestedSessionId: string | undefined,
|
||||
owner: DaemonPtyAdapter
|
||||
): void {
|
||||
recordValidatedDaemonSpawn(this.daemonRoutes, resultSessionId, requestedSessionId, owner)
|
||||
}
|
||||
|
||||
clearUnownedCollision(sessionId: string): void {
|
||||
if (!this.sessions.has(sessionId)) {
|
||||
this.clearCollision(sessionId)
|
||||
}
|
||||
}
|
||||
|
||||
shouldForwardWriteUnavailable(sessionId: string, owner: DaemonPtyAdapter): boolean {
|
||||
return (
|
||||
!this.spawns.hasLiveCandidate(sessionId) &&
|
||||
!this.sessions.has(sessionId) &&
|
||||
!this.collisionIds.has(sessionId) &&
|
||||
this.daemonRoutes.getOwned(sessionId) === owner
|
||||
)
|
||||
}
|
||||
|
||||
deleteSession(sessionId: string): void {
|
||||
this.sessions.delete(sessionId)
|
||||
this.epochs.delete(sessionId)
|
||||
|
||||
Reference in New Issue
Block a user