fix(daemon): retire empty current-generation daemons (#9277)

* fix(daemon): retire empty current-generation daemons

Co-authored-by: Orca <help@stably.ai>

* fix(daemon): retire empty daemons on disconnect

Co-authored-by: Orca <help@stably.ai>

* test(daemon): authenticate Windows lifecycle harness

Co-authored-by: Orca <help@stably.ai>

* test(daemon): assert remaining shutdown budget

Co-authored-by: Orca <help@stably.ai>

---------

Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
Jinwoo Hong 2026-07-19 19:38:44 -07:00 committed by GitHub
parent 3a847bfac9
commit 7adda25b0a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
23 changed files with 4494 additions and 554 deletions

View File

@ -30,19 +30,99 @@ function readProtocolVersion() {
return Number(match[1])
}
function createStreamSocket(socketPath, token, protocolVersion, clientId, onFailure) {
const socket = connect(socketPath)
let buffer = ''
return new Promise((resolveStream, rejectStream) => {
let handshakeComplete = false
const rejectHandshake = (error) => {
if (handshakeComplete) {
onFailure(error)
return
}
handshakeComplete = true
clearTimeout(timer)
socket.removeListener('data', onData)
socket.removeListener('error', onError)
socket.removeListener('close', onClose)
socket.destroy()
rejectStream(error)
}
const onError = (error) => rejectHandshake(error)
const onClose = () => rejectHandshake(new Error('Daemon stream socket closed'))
const onData = (chunk) => {
buffer += chunk.toString('utf8')
const newline = buffer.indexOf('\n')
if (newline === -1) {
return
}
const message = JSON.parse(buffer.slice(0, newline))
if (message.type !== 'hello') {
return
}
if (!message.ok) {
rejectHandshake(new Error(message.error ?? 'Daemon rejected stream hello'))
return
}
handshakeComplete = true
clearTimeout(timer)
socket.removeListener('data', onData)
// Why: drain terminal events even though this lifecycle repro only asserts through RPC.
socket.on('data', () => {})
resolveStream(socket)
}
const timer = setTimeout(
() => rejectHandshake(new Error('Daemon stream hello timed out')),
requestTimeoutMs
)
socket.on('error', onError)
socket.on('close', onClose)
socket.on('data', onData)
socket.once('connect', () => {
socket.write(
`${JSON.stringify({
type: 'hello',
version: protocolVersion,
token,
clientId,
role: 'stream'
})}\n`
)
})
})
}
function createRpcClient(socketPath, tokenPath) {
const socket = connect(socketPath)
const clientId = randomUUID()
const protocolVersion = readProtocolVersion()
const token = readFileSync(tokenPath, 'utf8').trim()
const pending = new Map()
let buffer = ''
let requestId = 0
let streamSocket
let connectionError
let helloResolve
let helloReject
let helloTimer
const hello = new Promise((resolveHello, rejectHello) => {
helloResolve = resolveHello
helloReject = rejectHello
helloResolve = () => {
clearTimeout(helloTimer)
resolveHello()
}
helloReject = (error) => {
clearTimeout(helloTimer)
rejectHello(error)
}
})
helloTimer = setTimeout(() => {
helloReject(new Error('Daemon control hello timed out'))
socket.destroy()
}, requestTimeoutMs)
const rejectPending = (error) => {
connectionError ??= error
helloReject(error)
for (const { reject, timer } of pending.values()) {
clearTimeout(timer)
@ -82,25 +162,40 @@ function createRpcClient(socketPath, tokenPath) {
}
})
const connected = new Promise((resolveConnected, rejectConnected) => {
const socketConnected = new Promise((resolveConnected, rejectConnected) => {
socket.once('connect', resolveConnected)
socket.once('error', rejectConnected)
}).then(() => {
socket.write(
`${JSON.stringify({
type: 'hello',
version: readProtocolVersion(),
token: readFileSync(tokenPath, 'utf8').trim(),
clientId: randomUUID(),
role: 'control'
})}\n`
})
const connected = Promise.all([
socketConnected.then(() => {
socket.write(
`${JSON.stringify({
type: 'hello',
version: protocolVersion,
token,
clientId,
role: 'control'
})}\n`
)
}),
hello
]).then(async () => {
// Why: v24 only admits terminals for the same complete control+stream pair as production.
streamSocket = await createStreamSocket(
socketPath,
token,
protocolVersion,
clientId,
rejectPending
)
return hello
})
return {
async request(type, payload) {
await connected
if (connectionError) {
throw connectionError
}
const id = `repro-${++requestId}`
return new Promise((resolveRequest, rejectRequest) => {
const timer = setTimeout(() => {
@ -112,6 +207,7 @@ function createRpcClient(socketPath, tokenPath) {
})
},
close() {
streamSocket?.destroy()
socket.destroy()
}
}

View File

@ -0,0 +1,279 @@
# Daemon lifecycle retirement for issue #9138
## Status
Implemented and locally validated for PR #9277.
The earlier full ownership/audit prototype is preserved at:
- branch: `Jinwoo-H/issue-9138-full-ownership-audit-snapshot`
- commit: `7c915909bd26670b8af36aa683cff85077395dd1`
- GitHub: <https://github.com/stablyai/orca/tree/Jinwoo-H/issue-9138-full-ownership-audit-snapshot>
That branch is the recovery point for ownership persistence, cross-profile raw extraction, startup
audit, the candidate journal, profile-transfer recovery, natural-exit reconciliation, and future
legacy enforcement. None of those systems ship in this PR.
Current `main` assigned protocol v23 to the macOS login-shell preparation change while this work was
in progress. The lifecycle contract therefore ships as v24; v23 is preserved as a legacy generation
alongside v22 and older versions.
## Inputs and scope decision
Issue: <https://github.com/stablyai/orca/issues/9138>
Reviewed design comment by AmethystLiang:
<https://github.com/stablyai/orca/issues/9138#issuecomment-5006601124>
The reviewed comment correctly identifies two different problems:
1. current-generation daemons have no daemon-owned empty lifecycle;
2. legacy sessions need complete cross-profile ownership evidence before an app-side reaper can act.
This PR solves only the first problem. The second problem is much larger, adds steady-state
persistence and startup-audit cost, and is audit-only until field evidence can justify enforcement.
Keeping it out makes the user-visible fix small enough to review and benchmark independently.
The narrow implementation retains these important rules from AmethystLiang's design:
- daemon lifecycle behavior requires a protocol bump because old running daemons cannot acquire it;
- live sessions always win over cleanup;
- absence from a worktree, pane layout, profile, or failed listing is never destructive evidence;
- exact PID, process-start time, and per-launch nonce identify a v24 endpoint incarnation;
- the daemon, not app startup, makes the atomic empty decision;
- pre-v24 daemons stay reattachable and are not automatically shut down;
- SSH, WSL, remote-runtime, degraded-provider, sleep/wake, and profile behavior stay unchanged.
The narrow implementation changes one policy from the original comment: it does not use a blanket
30-minute idle timer. Any loss of the last fully authenticated app client is an exact lifecycle
event, so a daemon retires as soon as it can atomically prove it is empty. This removes runtime
inactivity heuristics and periodic ownership work.
## User-visible behavior
```text
Before
app disconnects ──> daemon stays forever
├── live sessions stay (wanted)
└── zero sessions also stay (leak)
After
clean app detach ──> daemon atomically checks itself
├── any live session/work/client ──> stay alive
└── exactly empty ──> exit immediately
unexpected drop ──> daemon atomically checks itself
├── live session/work/connection ──> stay alive
└── exactly empty ──> exit immediately
v23 and older ──> existing reattach behavior; no automatic retirement
```
An end user with live terminals should notice no change. An end user who quits with no daemon-backed
terminals should no longer accumulate the new v24 generation. If Orca crashes or loses its socket,
live terminals still keep the daemon alive indefinitely. An empty daemon exits immediately; a later
app restart launches a fresh daemon instead of reusing an empty process.
## Protocol and lifecycle design
### Endpoint identity
Protocol v24 hello responses include:
```ts
type DaemonEndpointIdentity = {
pid: number
startedAtMs: number
launchNonce: string
}
```
The parent generates the launch nonce, passes the nonce and PID-record path to the daemon, and writes
the daemon's self-reported start time plus the same nonce to the PID record. Both authenticated client
sockets must report the same valid identity. v24 rejects a missing or malformed identity; v23 keeps
the previous identity-free handshake.
PID publication is fail-closed. Missing readiness identity, invalid PID, an existing PID record, or a
write failure terminates the new child and fails launch instead of leaving an untracked daemon.
### Clean detach
At the end of `DaemonPtyAdapter.disconnectOnly()`, after final checkpoints and producer resumes, a
v24 adapter establishes a full connection if necessary and sends `shutdownIfIdle` within one shared
250 ms budget. v23 and older adapters skip it.
Initialization establishes one authenticated v24 lifecycle lease even before the first terminal is
opened. This cancels the initial launch-adoption watchdog and ensures a never-used daemon can still
receive clean retirement on quit. If startup fallback has already won, the late daemon is not
installed; it instead receives the same bounded retirement attempt, which an adopted live session
will reject.
The daemon accepts retirement only when, in one event-loop turn:
- the requesting authenticated client has both control and stream sockets;
- it is the only authenticated client;
- every accepted transport belongs to that client;
- no `createOrAttach` operation is in flight;
- the terminal host has zero sessions.
When all conditions hold, the daemon synchronously closes the listening server before replying. That
is the admission fence: a new socket or terminal cannot appear after the empty proof. Cleanup then
runs asynchronously and the process exits. A failed RPC is non-fatal to app quit and falls back to
the same event-driven empty check when the authenticated sockets close.
### Initial adoption watchdog
A freshly launched v24 daemon gets up to two minutes to receive its first complete authenticated
client pair. Without this startup-only watchdog, the daemon would prove itself empty and exit in the
normal launch handoff before the parent could connect; without a bound, a parent crash during that
handoff would orphan the new daemon forever.
A complete pair permanently cancels this watchdog, and terminal admission requires that complete
pair. Raw and partial transports pause it without extending its original deadline. It is never
rearmed after adoption and is not a terminal inactivity or crash-reconnect timer.
### Unexpected disconnect
When the last client that completed both authenticated sockets loses its control connection, the
daemon records an event-driven retirement request with no wall-clock grace.
- A complete authenticated reconnect cancels the request if existing work or a transport kept the
daemon alive long enough to reconnect.
- Only a complete authenticated pair may admit a terminal; completing that pair cancels the request
before admission can begin.
- A raw socket, one-socket health probe, or partial authenticated connection blocks retirement but
cannot erase evidence that the last fully connected app left.
- Replacing a client ID first records the old full connection's loss; completing the replacement
stream cancels that evidence, while an incomplete replacement only blocks retirement.
- A live session prevents shutdown indefinitely. When the last session exits, the daemon immediately
rechecks every guard and retires only if it is then exactly empty.
The adapter remembers an authenticated unexpected disconnect. If self-retirement later removes the
token, a token-file `ENOENT` is respawnable only with that prior evidence. An initial missing token is
not broadened into destructive or respawn authority.
### Artifact cleanup
The daemon removes only artifacts it can claim as its own:
- token contents must match the daemon's in-memory token;
- PID and launch nonce must match the daemon process and launch nonce;
- cleanup first renames the canonical entry to a unique claim, validates that claim, and never
overwrites or unlinks a replacement installed at the canonical path.
Current-protocol external cleanup waits for v24 self-shutdown and does not unconditionally remove v24
PID/socket artifacts. Legacy cleanup behavior is unchanged.
## Performance design
There is no polling, profile enumeration, ownership checksum, candidate journal, process scan, or
steady-state persistence write in this PR.
The steady-state terminal hot path adds no timer work and no per-byte hashing. Initialization adds one
two-socket authenticated lifecycle handshake; the only new RPC is on app/provider detach, and its
connect-plus-request path shares a 250 ms cap, including when quit joins an existing connection
attempt; teardown fences that attempt from resurrecting sockets afterward. Unexpected-disconnect
bookkeeping changes only socket and session lifecycle events. Each unadopted daemon owns at most one
unref'ed startup watchdog, which is canceled permanently on adoption.
Validation compares current main and the branch for:
- daemon connect plus two-socket hello latency;
- repeated `listSessions` RPC latency;
- terminal echo/stream throughput through a real socket daemon;
- clean empty-detach latency;
- event-loop delay under repeated RPC/stream work;
- idle CPU/RSS and timer count where observable.
The acceptance target is no statistically meaningful terminal throughput regression and no new
steady-state disk writes. Results are recorded in the PR body.
### Local performance regression screen
The final local host was not quiet enough for publication-grade absolute numbers: load averages were
17-38 and unrelated Orca, browser, simulator, and VM processes occupied several cores. A paired
same-host screen still found no large regression. Five `main` v23 samples were bracketed by ten v24
branch samples; medians across sample medians were:
| Measure | `main` v23 | branch v24 |
| -------------------- | ---------: | ---------: |
| two-socket connect | 1.38 ms | 1.32 ms |
| `listSessions` RPC | 0.0366 ms | 0.0374 ms |
| terminal echo stream | 3.28 MiB/s | 3.14 MiB/s |
All three medians were within about 5%. Individual stream samples varied from 0.66 to 4.04 MiB/s and
event-loop-delay samples had similar load-driven outliers, so these results are a coarse regression
screen, not evidence of an exact performance delta. Static hot-path review confirms lifecycle work
runs on connection, disconnection, session admission/exit, and quit events, with no new work per PTY
byte and no steady-state persistence or polling.
## Verification and validation
### Focused unit and integration tests
- v24 requires valid matching endpoint identity on both sockets.
- v23 and v22 accept the prior identity-free handshake and remain listed as previous protocols.
- production launch passes PID path and nonce and writes the exact readiness identity.
- incomplete readiness identity or failed exclusive PID publication kills and rejects the child.
- clean empty detach exits immediately.
- a never-used current adapter connects and retires cleanly on quit.
- initial adoption cancels the launch watchdog and keeps first-terminal spawn working after its old
deadline.
- a live session, another client, raw transport, or in-flight admission rejects clean retirement.
- a control-only overlapping client blocks but cannot erase the last full-client retirement request.
- a same-client-ID control replacement cannot erase the prior full connection's retirement request.
- a control-only client cannot admit a terminal or erase startup/retirement evidence with a failed
request.
- startup fail-open performs bounded empty retirement without installing a late provider.
- quit remains bounded while a prior handshake is stalled and cannot resurrect client sockets later.
- the synchronous listener fence rejects post-fence connections as retryable.
- an unexpected empty disconnect retires immediately without a runtime inactivity timer.
- a real reconnect cancels pending retirement while live work keeps the daemon available.
- raw and health probes block but cannot erase pending retirement.
- final session exit triggers an immediate guarded retirement check.
- token/PID cleanup preserves malformed, stale, and replacement artifacts.
- authenticated token disappearance performs one coalesced respawn; initial token absence does not.
### Process/E2E tests
- start a real isolated v24 daemon with real socket/named-pipe, token, and PID artifacts;
- authenticate, disconnect the last empty client, and verify the exact process and owned artifacts
exit;
- prove a live session rejects retirement and remains reattachable;
- run a protocol-v22 fixture beside v24 and prove v22 remains connectable/reattachable;
- never target a production runtime directory or signal a process not created by the fixture.
### Repository validation
- Node typecheck;
- oxlint and repository max-lines policy;
- focused daemon, adapter, launcher, restart, and legacy-routing suites;
- full daemon test suite;
- desktop and web production builds;
- `git diff --check` and review of every changed file against `origin/main`;
- independent review-until-clean, with review loops recorded in `.orca/bug-factory.json`;
- packaged Windows/Linux validation where CI is available; local macOS process E2E before publication.
Final local results on macOS arm64 after the event-driven policy revision:
- full daemon suite: 56 files passed, 2 skipped; 939 tests passed, 5 skipped;
- process E2E: v22 remained live and reattachable while the exact empty v24 process and its owned
artifacts retired immediately after its final authenticated client disconnected;
- full Node typecheck, focused oxlint, max-lines ratchet, and `git diff --check` passed;
- full desktop, web, and native production build passed with existing build warnings;
- three independent post-revision review tracks covering architecture/state machines,
ownership/adoption, and process lifecycle ended clean after actionable findings were fixed.
The local shell used Node 26.5.0 while the repository requests Node 24; the commands completed
successfully, and repository CI remains responsible for the supported Node/platform matrix.
## Rollout and rollback
This PR changes only v24. Existing v23 and older daemons are preserved. Rolling back the app leaves
v24 as another legacy generation and does not give an older app authority to shut it down.
If the lifecycle behavior must be disabled, revert the v24 protocol/lifecycle commit. The separated
ownership/audit prototype remains recoverable from the archived branch and commit above; it should
return only as a separately reviewed, benchmarked follow-up.

View File

@ -68,6 +68,12 @@ describe('DaemonClient', () => {
onStreamHello?: (msg: HelloMessage) => void
rejectVersion?: boolean
suppressHelloResponse?: boolean
omitHelloIdentity?: boolean
helloIdentity?: (role: 'control' | 'stream') => {
pid: number
startedAtMs: number
launchNonce: string
}
}): Promise<void> {
return new Promise((resolve) => {
server = createServer((socket) => {
@ -103,7 +109,19 @@ describe('DaemonClient', () => {
socket.write(encodeNdjson({ type: 'hello', ok: false, error: 'Version mismatch' }))
return
}
socket.write(encodeNdjson({ type: 'hello', ok: true }))
socket.write(
encodeNdjson({
type: 'hello',
ok: true,
...(!opts?.omitHelloIdentity
? {
daemonIdentity: opts?.helloIdentity
? opts.helloIdentity(hello.role)
: { pid: 123, startedAtMs: 456, launchNonce: 'default-launch' }
}
: {})
})
)
if (hello.role === 'stream') {
opts?.onStreamHello?.(hello)
}
@ -136,6 +154,48 @@ describe('DaemonClient', () => {
await waitFor(() => hellos.length > 0)
})
it('captures one matching endpoint identity from both authenticated sockets', async () => {
const identity = { pid: 123, startedAtMs: 456, launchNonce: 'launch-a' }
await startMockDaemon({ helloIdentity: () => identity })
client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
expect(client.getDaemonIdentity()).toEqual(identity)
})
it('rejects a v24 daemon that omits endpoint identity', async () => {
await startMockDaemon({ omitHelloIdentity: true })
client = new DaemonClient({ socketPath, tokenPath })
await expect(client.ensureConnected()).rejects.toThrow('Invalid daemon identity')
})
it('allows a legacy v23 daemon to omit endpoint identity', async () => {
await startMockDaemon({ omitHelloIdentity: true })
client = new DaemonClient({ socketPath, tokenPath, protocolVersion: 23 })
await expect(client.ensureConnected()).resolves.toBeUndefined()
expect(client.getDaemonIdentity()).toBeNull()
})
it('rejects when control and stream sockets report different daemon identities', async () => {
await startMockDaemon({
helloIdentity: (role) => ({
pid: role === 'control' ? 123 : 124,
startedAtMs: 456,
launchNonce: 'launch-a'
})
})
client = new DaemonClient({ socketPath, tokenPath })
await expect(client.ensureConnected()).rejects.toThrow(
'Daemon identity changed during connection'
)
expect(client.getDaemonIdentity()).toBeNull()
})
it('removes socket startup listeners after connecting', async () => {
await startMockDaemon()
@ -189,6 +249,35 @@ describe('DaemonClient', () => {
}
})
it('bounds a waiter on an existing connection attempt and prevents socket resurrection', async () => {
let resolveHello: () => void = () => {}
const helloReceived = new Promise<void>((resolve) => {
resolveHello = resolve
})
await startMockDaemon({
suppressHelloResponse: true,
onHello: resolveHello
})
client = new DaemonClient({ socketPath, tokenPath })
const ownerAttempt = client.ensureConnected().catch((error: Error) => error)
await helloReceived
await expect(client.ensureConnectedWithin(25)).rejects.toThrow(
'Connection attempt wait timed out'
)
client.disconnect()
await expect(ownerAttempt).resolves.toBeInstanceOf(Error)
await new Promise((resolve) => setTimeout(resolve, 25))
const disconnected = client as unknown as {
controlSocket: Socket | null
streamSocket: Socket | null
}
expect(client.isConnected()).toBe(false)
expect(disconnected.controlSocket).toBeNull()
expect(disconnected.streamSocket).toBeNull()
})
it('removes hello startup listeners after timeout', async () => {
vi.useFakeTimers()
@ -200,11 +289,16 @@ describe('DaemonClient', () => {
socket.destroy = destroy as unknown as Socket['destroy']
const sendHello = (
client as unknown as {
sendHello(socket: Socket, token: string, role: 'control' | 'stream'): Promise<void>
sendHello(
socket: Socket,
token: string,
role: 'control' | 'stream',
timeoutMs: number
): Promise<void>
}
).sendHello.bind(client)
const promise = sendHello(socket, 'test-token-123', 'control')
const promise = sendHello(socket, 'test-token-123', 'control', 5000)
const rejection = expect(promise).rejects.toThrow('Hello response timed out')
await vi.advanceTimersByTimeAsync(5000)

View File

@ -4,11 +4,23 @@ import { readFileSync } from 'node:fs'
import { randomUUID } from 'node:crypto'
import { StringDecoder } from 'node:string_decoder'
import { encodeNdjson, createNdjsonParser } from './ndjson'
import { PROTOCOL_VERSION, NOTIFY_PREFIX, DaemonProtocolError } from './types'
import type { HelloMessage, HelloResponse, RpcResponse, DaemonEvent } from './types'
import {
CLEAN_DISCONNECT_PROTOCOL_VERSION,
PROTOCOL_VERSION,
NOTIFY_PREFIX,
DaemonProtocolError
} from './types'
import type {
DaemonEndpointIdentity,
HelloMessage,
HelloResponse,
RpcResponse,
DaemonEvent
} from './types'
import { addNodePtyRecoveryHint } from './node-pty-error-hints'
const CONNECT_TIMEOUT_MS = 5000
const CONNECTION_ATTEMPT_WAIT_MS = CONNECT_TIMEOUT_MS * 4
const REQUEST_TIMEOUT_MS = 30000
export type DaemonClientOptions = {
@ -42,6 +54,9 @@ export class DaemonClient {
// all call ensureConnected(). Without a lock, each starts a separate
// connection attempt, overwriting sockets and triggering "Connection lost".
private connectingPromise: Promise<void> | null = null
private connectionAttemptGeneration = 0
private daemonIdentity: DaemonEndpointIdentity | null = null
private observedAuthenticatedDisconnect = false
private pendingRequests = new Map<string, PendingRequest>()
private eventListeners: ((event: unknown) => void)[] = []
@ -59,15 +74,38 @@ export class DaemonClient {
return this.connected
}
getDaemonIdentity(): DaemonEndpointIdentity | null {
return this.daemonIdentity ? { ...this.daemonIdentity } : null
}
hasObservedAuthenticatedDisconnect(): boolean {
return this.observedAuthenticatedDisconnect
}
async ensureConnected(): Promise<void> {
return this.ensureConnectedWithTimeout(CONNECT_TIMEOUT_MS, false)
}
async ensureConnectedWithin(timeoutMs: number): Promise<void> {
return this.ensureConnectedWithTimeout(timeoutMs, true)
}
private async ensureConnectedWithTimeout(
timeoutMs: number,
sharedBudget: boolean
): Promise<void> {
if (this.connected) {
return
}
if (this.connectingPromise) {
return this.connectingPromise
// Why: a normal connection may legitimately consume one timeout for each
// socket and hello; bounded teardown calls instead keep their one shared budget.
const waiterTimeoutMs = sharedBudget ? timeoutMs : CONNECTION_ATTEMPT_WAIT_MS
return this.waitForConnectionAttempt(this.connectingPromise, waiterTimeoutMs)
}
this.connectingPromise = this.doConnect()
const attemptGeneration = this.connectionAttemptGeneration
this.connectingPromise = this.doConnect(timeoutMs, attemptGeneration, sharedBudget)
try {
await this.connectingPromise
} finally {
@ -75,8 +113,15 @@ export class DaemonClient {
}
}
private async doConnect(): Promise<void> {
private async doConnect(
timeoutMs: number,
attemptGeneration: number,
sharedBudget: boolean
): Promise<void> {
const token = readFileSync(this.tokenPath, 'utf-8').trim()
const deadlineMs = Date.now() + timeoutMs
const remainingMs = (): number =>
sharedBudget ? Math.max(1, deadlineMs - Date.now()) : timeoutMs
const pendingListenerCleanups: (() => void)[] = []
const cleanupPendingListeners = (): void => {
for (const cleanup of pendingListenerCleanups.splice(0)) {
@ -86,15 +131,32 @@ export class DaemonClient {
try {
// Sequential: control first, then stream
this.controlSocket = await this.connectSocket()
await this.sendHello(this.controlSocket, token, 'control')
const pendingControlSocket = await this.connectSocket(remainingMs())
this.assertConnectionAttemptCurrent(attemptGeneration, pendingControlSocket)
this.controlSocket = pendingControlSocket
const controlIdentity = await this.sendHello(
this.controlSocket,
token,
'control',
remainingMs()
)
this.assertConnectionAttemptCurrent(attemptGeneration, this.controlSocket)
pendingListenerCleanups.push(this.setupControlParser(this.controlSocket))
this.streamSocket = await this.connectSocket()
await this.sendHello(this.streamSocket, token, 'stream')
const pendingStreamSocket = await this.connectSocket(remainingMs())
this.assertConnectionAttemptCurrent(attemptGeneration, pendingStreamSocket)
this.streamSocket = pendingStreamSocket
const streamIdentity = await this.sendHello(this.streamSocket, token, 'stream', remainingMs())
this.assertConnectionAttemptCurrent(attemptGeneration, this.streamSocket)
if (!sameDaemonIdentity(controlIdentity, streamIdentity)) {
throw new DaemonProtocolError('Daemon identity changed during connection')
}
pendingListenerCleanups.push(this.setupStreamParser(this.streamSocket))
this.assertConnectionAttemptCurrent(attemptGeneration)
this.connected = true
this.observedAuthenticatedDisconnect = false
this.daemonIdentity = controlIdentity
this.disconnectArmed = true
this.connectionGeneration++
@ -120,12 +182,17 @@ export class DaemonClient {
this.controlSocket = null
this.streamSocket = null
this.connected = false
this.daemonIdentity = null
this.disconnectArmed = false
throw error
}
}
async request<T = unknown>(type: string, payload: unknown): Promise<T> {
async request<T = unknown>(
type: string,
payload: unknown,
timeoutMs = REQUEST_TIMEOUT_MS
): Promise<T> {
if (!this.connected || !this.controlSocket) {
throw new DaemonProtocolError('Not connected')
}
@ -136,8 +203,8 @@ export class DaemonClient {
return new Promise<T>((resolve, reject) => {
const timer = setTimeout(() => {
this.pendingRequests.delete(id)
reject(new DaemonProtocolError(`Request ${type} timed out after ${REQUEST_TIMEOUT_MS}ms`))
}, REQUEST_TIMEOUT_MS)
reject(new DaemonProtocolError(`Request ${type} timed out after ${timeoutMs}ms`))
}, timeoutMs)
this.pendingRequests.set(id, {
resolve: resolve as (value: unknown) => void,
@ -180,7 +247,9 @@ export class DaemonClient {
}
disconnect(): void {
this.connectionAttemptGeneration++
this.connected = false
this.daemonIdentity = null
this.disconnectArmed = false
this.cleanupActiveSocketListeners()
@ -196,7 +265,7 @@ export class DaemonClient {
this.streamSocket = null
}
private connectSocket(): Promise<Socket> {
private connectSocket(timeoutMs: number): Promise<Socket> {
return new Promise((resolve, reject) => {
const socket = connect(this.socketPath)
const cleanup = (): void => {
@ -216,14 +285,45 @@ export class DaemonClient {
cleanup()
socket.destroy()
reject(new DaemonProtocolError('Connection timed out'))
}, CONNECT_TIMEOUT_MS)
}, timeoutMs)
socket.on('connect', onConnect)
socket.on('error', onError)
})
}
private sendHello(socket: Socket, token: string, role: 'control' | 'stream'): Promise<void> {
private waitForConnectionAttempt(attempt: Promise<void>, timeoutMs: number): Promise<void> {
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
reject(new DaemonProtocolError('Connection attempt wait timed out'))
}, timeoutMs)
attempt.then(
() => {
clearTimeout(timer)
resolve()
},
(error) => {
clearTimeout(timer)
reject(error)
}
)
})
}
private assertConnectionAttemptCurrent(attemptGeneration: number, socket?: Socket): void {
if (attemptGeneration === this.connectionAttemptGeneration) {
return
}
socket?.destroy()
throw new DaemonProtocolError('Disconnected')
}
private sendHello(
socket: Socket,
token: string,
role: 'control' | 'stream',
timeoutMs: number
): Promise<DaemonEndpointIdentity | null> {
return new Promise((resolve, reject) => {
const hello: HelloMessage = {
type: 'hello',
@ -245,7 +345,7 @@ export class DaemonClient {
socket.removeListener('error', onError)
socket.removeListener('close', onClose)
}
const finish = (error?: Error): void => {
const finish = (error?: Error, identity: DaemonEndpointIdentity | null = null): void => {
if (settled) {
return
}
@ -255,7 +355,7 @@ export class DaemonClient {
reject(error)
return
}
resolve()
resolve(identity)
}
// Why: daemon socket chunks can split emoji/box-drawing UTF-8 bytes.
// Decoding each Buffer independently would permanently inject U+FFFD.
@ -271,7 +371,15 @@ export class DaemonClient {
try {
const response = JSON.parse(line) as HelloResponse
if (response.ok) {
finish()
const identity = parseDaemonEndpointIdentity(response.daemonIdentity)
if (
(this.protocolVersion >= CLEAN_DISCONNECT_PROTOCOL_VERSION && identity === null) ||
(response.daemonIdentity !== undefined && identity === null)
) {
finish(new DaemonProtocolError('Invalid daemon identity'))
return
}
finish(undefined, identity)
} else {
finish(
new DaemonProtocolError(addNodePtyRecoveryHint(response.error ?? 'Hello rejected'))
@ -290,7 +398,7 @@ export class DaemonClient {
// without a handshake timeout, startup waits forever on ensureConnected().
finish(new DaemonProtocolError('Hello response timed out'))
socket.destroy()
}, CONNECT_TIMEOUT_MS)
}, timeoutMs)
socket.on('data', onData)
socket.on('error', onError)
socket.on('close', onClose)
@ -352,7 +460,12 @@ export class DaemonClient {
return
}
this.disconnectArmed = false
this.connectionAttemptGeneration++
if (this.daemonIdentity) {
this.observedAuthenticatedDisconnect = true
}
this.connected = false
this.daemonIdentity = null
this.cleanupActiveSocketListeners()
for (const [id, pending] of this.pendingRequests) {
@ -377,3 +490,40 @@ export class DaemonClient {
cleanup?.()
}
}
function parseDaemonEndpointIdentity(value: unknown): DaemonEndpointIdentity | null {
if (!value || typeof value !== 'object') {
return null
}
const identity = value as { pid?: unknown; startedAtMs?: unknown; launchNonce?: unknown }
if (
!Number.isSafeInteger(identity.pid) ||
(identity.pid as number) <= 0 ||
typeof identity.startedAtMs !== 'number' ||
!Number.isFinite(identity.startedAtMs) ||
identity.startedAtMs <= 0 ||
typeof identity.launchNonce !== 'string' ||
identity.launchNonce.length === 0
) {
return null
}
return {
pid: identity.pid as number,
startedAtMs: identity.startedAtMs,
launchNonce: identity.launchNonce
}
}
function sameDaemonIdentity(
left: DaemonEndpointIdentity | null,
right: DaemonEndpointIdentity | null
): boolean {
return (
(left === null && right === null) ||
(left !== null &&
right !== null &&
left.pid === right.pid &&
left.startedAtMs === right.startedAtMs &&
left.launchNonce === right.launchNonce)
)
}

View File

@ -51,6 +51,49 @@ describe('daemon-entry parseArgs', () => {
})
})
it('parses the internal PID-record ownership pair', () => {
expect(
parseArgs([
'--socket',
'/tmp/t.sock',
'--token',
'/tmp/t.token',
'--pid-record',
'/tmp/t.pid',
'--launch-nonce',
'launch-a'
])
).toEqual({
socketPath: '/tmp/t.sock',
tokenPath: '/tmp/t.token',
pidPath: '/tmp/t.pid',
launchNonce: 'launch-a'
})
})
it('rejects either PID-record ownership argument without its pair', () => {
expect(() =>
parseArgs([
'--socket',
'/tmp/t.sock',
'--token',
'/tmp/t.token',
'--pid-record',
'/tmp/t.pid'
])
).toThrow('provided together')
expect(() =>
parseArgs([
'--socket',
'/tmp/t.sock',
'--token',
'/tmp/t.token',
'--launch-nonce',
'launch-a'
])
).toThrow('provided together')
})
it('still requires --socket and --token when --log-file is given', () => {
expect(() => parseArgs(['--log-file', '/tmp/daemon.log'])).toThrow('Usage:')
})

View File

@ -17,6 +17,8 @@ import { prepareMacosTccLoginShell } from '../providers/macos-tcc-login-shell'
export type ParsedDaemonArgs = {
socketPath: string
tokenPath: string
pidPath?: string
launchNonce?: string
/** Optional — absent for adopted old daemons and tests, which log nothing. */
logFilePath?: string
}
@ -25,6 +27,8 @@ export function parseArgs(argv: string[]): ParsedDaemonArgs {
let socketPath = ''
let tokenPath = ''
let logFilePath = ''
let pidPath = ''
let launchNonce = ''
for (let i = 0; i < argv.length; i++) {
if (argv[i] === '--socket' && argv[i + 1]) {
@ -36,6 +40,12 @@ export function parseArgs(argv: string[]): ParsedDaemonArgs {
} else if (argv[i] === '--log-file' && argv[i + 1]) {
logFilePath = argv[i + 1]
i++
} else if (argv[i] === '--pid-record' && argv[i + 1]) {
pidPath = argv[i + 1]
i++
} else if (argv[i] === '--launch-nonce' && argv[i + 1]) {
launchNonce = argv[i + 1]
i++
}
}
@ -43,7 +53,16 @@ export function parseArgs(argv: string[]): ParsedDaemonArgs {
throw new Error('Usage: daemon-entry --socket <path> --token <path> [--log-file <path>]')
}
return logFilePath ? { socketPath, tokenPath, logFilePath } : { socketPath, tokenPath }
if ((pidPath && !launchNonce) || (!pidPath && launchNonce)) {
throw new Error('Daemon PID record path and launch nonce must be provided together')
}
return {
socketPath,
tokenPath,
...(pidPath ? { pidPath, launchNonce } : {}),
...(logFilePath ? { logFilePath } : {})
}
}
async function main(): Promise<void> {
@ -54,7 +73,10 @@ async function main(): Promise<void> {
// an otherwise healthy detached daemon. Swallow it: stderr is diagnostic only.
process.stderr.on('error', () => {})
const { socketPath, tokenPath, logFilePath } = parseArgs(process.argv.slice(2))
const { socketPath, tokenPath, pidPath, launchNonce, logFilePath } = parseArgs(
process.argv.slice(2)
)
const startedAtMs = Date.now() - process.uptime() * 1000
// Fail-open: a broken log path must never block daemon startup.
const daemonLog = logFilePath ? createDaemonFileLog(logFilePath) : createNoopDaemonFileLog()
daemonLog.log('startup', { protocolVersion: PROTOCOL_VERSION, socketPath })
@ -126,16 +148,25 @@ async function main(): Promise<void> {
daemon = await startDaemon({
socketPath,
tokenPath,
...(pidPath ? { pidPath } : {}),
...(launchNonce ? { launchNonce } : {}),
...(pidPath ? { startedAtMs } : {}),
log: daemonLog,
preparePtySpawn: prepareMacosTccLoginShell,
spawnSubprocess: (opts) => createPtySubprocess(opts)
spawnSubprocess: (opts) => createPtySubprocess(opts),
onIdleShutdown: () => {
shuttingDown = true
daemonLog.log('shutdown', { reason: 'idle' })
daemonLog.close()
process.exit(0)
}
})
// Signal readiness to parent via IPC (if available)
if (process.send) {
// Why: Windows has no cheap OS query for a child's start time, so the
// daemon self-reports it here for the pid file's pid-recycling guard.
process.send({ type: 'ready', startedAtMs: Date.now() - process.uptime() * 1000 })
process.send({ type: 'ready', startedAtMs })
}
daemonLog.log('ready')

View File

@ -3,7 +3,9 @@ import { PREVIOUS_DAEMON_PROTOCOL_VERSIONS, PROTOCOL_VERSION } from './types'
describe('foreground-confirmation daemon protocol', () => {
it('rejects daemons from before the fresh-confirmation RPC', () => {
expect(PROTOCOL_VERSION).toBeGreaterThan(19)
expect(PROTOCOL_VERSION).toBe(24)
expect(PREVIOUS_DAEMON_PROTOCOL_VERSIONS).toContain(19)
expect(PREVIOUS_DAEMON_PROTOCOL_VERSIONS).toContain(22)
expect(PREVIOUS_DAEMON_PROTOCOL_VERSIONS).toContain(23)
})
})

View File

@ -0,0 +1,20 @@
export type HelloMessage = {
type: 'hello'
version: number
token: string
clientId: string
role: 'control' | 'stream'
}
export type DaemonEndpointIdentity = {
pid: number
startedAtMs: number
launchNonce: string
}
export type HelloResponse = {
type: 'hello'
ok: boolean
error?: string
daemonIdentity?: DaemonEndpointIdentity
}

View File

@ -0,0 +1,728 @@
import { EventEmitter } from 'node:events'
import { connect, type Socket } from 'node:net'
import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { DaemonClient } from './client'
import { DaemonPtyAdapter } from './daemon-pty-adapter'
import { DaemonServer } from './daemon-server'
import { PROTOCOL_VERSION } from './types'
import {
getDaemonPidPath,
getDaemonSocketPath,
serializeDaemonPidFile,
unlinkOwnedDaemonPidFile
} from './daemon-spawner'
import type { SubprocessHandle } from './session'
type ManualTimer = {
callback: () => void
dueAt: number
cancelled: boolean
}
class ManualIdleClock {
private nowMs = 0
private timers = new Set<ManualTimer>()
setTimeout(callback: () => void, delayMs: number): ManualTimer {
const timer = { callback, dueAt: this.nowMs + delayMs, cancelled: false }
this.timers.add(timer)
return timer
}
clearTimeout(handle: unknown): void {
const timer = handle as ManualTimer
timer.cancelled = true
this.timers.delete(timer)
}
now(): number {
return this.nowMs
}
advanceBy(ms: number): void {
this.nowMs += ms
for (const timer of [...this.timers].sort((a, b) => a.dueAt - b.dueAt)) {
if (timer.cancelled || timer.dueAt > this.nowMs) {
continue
}
this.timers.delete(timer)
timer.callback()
}
}
get pendingCount(): number {
return this.timers.size
}
}
function createMockSubprocess(): SubprocessHandle & { exit(code: number): void } {
let onExit: ((code: number) => void) | null = null
return {
pid: 9345,
getForegroundProcess: () => null,
write: vi.fn(),
resize: vi.fn(),
kill: vi.fn(),
forceKill: vi.fn(),
signal: vi.fn(),
onData: vi.fn(),
onExit(callback) {
onExit = callback
},
dispose: vi.fn(),
exit(code) {
onExit?.(code)
}
}
}
async function waitFor(predicate: () => boolean): Promise<void> {
const deadline = Date.now() + 2_000
while (!predicate()) {
if (Date.now() >= deadline) {
throw new Error('Timed out waiting for daemon idle state')
}
await new Promise((resolve) => setTimeout(resolve, 5))
}
}
async function requestOnRawSocket(
socket: Socket,
request: { id: string; type: string; payload: unknown }
): Promise<{ error?: string }> {
return new Promise((resolve, reject) => {
let buffer = ''
const cleanup = (): void => {
clearTimeout(timeout)
socket.off('data', onData)
}
const onData = (chunk: Buffer): void => {
buffer += chunk.toString('utf8')
for (;;) {
const newlineIndex = buffer.indexOf('\n')
if (newlineIndex === -1) {
return
}
const line = buffer.slice(0, newlineIndex)
buffer = buffer.slice(newlineIndex + 1)
const message = JSON.parse(line) as { id?: string; error?: string }
if (message.id === request.id) {
cleanup()
resolve(message)
return
}
}
}
const timeout = setTimeout(() => {
cleanup()
reject(new Error(`Timed out waiting for raw response ${request.id}`))
}, 2_000)
socket.on('data', onData)
socket.write(`${JSON.stringify(request)}\n`)
})
}
describe('current daemon lifecycle retirement', () => {
let dir: string
let socketPath: string
let tokenPath: string
let pidPath: string
let clock: ManualIdleClock
let server: DaemonServer | null
let subprocess: ReturnType<typeof createMockSubprocess>
let onIdleShutdown: ReturnType<typeof vi.fn<() => void>>
beforeEach(() => {
dir = mkdtempSync(join(tmpdir(), 'daemon-idle-shutdown-'))
socketPath = getDaemonSocketPath(dir)
tokenPath = join(dir, 'daemon.token')
pidPath = getDaemonPidPath(dir)
clock = new ManualIdleClock()
subprocess = createMockSubprocess()
onIdleShutdown = vi.fn<() => void>()
server = null
})
afterEach(async () => {
await server?.shutdown().catch(() => {})
rmSync(dir, { recursive: true, force: true })
})
async function startServer(
options: {
launchNonce?: string
protocolVersion?: number
} = {}
): Promise<void> {
server = new DaemonServer({
socketPath,
tokenPath,
...(options.launchNonce ? { pidPath, launchNonce: options.launchNonce } : {}),
...(options.protocolVersion !== undefined
? { protocolVersion: options.protocolVersion }
: {}),
initialAdoptionTestConfig: { timeoutMs: 100, clock },
onIdleShutdown,
spawnSubprocess: () => subprocess
})
await server.start()
}
it('retires immediately after an unexpected empty disconnect and removes owned artifacts', async () => {
const launchNonce = 'launch-a'
writeFileSync(
pidPath,
serializeDaemonPidFile({ pid: process.pid, startedAtMs: null, launchNonce })
)
await startServer({ launchNonce })
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
client.disconnect()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
expect(clock.pendingCount).toBe(0)
expect(existsSync(tokenPath)).toBe(false)
expect(existsSync(pidPath)).toBe(false)
if (process.platform !== 'win32') {
expect(existsSync(socketPath)).toBe(false)
}
})
it('retires a fresh daemon that is never adopted by a full client pair', async () => {
await startServer()
expect(clock.pendingCount).toBe(1)
clock.advanceBy(100)
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
})
it('retires immediately after an authenticated clean disconnect proves it is empty', async () => {
await startServer()
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
await expect(client.request('shutdownIfIdle', undefined)).resolves.toEqual({
retiring: true
})
client.disconnect()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
expect(clock.pendingCount).toBe(0)
})
it('keeps resources alive until the shutdownIfIdle reply write flushes', async () => {
await startServer()
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
const daemon = server as unknown as {
clients: Map<string, { controlSocket: Socket }>
host: { dispose: () => Promise<void> }
}
const controlSocket = [...daemon.clients.values()][0].controlSocket
const originalWrite = controlSocket.write.bind(controlSocket)
let replyFlushed: (() => void) | undefined
vi.spyOn(controlSocket, 'write').mockImplementation(((
chunk: string | Uint8Array,
...args: unknown[]
) => {
replyFlushed = args.find((arg) => typeof arg === 'function') as (() => void) | undefined
return originalWrite(chunk)
}) as unknown as Socket['write'])
const dispose = vi.spyOn(daemon.host, 'dispose')
await expect(client.request('shutdownIfIdle', undefined)).resolves.toEqual({ retiring: true })
expect(dispose).not.toHaveBeenCalled()
expect(onIdleShutdown).not.toHaveBeenCalled()
replyFlushed?.()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
expect(dispose).toHaveBeenCalledOnce()
client.disconnect()
})
it('finishes shutdownIfIdle when the peer closes before its reply callback', async () => {
await startServer()
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
const daemon = server as unknown as {
clients: Map<string, { controlSocket: Socket }>
host: { dispose: () => Promise<void> }
}
const controlSocket = [...daemon.clients.values()][0].controlSocket
const originalWrite = controlSocket.write.bind(controlSocket)
vi.spyOn(controlSocket, 'write').mockImplementation(((chunk: string | Uint8Array) =>
originalWrite(chunk)) as unknown as Socket['write'])
const dispose = vi.spyOn(daemon.host, 'dispose')
await expect(client.request('shutdownIfIdle', undefined)).resolves.toEqual({ retiring: true })
expect(dispose).not.toHaveBeenCalled()
client.disconnect()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
expect(dispose).toHaveBeenCalledOnce()
})
it('connects and retires a never-used adapter during clean disconnect', async () => {
await startServer()
const adapter = new DaemonPtyAdapter({ socketPath, tokenPath })
await adapter.disconnectOnly()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
expect(clock.pendingCount).toBe(0)
})
it('preserves a live session after clean detach and retires when that session exits', async () => {
await startServer()
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
await client.request('createOrAttach', { sessionId: 'preserved', cols: 80, rows: 24 })
await expect(client.request('shutdownIfIdle', undefined)).resolves.toEqual({
retiring: false
})
client.disconnect()
expect(onIdleShutdown).not.toHaveBeenCalled()
subprocess.exit(0)
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
expect(clock.pendingCount).toBe(0)
})
it('rejects clean retirement while create or attach is in flight', async () => {
await startServer()
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
const daemon = server as unknown as { createOrAttachInFlight: number }
daemon.createOrAttachInFlight = 1
await expect(client.request('shutdownIfIdle', undefined)).resolves.toEqual({
retiring: false
})
expect(onIdleShutdown).not.toHaveBeenCalled()
client.disconnect()
})
it('rejects clean retirement when an unknown transport is connected', async () => {
await startServer()
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
const rawSocket = connect(socketPath)
await new Promise<void>((resolve) => rawSocket.once('connect', resolve))
await expect(client.request('shutdownIfIdle', undefined)).resolves.toEqual({
retiring: false
})
rawSocket.destroy()
expect(onIdleShutdown).not.toHaveBeenCalled()
client.disconnect()
})
it('rejects clean retirement while another authenticated client is connected', async () => {
await startServer()
const first = new DaemonClient({ socketPath, tokenPath })
const second = new DaemonClient({ socketPath, tokenPath })
await Promise.all([first.ensureConnected(), second.ensureConnected()])
await expect(first.request('shutdownIfIdle', undefined)).resolves.toEqual({
retiring: false
})
expect(onIdleShutdown).not.toHaveBeenCalled()
first.disconnect()
second.disconnect()
})
it("preserves another client's live session when failed adoption disconnects", async () => {
await startServer()
const adoptingAdapter = new DaemonPtyAdapter({ socketPath, tokenPath })
const liveOwner = new DaemonClient({ socketPath, tokenPath })
await Promise.all([adoptingAdapter.establishLifecycleLease(), liveOwner.ensureConnected()])
await liveOwner.request('createOrAttach', {
sessionId: 'owned-by-second-client',
cols: 80,
rows: 24
})
await adoptingAdapter.disconnectOnly()
expect(onIdleShutdown).not.toHaveBeenCalled()
await expect(liveOwner.request('listSessions', undefined)).resolves.toMatchObject({
sessions: [expect.objectContaining({ sessionId: 'owned-by-second-client', isAlive: true })]
})
subprocess.exit(0)
liveOwner.disconnect()
})
it('lets an overlapping raw socket block but not erase empty retirement', async () => {
await startServer()
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
const rawSocket = connect(socketPath)
await new Promise<void>((resolve) => rawSocket.once('connect', resolve))
client.disconnect()
expect(onIdleShutdown).not.toHaveBeenCalled()
rawSocket.destroy()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
expect(clock.pendingCount).toBe(0)
})
it('retires after the last authenticated client disconnects with no sessions', async () => {
await startServer()
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
expect(clock.pendingCount).toBe(0)
client.disconnect()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
expect(clock.pendingCount).toBe(0)
})
it('lets a complete reconnect cancel retirement while a live session blocks it', async () => {
await startServer()
const first = new DaemonClient({ socketPath, tokenPath })
await first.ensureConnected()
await first.request('createOrAttach', { sessionId: 'reconnected', cols: 80, rows: 24 })
first.disconnect()
const daemon = server as unknown as { retirementRequested: boolean }
await waitFor(() => daemon.retirementRequested)
expect(onIdleShutdown).not.toHaveBeenCalled()
const second = new DaemonClient({ socketPath, tokenPath })
await second.ensureConnected()
await waitFor(() => !daemon.retirementRequested)
subprocess.exit(0)
expect(onIdleShutdown).not.toHaveBeenCalled()
second.disconnect()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
})
it('cancels the initial-adoption timeout before the adapter creates its first terminal', async () => {
await startServer()
await waitFor(() => clock.pendingCount === 1)
clock.advanceBy(50)
const adopted = new DaemonPtyAdapter({ socketPath, tokenPath })
await adopted.establishLifecycleLease()
expect(clock.pendingCount).toBe(0)
clock.advanceBy(100)
expect(onIdleShutdown).not.toHaveBeenCalled()
await expect(
adopted.spawn({
sessionId: 'first-after-adoption',
cols: 80,
rows: 24
})
).resolves.toMatchObject({ id: 'first-after-adoption' })
subprocess.exit(0)
adopted.dispose()
})
it('does not let repeated authenticated control probes extend the startup deadline', async () => {
await startServer()
const healthControl = connect(socketPath)
await new Promise<void>((resolve) => healthControl.once('connect', resolve))
healthControl.write(
`${JSON.stringify({
type: 'hello',
version: PROTOCOL_VERSION,
token: readFileSync(tokenPath, 'utf8').trim(),
clientId: 'startup-health-control',
role: 'control'
})}\n`
)
const daemon = server as unknown as { clients: Map<string, unknown> }
await waitFor(() => daemon.clients.has('startup-health-control'))
clock.advanceBy(1_000)
healthControl.destroy()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
})
it('does not let a control-only create cancel the initial-adoption timeout', async () => {
await startServer()
const control = connect(socketPath)
await new Promise<void>((resolve) => control.once('connect', resolve))
control.write(
`${JSON.stringify({
type: 'hello',
version: PROTOCOL_VERSION,
token: readFileSync(tokenPath, 'utf8').trim(),
clientId: 'startup-control-create',
role: 'control'
})}\n`
)
const daemon = server as unknown as {
clients: Map<string, unknown>
retirementRequested: boolean
}
await waitFor(() => daemon.clients.has('startup-control-create'))
const response = await requestOnRawSocket(control, {
id: 'control-only-create',
type: 'createOrAttach',
payload: { sessionId: 'must-not-start', cols: 80, rows: 24 }
})
expect(response.error).toContain('connection is incomplete')
expect(daemon.retirementRequested).toBe(false)
expect(clock.pendingCount).toBe(0)
clock.advanceBy(1_000)
expect(onIdleShutdown).not.toHaveBeenCalled()
control.destroy()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
})
it('keeps empty-retirement intent when a control-only client overlaps the last app', async () => {
await startServer()
const paired = new DaemonClient({ socketPath, tokenPath })
await paired.ensureConnected()
const incomplete = connect(socketPath)
await new Promise<void>((resolve) => incomplete.once('connect', resolve))
incomplete.write(
`${JSON.stringify({
type: 'hello',
version: PROTOCOL_VERSION,
token: readFileSync(tokenPath, 'utf8').trim(),
clientId: 'control-only-overlap',
role: 'control'
})}\n`
)
const daemon = server as unknown as {
clients: Map<string, unknown>
retirementRequested: boolean
}
await waitFor(() => daemon.clients.has('control-only-overlap'))
paired.disconnect()
await waitFor(() => daemon.retirementRequested)
expect(clock.pendingCount).toBe(0)
const response = await requestOnRawSocket(incomplete, {
id: 'overlap-control-create',
type: 'createOrAttach',
payload: { sessionId: 'must-not-start', cols: 80, rows: 24 }
})
expect(response.error).toContain('connection is incomplete')
expect(daemon.retirementRequested).toBe(true)
incomplete.destroy()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
})
it('keeps retirement intent when a same-client replacement never completes its stream', async () => {
await startServer()
const paired = new DaemonClient({ socketPath, tokenPath })
await paired.ensureConnected()
const clientId = (paired as unknown as { clientId: string }).clientId
const replacementControl = connect(socketPath)
await new Promise<void>((resolve) => replacementControl.once('connect', resolve))
replacementControl.write(
`${JSON.stringify({
type: 'hello',
version: PROTOCOL_VERSION,
token: readFileSync(tokenPath, 'utf8').trim(),
clientId,
role: 'control'
})}\n`
)
const daemon = server as unknown as {
retirementRequested: boolean
}
await waitFor(() => daemon.retirementRequested)
expect(clock.pendingCount).toBe(0)
replacementControl.destroy()
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
})
it('keeps a live session after clients disconnect, then retires on its exit', async () => {
await startServer()
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
await client.request('createOrAttach', { sessionId: 'live', cols: 80, rows: 24 })
client.disconnect()
const daemon = server as unknown as { retirementRequested: boolean }
await waitFor(() => daemon.retirementRequested)
expect(onIdleShutdown).not.toHaveBeenCalled()
expect(clock.pendingCount).toBe(0)
subprocess.exit(0)
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
})
it('uses the direct-construction protocol fixture version for hello compatibility', async () => {
await startServer({ protocolVersion: 22 })
const client = new DaemonClient({ socketPath, tokenPath, protocolVersion: 22 })
await expect(client.ensureConnected()).resolves.toBeUndefined()
client.disconnect()
})
it('requests clean retirement only for protocol v24 and newer adapters', async () => {
await startServer()
const current = new DaemonPtyAdapter({ socketPath, tokenPath })
await current.listProcesses()
const currentClient = (
current as unknown as {
client: DaemonClient
}
).client
const currentRequest = vi.spyOn(currentClient, 'request')
await current.disconnectOnly()
expect(currentRequest).toHaveBeenCalledOnce()
expect(currentRequest.mock.calls[0]?.slice(0, 2)).toEqual(['shutdownIfIdle', undefined])
const timeoutMs = currentRequest.mock.calls[0]?.[2]
// Why: connection and RPC share a wall-clock budget, so elapsed setup time is expected.
expect(timeoutMs).toEqual(expect.any(Number))
expect(timeoutMs).toBeGreaterThan(0)
expect(timeoutMs).toBeLessThanOrEqual(250)
})
it('does not send the v24 clean-disconnect RPC to a legacy daemon', async () => {
await startServer({ protocolVersion: 23 })
const legacy = new DaemonPtyAdapter({ socketPath, tokenPath, protocolVersion: 23 })
await legacy.listProcesses()
const legacyClient = (
legacy as unknown as {
client: DaemonClient
}
).client
const legacyRequest = vi.spyOn(legacyClient, 'request')
await legacy.disconnectOnly()
expect(legacyRequest.mock.calls.map(([type]) => type)).not.toContain('shutdownIfIdle')
})
it('rejects create or attach once the idle admission fence is pending', async () => {
await startServer()
const daemon = server as unknown as {
idleShutdownState: string
routeRequest(clientId: string, request: unknown): Promise<unknown>
}
daemon.idleShutdownState = 'idle-shutdown-pending'
await expect(
daemon.routeRequest('late-client', {
id: 'late-create',
type: 'createOrAttach',
payload: { sessionId: 'late', cols: 80, rows: 24 }
})
).rejects.toThrow('temporarily unavailable; reconnect')
})
it('aborts the pending shutdown when create or attach started before the fence', async () => {
await startServer()
const daemon = server as unknown as {
createOrAttachInFlight: number
idleShutdownState: string
beginIdleShutdown(): void
}
daemon.createOrAttachInFlight = 1
daemon.beginIdleShutdown()
expect(daemon.idleShutdownState).toBe('running')
expect(onIdleShutdown).not.toHaveBeenCalled()
})
it('explicitly marks a post-fence accepted transport as retryable', async () => {
await startServer()
const daemon = server as unknown as {
idleShutdownState: string
handleConnection(socket: Socket): void
}
daemon.idleShutdownState = 'idle-shutdown-pending'
const socket = new EventEmitter() as Socket
socket.end = vi.fn() as unknown as Socket['end']
daemon.handleConnection(socket)
const payload = vi.mocked(socket.end).mock.calls[0]?.[0]
expect(JSON.parse(String(payload))).toMatchObject({ ok: false, retryable: true })
socket.emit('close')
})
it('preserves a replacement PID record during otherwise successful idle cleanup', async () => {
writeFileSync(
pidPath,
serializeDaemonPidFile({ pid: process.pid, startedAtMs: null, launchNonce: 'replacement' })
)
await startServer({ launchNonce: 'mine' })
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
await client.request('shutdownIfIdle', undefined)
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
expect(JSON.parse(readFileSync(pidPath, 'utf8'))).toMatchObject({
pid: process.pid,
launchNonce: 'replacement'
})
})
it('preserves a token file replaced before idle cleanup', async () => {
await startServer()
const client = new DaemonClient({ socketPath, tokenPath })
await client.ensureConnected()
writeFileSync(tokenPath, 'replacement-token')
await client.request('shutdownIfIdle', undefined)
await waitFor(() => onIdleShutdown.mock.calls.length === 1)
expect(readFileSync(tokenPath, 'utf8')).toBe('replacement-token')
})
})
describe('daemon PID record ownership cleanup', () => {
let dir: string
let pidPath: string
beforeEach(() => {
dir = mkdtempSync(join(tmpdir(), 'daemon-pid-ownership-'))
pidPath = join(dir, 'daemon.pid')
})
afterEach(() => rmSync(dir, { recursive: true, force: true }))
it('unlinks only an exact PID and launch nonce match', () => {
writeFileSync(
pidPath,
serializeDaemonPidFile({ pid: 123, startedAtMs: null, launchNonce: 'mine' })
)
expect(unlinkOwnedDaemonPidFile(pidPath, 123, 'mine')).toBe(true)
expect(existsSync(pidPath)).toBe(false)
})
it.each([
['malformed', '{'],
['stale PID', serializeDaemonPidFile({ pid: 456, startedAtMs: null, launchNonce: 'mine' })],
[
'replacement nonce',
serializeDaemonPidFile({ pid: 123, startedAtMs: null, launchNonce: 'new' })
]
])('preserves a %s record', (_label, contents) => {
writeFileSync(pidPath, contents)
expect(unlinkOwnedDaemonPidFile(pidPath, 123, 'mine')).toBe(false)
expect(readFileSync(pidPath, 'utf8')).toBe(contents)
})
it('leaves a missing record missing', () => {
expect(unlinkOwnedDaemonPidFile(pidPath, 123, 'mine')).toBe(false)
expect(existsSync(pidPath)).toBe(false)
})
})

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@ -4,9 +4,16 @@ import type { DaemonFileLog } from './daemon-file-log'
export type DaemonStartOptions = {
socketPath: string
tokenPath: string
pidPath?: string
launchNonce?: string
startedAtMs?: number
/** Direct-construction seam for versioned protocol fixtures; never CLI/env configured. */
protocolVersion?: number
spawnSubprocess: DaemonServerOptions['spawnSubprocess']
preparePtySpawn?: DaemonServerOptions['preparePtySpawn']
log?: DaemonFileLog
onIdleShutdown?: () => void
initialAdoptionTestConfig?: DaemonServerOptions['initialAdoptionTestConfig']
}
export type DaemonHandle = {
@ -17,9 +24,17 @@ export async function startDaemon(opts: DaemonStartOptions): Promise<DaemonHandl
const server = new DaemonServer({
socketPath: opts.socketPath,
tokenPath: opts.tokenPath,
...(opts.pidPath ? { pidPath: opts.pidPath } : {}),
...(opts.launchNonce ? { launchNonce: opts.launchNonce } : {}),
...(opts.startedAtMs ? { startedAtMs: opts.startedAtMs } : {}),
...(opts.protocolVersion !== undefined ? { protocolVersion: opts.protocolVersion } : {}),
spawnSubprocess: opts.spawnSubprocess,
...(opts.preparePtySpawn ? { preparePtySpawn: opts.preparePtySpawn } : {}),
...(opts.log ? { log: opts.log } : {})
...(opts.log ? { log: opts.log } : {}),
...(opts.onIdleShutdown ? { onIdleShutdown: opts.onIdleShutdown } : {}),
...(opts.initialAdoptionTestConfig
? { initialAdoptionTestConfig: opts.initialAdoptionTestConfig }
: {})
})
await server.start()

View File

@ -11,6 +11,7 @@ import { mintPtySessionId, parsePtySessionId } from './pty-session-id'
import { supportsPtyStartupBarrier } from './shell-ready'
import { CODEX_SHELL_READY_TIMEOUT_MS } from './session'
import {
CLEAN_DISCONNECT_PROTOCOL_VERSION,
GIT_CREDENTIAL_GUARD_HOST_PROTOCOL_VERSION,
PROTOCOL_VERSION,
type CreateOrAttachResult,
@ -72,7 +73,7 @@ export type DaemonPtyAdapterOptions = {
historyPath?: string
/** Called when the daemon socket is unreachable (process died). Expected to
* fork a fresh daemon so the next connection attempt can succeed. */
respawn?: () => Promise<void>
respawn?: () => Promise<void | (() => void)>
}
const MAX_TOMBSTONES = 1000
@ -92,7 +93,9 @@ export class DaemonPtyAdapter implements IPtyProvider {
private client: DaemonClient
private historyManager: HistoryManager | null
private historyReader: HistoryReader | null
private respawnFn: (() => Promise<void>) | null
private respawnFn: (() => Promise<void | (() => void)>) | null
private pendingRespawnAdoptionRelease: (() => void) | null = null
private respawnAdoptionClosed = false
// Why: multiple pane mounts can call spawn() concurrently. If the daemon is
// dead, all calls enter withDaemonRetry's catch block at once. Without a
// lock, each would fork its own daemon process. This promise coalesces
@ -958,6 +961,8 @@ export class DaemonPtyAdapter implements IPtyProvider {
}
dispose(): void {
this.respawnAdoptionClosed = true
this.releasePendingRespawnAdoptionLease()
this.stopCheckpointTimer()
this.dirtySessionVersions.clear()
this.lastFullCheckpointAt.clear()
@ -978,6 +983,15 @@ export class DaemonPtyAdapter implements IPtyProvider {
this.client.disconnect()
}
async establishLifecycleLease(): Promise<void> {
if (this.protocolVersion < CLEAN_DISCONNECT_PROTOCOL_VERSION) {
return
}
// Why: an authenticated pair cancels the launch-adoption watchdog and gives
// a never-used adapter authority to retire its empty daemon during clean quit.
await this.client.ensureConnected()
}
// Why: for in-process daemon mode, disconnect without flushing history.
// dispose() writes endedAt for all sessions, which would prevent cold
// restore. disconnectOnly() leaves history files in unclean state so
@ -985,6 +999,8 @@ export class DaemonPtyAdapter implements IPtyProvider {
// We write a final checkpoint before disconnecting so that if the daemon
// later crashes while Orca is closed, checkpoint.json has recovery data.
async disconnectOnly(): Promise<void> {
this.respawnAdoptionClosed = true
this.releasePendingRespawnAdoptionLease()
this.stopCheckpointTimer()
// Why: wait for any in-flight timer pass to finish before starting
// the final checkpoint. Otherwise both passes race on the shared tmp
@ -1011,11 +1027,31 @@ export class DaemonPtyAdapter implements IPtyProvider {
this.producerResumesOwedOnReconnect.clear()
this.removeEventListener?.()
this.removeEventListener = null
if (this.protocolVersion >= CLEAN_DISCONNECT_PROTOCOL_VERSION) {
try {
// Why: only the authenticated daemon can atomically prove it is empty;
// one shared budget keeps a first connection plus retirement off quit's critical path.
const deadlineMs = Date.now() + 250
if (!this.client.isConnected()) {
await this.client.ensureConnectedWithin(Math.max(1, deadlineMs - Date.now()))
}
await this.client.request('shutdownIfIdle', undefined, Math.max(1, deadlineMs - Date.now()))
} catch {
// An unreachable daemon falls back to event-driven retirement when its
// authenticated sockets close and it can prove itself empty.
}
}
this.client.disconnect()
}
private async ensureConnected(): Promise<void> {
await this.client.ensureConnected()
try {
await this.client.ensureConnected()
} finally {
// Why: a respawn launcher holds a temporary full pair until this adapter
// has attempted its permanent reconnect, preventing both gaps and leaks.
this.releasePendingRespawnAdoptionLease()
}
// Why sampled before setupEventRouting: routing is (re)installed exactly
// once per connection, so "no listener yet" identifies a fresh connect —
// the only time the daemon-side backgrounded set needs a resync (it is
@ -1298,7 +1334,15 @@ export class DaemonPtyAdapter implements IPtyProvider {
try {
return await fn()
} catch (err) {
if (!this.respawnFn || !isDaemonGoneError(err)) {
// Why: self-retirement removes the token only after an authenticated
// endpoint dropped; an initial missing token may still hide a live daemon.
const missingRetiredEndpointToken =
isMissingTokenFileError(err) && this.client.hasObservedAuthenticatedDisconnect()
if (
this.respawnAdoptionClosed ||
!this.respawnFn ||
(!isDaemonGoneError(err) && !missingRetiredEndpointToken)
) {
throw err
}
if (!this.respawnPromise) {
@ -1307,7 +1351,13 @@ export class DaemonPtyAdapter implements IPtyProvider {
})
}
await this.respawnPromise
return await fn()
try {
return await fn()
} finally {
// Why: the retried operation may reject before it reaches a connection
// attempt (for example, a tombstone racing respawn).
this.releasePendingRespawnAdoptionLease()
}
}
}
@ -1371,7 +1421,20 @@ export class DaemonPtyAdapter implements IPtyProvider {
this.removeEventListener?.()
this.removeEventListener = null
this.client.disconnect()
await this.respawnFn!()
const releaseAdoptionLease = await this.respawnFn!()
if (this.respawnAdoptionClosed) {
// Why: app teardown may win while the launcher is still acquiring its
// temporary pair; a late result must not reinstall a lease nobody owns.
releaseAdoptionLease?.()
throw new Error('Daemon adapter closed during respawn')
}
this.pendingRespawnAdoptionRelease = releaseAdoptionLease ?? null
}
private releasePendingRespawnAdoptionLease(): void {
const release = this.pendingRespawnAdoptionRelease
this.pendingRespawnAdoptionRelease = null
release?.()
}
private setupEventRouting(): void {
@ -1476,5 +1539,18 @@ function isDaemonGoneError(err: unknown): boolean {
return true
}
const msg = err.message
return msg === 'Connection lost' || msg === 'Not connected' || msg === 'Hello response timed out'
return (
msg === 'Connection lost' ||
msg === 'Not connected' ||
msg === 'Hello response timed out' ||
msg === 'Daemon temporarily unavailable; reconnect'
)
}
function isMissingTokenFileError(err: unknown): boolean {
if (!(err instanceof Error)) {
return false
}
const errno = err as NodeJS.ErrnoException
return errno.code === 'ENOENT' && errno.syscall === 'open'
}

View File

@ -0,0 +1,228 @@
import { existsSync, mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { DaemonPtyAdapter } from './daemon-pty-adapter'
import { DaemonServer } from './daemon-server'
import { getDaemonSocketPath } from './daemon-spawner'
import type { SubprocessHandle } from './session'
function fixtureSubprocess(): SubprocessHandle {
let onExit: ((code: number) => void) | null = null
return {
pid: process.pid,
getForegroundProcess: () => null,
write: () => {},
resize: () => {},
kill: () => queueMicrotask(() => onExit?.(0)),
forceKill: () => queueMicrotask(() => onExit?.(137)),
signal: () => {},
onData: () => {},
onExit: (callback) => {
onExit = callback
},
dispose: () => {}
}
}
async function waitFor(predicate: () => boolean): Promise<void> {
const deadline = Date.now() + 2_000
while (!predicate()) {
if (Date.now() >= deadline) {
throw new Error('Timed out waiting for daemon disconnect')
}
await new Promise((resolve) => setTimeout(resolve, 5))
}
}
describe('daemon self-retirement respawn', () => {
let dir: string
let socketPath: string
let tokenPath: string
let server: DaemonServer | null
beforeEach(() => {
dir = mkdtempSync(join(tmpdir(), 'daemon-retirement-respawn-'))
socketPath = getDaemonSocketPath(dir)
tokenPath = join(dir, 'daemon.token')
server = null
})
afterEach(async () => {
await server?.shutdown().catch(() => {})
rmSync(dir, { recursive: true, force: true })
})
async function startServer(): Promise<DaemonServer> {
const next = new DaemonServer({
socketPath,
tokenPath,
spawnSubprocess: () => fixtureSubprocess()
})
await next.start()
server = next
return next
}
it('coalesces respawn after an authenticated endpoint removes its token', async () => {
const original = await startServer()
const respawn = vi.fn(async () => {
await startServer()
})
const adapter = new DaemonPtyAdapter({ socketPath, tokenPath, respawn })
await adapter.listProcesses()
const client = (
adapter as unknown as {
client: { hasObservedAuthenticatedDisconnect(): boolean }
}
).client
await original.shutdown()
await waitFor(() => client.hasObservedAuthenticatedDisconnect())
await Promise.all([
adapter.spawn({ sessionId: 'first', cols: 80, rows: 24 }),
adapter.spawn({ sessionId: 'second', cols: 80, rows: 24 })
])
expect(respawn).toHaveBeenCalledTimes(1)
adapter.dispose()
})
it('releases the temporary respawn lease before clean retirement', async () => {
const original = await startServer()
let temporaryAdapter: DaemonPtyAdapter | null = null
const releaseTemporaryLease = vi.fn(() => temporaryAdapter?.dispose())
const respawn = vi.fn(async () => {
await startServer()
temporaryAdapter = new DaemonPtyAdapter({ socketPath, tokenPath })
await temporaryAdapter.establishLifecycleLease()
return releaseTemporaryLease
})
const adapter = new DaemonPtyAdapter({ socketPath, tokenPath, respawn })
await adapter.listProcesses()
const client = (
adapter as unknown as {
client: { hasObservedAuthenticatedDisconnect(): boolean }
}
).client
await original.shutdown()
await waitFor(() => client.hasObservedAuthenticatedDisconnect())
await adapter.spawn({ sessionId: 'respawned', cols: 80, rows: 24 })
expect(releaseTemporaryLease).toHaveBeenCalledOnce()
await adapter.shutdown('respawned', { immediate: true })
await adapter.disconnectOnly()
await waitFor(() => !existsSync(tokenPath))
})
it('releases a temporary lease when a tombstone wins the respawn race', async () => {
const original = await startServer()
let temporaryAdapter: DaemonPtyAdapter | null = null
const releaseTemporaryLease = vi.fn(() => temporaryAdapter?.dispose())
let adapter!: DaemonPtyAdapter
const respawn = vi.fn(async () => {
await startServer()
temporaryAdapter = new DaemonPtyAdapter({ socketPath, tokenPath })
await temporaryAdapter.establishLifecycleLease()
const tombstones = (adapter as unknown as { killedSessionTombstones: Map<string, number> })
.killedSessionTombstones
tombstones.set('closed-during-respawn', Date.now())
return releaseTemporaryLease
})
adapter = new DaemonPtyAdapter({ socketPath, tokenPath, respawn })
await adapter.listProcesses()
const client = (
adapter as unknown as {
client: { hasObservedAuthenticatedDisconnect(): boolean }
}
).client
await original.shutdown()
await waitFor(() => client.hasObservedAuthenticatedDisconnect())
await expect(
adapter.spawn({ sessionId: 'closed-during-respawn', cols: 80, rows: 24 })
).rejects.toThrow('was explicitly killed')
expect(releaseTemporaryLease).toHaveBeenCalledOnce()
adapter.dispose()
})
it('releases a lease returned after disposal and does not reconnect', async () => {
const original = await startServer()
let temporaryAdapter: DaemonPtyAdapter | null = null
const releaseTemporaryLease = vi.fn(() => temporaryAdapter?.dispose())
let returnRespawnLease!: () => void
const respawn = vi.fn(async () => {
await startServer()
temporaryAdapter = new DaemonPtyAdapter({ socketPath, tokenPath })
await temporaryAdapter.establishLifecycleLease()
await new Promise<void>((resolve) => {
returnRespawnLease = resolve
})
return releaseTemporaryLease
})
const adapter = new DaemonPtyAdapter({ socketPath, tokenPath, respawn })
await adapter.listProcesses()
const client = (
adapter as unknown as {
client: { hasObservedAuthenticatedDisconnect(): boolean; isConnected(): boolean }
}
).client
await original.shutdown()
await waitFor(() => client.hasObservedAuthenticatedDisconnect())
const spawn = adapter.spawn({ sessionId: 'disposed', cols: 80, rows: 24 })
await waitFor(() => returnRespawnLease !== undefined)
adapter.dispose()
returnRespawnLease()
await expect(spawn).rejects.toThrow('closed during respawn')
expect(releaseTemporaryLease).toHaveBeenCalledOnce()
expect(client.isConnected()).toBe(false)
})
it('does not start a respawn after disposal rejects an in-flight operation', async () => {
await startServer()
const respawn = vi.fn(async () => {})
const adapter = new DaemonPtyAdapter({ socketPath, tokenPath, respawn })
await adapter.establishLifecycleLease()
let rejectCreate!: (error: Error) => void
const client = (
adapter as unknown as {
client: {
request: (method: string, params: unknown) => Promise<unknown>
}
}
).client
vi.spyOn(client, 'request').mockImplementationOnce(
() =>
new Promise((_, reject) => {
rejectCreate = reject
})
)
const spawn = adapter.spawn({ sessionId: 'disposed-in-flight', cols: 80, rows: 24 })
await waitFor(() => rejectCreate !== undefined)
adapter.dispose()
rejectCreate(new Error('Connection lost'))
await expect(spawn).rejects.toThrow('Connection lost')
expect(respawn).not.toHaveBeenCalled()
})
it('does not treat an initial missing token as respawn authority', async () => {
const respawn = vi.fn(async () => {})
const adapter = new DaemonPtyAdapter({ socketPath, tokenPath, respawn })
await expect(adapter.spawn({ sessionId: 'missing', cols: 80, rows: 24 })).rejects.toMatchObject(
{
code: 'ENOENT'
}
)
expect(respawn).not.toHaveBeenCalled()
adapter.dispose()
})
})

View File

@ -2,13 +2,13 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { connect, type Server, type Socket } from 'node:net'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { existsSync, mkdtempSync, rmSync, readFileSync } from 'node:fs'
import { existsSync, mkdtempSync, rmSync, readFileSync, writeFileSync } from 'node:fs'
import { DaemonServer } from './daemon-server'
import { DaemonClient } from './client'
import { encodeNdjson } from './ndjson'
import { PROTOCOL_VERSION, type DaemonRequest } from './types'
import type { SubprocessHandle } from './session'
import { getDaemonSocketPath } from './daemon-spawner'
import { getDaemonPidPath, getDaemonSocketPath, serializeDaemonPidFile } from './daemon-spawner'
const confirmForegroundProcessMock = vi.fn(async () => 'droid')
@ -58,6 +58,7 @@ type DaemonServerPrivate = {
clientId: string
controlSocket: Socket
streamSocket: Socket | null
authenticatedPairEstablished: boolean
}
>
routeRequest(clientId: string, request: DaemonRequest): Promise<unknown>
@ -67,6 +68,7 @@ describe('DaemonServer', () => {
let dir: string
let socketPath: string
let tokenPath: string
let pidPath: string
let server: DaemonServer
let client: DaemonClient
@ -75,6 +77,7 @@ describe('DaemonServer', () => {
dir = createTestDir()
socketPath = getDaemonSocketPath(dir)
tokenPath = join(dir, 'test.token')
pidPath = getDaemonPidPath(dir)
})
afterEach(async () => {
@ -83,10 +86,11 @@ describe('DaemonServer', () => {
rmSync(dir, { recursive: true, force: true })
})
async function startServer(): Promise<void> {
async function startServer(launchNonce?: string): Promise<void> {
server = new DaemonServer({
socketPath,
tokenPath,
...(launchNonce ? { pidPath, launchNonce } : {}),
spawnSubprocess: () => createMockSubprocess()
})
await server.start()
@ -290,12 +294,12 @@ describe('DaemonServer', () => {
spawnSubprocess
})
await server.start()
const daemon = server as unknown as DaemonServerPrivate
const c = await connectClient()
const create = daemon.routeRequest('shutdown-client', {
id: 'shutdown-create',
type: 'createOrAttach',
payload: { sessionId: 'shutdown-pending', cols: 80, rows: 24 }
const create = c.request('createOrAttach', {
sessionId: 'shutdown-pending',
cols: 80,
rows: 24
})
const canceledCreate = expect(create).rejects.toThrow(
'Attach canceled for session shutdown-pending'
@ -535,7 +539,8 @@ describe('DaemonServer', () => {
daemon.clients.set('client-1', {
clientId: 'client-1',
controlSocket,
streamSocket
streamSocket,
authenticatedPairEstablished: true
})
await daemon.routeRequest('client-1', {
@ -591,7 +596,8 @@ describe('DaemonServer', () => {
daemon.clients.set('client-1', {
clientId: 'client-1',
controlSocket,
streamSocket
streamSocket,
authenticatedPairEstablished: true
})
await daemon.routeRequest('client-1', {
@ -648,7 +654,8 @@ describe('DaemonServer', () => {
daemon.clients.set('client-1', {
clientId: 'client-1',
controlSocket,
streamSocket
streamSocket,
authenticatedPairEstablished: true
})
await daemon.routeRequest('client-1', {
id: 'req-1',
@ -780,6 +787,68 @@ describe('DaemonServer', () => {
})
describe('shutdown', () => {
it('waits for the ordinary shutdown reply write before destroying resources', async () => {
await startServer()
const c = await connectClient()
const daemon = server as unknown as DaemonServerPrivate & {
host: { dispose: () => Promise<void> }
}
const controlSocket = [...daemon.clients.values()][0].controlSocket
const originalWrite = controlSocket.write.bind(controlSocket)
let replyFlushed: (() => void) | undefined
vi.spyOn(controlSocket, 'write').mockImplementation(((
chunk: string | Uint8Array,
...args: unknown[]
) => {
replyFlushed = args.find((arg) => typeof arg === 'function') as (() => void) | undefined
return originalWrite(chunk)
}) as unknown as Socket['write'])
const dispose = vi.spyOn(daemon.host, 'dispose')
await expect(c.request('shutdown', { killSessions: false })).resolves.toEqual({})
expect(dispose).not.toHaveBeenCalled()
expect(existsSync(tokenPath)).toBe(true)
replyFlushed?.()
await waitFor(() => !existsSync(tokenPath))
expect(dispose).toHaveBeenCalledOnce()
})
it('removes only its owned token and PID record', async () => {
const launchNonce = 'ordinary-shutdown'
writeFileSync(
pidPath,
serializeDaemonPidFile({ pid: process.pid, startedAtMs: null, launchNonce })
)
await startServer(launchNonce)
await server.shutdown()
expect(existsSync(tokenPath)).toBe(false)
expect(existsSync(pidPath)).toBe(false)
})
it('preserves token and PID artifacts replaced before ordinary cleanup', async () => {
await startServer('mine')
writeFileSync(tokenPath, 'replacement-token')
writeFileSync(
pidPath,
serializeDaemonPidFile({
pid: process.pid,
startedAtMs: null,
launchNonce: 'replacement'
})
)
await server.shutdown()
expect(readFileSync(tokenPath, 'utf8')).toBe('replacement-token')
expect(JSON.parse(readFileSync(pidPath, 'utf8'))).toMatchObject({
pid: process.pid,
launchNonce: 'replacement'
})
})
it('stops accepting connections after shutdown', async () => {
await startServer()
await server.shutdown()
@ -800,9 +869,7 @@ describe('DaemonServer', () => {
)
const c = await connectClient()
// The daemon may self-terminate before the reply flushes; callers treat
// that as success, so only the observable teardown below is asserted.
await c.request('shutdown', { killSessions: true }).catch(() => {})
await expect(c.request('shutdown', { killSessions: true })).resolves.toEqual({})
await waitFor(() => daemon.server === null)
await waitFor(() => !existsSync(socketPath))

View File

@ -4,7 +4,7 @@
import { createServer, type Server, type Socket } from 'node:net'
import { randomUUID } from 'node:crypto'
import { performance } from 'node:perf_hooks'
import { writeFileSync, chmodSync, unlinkSync } from 'node:fs'
import { writeFileSync, chmodSync } from 'node:fs'
import { StringDecoder } from 'node:string_decoder'
import { encodeNdjson, createNdjsonParser } from './ndjson'
import { TerminalHost } from './terminal-host'
@ -23,7 +23,9 @@ import type { SubprocessHandle } from './session'
import { checkPtySpawnHealth } from './pty-subprocess'
import { createNoopDaemonFileLog, type DaemonFileLog } from './daemon-file-log'
import { isTuiAgent } from '../../shared/tui-agent-config'
import { unlinkOwnedDaemonPidFile, unlinkOwnedDaemonTokenFile } from './daemon-spawner'
import {
CLEAN_DISCONNECT_PROTOCOL_VERSION,
PROTOCOL_VERSION,
NOTIFY_PREFIX,
SessionNotFoundError,
@ -35,6 +37,21 @@ import {
export type DaemonServerOptions = {
socketPath: string
tokenPath: string
pidPath?: string
launchNonce?: string
startedAtMs?: number
/** Direct-construction seam for protocol fixture tests; production never overrides it. */
protocolVersion?: number
onIdleShutdown?: () => void
/** Direct-construction-only controls; production uses the compiled initial-adoption timeout. */
initialAdoptionTestConfig?: {
timeoutMs: number
clock: {
setTimeout(callback: () => void, delayMs: number): unknown
clearTimeout(handle: unknown): void
now(): number
}
}
ptySpawnHealthCheck?: () => Promise<void>
preparePtySpawn?: () => Promise<void>
log?: DaemonFileLog
@ -53,21 +70,46 @@ type ConnectedClient = {
clientId: string
controlSocket: Socket
streamSocket: Socket | null
authenticatedPairEstablished: boolean
}
type PendingPtySpawnPreparation = {
canceled: boolean
}
type PendingShutdownReply = {
start: () => void
}
export class DaemonServer {
// Why: a new daemon must survive long enough for its first client pair, but
// a parent crash between launch and adoption must not orphan it forever.
private static readonly INITIAL_ADOPTION_TIMEOUT_MS = 2 * 60 * 1000
private static readonly SHUTDOWN_REPLY_FLUSH_TIMEOUT_MS = 1_000
private server: Server | null = null
private token: string
private host: TerminalHost
private socketPath: string
private tokenPath: string
private pidPath: string | null
private launchNonce: string | null
private startedAtMs: number | null
private protocolVersion: number
private onIdleShutdown: () => void
private ptySpawnHealthCheck: () => Promise<void>
private preparePtySpawn: () => Promise<void>
private log: DaemonFileLog
private transportSockets = new Set<Socket>()
private createOrAttachInFlight = 0
private idleShutdownState: 'running' | 'idle-shutdown-pending' | 'shutting-down' = 'running'
private initialAdoptionTimer: unknown | null = null
private initialAdoptionDeadlineMs: number | null = null
private retirementRequested = false
private shutdownPromise: Promise<void> | null = null
private ordinaryShutdownServerClose: Promise<void> | null = null
private pendingShutdownReplies = new Map<string, PendingShutdownReply>()
private initialAdoptionTimeoutMs: number
private lifecycleClock: NonNullable<DaemonServerOptions['initialAdoptionTestConfig']>['clock']
private clients = new Map<string, ConnectedClient>()
private streamDataBatcher = new DaemonStreamDataBatcher(
@ -116,6 +158,28 @@ export class DaemonServer {
constructor(opts: DaemonServerOptions) {
this.socketPath = opts.socketPath
this.tokenPath = opts.tokenPath
this.pidPath = opts.pidPath ?? null
this.protocolVersion = opts.protocolVersion ?? PROTOCOL_VERSION
this.launchNonce =
opts.launchNonce ??
(this.protocolVersion >= CLEAN_DISCONNECT_PROTOCOL_VERSION ? randomUUID() : null)
this.startedAtMs =
opts.startedAtMs ??
(this.protocolVersion >= CLEAN_DISCONNECT_PROTOCOL_VERSION
? Date.now() - process.uptime() * 1000
: null)
this.onIdleShutdown = opts.onIdleShutdown ?? (() => {})
this.initialAdoptionTimeoutMs =
opts.initialAdoptionTestConfig?.timeoutMs ?? DaemonServer.INITIAL_ADOPTION_TIMEOUT_MS
this.lifecycleClock = opts.initialAdoptionTestConfig?.clock ?? {
setTimeout: (callback, delayMs) => {
const timer = setTimeout(callback, delayMs)
timer.unref()
return timer
},
clearTimeout: (handle) => clearTimeout(handle as ReturnType<typeof setTimeout>),
now: () => Date.now()
}
this.token = randomUUID()
this.host = new TerminalHost({ spawnSubprocess: opts.spawnSubprocess })
this.ptySpawnHealthCheck = opts.ptySpawnHealthCheck ?? checkPtySpawnHealth
@ -150,12 +214,48 @@ export class DaemonServer {
} catch {
// Best-effort on platforms that support it
}
if (this.protocolVersion >= CLEAN_DISCONNECT_PROTOCOL_VERSION) {
// Why: a parent crash before the first full client pair must not leave
// a freshly published, empty daemon alive forever.
this.armInitialAdoptionTimeout()
}
resolve()
})
})
}
async shutdown(): Promise<void> {
if (this.shutdownPromise) {
return this.shutdownPromise
}
const serverClose = this.beginOrdinaryShutdownFence()
this.shutdownPromise = this.finishOrdinaryShutdown(serverClose)
return this.shutdownPromise
}
private beginOrdinaryShutdownFence(): Promise<void> {
this.idleShutdownState = 'shutting-down'
this.cancelInitialAdoptionTimer()
this.ordinaryShutdownServerClose ??= this.beginServerClose()
return this.ordinaryShutdownServerClose
}
private async finishOrdinaryShutdown(serverClose: Promise<void>): Promise<void> {
this.unlinkOwnedEndpointArtifacts()
await this.disposeDaemonResources()
await serverClose
}
private unlinkOwnedEndpointArtifacts(): void {
// Why: close has already fenced this endpoint, but ownership checks still
// prevent a late replacement's canonical token or PID record from removal.
unlinkOwnedDaemonTokenFile(this.tokenPath, this.token)
if (this.pidPath && this.launchNonce) {
unlinkOwnedDaemonPidFile(this.pidPath, process.pid, this.launchNonce)
}
}
private async disposeDaemonResources(): Promise<void> {
this.stopStreamBacklogProbe()
this.transientFactRelay.dispose()
this.cancelAllPendingPtySpawnPreparations()
@ -169,29 +269,142 @@ export class DaemonServer {
})
}
this.streamDataBatcher.clear()
this.pendingShutdownReplies.clear()
for (const [, client] of this.clients) {
client.controlSocket.destroy()
client.streamSocket?.destroy()
}
this.clients.clear()
for (const socket of this.transportSockets) {
socket.destroy()
}
this.transportSockets.clear()
}
private beginServerClose(): Promise<void> {
const server = this.server
this.server = null
if (!server) {
return Promise.resolve()
}
return new Promise<void>((resolve) => {
if (this.server) {
this.server.close(() => {
try {
unlinkSync(this.socketPath)
} catch {}
resolve()
})
this.server = null
} else {
// Why: call close synchronously before any awaited cleanup so no new
// transport can enter after the idle fence is proven empty.
server.close(() => {
// Node owns unlinking its Unix listener. An extra check-then-unlink here could
// delete a replacement endpoint installed concurrently after close.
resolve()
}
})
})
}
private isIdle(): boolean {
return (
this.transportSockets.size === 0 &&
this.clients.size === 0 &&
this.createOrAttachInFlight === 0 &&
this.host.listSessions().length === 0
)
}
private reevaluateIdleShutdown(): void {
if (this.idleShutdownState !== 'running') {
return
}
if (this.retirementRequested) {
this.cancelInitialAdoptionTimer()
if (this.isIdle()) {
this.beginIdleShutdown()
}
return
}
if (!this.isIdle() || this.initialAdoptionDeadlineMs === null) {
this.cancelInitialAdoptionTimer()
return
}
if (this.initialAdoptionTimer !== null) {
return
}
const remainingMs = Math.max(0, this.initialAdoptionDeadlineMs - this.lifecycleClock.now())
if (remainingMs === 0) {
this.initialAdoptionDeadlineMs = null
this.retirementRequested = true
this.beginIdleShutdown()
return
}
this.initialAdoptionTimer = this.lifecycleClock.setTimeout(() => {
this.initialAdoptionTimer = null
this.initialAdoptionDeadlineMs = null
this.retirementRequested = true
this.beginIdleShutdown()
}, remainingMs)
}
private armInitialAdoptionTimeout(): void {
this.initialAdoptionDeadlineMs = this.lifecycleClock.now() + this.initialAdoptionTimeoutMs
this.reevaluateIdleShutdown()
}
private cancelInitialAdoptionTimer(): void {
if (this.initialAdoptionTimer === null) {
return
}
this.lifecycleClock.clearTimeout(this.initialAdoptionTimer)
this.initialAdoptionTimer = null
}
private beginIdleShutdown(): void {
this.initialAdoptionTimer = null
if (this.idleShutdownState !== 'running') {
return
}
this.idleShutdownState = 'idle-shutdown-pending'
if (!this.isIdle()) {
// Why: work admitted before the fence wins. Clearing the pending state
// keeps that already-started client/session fully usable.
this.idleShutdownState = 'running'
this.reevaluateIdleShutdown()
return
}
this.idleShutdownState = 'shutting-down'
// beginServerClose() runs synchronously up to server.close(), before host
// disposal or file cleanup can yield to a racing connection.
const serverClose = this.beginServerClose()
this.shutdownPromise = this.finishIdleShutdown(serverClose)
}
private async finishIdleShutdown(serverClose: Promise<void>): Promise<void> {
this.unlinkOwnedEndpointArtifacts()
await this.disposeDaemonResources()
await serverClose
this.onIdleShutdown()
}
private handleConnection(socket: Socket): void {
this.cancelInitialAdoptionTimer()
this.transportSockets.add(socket)
const removeTransport = (): void => {
this.transportSockets.delete(socket)
this.reevaluateIdleShutdown()
}
socket.once('close', removeTransport)
socket.on('error', () => socket.destroy())
if (this.idleShutdownState !== 'running') {
// Why: an accepted connection queued just before server.close() must get
// an explicit retry signal instead of appearing authenticated then dying.
socket.end(
encodeNdjson({
type: 'hello',
ok: false,
error: 'Daemon temporarily unavailable; reconnect',
retryable: true
})
)
return
}
// Why: clients can send multibyte prompt/input text split across socket
// chunks; keep UTF-8 sequences intact before NDJSON parsing.
const decoder = new StringDecoder('utf8')
@ -203,7 +416,6 @@ export class DaemonServer {
)
socket.on('data', (chunk) => parser.feed(decoder.write(chunk)))
socket.on('error', () => socket.destroy())
}
private handleFirstMessage(
@ -219,7 +431,7 @@ export class DaemonServer {
return
}
if (hello.version !== PROTOCOL_VERSION) {
if (hello.version !== this.protocolVersion) {
this.log.log('client-hello-rejected', {
reason: 'protocol-mismatch',
clientVersion: hello.version
@ -237,18 +449,34 @@ export class DaemonServer {
}
this.log.log('client-hello-accepted', { role: hello.role, clientId: hello.clientId })
socket.write(encodeNdjson({ type: 'hello', ok: true }))
socket.write(
encodeNdjson({
type: 'hello',
ok: true,
...(this.launchNonce && this.startedAtMs
? {
daemonIdentity: {
pid: process.pid,
startedAtMs: this.startedAtMs,
launchNonce: this.launchNonce
}
}
: {})
})
)
if (hello.role === 'control') {
const previous = this.clients.get(hello.clientId)
const client: ConnectedClient = {
clientId: hello.clientId,
controlSocket: socket,
streamSocket: null
streamSocket: null,
authenticatedPairEstablished: false
}
this.clients.set(hello.clientId, client)
this.setupControlSocket(socket, hello.clientId)
if (previous) {
this.recordFullyAuthenticatedDisconnect(previous.authenticatedPairEstablished)
// Why: a reconnect can reuse a clientId before the old sockets notice
// their close. Tear them down after installing the new owner so stale
// close events cannot delete the replacement client entry.
@ -264,6 +492,12 @@ export class DaemonServer {
return
}
this.setupStreamSocket(socket, client)
client.authenticatedPairEstablished = true
// A complete app connection, unlike a health or raw socket probe, owns
// the endpoint again and cancels pending event-driven retirement.
this.initialAdoptionDeadlineMs = null
this.retirementRequested = false
this.cancelInitialAdoptionTimer()
}
}
@ -285,12 +519,28 @@ export class DaemonServer {
if (client?.controlSocket !== socket) {
return
}
const wasFullyAuthenticated = client.authenticatedPairEstablished
this.streamDataBatcher.clear(clientId)
client.streamSocket?.destroy()
this.clients.delete(clientId)
this.recordFullyAuthenticatedDisconnect(wasFullyAuthenticated)
this.reevaluateIdleShutdown()
})
}
private recordFullyAuthenticatedDisconnect(wasFullyAuthenticated: boolean): void {
if (
!wasFullyAuthenticated ||
[...this.clients.values()].some((remaining) => remaining.authenticatedPairEstablished) ||
this.idleShutdownState !== 'running'
) {
return
}
// Why: once the last full client is gone, exact daemon-side emptiness is
// sufficient; incomplete transports may block but never erase this request.
this.retirementRequested = true
}
private setupStreamSocket(socket: Socket, client: ConnectedClient): void {
const previous = client.streamSocket
socket.removeAllListeners('data')
@ -330,7 +580,12 @@ export class DaemonServer {
try {
const result = await this.routeRequest(clientId, request)
if (!isNotify) {
socket.write(encodeNdjson({ id: request.id, ok: true, payload: result }))
const pendingShutdown = this.pendingShutdownReplies.get(
this.shutdownReplyKey(clientId, request.id)
)
socket.write(encodeNdjson({ id: request.id, ok: true, payload: result }), () => {
pendingShutdown?.start()
})
}
} catch (err) {
if (!isNotify) {
@ -345,6 +600,41 @@ export class DaemonServer {
}
}
private shutdownReplyKey(clientId: string, requestId: string): string {
return `${clientId}\u0000${requestId}`
}
private deferShutdownUntilReply(
clientId: string,
requestId: string,
socket: Socket,
finish: () => Promise<void>
): void {
const key = this.shutdownReplyKey(clientId, requestId)
let started = false
let timer: ReturnType<typeof setTimeout>
const start = (): void => {
if (started) {
return
}
started = true
clearTimeout(timer)
socket.off('close', start)
socket.off('error', start)
this.pendingShutdownReplies.delete(key)
if (!this.shutdownPromise) {
this.shutdownPromise = finish()
}
}
// Why: a non-reading authenticated peer must not pin a fenced daemon by
// holding its acknowledgement behind permanent socket backpressure.
timer = setTimeout(start, DaemonServer.SHUTDOWN_REPLY_FLUSH_TIMEOUT_MS)
timer.unref()
socket.once('close', start)
socket.once('error', start)
this.pendingShutdownReplies.set(key, { start })
}
private async preparePtySpawnUnlessCanceled(sessionId: string): Promise<void> {
const preparation: PendingPtySpawnPreparation = { canceled: false }
const pending = this.pendingPtySpawnPreparations.get(sessionId) ?? new Set()
@ -387,64 +677,80 @@ export class DaemonServer {
switch (request.type) {
case 'createOrAttach': {
if (this.idleShutdownState !== 'running') {
throw new Error('Daemon temporarily unavailable; reconnect')
}
if (!client?.authenticatedPairEstablished || client.streamSocket === null) {
// Why: a control-only replacement cannot own terminal admission or
// erase the prior full client's monotonic retirement request.
throw new Error('Daemon client connection is incomplete; reconnect')
}
this.createOrAttachInFlight++
const p = request.payload
await this.preparePtySpawnUnlessCanceled(p.sessionId)
const result = await this.host.createOrAttach({
sessionId: p.sessionId,
cols: p.cols,
rows: p.rows,
cwd: p.cwd,
env: p.env,
envToDelete: p.envToDelete,
command: p.command,
startupCommandDelivery: p.startupCommandDelivery,
// Why: daemon RPC payloads are untrusted JSON. Persist only the
// allowlisted enum used for byte routing, never arbitrary identity.
...(isTuiAgent(p.launchAgent) ? { launchAgent: p.launchAgent } : {}),
shellOverride: p.shellOverride,
terminalWindowsWslDistro: p.terminalWindowsWslDistro,
terminalWindowsPowerShellImplementation: p.terminalWindowsPowerShellImplementation,
shellReadySupported: p.shellReadySupported,
historySeed: p.historySeed,
...(p.shellReadyTimeoutMs !== undefined
? { shellReadyTimeoutMs: p.shellReadyTimeoutMs }
: {}),
streamClient: {
onData: (data) => {
// Scan BEFORE enqueue: the batcher may keep-tail drop this
// chunk, but its facts must be captured regardless.
this.transientFactRelay.onSessionData(p.sessionId, data)
const lastInputAt = this.lastInputAtBySessionId.get(p.sessionId)
const isInteractiveOutput =
data.length <= DaemonServer.INTERACTIVE_OUTPUT_MAX_CHARS &&
lastInputAt !== undefined &&
performance.now() - lastInputAt <= DaemonServer.INTERACTIVE_OUTPUT_WINDOW_MS
this.streamDataBatcher.enqueue(clientId, p.sessionId, data, {
flushImmediately: isInteractiveOutput,
flushMaxChars: DaemonServer.INTERACTIVE_OUTPUT_MAX_CHARS
})
},
onExit: (code) => {
// Why: exit tears down renderer handlers, so it must ride the
// ordered queue behind final output even when the shallow socket
// gate holds that output for a later drain pass.
this.log.log('session-exited', { sessionId: p.sessionId, code })
this.streamDataBatcher.enqueueControlEvent(clientId, p.sessionId, {
type: 'event',
event: 'exit',
sessionId: p.sessionId,
payload: { code }
})
this.streamDataBatcher.flush(clientId)
recordDaemonStreamBacklogEvent('sessionExit', {
sessionIdSuffix: p.sessionId.slice(-10)
})
this.transientFactRelay.onSessionExit(p.sessionId)
this.streamClientIdBySessionId.delete(p.sessionId)
this.lastInputAtBySessionId.delete(p.sessionId)
let result: Awaited<ReturnType<TerminalHost['createOrAttach']>>
try {
await this.preparePtySpawnUnlessCanceled(p.sessionId)
result = await this.host.createOrAttach({
sessionId: p.sessionId,
cols: p.cols,
rows: p.rows,
cwd: p.cwd,
env: p.env,
envToDelete: p.envToDelete,
command: p.command,
startupCommandDelivery: p.startupCommandDelivery,
// Why: daemon RPC payloads are untrusted JSON. Persist only the
// allowlisted enum used for byte routing, never arbitrary identity.
...(isTuiAgent(p.launchAgent) ? { launchAgent: p.launchAgent } : {}),
shellOverride: p.shellOverride,
terminalWindowsWslDistro: p.terminalWindowsWslDistro,
terminalWindowsPowerShellImplementation: p.terminalWindowsPowerShellImplementation,
shellReadySupported: p.shellReadySupported,
historySeed: p.historySeed,
...(p.shellReadyTimeoutMs !== undefined
? { shellReadyTimeoutMs: p.shellReadyTimeoutMs }
: {}),
streamClient: {
onData: (data) => {
// Scan BEFORE enqueue: the batcher may keep-tail drop this
// chunk, but its facts must be captured regardless.
this.transientFactRelay.onSessionData(p.sessionId, data)
const lastInputAt = this.lastInputAtBySessionId.get(p.sessionId)
const isInteractiveOutput =
data.length <= DaemonServer.INTERACTIVE_OUTPUT_MAX_CHARS &&
lastInputAt !== undefined &&
performance.now() - lastInputAt <= DaemonServer.INTERACTIVE_OUTPUT_WINDOW_MS
this.streamDataBatcher.enqueue(clientId, p.sessionId, data, {
flushImmediately: isInteractiveOutput,
flushMaxChars: DaemonServer.INTERACTIVE_OUTPUT_MAX_CHARS
})
},
onExit: (code) => {
// Why: exit tears down renderer handlers, so it must ride the
// ordered queue behind final output even when the shallow socket
// gate holds that output for a later drain pass.
this.log.log('session-exited', { sessionId: p.sessionId, code })
this.streamDataBatcher.enqueueControlEvent(clientId, p.sessionId, {
type: 'event',
event: 'exit',
sessionId: p.sessionId,
payload: { code }
})
this.streamDataBatcher.flush(clientId)
recordDaemonStreamBacklogEvent('sessionExit', {
sessionIdSuffix: p.sessionId.slice(-10)
})
this.transientFactRelay.onSessionExit(p.sessionId)
this.streamClientIdBySessionId.delete(p.sessionId)
this.lastInputAtBySessionId.delete(p.sessionId)
this.reevaluateIdleShutdown()
}
}
}
})
})
} finally {
this.createOrAttachInFlight--
this.reevaluateIdleShutdown()
}
this.streamClientIdBySessionId.set(p.sessionId, clientId)
// Why an attach-time marker: the adapter resyncs the background set on
// a fresh connection, which can precede this attach — main's scan
@ -599,6 +905,35 @@ export class DaemonServer {
case 'listSessions':
return { sessions: this.host.listSessions() }
case 'shutdownIfIdle': {
const authenticatedClient = this.clients.get(clientId)
const retiring =
authenticatedClient !== undefined &&
authenticatedClient.streamSocket !== null &&
this.clients.size === 1 &&
this.createOrAttachInFlight === 0 &&
this.host.listSessions().length === 0 &&
[...this.transportSockets].every(
(transport) =>
transport === authenticatedClient.controlSocket ||
transport === authenticatedClient.streamSocket
)
if (!retiring) {
return { retiring: false }
}
this.idleShutdownState = 'shutting-down'
this.initialAdoptionDeadlineMs = null
this.retirementRequested = false
this.cancelInitialAdoptionTimer()
// Why: close before acknowledging retirement so no new terminal can
// race between the empty proof and daemon disposal.
const serverClose = this.beginServerClose()
this.deferShutdownUntilReply(clientId, request.id, authenticatedClient.controlSocket, () =>
this.finishIdleShutdown(serverClose)
)
return { retiring: true }
}
case 'getSnapshot': {
const snapshotStart = performance.now()
const requestedScrollbackRows = request.payload.scrollbackRows
@ -644,11 +979,12 @@ export class DaemonServer {
await this.ptySpawnHealthCheck()
return { healthy: true }
case 'shutdown':
case 'shutdown': {
this.log.log('shutdown', {
reason: 'rpc',
killSessions: request.payload.killSessions === true
})
const serverClose = this.beginOrdinaryShutdownFence()
if (request.payload.killSessions) {
try {
await this.host.dispose()
@ -661,8 +997,16 @@ export class DaemonServer {
})
}
}
process.nextTick(() => this.shutdown())
const controlSocket = this.clients.get(clientId)?.controlSocket
if (controlSocket) {
this.deferShutdownUntilReply(clientId, request.id, controlSocket, () =>
this.finishOrdinaryShutdown(serverClose)
)
} else if (!this.shutdownPromise) {
this.shutdownPromise = this.finishOrdinaryShutdown(serverClose)
}
return {}
}
}
throw new Error(`Unknown request type: ${(request as { type: string }).type}`)
}

View File

@ -1,12 +1,13 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { mkdtempSync, rmSync } from 'node:fs'
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
import {
DaemonSpawner,
getDaemonPidPath,
getDaemonSocketPath,
getDaemonTokenPath
getDaemonTokenPath,
restoreClaimedDaemonArtifact
} from './daemon-spawner'
import { startDaemon, type DaemonHandle } from './daemon-main'
import { DaemonClient } from './client'
@ -70,6 +71,20 @@ describe('DaemonSpawner', () => {
}
describe('ensureRunning', () => {
it('passes the scoped PID path and a fresh launch nonce to the launcher', async () => {
const launcher = vi.fn(async () => ({ shutdown: vi.fn(async () => {}) }))
spawner = new DaemonSpawner({ runtimeDir: dir, launcher })
await spawner.ensureRunning()
expect(launcher).toHaveBeenCalledWith(
getDaemonSocketPath(dir),
getDaemonTokenPath(dir),
getDaemonPidPath(dir),
expect.stringMatching(/^[0-9a-f-]{36}$/)
)
})
it('uses protocol-scoped socket and token paths', () => {
const socketPath = getDaemonSocketPath(dir)
const tokenPath = getDaemonTokenPath(dir)
@ -172,3 +187,51 @@ describe('DaemonSpawner', () => {
})
})
})
describe('restoreClaimedDaemonArtifact', () => {
it('retains the unique claim when restoration fails without a replacement', () => {
expect(
restoreClaimedDaemonArtifact('/claimed', '/canonical', {
copyExclusive: () => {
throw new Error('injected ENOSPC')
},
canonicalExists: () => false
})
).toBe(false)
})
it('retains the unique claim when a failed copy leaves a partial canonical file', () => {
const restoreDir = createTestDir()
const canonicalPath = join(restoreDir, 'partial-canonical')
try {
expect(
restoreClaimedDaemonArtifact('/claimed', canonicalPath, {
copyExclusive: () => {
writeFileSync(canonicalPath, 'partial')
throw Object.assign(new Error('injected ENOSPC'), { code: 'ENOSPC' })
},
canonicalExists: () => true
})
).toBe(false)
} finally {
rmSync(restoreDir, { recursive: true, force: true })
}
})
it('allows claim cleanup after successful restore or a confirmed replacement', () => {
expect(
restoreClaimedDaemonArtifact('/claimed', '/canonical', {
copyExclusive: () => {},
canonicalExists: () => false
})
).toBe(true)
expect(
restoreClaimedDaemonArtifact('/claimed', '/canonical', {
copyExclusive: () => {
throw Object.assign(new Error('injected EEXIST'), { code: 'EEXIST' })
},
canonicalExists: () => true
})
).toBe(true)
})
})

View File

@ -1,4 +1,5 @@
import { createHash } from 'node:crypto'
import { createHash, randomUUID } from 'node:crypto'
import { constants, copyFileSync, existsSync, readFileSync, renameSync, unlinkSync } from 'node:fs'
import { join } from 'node:path'
import { PROTOCOL_VERSION } from './types'
@ -12,14 +13,21 @@ export type DaemonPidFile = {
startedAtMs: number | null
entryPath?: string
appVersion?: string
launchNonce?: string
}
export type DaemonProcessHandle = {
mode?: 'degraded-new-pty-fallback'
releaseAdoptionLease?(): void
shutdown(): Promise<void>
}
export type DaemonLauncher = (socketPath: string, tokenPath: string) => Promise<DaemonProcessHandle>
export type DaemonLauncher = (
socketPath: string,
tokenPath: string,
pidPath?: string,
launchNonce?: string
) => Promise<DaemonProcessHandle>
export type DaemonSpawnerOptions = {
runtimeDir: string
@ -32,12 +40,14 @@ export class DaemonSpawner {
private handle: DaemonProcessHandle | null = null
private socketPath: string
private tokenPath: string
private pidPath: string
constructor(opts: DaemonSpawnerOptions) {
this.runtimeDir = opts.runtimeDir
this.launcher = opts.launcher
this.socketPath = getDaemonSocketPath(this.runtimeDir)
this.tokenPath = getDaemonTokenPath(this.runtimeDir)
this.pidPath = getDaemonPidPath(this.runtimeDir)
}
async ensureRunning(): Promise<DaemonConnectionInfo> {
@ -45,7 +55,9 @@ export class DaemonSpawner {
return { socketPath: this.socketPath, tokenPath: this.tokenPath }
}
this.handle = await this.launcher(this.socketPath, this.tokenPath)
// Why: a detached daemon may clean up after its parent exits. A unique
// launch identity keeps it from deleting a replacement daemon's PID file.
this.handle = await this.launcher(this.socketPath, this.tokenPath, this.pidPath, randomUUID())
return { socketPath: this.socketPath, tokenPath: this.tokenPath }
}
@ -96,3 +108,83 @@ export function getDaemonPidPath(runtimeDir: string, protocolVersion = PROTOCOL_
export function serializeDaemonPidFile(pidFile: DaemonPidFile): string {
return JSON.stringify(pidFile)
}
export function unlinkOwnedDaemonPidFile(
pidPath: string,
expectedPid: number,
expectedLaunchNonce: string
): boolean {
return claimAndUnlinkOwnedFile(pidPath, (content) => {
try {
const parsed = JSON.parse(content) as { pid?: unknown; launchNonce?: unknown }
return parsed.pid === expectedPid && parsed.launchNonce === expectedLaunchNonce
} catch {
return false
}
})
}
export function unlinkOwnedDaemonTokenFile(tokenPath: string, expectedToken: string): boolean {
return claimAndUnlinkOwnedFile(tokenPath, (content) => content.trim() === expectedToken)
}
function claimAndUnlinkOwnedFile(
filePath: string,
ownsContent: (content: string) => boolean
): boolean {
const claimedPath = `${filePath}.cleanup-${process.pid}-${randomUUID()}`
try {
// Why: rename claims one exact directory entry before inspection, so a replacement
// installed afterward stays at the canonical path and cannot be unlinked by us.
renameSync(filePath, claimedPath)
} catch {
return false
}
try {
if (ownsContent(readFileSync(claimedPath, 'utf8'))) {
unlinkSync(claimedPath)
return true
}
} catch {
// Restore below when the claimed file cannot be validated as ours.
}
const restoredOrReplaced = restoreClaimedDaemonArtifact(claimedPath, filePath)
if (restoredOrReplaced) {
try {
unlinkSync(claimedPath)
} catch {
// A uniquely named unowned claim is safer to leave than overwriting a replacement.
}
}
return false
}
export function restoreClaimedDaemonArtifact(
claimedPath: string,
canonicalPath: string,
operations: {
copyExclusive?: (source: string, target: string) => void
canonicalExists?: (path: string) => boolean
} = {}
): boolean {
const copyExclusive =
operations.copyExclusive ??
((source: string, target: string) => copyFileSync(source, target, constants.COPYFILE_EXCL))
const canonicalExists = operations.canonicalExists ?? existsSync
try {
// Why: exclusive restore never overwrites a newer canonical replacement.
copyExclusive(claimedPath, canonicalPath)
return true
} catch (error) {
// Why: copy failures can leave a partial canonical file. Only EEXIST proves
// another owner had already installed a replacement before our copy.
return (
typeof error === 'object' &&
error !== null &&
'code' in error &&
error.code === 'EEXIST' &&
canonicalExists(canonicalPath)
)
}
}

View File

@ -1,7 +1,7 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { mkdtempSync, rmSync } from 'node:fs'
import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
import { createProductionLauncher } from './production-launcher'
import { startDaemon, type DaemonHandle } from './daemon-main'
import { DaemonClient } from './client'
@ -63,6 +63,20 @@ describe('createProductionLauncher', () => {
expect(typeof launcher).toBe('function')
})
it('rejects either ownership argument without its pair before forking', async () => {
const launcher = createProductionLauncher({
getDaemonEntryPath: () => '/fake/path.js'
})
await expect(
launcher(socketPathFor(dir), tokenPathFor(dir), join(dir, 'daemon.pid'))
).rejects.toThrow('provided together')
await expect(
launcher(socketPathFor(dir), tokenPathFor(dir), undefined, 'launch-a')
).rejects.toThrow('provided together')
expect(forkMock).not.toHaveBeenCalled()
})
it('can be used with DaemonSpawner (in-process fallback)', async () => {
// Use in-process launcher for testing (same as DaemonSpawner tests)
const launcher = async (socketPath: string, tokenPath: string) => {
@ -115,8 +129,9 @@ describe('createProductionLauncher', () => {
getDaemonEntryPath: () => join(dir, 'daemon-entry.js')
})
const launch = launcher(socketPathFor(dir), tokenPathFor(dir))
handlers.message[0]?.({ type: 'ready' })
const pidPath = join(dir, 'daemon.pid')
const launch = launcher(socketPathFor(dir), tokenPathFor(dir), pidPath, 'launch-a')
handlers.message[0]?.({ type: 'ready', startedAtMs: 123_456 })
const handle = await launch
expect(handle.shutdown).toEqual(expect.any(Function))
@ -125,9 +140,73 @@ describe('createProductionLauncher', () => {
expect(handlers.exit).toHaveLength(0)
expect(child.disconnect).toHaveBeenCalled()
expect(child.unref).toHaveBeenCalled()
expect(JSON.parse(readFileSync(pidPath, 'utf8'))).toEqual({
pid: 12345,
startedAtMs: 123_456,
entryPath: join(dir, 'daemon-entry.js'),
launchNonce: 'launch-a'
})
expect(forkMock).toHaveBeenCalledWith(
join(dir, 'daemon-entry.js'),
expect.arrayContaining(['--pid-record', pidPath, '--launch-nonce', 'launch-a']),
expect.objectContaining({ stdio: ['ignore', 'ignore', 'ignore', 'ipc'] })
)
})
it('removes shutdown exit listener when force-kill timeout settles first', async () => {
it('resolves shutdown only after observing child exit', async () => {
const handlers: Record<string, ((arg?: unknown) => void)[]> = {
message: [],
error: [],
exit: []
}
const child = {
pid: 12345,
connected: true,
exitCode: null as number | null,
signalCode: null as NodeJS.Signals | null,
on: vi.fn((event: string, callback: (arg?: unknown) => void) => {
handlers[event]?.push(callback)
return child
}),
once: vi.fn((event: string, callback: (arg?: unknown) => void) => {
handlers[event]?.push(callback)
return child
}),
off: vi.fn((event: string, callback: (arg?: unknown) => void) => {
handlers[event] = handlers[event]?.filter((handler) => handler !== callback) ?? []
return child
}),
kill: vi.fn((signal: NodeJS.Signals) => {
if (signal === 'SIGTERM') {
queueMicrotask(() => {
child.exitCode = 0
for (const callback of handlers.exit.slice()) {
callback(0)
}
})
}
return true
}),
disconnect: vi.fn(() => {
child.connected = false
}),
unref: vi.fn()
}
forkMock.mockReturnValueOnce(child)
const launcher = createProductionLauncher({
getDaemonEntryPath: () => join(dir, 'daemon-entry.js')
})
const launch = launcher(socketPathFor(dir), tokenPathFor(dir))
handlers.message[0]?.({ type: 'ready', startedAtMs: 123_456 })
const handle = await launch
await expect(handle.shutdown()).resolves.toBeUndefined()
expect(child.kill).toHaveBeenCalledWith('SIGTERM')
expect(child.exitCode).toBe(0)
})
it('rejects shutdown and releases child handles when SIGKILL never produces exit', async () => {
vi.useFakeTimers()
try {
const handlers: Record<string, ((arg?: unknown) => void)[]> = {
@ -138,6 +217,9 @@ describe('createProductionLauncher', () => {
const child = {
pid: 12345,
killed: false,
connected: false,
exitCode: null,
signalCode: null,
on: vi.fn((event: string, cb: (arg?: unknown) => void) => {
handlers[event]?.push(cb)
return child
@ -150,8 +232,10 @@ describe('createProductionLauncher', () => {
handlers[event] = handlers[event]?.filter((handler) => handler !== cb) ?? []
return child
}),
kill: vi.fn(),
disconnect: vi.fn(),
kill: vi.fn(() => true),
disconnect: vi.fn(() => {
child.connected = false
}),
unref: vi.fn()
}
forkMock.mockReturnValueOnce(child)
@ -161,22 +245,135 @@ describe('createProductionLauncher', () => {
})
const launch = launcher(socketPathFor(dir), tokenPathFor(dir))
handlers.message[0]?.({ type: 'ready' })
handlers.message[0]?.({ type: 'ready', startedAtMs: 123_456 })
const handle = await launch
const shutdown = handle.shutdown()
const shutdown = expect(handle.shutdown()).rejects.toThrow(
'Daemon did not exit after SIGKILL'
)
expect(handlers.exit).toHaveLength(1)
await vi.advanceTimersByTimeAsync(5000)
await vi.advanceTimersByTimeAsync(6000)
await shutdown
expect(child.kill).toHaveBeenNthCalledWith(1, 'SIGTERM')
expect(child.kill).toHaveBeenNthCalledWith(2, 'SIGKILL')
expect(handlers.exit).toHaveLength(0)
expect(child.unref).toHaveBeenCalledTimes(2)
} finally {
vi.useRealTimers()
}
})
it('preserves readiness and signaling failures while releasing startup IPC', async () => {
vi.useFakeTimers()
try {
const handlers: Record<string, ((arg?: unknown) => void)[]> = {
message: [],
error: [],
exit: []
}
const signalError = Object.assign(new Error('permission denied'), { code: 'EACCES' })
const child = {
pid: 12345,
connected: true,
exitCode: null,
signalCode: null,
on: vi.fn((event: string, cb: (arg?: unknown) => void) => {
handlers[event]?.push(cb)
return child
}),
once: vi.fn((event: string, cb: (arg?: unknown) => void) => {
handlers[event]?.push(cb)
return child
}),
off: vi.fn((event: string, cb: (arg?: unknown) => void) => {
handlers[event] = handlers[event]?.filter((handler) => handler !== cb) ?? []
return child
}),
kill: vi.fn(() => {
throw signalError
}),
disconnect: vi.fn(() => {
child.connected = false
}),
unref: vi.fn()
}
forkMock.mockReturnValueOnce(child)
const launcher = createProductionLauncher({
getDaemonEntryPath: () => join(dir, 'daemon-entry.js')
})
const launch = launcher(socketPathFor(dir), tokenPathFor(dir))
handlers.message[0]?.({ type: 'ready' })
const error = await launch.catch((caught: unknown) => caught)
expect(error).toBeInstanceOf(AggregateError)
expect((error as AggregateError).errors).toEqual([
expect.objectContaining({ message: 'Daemon readiness identity is incomplete' }),
signalError
])
expect(child.disconnect).toHaveBeenCalledOnce()
expect(child.unref).toHaveBeenCalledOnce()
expect(handlers.message).toHaveLength(0)
expect(handlers.error).toHaveLength(0)
expect(handlers.exit).toHaveLength(0)
} finally {
vi.useRealTimers()
}
})
it('preserves PID publication and cleanup failures', async () => {
const handlers: Record<string, ((arg?: unknown) => void)[]> = {
message: [],
error: [],
exit: []
}
const signalError = Object.assign(new Error('signal blocked'), { code: 'EPERM' })
const child = {
pid: 12345,
connected: true,
exitCode: null,
signalCode: null,
on: vi.fn((event: string, cb: (arg?: unknown) => void) => {
handlers[event]?.push(cb)
return child
}),
once: vi.fn((event: string, cb: (arg?: unknown) => void) => {
handlers[event]?.push(cb)
return child
}),
off: vi.fn((event: string, cb: (arg?: unknown) => void) => {
handlers[event] = handlers[event]?.filter((handler) => handler !== cb) ?? []
return child
}),
kill: vi.fn(() => {
throw signalError
}),
disconnect: vi.fn(() => {
child.connected = false
}),
unref: vi.fn()
}
forkMock.mockReturnValueOnce(child)
const pidPath = join(dir, 'occupied.pid')
writeFileSync(pidPath, 'occupied')
const launcher = createProductionLauncher({
getDaemonEntryPath: () => join(dir, 'daemon-entry.js')
})
const launch = launcher(socketPathFor(dir), tokenPathFor(dir), pidPath, 'launch-b')
handlers.message[0]?.({ type: 'ready', startedAtMs: 123_456 })
const error = await launch.catch((caught: unknown) => caught)
expect(error).toBeInstanceOf(AggregateError)
expect((error as AggregateError).errors).toEqual([
expect.objectContaining({ code: 'EEXIST' }),
signalError
])
expect(child.disconnect).toHaveBeenCalledOnce()
expect(child.unref).toHaveBeenCalledOnce()
})
})
function socketPathFor(dir: string): string {

View File

@ -1,5 +1,10 @@
import { fork, type ChildProcess } from 'node:child_process'
import type { DaemonLauncher, DaemonProcessHandle } from './daemon-spawner'
import { writeFileSync } from 'node:fs'
import {
serializeDaemonPidFile,
type DaemonLauncher,
type DaemonProcessHandle
} from './daemon-spawner'
const READY_TIMEOUT_MS = 10_000
@ -8,17 +13,63 @@ export type ProductionLauncherOptions = {
}
export function createProductionLauncher(opts: ProductionLauncherOptions): DaemonLauncher {
return async (socketPath: string, tokenPath: string): Promise<DaemonProcessHandle> => {
return async (
socketPath: string,
tokenPath: string,
pidPath?: string,
launchNonce?: string
): Promise<DaemonProcessHandle> => {
if ((pidPath === undefined) !== (launchNonce === undefined)) {
// Why: partial ownership metadata would launch a v24 daemon that cannot
// prove which PID record it may remove during self-retirement.
throw new Error('Daemon PID path and launch nonce must be provided together')
}
const entryPath = opts.getDaemonEntryPath()
const child = fork(entryPath, ['--socket', socketPath, '--token', tokenPath], {
stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
detached: true,
env: { ...process.env },
...(process.platform === 'win32' ? { windowsHide: true } : {})
})
const child = fork(
entryPath,
[
'--socket',
socketPath,
'--token',
tokenPath,
...(pidPath && launchNonce ? ['--pid-record', pidPath, '--launch-nonce', launchNonce] : [])
],
{
// Why: detached daemon output is not consumed; ignored streams cannot
// keep Electron alive after the child and IPC channel are unreferenced.
stdio: ['ignore', 'ignore', 'ignore', 'ipc'],
detached: true,
env: { ...process.env },
...(process.platform === 'win32' ? { windowsHide: true } : {})
}
)
await waitForReady(child)
let startedAtMs: number
try {
startedAtMs = await waitForReady(child)
} catch (error) {
return rejectAfterChildCleanup(child, error)
}
if (pidPath && launchNonce) {
if (!Number.isSafeInteger(child.pid) || (child.pid as number) <= 0) {
return rejectAfterChildCleanup(child, new Error('Daemon readiness identity is incomplete'))
}
try {
writeFileSync(
pidPath,
serializeDaemonPidFile({
pid: child.pid as number,
startedAtMs,
entryPath,
launchNonce
}),
{ mode: 0o600, flag: 'wx' }
)
} catch (error) {
return rejectAfterChildCleanup(child, error)
}
}
// Unref so the Electron process can exit without waiting for the daemon
child.unref()
@ -30,7 +81,7 @@ export function createProductionLauncher(opts: ProductionLauncherOptions): Daemo
}
}
function waitForReady(child: ChildProcess): Promise<void> {
function waitForReady(child: ChildProcess): Promise<number> {
return new Promise((resolve, reject) => {
let timeout: ReturnType<typeof setTimeout> | undefined
let settled = false
@ -42,15 +93,12 @@ function waitForReady(child: ChildProcess): Promise<void> {
child.off('error', onError)
child.off('exit', onExit)
}
function fail(error: Error, killChild = false): void {
function fail(error: Error): void {
if (settled) {
return
}
settled = true
cleanupStartupListeners()
if (killChild) {
child.kill('SIGTERM')
}
reject(error)
}
function onMessage(msg: unknown): void {
@ -58,11 +106,16 @@ function waitForReady(child: ChildProcess): Promise<void> {
if (settled) {
return
}
const startedAtMs = (msg as { startedAtMs?: unknown }).startedAtMs
if (typeof startedAtMs !== 'number' || !Number.isFinite(startedAtMs) || startedAtMs <= 0) {
fail(new Error('Daemon readiness identity is incomplete'))
return
}
settled = true
// Why: the daemon is detached after readiness, so startup listeners
// must not keep the child process closure alive for the daemon lifetime.
cleanupStartupListeners()
resolve()
resolve(startedAtMs)
}
}
function onError(err: Error): void {
@ -73,7 +126,7 @@ function waitForReady(child: ChildProcess): Promise<void> {
}
timeout = setTimeout(() => {
fail(new Error('Daemon failed to signal readiness within timeout'), true)
fail(new Error('Daemon failed to signal readiness within timeout'))
}, READY_TIMEOUT_MS)
child.on('message', onMessage)
@ -82,35 +135,88 @@ function waitForReady(child: ChildProcess): Promise<void> {
})
}
function shutdownChild(child: ChildProcess): Promise<void> {
return new Promise((resolve) => {
if (child.killed) {
resolve()
async function shutdownChild(child: ChildProcess): Promise<void> {
try {
if (
(child.exitCode !== null && child.exitCode !== undefined) ||
(child.signalCode !== null && child.signalCode !== undefined)
) {
return
}
let settled = false
let timeout: ReturnType<typeof setTimeout>
function finish(): void {
if (settled) {
return
await new Promise<void>((resolve, reject) => {
let settled = false
let timeout: ReturnType<typeof setTimeout>
let forceTimeout: ReturnType<typeof setTimeout> | undefined
function finish(error?: unknown): void {
if (settled) {
return
}
settled = true
clearTimeout(timeout)
if (forceTimeout) {
clearTimeout(forceTimeout)
}
child.off('exit', onExit)
if (error) {
reject(error)
} else {
resolve()
}
}
settled = true
clearTimeout(timeout)
child.off('exit', onExit)
resolve()
function onExit(): void {
finish()
}
timeout = setTimeout(() => {
try {
if (child.kill('SIGKILL') === false) {
finish(new Error('Failed to deliver SIGKILL to daemon'))
return
}
} catch (error) {
finish(isNoSuchProcessError(error) ? undefined : error)
return
}
// Why: signal delivery is not process exit; keep the listener for one
// bounded interval before releasing launcher-owned handles.
if (!settled) {
forceTimeout = setTimeout(
() => finish(new Error('Daemon did not exit after SIGKILL')),
1000
)
}
}, 5000)
child.once('exit', onExit)
try {
if (child.kill('SIGTERM') === false) {
finish(new Error('Failed to deliver SIGTERM to daemon'))
}
} catch (error) {
finish(isNoSuchProcessError(error) ? undefined : error)
}
})
} finally {
if (child.connected) {
child.disconnect()
}
function onExit(): void {
finish()
}
timeout = setTimeout(() => {
child.kill('SIGKILL')
finish()
}, 5000)
child.once('exit', onExit)
child.kill('SIGTERM')
})
child.unref()
}
}
async function rejectAfterChildCleanup(child: ChildProcess, launchError: unknown): Promise<never> {
try {
await shutdownChild(child)
} catch (cleanupError) {
throw new AggregateError(
[launchError, cleanupError],
'Daemon launch and child cleanup both failed'
)
}
throw launchError
}
function isNoSuchProcessError(error: unknown): boolean {
return typeof error === 'object' && error !== null && 'code' in error && error.code === 'ESRCH'
}

View File

@ -16,10 +16,11 @@ import type { TuiAgent } from '../../shared/types'
// when daemon-baked behavior cannot be delivered by on-disk wrapper refresh.
// Why: bump when adding daemon wire behavior so same-version old daemons do
// not silently accept the handshake and then reject new RPCs.
export const PROTOCOL_VERSION = 23
export const PROTOCOL_VERSION = 24
export const GIT_CREDENTIAL_GUARD_HOST_PROTOCOL_VERSION = 22
export const CLEAN_DISCONNECT_PROTOCOL_VERSION = 24
export const PREVIOUS_DAEMON_PROTOCOL_VERSIONS = [
1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22
1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23
] as const
// ─── Session State Machine ──────────────────────────────────────────
@ -80,19 +81,7 @@ export type { TerminalCheckpointFile } from './daemon-checkpoint-file'
// ─── NDJSON Protocol Messages ───────────────────────────────────────
// Hello handshake (first message on each socket)
export type HelloMessage = {
type: 'hello'
version: number
token: string
clientId: string
role: 'control' | 'stream'
}
export type HelloResponse = {
type: 'hello'
ok: boolean
error?: string
}
export type { DaemonEndpointIdentity, HelloMessage, HelloResponse } from './daemon-hello-protocol'
// ─── RPC Requests (Client → Daemon, on control socket) ─────────────
@ -211,6 +200,11 @@ export type ListSessionsRequest = {
type: 'listSessions'
}
export type ShutdownIfIdleRequest = {
id: string
type: 'shutdownIfIdle'
}
export type DetachRequest = {
id: string
type: 'detach'
@ -330,6 +324,7 @@ export type DaemonRequest =
| KillRequest
| SignalRequest
| ListSessionsRequest
| ShutdownIfIdleRequest
| DetachRequest
| GetCwdRequest
| GetForegroundProcessRequest
@ -368,6 +363,10 @@ export type ListSessionsResult = {
sessions: SessionInfo[]
}
export type ShutdownIfIdleResult = {
retiring: boolean
}
export type SystemResolverHealth = 'healthy' | 'unhealthy' | 'unknown'
export type SystemResolverHealthResult = {

View File

@ -0,0 +1,241 @@
import { fork, type ChildProcess } from 'node:child_process'
import { randomUUID } from 'node:crypto'
import { existsSync, mkdirSync, mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import path from 'node:path'
import { build } from 'esbuild'
import { expect, test } from '@playwright/test'
import { DaemonClient } from '../../src/main/daemon/client'
import { DaemonPtyAdapter } from '../../src/main/daemon/daemon-pty-adapter'
import {
getDaemonPidPath,
getDaemonSocketPath,
getDaemonTokenPath
} from '../../src/main/daemon/daemon-spawner'
import { PROTOCOL_VERSION } from '../../src/main/daemon/types'
type FixtureDaemon = {
child: ChildProcess
protocolVersion: number
socketPath: string
tokenPath: string
pidPath: string
}
async function waitFor(
label: string,
predicate: () => boolean | Promise<boolean>,
timeoutMs = 10_000
): Promise<void> {
const deadline = Date.now() + timeoutMs
while (!(await predicate())) {
if (Date.now() >= deadline) {
throw new Error(`Timed out waiting for ${label}`)
}
await new Promise((resolve) => setTimeout(resolve, 25))
}
}
async function launchFixture(
entryPath: string,
daemonDir: string,
protocolVersion: number
): Promise<FixtureDaemon> {
const socketPath = getDaemonSocketPath(daemonDir, protocolVersion)
const tokenPath = getDaemonTokenPath(daemonDir, protocolVersion)
const pidPath = getDaemonPidPath(daemonDir, protocolVersion)
const launchNonce = randomUUID()
const child = fork(
entryPath,
[
'--protocol',
String(protocolVersion),
'--socket',
socketPath,
'--token',
tokenPath,
...(protocolVersion >= PROTOCOL_VERSION
? ['--pid-record', pidPath, '--launch-nonce', launchNonce]
: [])
],
{
stdio: ['ignore', 'ignore', 'pipe', 'ipc'],
env: {
...process.env,
NODE_PATH: [path.join(process.cwd(), 'node_modules'), process.env.NODE_PATH]
.filter(Boolean)
.join(path.delimiter)
}
}
)
let stderr = ''
child.stderr?.on('data', (chunk) => {
stderr = `${stderr}${String(chunk)}`.slice(-8_192)
})
try {
await new Promise<void>((resolve, reject) => {
const cleanup = (): void => {
clearTimeout(timeout)
child.off('message', onMessage)
child.off('error', onError)
child.off('exit', onExit)
}
const onMessage = (message: unknown): void => {
if ((message as { type?: unknown }).type !== 'ready') {
return
}
cleanup()
resolve()
}
const onError = (error: Error): void => {
cleanup()
reject(error)
}
const onExit = (code: number | null): void => {
cleanup()
reject(new Error(`Lifecycle fixture exited with ${code}: ${stderr.trim()}`))
}
const timeout = setTimeout(() => {
cleanup()
reject(new Error('Lifecycle fixture startup timed out'))
}, 10_000)
child.on('message', onMessage)
child.on('error', onError)
child.on('exit', onExit)
})
} catch (error) {
try {
await stopChild(child, `fixture v${protocolVersion}`)
} catch (cleanupError) {
throw new AggregateError([error, cleanupError], 'Fixture startup and cleanup both failed')
}
throw error
}
return { child, protocolVersion, socketPath, tokenPath, pidPath }
}
async function stopChild(child: ChildProcess, label: string): Promise<void> {
if (child.exitCode !== null || child.signalCode !== null) {
return
}
child.kill('SIGTERM')
try {
await waitFor(
`${label} graceful exit`,
() => child.exitCode !== null || child.signalCode !== null,
3_000
)
return
} catch {
child.kill('SIGKILL')
}
await waitFor(
`${label} forced exit`,
() => child.exitCode !== null || child.signalCode !== null,
3_000
)
}
async function stopFixture(fixture: FixtureDaemon): Promise<void> {
await stopChild(fixture.child, `fixture v${fixture.protocolVersion}`)
}
test('v22 stays reattachable while v24 retires after its last empty client disconnects', async () => {
const rootDir = mkdtempSync(path.join(tmpdir(), 'orca-daemon-lifecycle-'))
const daemonDir = path.join(rootDir, 'daemon')
mkdirSync(daemonDir, { recursive: true })
const entryPath = path.join(rootDir, 'daemon-lifecycle-entry.cjs')
const fixtures: FixtureDaemon[] = []
let testError: unknown
try {
await build({
entryPoints: [path.join(process.cwd(), 'tests/e2e/fixtures/daemon-lifecycle-entry.ts')],
outfile: entryPath,
bundle: true,
platform: 'node',
format: 'cjs',
target: 'node20',
external: ['node-pty'],
logLevel: 'silent'
})
const legacy = await launchFixture(entryPath, daemonDir, 22)
fixtures.push(legacy)
const legacyClient = new DaemonClient({
socketPath: legacy.socketPath,
tokenPath: legacy.tokenPath,
protocolVersion: 22
})
await legacyClient.ensureConnected()
await expect(
legacyClient.request('createOrAttach', { sessionId: 'legacy-live', cols: 80, rows: 24 })
).resolves.toMatchObject({ isNew: true })
legacyClient.disconnect()
const current = await launchFixture(entryPath, daemonDir, PROTOCOL_VERSION)
fixtures.push(current)
const reattachClient = new DaemonClient({
socketPath: legacy.socketPath,
tokenPath: legacy.tokenPath,
protocolVersion: 22
})
await reattachClient.ensureConnected()
await expect(
reattachClient.request('createOrAttach', {
sessionId: 'legacy-live',
cols: 80,
rows: 24
})
).resolves.toMatchObject({ isNew: false })
reattachClient.disconnect()
const currentAdapter = new DaemonPtyAdapter({
socketPath: current.socketPath,
tokenPath: current.tokenPath
})
const secondCurrentClient = new DaemonClient({
socketPath: current.socketPath,
tokenPath: current.tokenPath
})
await secondCurrentClient.ensureConnected()
// Why: failed adoption may overlap another authenticated app client, so
// only daemon-owned idle retirement can safely decide whether to exit.
await currentAdapter.disconnectOnly()
expect(current.child.exitCode).toBeNull()
await expect(secondCurrentClient.request('listSessions', undefined)).resolves.toEqual({
sessions: []
})
secondCurrentClient.disconnect()
await waitFor('v24 process exit', () => current.child.exitCode !== null)
expect(existsSync(current.tokenPath)).toBe(false)
expect(existsSync(current.pidPath)).toBe(false)
if (process.platform !== 'win32') {
expect(existsSync(current.socketPath)).toBe(false)
}
expect(legacy.child.exitCode).toBeNull()
} catch (error) {
testError = error
}
const results = await Promise.allSettled(fixtures.map((fixture) => stopFixture(fixture)))
const cleanupErrors = results.flatMap((result) =>
result.status === 'rejected' ? [result.reason] : []
)
try {
rmSync(rootDir, { recursive: true, force: true })
} catch (error) {
cleanupErrors.push(error)
}
if (testError !== undefined && cleanupErrors.length > 0) {
throw new AggregateError([testError, ...cleanupErrors], 'Lifecycle test and cleanup failed')
}
if (testError !== undefined) {
throw testError
}
if (cleanupErrors.length > 0) {
throw new AggregateError(cleanupErrors, 'Lifecycle fixture cleanup failed')
}
})

View File

@ -0,0 +1,122 @@
import process from 'node:process'
import { writeFileSync } from 'node:fs'
import { startDaemon, type DaemonHandle } from '../../../src/main/daemon/daemon-main'
import { serializeDaemonPidFile } from '../../../src/main/daemon/daemon-spawner'
import type { SubprocessHandle } from '../../../src/main/daemon/session'
type FixtureArgs = {
protocolVersion: number
socketPath: string
tokenPath: string
pidPath?: string
launchNonce?: string
}
function parseArgs(argv: string[]): FixtureArgs {
const values = new Map<string, string>()
for (let index = 0; index < argv.length; index += 2) {
const key = argv[index]
const value = argv[index + 1]
if (!key || !value) {
throw new Error('Lifecycle fixture arguments must be key/value pairs')
}
values.set(key, value)
}
const protocolVersion = Number(values.get('--protocol'))
const socketPath = values.get('--socket')
const tokenPath = values.get('--token')
const pidPath = values.get('--pid-record')
const launchNonce = values.get('--launch-nonce')
if (
!Number.isInteger(protocolVersion) ||
protocolVersion < 1 ||
!socketPath ||
!tokenPath ||
Boolean(pidPath) !== Boolean(launchNonce)
) {
throw new Error('Invalid lifecycle fixture arguments')
}
return {
protocolVersion,
socketPath,
tokenPath,
...(pidPath && launchNonce ? { pidPath, launchNonce } : {})
}
}
function createFixtureSubprocess(): SubprocessHandle {
let onData: ((data: string) => void) | null = null
let onExit: ((code: number) => void) | null = null
let exited = false
const exit = (code: number): void => {
if (exited) {
return
}
exited = true
onExit?.(code)
}
return {
pid: process.pid,
getForegroundProcess: () => null,
write: (data) => onData?.(data),
resize: () => {},
kill: () => exit(0),
forceKill: () => exit(137),
signal: () => {},
onData: (callback) => {
onData = callback
},
onExit: (callback) => {
onExit = callback
},
dispose: () => exit(0)
}
}
async function main(): Promise<void> {
const args = parseArgs(process.argv.slice(2))
const startedAtMs = Date.now() - process.uptime() * 1000
let daemon: DaemonHandle | null = await startDaemon({
protocolVersion: args.protocolVersion,
socketPath: args.socketPath,
tokenPath: args.tokenPath,
...(args.pidPath ? { pidPath: args.pidPath } : {}),
...(args.launchNonce ? { launchNonce: args.launchNonce } : {}),
startedAtMs,
spawnSubprocess: () => createFixtureSubprocess(),
onIdleShutdown: () => process.exit(0)
})
if (args.pidPath && args.launchNonce) {
writeFileSync(
args.pidPath,
serializeDaemonPidFile({
pid: process.pid,
startedAtMs,
launchNonce: args.launchNonce
}),
{ mode: 0o600, flag: 'wx' }
)
}
let shuttingDown = false
const shutdown = async (): Promise<void> => {
if (shuttingDown) {
return
}
shuttingDown = true
try {
await daemon?.shutdown()
daemon = null
} finally {
process.exit(0)
}
}
process.on('SIGTERM', () => void shutdown())
process.on('SIGINT', () => void shutdown())
process.send?.({ type: 'ready', pid: process.pid, startedAtMs })
}
void main().catch((error) => {
console.error(error)
process.exit(1)
})