diff --git a/src/application/binary/BinarySession.ts b/src/application/binary/BinarySession.ts index b12b310ff2..e604256d99 100644 --- a/src/application/binary/BinarySession.ts +++ b/src/application/binary/BinarySession.ts @@ -98,6 +98,30 @@ export class BinarySession this.#providerRouter = providerRouter; } + /** + * Fast-path liveness gate (#1719): short-circuit only when the retained + * client is genuinely alive. The probe reads the child's operation health + * rather than the composite's unconditionally-ok `health`, so a dead child + * falls through and a real client is started; an aborted caller leaves the + * session untouched. Returns the fast-path result, or undefined to fall + * through to the slow path. + */ + async #tryRetainedFastPath( + retained: SessionBinding, + target: BinaryTarget, + options: Pick, + ): Promise | undefined> { + const live = await this.#retainedClientLive(retained.client, options); + if (isAborted(options.signal)) + return err(new AnalysisCancelledError("open_binary")); + if (!live) return undefined; + if (options.snapshot !== undefined) { + const imported = this.importAnalysisSnapshot(options.snapshot); + if (!imported.ok) return imported; + } + return ok(target); + } + /** Identify the provider producing evidence for this session. */ providerIdentity(operation?: AnalysisOperation): ProviderIdentity { const route = this.#currentRoute(); @@ -247,12 +271,14 @@ export class BinarySession if (!pendingClosed.ok) return pendingClosed; if (isAborted(options.signal)) return err(new AnalysisCancelledError("open_binary")); - if (sameTarget && sameProfile) { - if (options.snapshot !== undefined) { - const imported = this.importAnalysisSnapshot(options.snapshot); - if (!imported.ok) return imported; - } - return ok(target); + const retained = sameTarget && sameProfile ? this.#active : undefined; + if (retained !== undefined) { + const fastPath = await this.#tryRetainedFastPath( + retained, + target, + options, + ); + if (fastPath !== undefined) return fastPath; } const previous = this.#active; this.#active = undefined; @@ -703,6 +729,36 @@ export class BinarySession ); } + /** + * Probe whether a retained provider client is still live. + * + * On production routes the client is a CompositeProvider whose `health` + * answers unconditionally ok without touching a child, so `health` alone + * cannot see a dead child. `operationHealthSnapshot` reaches the + * instantiated children instead. Only the known-live states allow the + * fast path: `exited`/`unreachable` are dead, `unknown` is never + * evidence of health (it is what a recorded but uncharacterized failure + * reports), and `not_started` is a startup that never completed — in the + * Hopper timeout case the client holds retained resources that block the + * next `start()` until `close()` retries cleanup, so treating it as live + * would skip the retire path and hand the wedged client to subsequent + * analysis. The gate fails closed: without a snapshot signal, fall back + * to the `health` admission probe. + */ + async #retainedClientLive( + client: AnalysisClient, + options: Pick, + ): Promise { + const snapshot = client.operationHealthSnapshot?.(); + if (snapshot !== undefined) + return snapshot.state === "idle" || snapshot.state === "busy"; + try { + return (await client.execute("health", {}, options)).ok; + } catch { + return false; + } + } + async #restore( previous: SessionBinding | undefined, ): Promise> { diff --git a/src/application/binary/BinarySessionLiveness.test.ts b/src/application/binary/BinarySessionLiveness.test.ts new file mode 100644 index 0000000000..88915f4c51 --- /dev/null +++ b/src/application/binary/BinarySessionLiveness.test.ts @@ -0,0 +1,338 @@ +import { describe, expect, it } from "vitest"; + +import { + createAnalysisExecution, + type AnalysisClient, + type AnalysisOperation, + type AnalysisProvider, + type AnalysisProviderCandidate, + type CapabilityDescriptor, + type ProviderAvailability, + type ProviderIdentity, + type ProviderTargetSupport, +} from "../AnalysisProvider.js"; +import { createAnalysisProfile } from "../../domain/analysisProfile.js"; +import { ProviderAdapterError } from "../../domain/providerAdapterError.js"; +import type { ProviderOperationHealth } from "../../domain/providerOperationHealth.js"; +import { err, ok } from "../../domain/result.js"; +import { InvestigationRecords } from "../investigation/InvestigationRecords.js"; +import { AnalysisProviderRegistry } from "./AnalysisProviderRegistry.js"; +import { BinarySession } from "./BinarySession.js"; +import { SessionProviderRouter } from "./SessionProviderRouter.js"; + +/** + * Regression coverage for https://github.com/morluto/rea/issues/1719: + * `BinarySession#open` must not short-circuit on the `sameTarget && + * sameProfile` fast path when the retained provider client is dead. It must + * start a fresh client instead. + * + * The fixtures are production-composed: a deep provider plus an auxiliary + * provider, so `SessionProviderRouter` wraps them in a real + * `CompositeProvider` exactly like `src/composition/binary.ts` does for the + * CLI/MCP runtime. The composite's `health` answers unconditionally ok + * without touching a child, so a probe that only calls `health` cannot see a + * dead child — the liveness signal must reach the actual child through + * `operationHealthSnapshot`, the way `HopperClient.operationHealth` reports a + * recorded process exit. + */ +describe("BinarySession open liveness", () => { + it("reuses the retained client while the child stays live", async () => { + const { session, deep } = openSession(); + expect((await session.open(process.execPath)).ok).toBe(true); + // Instantiate the deep child inside the composite, like a real query would. + expect((await session.execute("address_name", {})).ok).toBe(true); + expect(deep.created).toHaveLength(1); + + const second = await session.open(process.execPath); + expect(second.ok).toBe(true); + // Still the same composite and child: nothing was retired or recreated. + expect(deep.created).toHaveLength(1); + expect(deep.created[0]?.closeCalled).toBe(false); + expect((await session.execute("address_name", {})).ok).toBe(true); + expect(deep.created).toHaveLength(1); + }); + + it("starts a fresh client when the retained child died", async () => { + const { session, deep } = openSession(); + expect((await session.open(process.execPath)).ok).toBe(true); + expect((await session.execute("address_name", {})).ok).toBe(true); + expect(deep.created).toHaveLength(1); + + // Kill the provider process out from under the session, then re-open + // the same target: the session must not short-circuit on the dead + // child — it must retire it and start a real one. + const dead = deep.created[0]; + dead?.kill(); + + const second = await session.open(process.execPath); + expect(second.ok).toBe(true); + // The dead binding was retired, not retained. + expect(dead?.closeCalled).toBe(true); + // The next query runs against a freshly created child, not the dead one. + expect((await session.execute("address_name", {})).ok).toBe(true); + expect(deep.created).toHaveLength(2); + }); + + it("treats an uncharacterized operation failure as dead", async () => { + // A recorded failure the provider could not characterize ("unknown", + // e.g. a Hopper spawn failure with no observed exit code) is never + // evidence of health: the gate fails closed. + const { session, deep } = openSession(); + expect((await session.open(process.execPath)).ok).toBe(true); + expect((await session.execute("address_name", {})).ok).toBe(true); + expect(deep.created).toHaveLength(1); + + const dead = deep.created[0]; + dead?.kill("unknown"); + + const second = await session.open(process.execPath); + expect(second.ok).toBe(true); + expect(dead?.closeCalled).toBe(true); + expect((await session.execute("address_name", {})).ok).toBe(true); + expect(deep.created).toHaveLength(2); + }); + + it("retires a retained client whose startup never completed", async () => { + // A Hopper startup timeout leaves the operation-health snapshot in + // "not_started" while the retained resources block the next start() + // until close() retries cleanup. Classifying that state as live would return success + // without entering the retire path, so subsequent analysis still hits + // the wedged client. + const { session, deep } = openSession(); + expect((await session.open(process.execPath)).ok).toBe(true); + expect((await session.execute("address_name", {})).ok).toBe(true); + expect(deep.created).toHaveLength(1); + + const wedged = deep.created[0]; + wedged?.kill("not_started"); + + const second = await session.open(process.execPath); + expect(second.ok).toBe(true); + // The wedged binding was retired (close ran, so cleanup is retried), + // not reused. + expect(wedged?.closeCalled).toBe(true); + // The next query runs against a freshly created child, not the wedged one. + expect((await session.execute("address_name", {})).ok).toBe(true); + expect(deep.created).toHaveLength(2); + }); + + it("falls back to the health probe when the child reports no operation health", async () => { + // Without an operation-health hook there is no child-liveness signal, + // so the probe keeps the previous `health` admission behavior. + const { session, deep } = openSession({ operationHealth: false }); + expect((await session.open(process.execPath)).ok).toBe(true); + expect((await session.execute("address_name", {})).ok).toBe(true); + expect(deep.created).toHaveLength(1); + + const second = await session.open(process.execPath); + expect(second.ok).toBe(true); + // The composite's `health` answers ok, so the live client is reused. + expect(deep.created).toHaveLength(1); + expect(deep.created[0]?.closeCalled).toBe(false); + }); + + it("reports cancellation without touching the session when aborted mid-probe", async () => { + const { session, deep, onProbe } = openSession(); + expect((await session.open(process.execPath)).ok).toBe(true); + expect((await session.execute("address_name", {})).ok).toBe(true); + + const controller = new AbortController(); + // Fire the abort while the fast-path liveness probe is in flight: the + // hook runs synchronously inside the snapshot read, so the abort lands + // after the probe started but before the fast-path decision completes. + let fired = false; + onProbe(() => { + if (!fired) { + fired = true; + controller.abort(); + } + }); + + const cancelled = await session.open(process.execPath, { + signal: controller.signal, + }); + expect(cancelled).toMatchObject({ + ok: false, + error: { _tag: "AnalysisCancelledError" }, + }); + // The dead-or-live question is moot: the session must be untouched. + expect(deep.created).toHaveLength(1); + expect(deep.created[0]?.closeCalled).toBe(false); + const reopened = await session.open(process.execPath); + expect(reopened.ok).toBe(true); + expect(deep.created).toHaveLength(1); + }); +}); + +interface KillableChild extends AnalysisClient { + kill(state?: "exited" | "unreachable" | "unknown" | "not_started"): void; + closeCalled: boolean; +} + +const deepIdentity: ProviderIdentity = { + id: "liveness-deep", + name: "Liveness deep provider", + version: "1", +}; + +const auxIdentity: ProviderIdentity = { + id: "liveness-aux", + name: "Liveness auxiliary provider", + version: "1", +}; + +const capability = ( + identity: ProviderIdentity, + operation: Exclude, +): CapabilityDescriptor => ({ + provider: identity, + operation, + available: true, + reason: null, + effects: { + mutatesArtifact: false, + launchesProcess: false, + mayShowUi: false, + mayAccessNetwork: false, + mayWriteFilesystem: false, + changesPermissions: false, + requiresRoot: false, + }, + limitations: [], +}); + +const openSession = ( + options: { operationHealth?: boolean } = {}, +): { + session: BinarySession; + deep: { created: KillableChild[] }; + onProbe: (hook: () => void) => void; +} => { + const { provider, created, onProbe } = killableDeepProvider(options); + // Mirror the production composition: a deep provider plus auxiliary + // providers, so the route wraps them in a real CompositeProvider (whose + // trivially-ok `health` must not mask a dead child). + const aux: AnalysisProvider = { + identity: () => auxIdentity, + capabilities: () => [], + createClient: (): AnalysisClient => ({ + execute: (operation) => + Promise.resolve( + ok(createAnalysisExecution(`aux:${operation}`, auxIdentity)), + ), + close: () => Promise.resolve(ok(null)), + }), + }; + const session = new BinarySession( + SessionProviderRouter.selectable(new AnalysisProviderRegistry([provider]), [ + aux, + ]), + new InvestigationRecords(), + ); + return { session, deep: { created }, onProbe }; +}; + +const killableDeepProvider = (options: { + operationHealth?: boolean; +}): { + provider: AnalysisProviderCandidate; + created: KillableChild[]; + onProbe: (hook: () => void) => void; +} => { + const created: KillableChild[] = []; + let probeHook: (() => void) | undefined; + const withOperationHealth = options.operationHealth !== false; + const provider: AnalysisProviderCandidate = { + identity: () => deepIdentity, + capabilities: () => [capability(deepIdentity, "address_name")], + inspectAvailability: (): ProviderAvailability => ({ + status: "available", + code: null, + reason: null, + diagnostics: {}, + }), + inspectTargetSupport: (): ProviderTargetSupport => ({ + status: "supported", + code: null, + reason: null, + diagnostics: {}, + }), + resolveAnalysisProfile: () => + Promise.resolve( + ok({ + profile: createAnalysisProfile( + { id: deepIdentity.id, name: deepIdentity.name, version: "1" }, + { fixture: "liveness" }, + ), + }), + ), + createClient: (): KillableChild => { + let alive = true; + let closeCalled = false; + let deadState: "exited" | "unreachable" | "unknown" | "not_started" = + "exited"; + const client: KillableChild = { + execute: (operation) => { + return alive + ? Promise.resolve( + ok( + createAnalysisExecution( + operation === "health" ? null : `liveness:${operation}`, + deepIdentity, + ), + ), + ) + : Promise.resolve( + err( + new ProviderAdapterError(deepIdentity.id, operation, { + diagnostics: { reason: "process_killed" }, + }), + ), + ); + }, + // Genuine child-liveness signal, like HopperClient.operationHealth: + // a recorded process exit stays visible after the kill. + ...(withOperationHealth + ? { + operationHealthSnapshot: (): ProviderOperationHealth => { + probeHook?.(); + return { + state: alive ? "idle" : deadState, + stage: alive ? null : "connection", + retryAction: alive ? null : "restart_provider", + exitCode: alive ? null : 1, + requests: [], + }; + }, + } + : {}), + close: () => { + closeCalled = true; + return Promise.resolve(ok(null)); + }, + get closeCalled() { + return closeCalled; + }, + kill: ( + state: + | "exited" + | "unreachable" + | "unknown" + | "not_started" = "exited", + ) => { + alive = false; + deadState = state; + }, + }; + created.push(client); + return client; + }, + }; + return { + provider, + created, + onProbe: (hook: () => void) => { + probeHook = hook; + }, + }; +}; diff --git a/tests/boundary/providers/hopper/hopperClient.cleanupRetry.test.ts b/tests/boundary/providers/hopper/hopperClient.cleanupRetry.test.ts index e6f8215a03..030a192ac6 100644 --- a/tests/boundary/providers/hopper/hopperClient.cleanupRetry.test.ts +++ b/tests/boundary/providers/hopper/hopperClient.cleanupRetry.test.ts @@ -249,6 +249,32 @@ it("blocks a fresh launch while cleanup remains unverified", async () => { }); }); +it("preserves failed startup state through a blocked retry until cleanup succeeds", async () => { + const launcher = new RetryCleanupLauncher(); + const { client } = await failedStartup(launcher); + // The startup timeout leaves the client in not_started with retained + // resources; cleanup is still incomplete. + expect(client.operationHealth()).toMatchObject({ state: "not_started" }); + // An intervening retry is blocked by the retained resources, and now + // surfaces the recorded startup failure (upstream 069308bb) instead of + // a generic protocol error... + await expect(client.start()).resolves.toMatchObject({ + ok: false, + error: { + _tag: "ProviderAdapterError", + cleanupIncomplete: true, + cause: { _tag: "HopperTimeoutError" }, + }, + }); + // ...but must not wipe the failed-startup state: the client still + // reports not_started (never idle) until cleanup succeeds, so a + // same-target open_binary keeps retiring the wedged client instead of + // fast-pathing it. + expect(client.operationHealth()).toMatchObject({ state: "not_started" }); + launcher.allowCleanup = true; + await expect(client.close()).resolves.toEqual({ ok: true, value: null }); +}); + it("permits a fresh launch after stopping an owned bridge-request provider without an acknowledgement", async () => { const fixture = new RetryCleanupLauncher(); fixture.allowCleanup = true; diff --git a/tests/boundary/providers/hopper/hopperClient.startupFailures.test.ts b/tests/boundary/providers/hopper/hopperClient.startupFailures.test.ts index 0388346b92..3d96ddb67b 100644 --- a/tests/boundary/providers/hopper/hopperClient.startupFailures.test.ts +++ b/tests/boundary/providers/hopper/hopperClient.startupFailures.test.ts @@ -1,5 +1,6 @@ import { spawn, type ChildProcess } from "node:child_process"; import { access } from "node:fs/promises"; +import { createServer, type Server } from "node:net"; import { afterEach, describe, expect, it } from "vitest"; @@ -119,6 +120,80 @@ class LateSilentLauncher implements BridgeLauncher { } } +/** + * Bridge fixture whose health handshake returns an invalid payload, so + * parseHopperServerInfo rejects it with HopperProtocolError. + * Note: assumes the client's first bridge request is `health`; any + * auth/preamble request added before it would hit the -32601 branch. + */ +class BadHealthLauncher implements BridgeLauncher { + #server: Server | undefined; + #process: ChildProcess | undefined; + + async launch(session: BridgeSession) { + const server = createServer((socket) => { + let buffer = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk: string) => { + buffer += chunk; + let newline = buffer.indexOf("\n"); + while (newline >= 0) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + try { + const request = JSON.parse(line) as { + id: number; + method: string; + }; + if (request.method === "health") { + socket.write( + `${JSON.stringify({ id: request.id, result: { bogus: true } })}\n`, + ); + } else { + socket.write( + `${JSON.stringify({ + id: request.id, + error: { code: -32601, message: "unknown method" }, + })}\n`, + ); + } + } catch { + // Ignore malformed input; the client only sends NDJSON. + } + newline = buffer.indexOf("\n"); + } + }); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(session.socketPath, () => { + server.off("error", reject); + resolve(); + }); + }); + this.#server = server; + const child = spawn( + process.execPath, + ["-e", "setInterval(() => {}, 1000)"], + { + stdio: "ignore", + }, + ); + this.#process = child; + return ok({ + process: child, + ownsProcessLifetime: true as const, + providerLifetime: "launcher-process" as const, + shutdownMode: "process-cleanup" as const, + cleanup: async () => { + child.kill("SIGTERM"); + await new Promise((resolve) => server.close(() => resolve())); + return { cleaned: true as const, signaled: true }; + }, + }); + } +} + const clients: HopperClient[] = []; afterEach(async () => { @@ -220,6 +295,38 @@ describe("HopperClient startup failures", () => { }); }); + it("records unknown when startup is cancelled before launch", async () => { + const launcher = new FixtureLauncher(); + const client = new HopperClient({ launcher, startupTimeoutMs: 30_000 }); + clients.push(client); + const controller = new AbortController(); + controller.abort(); + + const result = await client.start(controller.signal); + + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error._tag).toBe("HopperCancelledError"); + expect(launcher.processes).toHaveLength(0); + // A cancelled startup must not report idle: the same-target fast path + // would otherwise treat the client as live and skip the retire path. + expect(client.operationHealth()).toMatchObject({ state: "unknown" }); + }); + + it("records unknown when the health handshake is invalid", async () => { + const launcher = new BadHealthLauncher(); + const client = new HopperClient({ launcher, startupTimeoutMs: 30_000 }); + clients.push(client); + + const result = await client.start(); + + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error._tag).toBe("HopperProtocolError"); + // parseHopperServerInfo failures were never recorded, so operationHealth() + // reported idle and the fast path kept a client whose next start stays + // blocked by retained resources. + expect(client.operationHealth()).toMatchObject({ state: "unknown" }); + }); + it.each([70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80])( "reports Linux adapter exit %i during bridge startup", async (exitCode) => {