diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts index db8b3c16d4..5d0fb4e4fb 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts @@ -511,7 +511,8 @@ test('fences transcript range failures to the current registration and Host sour }); const request = (consumerId: string, generation: string) => ({ consumerId, - generation, + sessionId: 'session-1', + hostEpoch: `host-${generation}`, anchorSequence: null, maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES, }); @@ -620,9 +621,10 @@ test('fences transcript range failures across same-source replica recovery', asy const observations = new RuntimeHostSessionObservationRegistry(); const batches: DesktopTranscriptBatch[] = []; const consumerId = 'consumer-replica-recovery'; - const request = (generation: string) => ({ + const request = (hostEpoch: string) => ({ consumerId, - generation, + sessionId: 'session-1', + hostEpoch, anchorSequence: 1, maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES, }); @@ -644,7 +646,7 @@ test('fences transcript range failures across same-source replica recovery', asy }; await observations.attach(observer); const opened = await observations.openTranscript('session-1', consumerId, target); - const staleLoad = observations.loadTranscriptBefore(request(opened.generation), target.id); + const staleLoad = observations.loadTranscriptBefore(request(opened.hostEpoch), target.id); await waitFor(() => staleRangeStarted); firstEvents.push({ @@ -658,9 +660,8 @@ test('fences transcript range failures across same-source replica recovery', asy staleRange.reject(new Error('stale replica rejected its range')); await assert.doesNotReject(staleLoad); - const generation = batches.at(-1)!.generation; await assert.rejects( - observations.loadTranscriptBefore(request(generation), target.id), + observations.loadTranscriptBefore(request(opened.hostEpoch), target.id), (error) => error === currentFailure, ); await observations.close(); @@ -861,7 +862,7 @@ test('keeps a bounded transcript batch window in flight until the renderer ackno await observer.close(); }); -test('finishes transcript open against a replacement that arrives while reset delivery waits', async () => { +test('finishes transcript open and replays a stale range request after replacement', async () => { const firstEvents = new AsyncFrameQueue(); const secondEvents = new AsyncFrameQueue(); const message: StoredMessage = { @@ -873,6 +874,8 @@ test('finishes transcript open against a replacement that arrives while reset de modelId: 'test-model', }; let opens = 0; + let rangeLoads = 0; + const requestedAnchors: Array = []; const observer = new RuntimeHostSessionObserver({ client: { openSession: async () => { @@ -882,6 +885,49 @@ test('finishes transcript open against a replacement that arrives while reset de snapshot: continuitySnapshot(), transcript: Promise.resolve([message]), events, + transcriptBootstrap: { + throughSequence: 0, + overlayMessageCount: 0, + durable: { + kind: 'page', + sessionId: 'session-1', + source: 'durable', + direction: 'older', + throughSequence: 0, + rawBytes: 1, + fragments: [], + nextCursor: 'older', + }, + overlay: { + kind: 'page', + sessionId: 'session-1', + source: 'overlay', + direction: 'older', + throughSequence: null, + rawBytes: 0, + fragments: [], + nextCursor: null, + }, + }, + loadTranscriptOverlay: async () => [], + decodeTranscriptPage: async (page) => ({ + messages: page.rawBytes === 1 ? [{ identity: 0, message }] : [], + nextCursor: page.nextCursor, + }), + loadTranscriptPage: async (input) => { + rangeLoads += 1; + requestedAnchors.push(input.anchorSequence); + return { + kind: 'page', + sessionId: 'session-1', + source: input.source, + direction: input.direction, + throughSequence: input.throughSequence, + rawBytes: 0, + fragments: [], + nextCursor: null, + }; + }, async close() { events.end(); }, @@ -891,10 +937,21 @@ test('finishes transcript open against a replacement that arrives while reset de emitSessionsChanged() {}, }); const batches: DesktopTranscriptBatch[] = []; + let autoAcknowledge = false; const opening = observer.openTranscript('session-1', 'consumer-recovery', { id: 22, send(_channel, batch) { batches.push(batch); + if (autoAcknowledge) { + queueMicrotask(() => + observer.acknowledgeTranscript( + 'consumer-recovery', + batch.generation, + batch.deliverySequence, + 22, + ), + ); + } }, once() {}, off() {}, @@ -905,6 +962,7 @@ test('finishes transcript open against a replacement that arrives while reset de ); await waitFor(() => batches.length === 4); + const staleGeneration = batches[0]!.generation; firstEvents.push({ kind: 'subscription.closed', hostEpoch: 'host-1', @@ -937,6 +995,58 @@ test('finishes transcript open against a replacement that arrives while reset de const opened = await result; assert.equal(opened.error, undefined); assert.equal(opened.value?.generation, batches.at(-1)?.generation); + assert.notEqual(opened.value?.generation, staleGeneration); + rangeLoads = 0; + requestedAnchors.length = 0; + autoAcknowledge = true; + // The renderer dispatched this range request before the replacement replica + // was installed; the same Session and Host epoch continue the read against + // the current replica, and the requested slice is what the replica loads. + await assert.doesNotReject(() => + observer.loadTranscriptBefore( + { + consumerId: 'consumer-recovery', + sessionId: 'session-1', + hostEpoch: 'host-1', + anchorSequence: 0, + maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES, + }, + 22, + ), + ); + assert.equal(rangeLoads, 1); + assert.deepEqual(requestedAnchors, [0]); + // Durable sequence identity is Session- and Host-epoch-scoped: a request + // for a different Session or Host epoch must reject instead of silently + // reading a different slice of the transcript. + await assert.rejects( + () => + observer.loadTranscriptBefore( + { + consumerId: 'consumer-recovery', + sessionId: 'session-1', + hostEpoch: 'other-host', + anchorSequence: 0, + maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES, + }, + 22, + ), + /Desktop transcript host epoch changed/, + ); + await assert.rejects( + () => + observer.loadTranscriptBefore( + { + consumerId: 'consumer-recovery', + sessionId: 'other-session', + hostEpoch: 'host-1', + anchorSequence: 0, + maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES, + }, + 22, + ), + /Desktop transcript consumer belongs to another session/, + ); await observer.close(); }); @@ -1227,7 +1337,8 @@ test('keeps a transcript consumer available after a delivery fails', async () => observer.loadTranscriptAround( { consumerId, - generation: opened.generation, + sessionId: opened.sessionId, + hostEpoch: opened.hostEpoch, anchorSequence: 0, maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES, }, diff --git a/apps/desktop/src/main/__tests__/transcript-identity.test.ts b/apps/desktop/src/main/__tests__/transcript-identity.test.ts new file mode 100644 index 0000000000..bd853f7083 --- /dev/null +++ b/apps/desktop/src/main/__tests__/transcript-identity.test.ts @@ -0,0 +1,85 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import test from 'node:test'; +import type { DesktopTranscriptBatch } from '../../preload/transcript-contract.js'; +import { + adoptTranscriptIdentity, + type DesktopTranscriptIdentity, +} from '../../preload/transcript-identity.js'; + +function batch(overrides: Partial = {}): DesktopTranscriptBatch { + return { + sessionId: 'session-1', + generation: 'generation-1', + hostEpoch: 'host-1', + durableThrough: null, + fragments: [], + evictedDurableSequences: [], + completedOverlayMessageIds: [], + hasOlder: false, + hasNewer: false, + reset: false, + ready: true, + deliverySequence: 1, + ...overrides, + }; +} + +test('initializes the tracked identity from the first accepted batch', () => { + const first = batch({ generation: 'generation-1', hostEpoch: 'host-1' }); + assert.deepEqual(adoptTranscriptIdentity(undefined, first), { + generation: 'generation-1', + hostEpoch: 'host-1', + }); +}); + +test('keeps the current identity for a regular batch', () => { + const current: DesktopTranscriptIdentity = { generation: 'generation-1', hostEpoch: 'host-1' }; + assert.equal(adoptTranscriptIdentity(current, batch()), current); +}); + +test('adopts a reset batch identity so the next range request targets the replacement Host', () => { + const current: DesktopTranscriptIdentity = { generation: 'generation-1', hostEpoch: 'host-1' }; + const replacement = batch({ + generation: 'generation-2', + hostEpoch: 'host-2', + reset: true, + }); + const next = adoptTranscriptIdentity(current, replacement); + assert.deepEqual(next, { generation: 'generation-2', hostEpoch: 'host-2' }); + // The adopted identity stays in effect for later regular batches. + assert.equal( + adoptTranscriptIdentity(next, batch({ generation: 'generation-2', hostEpoch: 'host-2' })), + next, + ); +}); + +test('does not mutate an identity already captured by a dispatched request', () => { + const dispatched: DesktopTranscriptIdentity = { generation: 'generation-1', hostEpoch: 'host-1' }; + const replacement = batch({ + generation: 'generation-2', + hostEpoch: 'host-2', + reset: true, + }); + adoptTranscriptIdentity(dispatched, replacement); + // The host-1 request keeps its epoch; the Main-process guard fails it closed. + assert.deepEqual(dispatched, { generation: 'generation-1', hostEpoch: 'host-1' }); +}); diff --git a/apps/desktop/src/main/app-ipc-main.ts b/apps/desktop/src/main/app-ipc-main.ts index cb14099712..d43f899999 100644 --- a/apps/desktop/src/main/app-ipc-main.ts +++ b/apps/desktop/src/main/app-ipc-main.ts @@ -95,7 +95,7 @@ export function registerAppIpc( // observe the latest selection, not a snapshot taken at registration. const currentProjectRoot = (): Promise => projectRoot.current(); - targetIpc.handle('app:info', async () => { + handleReconnectableRead(targetIpc, 'app:info', async () => { const selection = await deps.projectManagement.current(); const projectPath = allowLocalProjectPaths ? selection.path : ''; return { diff --git a/apps/desktop/src/main/runtime-host-desktop-candidate.ts b/apps/desktop/src/main/runtime-host-desktop-candidate.ts index db8bab7e58..cb447f5ae3 100644 --- a/apps/desktop/src/main/runtime-host-desktop-candidate.ts +++ b/apps/desktop/src/main/runtime-host-desktop-candidate.ts @@ -36,6 +36,7 @@ import { type RuntimeHostCandidateLaunchBarrier, type RuntimeHostSpawnedProcess, type RemoteRuntimeHostProfile, + type CandidateExitDetails, } from "@maka/runtime-host/client"; import { INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID, @@ -167,6 +168,8 @@ export interface DesktopRuntimeHostCandidateStartInput readonly generation?: string; readonly takeoverHostEpoch?: string; readonly signal?: AbortSignal; + /** Candidate-exit sink forwarded to the launcher; the Desktop owns the sink. */ + readonly onExit?: (details: CandidateExitDetails) => void; readonly candidateLaunchBarrier?: RuntimeHostCandidateLaunchBarrier; readonly remote?: { readonly profile: RemoteRuntimeHostProfile; @@ -781,6 +784,7 @@ function connectInput( ? {} : { handshakeTimeoutMs: input.handshakeTimeoutMs }), ...(input.signal === undefined ? {} : { signal: input.signal }), + ...(input.onExit === undefined ? {} : { onExit: input.onExit }), }; } diff --git a/apps/desktop/src/main/runtime-host-desktop-manager.ts b/apps/desktop/src/main/runtime-host-desktop-manager.ts index 76c5c82aaa..61ab1868a2 100644 --- a/apps/desktop/src/main/runtime-host-desktop-manager.ts +++ b/apps/desktop/src/main/runtime-host-desktop-manager.ts @@ -27,6 +27,7 @@ import { LOCAL_RUNTIME_HOST_PROFILE, sameResolvedRuntimeHostProfileTarget, startRuntimeHostReconnectLifecycle, + type CandidateExitDetails, type ResolvedRuntimeHostProfile, type RuntimeHostReconnectBackoff, type RuntimeHostReconnectLifecycle, @@ -665,6 +666,9 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { signal, starting ? target.input.remote?.sshInteraction : 'batch', ), + onReconnectError: (error) => { + console.warn('[runtime-host] reconnect attempt failed:', error); + }, onFatalError: (error) => { if (!starting && target.valid) { target.valid = false; @@ -693,10 +697,12 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { sshInteraction: RuntimeHostSshInteraction | undefined, ): Promise { let takeoverHostEpoch: string | undefined; + const inheritedExit = target.input.onExit; while (true) { const result = await this.startCandidate( { ...target.input, + onExit: (details) => this.#reportCandidateExit(inheritedExit, details), ...(target.input.remote ? { remote: { @@ -776,6 +782,19 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { return target.lifecycle; } + /** Desktop-owned candidate-exit diagnostics; honors an embedder-supplied sink. */ + #reportCandidateExit( + inherited: ((details: CandidateExitDetails) => void) | undefined, + details: CandidateExitDetails, + ): void { + inherited?.(details); + if (details.code === 0 && details.signal === null) { + console.info('[runtime-host] candidate exited cleanly', details); + return; + } + console.error('[runtime-host] candidate exited unexpectedly', details); + } + async #waitForReadyCandidate( lifecycle: RuntimeHostReconnectLifecycle, previous?: DesktopRuntimeHostCandidate, diff --git a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts index 4aa110e773..8e42f395c9 100644 --- a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts @@ -780,7 +780,8 @@ function normalizeTranscriptRangeRequest(input: unknown): DesktopTranscriptRange } return { consumerId: requiredId(value.consumerId, 'Transcript consumer'), - generation: requiredId(value.generation, 'Transcript generation'), + sessionId: requiredId(value.sessionId, 'Session'), + hostEpoch: requiredId(value.hostEpoch, 'Host epoch'), anchorSequence: anchorSequence as number | null, maxBytes: maxBytes as number, }; diff --git a/apps/desktop/src/main/runtime-host-session-observer.ts b/apps/desktop/src/main/runtime-host-session-observer.ts index e58be3b3f0..c14addfd01 100644 --- a/apps/desktop/src/main/runtime-host-session-observer.ts +++ b/apps/desktop/src/main/runtime-host-session-observer.ts @@ -355,8 +355,7 @@ export class RuntimeHostSessionObserver { const { state, replica, consumer } = this.#requireTranscriptConsumer(request, targetId); const isCurrent = () => state.replica === replica && - state.transcriptConsumers.get(request.consumerId) === consumer && - consumer.generation === request.generation; + state.transcriptConsumers.get(request.consumerId) === consumer; const task = operation(replica); try { await task; @@ -1348,8 +1347,19 @@ export class RuntimeHostSessionObserver { if (targetId !== undefined && consumer.target.id !== targetId) { throw new Error('Desktop transcript consumer belongs to another renderer'); } - if (consumer.generation !== request.generation || replica.generation !== request.generation) { - throw new Error('Desktop transcript generation changed'); + // Durable transcript sequence identities belong to the Session and the + // Runtime Host epoch, not to a Desktop replica generation. Recovery may + // install a replacement replica (new generation, same session and host + // epoch) after the renderer dispatches a range request; continue that + // read against the current replica so navigation completes across + // reconnect. When the Host itself is replaced, sequence identity is not + // preserved, so reject the stale request instead of silently reading a + // different slice. + if (state.sessionId !== request.sessionId) { + throw new Error('Desktop transcript consumer belongs to another session'); + } + if (replica.hostEpoch !== request.hostEpoch) { + throw new Error('Desktop transcript host epoch changed; reopen the transcript'); } return { state, replica, consumer }; } diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index 1671c2da6f..7fc166f717 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -79,6 +79,10 @@ import { type DesktopTranscriptHandle, type DesktopTranscriptOpenResult, } from './transcript-contract.js'; +import { + adoptTranscriptIdentity, + type DesktopTranscriptIdentity, +} from './transcript-identity.js'; import type { DesktopDiagnosticInput, DesktopErrorDiagnosticWireInput, @@ -1981,7 +1985,7 @@ const makaBridge = { ): Promise { const consumerId = crypto.randomUUID(); const channel = `sessions:transcript:${consumerId}`; - let generation: string | undefined; + let identity: DesktopTranscriptIdentity | undefined; let closed = false; let requestClose = () => {}; let consumerScope: DesktopTargetScope | undefined; @@ -2000,11 +2004,12 @@ const makaBridge = { host.targetEpoch !== consumerScope.targetEpoch ) return; batch = assertDesktopTranscriptBatch(value); - if (batch.reset || generation === undefined) { - generation = batch.generation; + const adopted = adoptTranscriptIdentity(identity, batch); + if (adopted !== identity) { + identity = adopted; consumerScope = host; } - if (batch.generation === generation) handler(batch); + if (identity !== undefined && batch.generation === identity.generation) handler(batch); } catch (error) { requestClose(); throw error; @@ -2051,18 +2056,24 @@ const makaBridge = { throw error; } if (closed) throw new Error('Desktop transcript open was cancelled'); - generation ??= opened.generation; + identity ??= { generation: opened.generation, hostEpoch: opened.hostEpoch }; const range = ( operation: 'sessions:transcript:load-before' | 'sessions:transcript:load-around', anchorSequence: number | null, maxBytes = DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES, - ): Promise => - ipcRenderer.invoke(operation, consumerScope, { + ): Promise => { + const currentIdentity = identity; + if (!currentIdentity) { + throw new Error('Desktop transcript identity is unavailable'); + } + return ipcRenderer.invoke(operation, consumerScope, { consumerId, - generation, + sessionId: opened.sessionId, + hostEpoch: currentIdentity.hostEpoch, anchorSequence, maxBytes, }) as Promise; + }; return { ...opened, sessionId, diff --git a/apps/desktop/src/preload/transcript-contract.ts b/apps/desktop/src/preload/transcript-contract.ts index 3d7790fe8c..51b1880233 100644 --- a/apps/desktop/src/preload/transcript-contract.ts +++ b/apps/desktop/src/preload/transcript-contract.ts @@ -60,7 +60,8 @@ export interface DesktopTranscriptOpenResult { export interface DesktopTranscriptRangeRequest { readonly consumerId: string; - readonly generation: string; + readonly sessionId: string; + readonly hostEpoch: string; readonly anchorSequence: number | null; readonly maxBytes: number; } diff --git a/apps/desktop/src/preload/transcript-identity.ts b/apps/desktop/src/preload/transcript-identity.ts new file mode 100644 index 0000000000..d02ec35d53 --- /dev/null +++ b/apps/desktop/src/preload/transcript-identity.ts @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import type { DesktopTranscriptBatch } from './transcript-contract.js'; + +/** + * Durable transcript sequence identity: the Session's generation and the + * Runtime Host epoch that produced it. + * + * The renderer keeps this alongside the open handle so range requests always + * carry the epoch of the Host the renderer currently accepts batches from. + * A replacement Host sends a reset batch with a new generation and epoch; + * adopting that identity lets the next range request pass the Main-process + * guard, while requests already dispatched with the previous epoch still fail + * closed on the old Host. + */ +export interface DesktopTranscriptIdentity { + readonly generation: string; + readonly hostEpoch: string; +} + +/** + * Adopts a batch's identity when none is tracked yet or when the batch is a + * reset; otherwise keeps the current identity. Returns the current identity + * by reference when nothing changed so callers can detect adoption. + */ +export function adoptTranscriptIdentity( + current: DesktopTranscriptIdentity | undefined, + batch: DesktopTranscriptBatch, +): DesktopTranscriptIdentity { + if (current !== undefined && !batch.reset) return current; + return { generation: batch.generation, hostEpoch: batch.hostEpoch }; +} diff --git a/packages/runtime-host/src/__tests__/owned-candidate.test.ts b/packages/runtime-host/src/__tests__/owned-candidate.test.ts index 87c8200ce3..f0dd87c0ba 100644 --- a/packages/runtime-host/src/__tests__/owned-candidate.test.ts +++ b/packages/runtime-host/src/__tests__/owned-candidate.test.ts @@ -31,7 +31,11 @@ import { connectOwnedRuntimeHostWithDependencies, } from '../client/connect-or-spawn.js'; import { runHostedExecution } from '../client/hosted-execution.js'; -import { launchOwnedRuntimeHostCandidate, type OwnedCandidateAttempt } from '../client/launcher.js'; +import { + launchOwnedRuntimeHostCandidate, + type CandidateExitDetails, + type OwnedCandidateAttempt, +} from '../client/launcher.js'; test('owned connection keeps a fresh Host alive for its full election window', async () => { const rootPath = await mkdtemp(join(tmpdir(), 'maka-owned-first-connection-')); @@ -335,6 +339,37 @@ test('owned candidate settlement requires a clean process exit', async () => { }); }); +test('reports unexpected candidate exit through the caller-provided onExit sink', async () => { + const rootPath = await mkdtemp(join(tmpdir(), 'maka-owned-on-exit-')); + const exited = deferred(); + const launch = launchOwnedRuntimeHostCandidate({ + rootPath, + expectedRootId: '00000000-0000-4000-8000-000000000001', + entrypoint: new URL('./fixtures/owned-candidate-exit.js', import.meta.url), + env: { MAKA_TEST_EXIT_CODE: '1' }, + onExit: (details) => exited.resolve(details), + }); + + const candidate = await launch.spawned; + assert.equal(await candidate.settle(2_000), false); + assert.deepEqual(await exited.promise, { pid: candidate.pid, code: 1, signal: null }); +}); + +test('reports clean candidate exit through the caller-provided onExit sink', async () => { + const rootPath = await mkdtemp(join(tmpdir(), 'maka-owned-on-exit-clean-')); + const exited = deferred(); + const launch = launchOwnedRuntimeHostCandidate({ + rootPath, + expectedRootId: '00000000-0000-4000-8000-000000000001', + entrypoint: new URL('./fixtures/owned-candidate-exit.js', import.meta.url), + onExit: (details) => exited.resolve(details), + }); + + const candidate = await launch.spawned; + assert.equal(await candidate.settle(2_000), true); + assert.deepEqual(await exited.promise, { pid: candidate.pid, code: 0, signal: null }); +}); + test('owned candidate can be released to the enclosing environment without termination', async () => { const rootPath = await mkdtemp(join(tmpdir(), 'maka-owned-candidate-')); const launch = launchOwnedRuntimeHostCandidate({ @@ -418,3 +453,14 @@ async function waitForDefined( }); } } + +function deferred(): { + readonly promise: Promise; + resolve(value: T): void; +} { + let resolve!: (value: T) => void; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} diff --git a/packages/runtime-host/src/candidate-entry.ts b/packages/runtime-host/src/candidate-entry.ts index a09b5e1566..7c6a4c8077 100644 --- a/packages/runtime-host/src/candidate-entry.ts +++ b/packages/runtime-host/src/candidate-entry.ts @@ -17,6 +17,7 @@ * under the License. */ +import { generalizedErrorMessage } from '@maka/core/redaction'; import { candidateStartupFailureExitCode, classifyCandidateStartupFailure, @@ -86,7 +87,12 @@ export async function runExecutionCandidateEntry( const stopWatch = hooks.onWon?.(result.host); try { await runRuntimeHostProcessLifecycle(result.host); - } catch { + } catch (error) { + // Log the redacted, generalized message only: a full error object can + // carry paths and spawn arguments in its message or stack. + console.error( + `[runtime-host] lifecycle failed: ${generalizedErrorMessage(error, 'Runtime Host lifecycle failed')}`, + ); process.exitCode = 1; } finally { stopWatch?.(); diff --git a/packages/runtime-host/src/client/connect-or-spawn.ts b/packages/runtime-host/src/client/connect-or-spawn.ts index 941620395a..0c54bbd0f3 100644 --- a/packages/runtime-host/src/client/connect-or-spawn.ts +++ b/packages/runtime-host/src/client/connect-or-spawn.ts @@ -40,11 +40,14 @@ import { import { launchDetachedRuntimeHostCandidate, launchOwnedRuntimeHostCandidate, + type CandidateExitDetails, type CandidateProcessExit, type CandidateLauncher, type DetachedCandidateAttempt, type OwnedCandidateAttempt, } from './launcher.js'; + +export type { CandidateExitDetails } from './launcher.js'; import { isPermanentCandidateStartupFailure, type CandidateStartupFailure, @@ -74,6 +77,8 @@ export interface ConnectOrSpawnRuntimeHostInput { handshakeTimeoutMs?: number; candidateEntrypoint: string | URL; signal?: AbortSignal; + /** Candidate-exit sink forwarded to the launcher; the embedder owns the sink. */ + onExit?: (details: CandidateExitDetails) => void; } interface ConnectOrSpawnRuntimeHostDependencies { @@ -412,6 +417,7 @@ export async function connectOrSpawnRuntimeHostWithDependencies( entrypoint: input.candidateEntrypoint, initialConnectionTimeoutMs: Math.ceil(remaining), ...(input.generation === undefined ? {} : { generation: input.generation }), + ...(input.onExit === undefined ? {} : { onExit: input.onExit }), }); candidateLaunches.add(launch); const attempt = await settleBeforeDeadline(launch.spawned, deadline, input.signal); diff --git a/packages/runtime-host/src/client/index.ts b/packages/runtime-host/src/client/index.ts index 138ea62296..7f73e67acf 100644 --- a/packages/runtime-host/src/client/index.ts +++ b/packages/runtime-host/src/client/index.ts @@ -97,6 +97,7 @@ export { } from './catalog-reader.js'; export { connectOrSpawnRuntimeHost, + type CandidateExitDetails, type ConnectOrSpawnRuntimeHostInput, type ConnectOrSpawnRuntimeHostResult, type RuntimeHostElectionDiagnostic, diff --git a/packages/runtime-host/src/client/launcher.ts b/packages/runtime-host/src/client/launcher.ts index b02c06eb60..3fa18c8d72 100644 --- a/packages/runtime-host/src/client/launcher.ts +++ b/packages/runtime-host/src/client/launcher.ts @@ -29,6 +29,12 @@ import { RUNTIME_HOST_STDERR_PIPE_ENV } from '../process-diagnostics.js'; const CANDIDATE_STDERR_MAX_BYTES = 4 * 1024; +export interface CandidateExitDetails { + readonly pid: number | undefined; + readonly code: number | null; + readonly signal: NodeJS.Signals | null; +} + export interface DetachedCandidateInput { rootPath: string; expectedRootId: string; @@ -39,6 +45,8 @@ export interface DetachedCandidateInput { executable?: string; entrypoint: string | URL; env?: NodeJS.ProcessEnv; + /** Called with the candidate's exit details; the embedder owns the sink. */ + readonly onExit?: (details: CandidateExitDetails) => void; } export interface DetachedCandidateAttempt { @@ -72,6 +80,7 @@ export function launchDetachedRuntimeHostCandidate( const startupAttemptId = randomUUID(); const child = spawnCandidate(input, true, startupAttemptId); const exited = observeCandidateExit(child); + notifyCandidateExit(child, exited, input.onExit); const startupFailure = readStartupFailure(exited, startupAttemptId); const spawned = spawnedPid(child).then(({ pid }) => { child.unref(); @@ -86,6 +95,7 @@ export function launchOwnedRuntimeHostCandidate(input: DetachedCandidateInput): const startupAttemptId = randomUUID(); const child = spawnCandidate(input, false, startupAttemptId); const exited = observeCandidateExit(child); + notifyCandidateExit(child, exited, input.onExit); const startupFailure = readStartupFailure(exited, startupAttemptId); return { spawned: spawnedPid(child).then(({ pid }) => ({ @@ -165,6 +175,21 @@ function spawnedPid(child: ReturnType): Promise<{ pid: number }> { }); } +function notifyCandidateExit( + child: ChildProcess, + exited: Promise, + onExit: DetachedCandidateInput['onExit'], +): void { + if (!onExit) return; + void exited.then(({ code, signal }) => { + try { + onExit({ pid: child.pid, code, signal }); + } catch { + // The embedder owns this diagnostics sink; it must not affect process settlement. + } + }); +} + function readStartupFailure( exited: Promise, startupAttemptId: string,