From 77ebfb5f8a9b1b7dd86a51a2ad67fcefe0f05fb3 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sat, 24 Oct 2026 04:13:07 +0800 Subject: [PATCH] fix(environment): retain primary facts through server cleanup failure --- docs/cli.md | 6 + src/cli/environmentCommands.ts | 36 ++- src/server/environment/serveEnvironment.ts | 116 +++++++-- .../environmentFinalizationCli.test.ts | 220 ++++++++++++++++++ 4 files changed, 346 insertions(+), 32 deletions(-) create mode 100644 tests/process-global/environmentFinalizationCli.test.ts diff --git a/docs/cli.md b/docs/cli.md index 1b81ff91af..c4f70aeb32 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -850,6 +850,12 @@ Refusals before binding (unreadable files, non-Evidence JSON, capture Evidence in a format other than HAR, a manifest or state whose parent is not the supplied evidence, an unbindable port) print a structured error and exit `1`, as does a listener that fails after binding. +If shutdown fails, the command reports `cleanup_incomplete` with the retained +server resources. It preserves a primary listener diagnostic, or a +`shutdown_observation` naming the requested shutdown and its URL, environment, +tool count and Evidence identifiers. The successful `stopped` document is +returned only after cleanup is confirmed. The same in-process server owner +retains failed stages for retry; the CLI exposes no separate retry command. `rea environment-export` is one-shot: it prints the written files and exits `0`, or prints a structured `invalid_request` error and exits `1` for the same pair refusals, a non-empty or file-occupied directory, a missing package diff --git a/src/cli/environmentCommands.ts b/src/cli/environmentCommands.ts index 81af65f837..ea3bbce52d 100644 --- a/src/cli/environmentCommands.ts +++ b/src/cli/environmentCommands.ts @@ -15,7 +15,8 @@ import { mcpHttpListenerPortSchema, type McpHttpEndpoint, } from "../config/mcpHttpEndpoint.js"; -import type { AnalysisError } from "../domain/analysisErrorBase.js"; +import { AnalysisError } from "../domain/analysisErrorBase.js"; +import { analysisErrorWithCleanupFailure } from "../domain/analysisErrorCleanup.js"; import type { StreamableHttpListenError } from "../server/http/streamableHttpListenError.js"; import { AnalysisCancelledError, @@ -215,16 +216,10 @@ const serveEnvironmentCommand = async ( if (!(cause instanceof StreamableHttpListenError)) throw cause; return cliError(transportError(endpoint.value, cause)); } - let failure: StreamableHttpListenError | undefined; - try { - failure = await Promise.race([untilAborted(signal), listenerFailed]); - } finally { - await served.close(); - } - if (failure !== undefined) - return cliError(transportError(endpoint.value, failure)); - return { - status: "stopped", + const failure = await Promise.race([untilAborted(signal), listenerFailed]); + const listenerError = + failure === undefined ? undefined : transportError(endpoint.value, failure); + const observation = { url: served.url, environment: served.card.name, tools: served.card.capabilities.tools.length, @@ -232,6 +227,25 @@ const serveEnvironmentCommand = async ( manifest_evidence_id: source.manifestEvidenceId, state: report, }; + try { + await served.close(); + } catch (cause: unknown) { + if (!(cause instanceof AnalysisError)) throw cause; + if (listenerError !== undefined) + return cliError( + analysisErrorWithCleanupFailure( + listenerError, + cause, + CLI_COMMANDS.environmentServe, + ).bindCleanupDependencies([cause]), + ); + cause.retainDiagnostics({ + shutdown_observation: { status: "shutdown_requested", ...observation }, + }); + return cliError(cause); + } + if (listenerError !== undefined) return cliError(listenerError); + return { status: "stopped", ...observation }; }; interface ExportArguments { diff --git a/src/server/environment/serveEnvironment.ts b/src/server/environment/serveEnvironment.ts index d98958eee2..1931937047 100644 --- a/src/server/environment/serveEnvironment.ts +++ b/src/server/environment/serveEnvironment.ts @@ -28,7 +28,11 @@ import { } from "../../application/environment/environmentTools.js"; import { projectAnalysisError } from "../../domain/analysisErrorProjection.js"; import { safeParseJson } from "../../domain/safeJson.js"; -import type { AnalysisResourceConstraintError } from "../../domain/analysisErrorCore.js"; +import { + AnalysisOutputError, + type AnalysisResourceConstraintError, +} from "../../domain/analysisErrorCore.js"; +import type { AnalysisError } from "../../domain/analysisErrorBase.js"; import type { StreamableHttpListenError } from "../http/streamableHttpListenError.js"; import { ok, type Result } from "../../domain/result.js"; import { PRODUCT_IDENTITY } from "../../identity.js"; @@ -439,6 +443,14 @@ export interface ServedEnvironment extends StreamableHttpServer { readonly card: EnvironmentCard; } +interface EnvironmentShutdownOwner { + readonly cleanupResources: readonly string[]; + close(): Promise; +} + +// A failed shutdown must retain the same staged HTTP and state-file owners. +const retainedEnvironmentShutdowns = new Set(); + /** Serve one manifest as an agent environment until closed. */ export const serveEnvironment = async ( options: ServeEnvironmentOptions, @@ -492,6 +504,87 @@ export const serveEnvironment = async ( options.onerror?.(error); }, }); + let httpClosePending = false; + let stateClosePending = false; + let closing: Promise | undefined; + const shutdownFailures = new Set(); + const owner: EnvironmentShutdownOwner = { + get cleanupResources() { + const stateResources = stateFiles.cleanupResources; + return [ + ...(httpClosePending ? [served.url] : []), + ...(stateClosePending + ? stateResources.length > 0 + ? stateResources + : ["environment state-file scope"] + : []), + ]; + }, + close: () => { + if (closing !== undefined) return closing; + const attempt = (async () => { + stop(); + const failures: { + readonly stage: "http-server" | "state-files"; + readonly cause: unknown; + }[] = []; + try { + await served.close(); + httpClosePending = false; + } catch (cause: unknown) { + httpClosePending = true; + failures.push({ stage: "http-server", cause }); + } + try { + await stateFiles.close(); + stateClosePending = false; + } catch (cause: unknown) { + stateClosePending = true; + failures.push({ stage: "state-files", cause }); + } + if (failures.length > 0) { + retainedEnvironmentShutdowns.add(owner); + const failure = new AnalysisOutputError( + "environment-serve", + failures + .map( + ({ stage, cause }) => + `${stage}: ${cause instanceof Error ? cause.message : "Non-Error cleanup failure (original cause retained)."}`, + ) + .join("; "), + { + cause: + failures.length === 1 + ? failures[0]?.cause + : new AggregateError( + failures.map(({ cause }) => cause), + "Environment server and state-file cleanup both failed", + ), + diagnostics: { + cleanup_stages: failures.map(({ stage }) => stage), + }, + cleanup: { + reason: "Environment shutdown remains incomplete.", + resources: owner.cleanupResources, + }, + remediationAction: + "Keep this REA process available for cleanup. Confirm the reported resources are released before starting a replacement environment server.", + }, + ).bindCleanupResourceOwner(owner); + shutdownFailures.add(failure); + throw failure; + } + retainedEnvironmentShutdowns.delete(owner); + for (const failure of shutdownFailures) failure.resolveCleanup(); + shutdownFailures.clear(); + })(); + closing = attempt; + void attempt.catch(() => { + if (closing === attempt) closing = undefined; + }); + return attempt; + }, + }; return ok({ ...served, card, @@ -499,25 +592,6 @@ export const serveEnvironment = async ( stop(); return served.prepareShutdown(); }, - close: async () => { - stop(); - const failures: unknown[] = []; - try { - await served.close(); - } catch (cause: unknown) { - failures.push(cause); - } - try { - await stateFiles.close(); - } catch (cause: unknown) { - failures.push(cause); - } - if (failures.length === 1) throw failures[0]; - if (failures.length > 1) - throw new AggregateError( - failures, - "Environment server and state-file cleanup both failed", - ); - }, + close: owner.close, }); }; diff --git a/tests/process-global/environmentFinalizationCli.test.ts b/tests/process-global/environmentFinalizationCli.test.ts new file mode 100644 index 0000000000..b3116b1619 --- /dev/null +++ b/tests/process-global/environmentFinalizationCli.test.ts @@ -0,0 +1,220 @@ +import { expect, it, vi } from "vitest"; +import { AnalysisError } from "../../src/domain/analysisErrorBase.js"; +import { createCli } from "../../src/cli.js"; +import type { ServedEnvironment } from "../../src/server/environment/serveEnvironment.js"; +import type { ServeStreamableHttpOptions } from "../../src/server/http/serveStreamableHttp.js"; +import { StreamableHttpListenError } from "../../src/server/http/streamableHttpListenError.js"; +import { createTestTempDirectory } from "../fixtures/temporaryDirectory.js"; +import { writeEnvironmentServeFiles } from "../fixtures/environmentServe.js"; + +const control = vi.hoisted(() => ({ + ready: undefined as (() => void) | undefined, + served: undefined as ServedEnvironment | undefined, + onerror: undefined as ServeStreamableHttpOptions["onerror"], + failClose: false, + closeCalls: 0, + cleanupFailure: undefined as AnalysisError | undefined, +})); + +vi.mock( + "../../src/server/http/serveStreamableHttp.js", + async (importOriginal) => { + const actual = + await importOriginal< + typeof import("../../src/server/http/serveStreamableHttp.js") + >(); + return { + ...actual, + serveStreamableHttp: (options: ServeStreamableHttpOptions) => { + control.onerror = options.onerror; + return actual.serveStreamableHttp({ + ...options, + createMcpServer: async (context) => { + const server = await options.createMcpServer(context); + const close = server.close.bind(server); + server.close = async () => { + control.closeCalls += 1; + if (control.failClose) { + control.failClose = false; + throw new Error("controlled MCP cleanup failure"); + } + await close(); + }; + return server; + }, + }); + }, + }; + }, +); + +vi.mock( + "../../src/server/environment/serveEnvironment.js", + async (importOriginal) => { + const actual = + await importOriginal< + typeof import("../../src/server/environment/serveEnvironment.js") + >(); + return { + ...actual, + serveEnvironment: async ( + ...args: Parameters + ) => { + const started = await actual.serveEnvironment(...args); + if (!started.ok) return started; + const served = { + ...started.value, + close: async () => { + try { + await started.value.close(); + } catch (cause) { + if (cause instanceof AnalysisError) + control.cleanupFailure = cause; + throw cause; + } + }, + }; + control.served = served; + control.ready?.(); + return { ...started, value: served }; + }, + }; + }, +); + +it.each([ + { ending: "signal", fails: false }, + { ending: "signal", fails: true }, + { ending: "listener", fails: false }, + { ending: "listener", fails: true }, +] as const)( + "retains environment $ending facts when final cleanup fails=$fails", + async ({ ending, fails }) => { + const files = await writeEnvironmentServeFiles( + await createTestTempDirectory("rea-environment-finalization-"), + ); + const originalExitCode = process.exitCode; + const originalListeners = new Set(process.listeners("SIGINT")); + let ready!: () => void; + const started = new Promise((resolve) => { + ready = resolve; + }); + control.ready = ready; + control.served = undefined; + control.failClose = fails; + control.closeCalls = 0; + control.cleanupFailure = undefined; + let stdout = ""; + const pending = createCli({ REA_LOG_LEVEL: "silent" }).serve( + [ + "environment-serve", + files.manifestPath, + "--evidence", + files.capturePath, + "--port", + "0", + "--json", + ], + { + env: {}, + exit: () => {}, + stdout: (text) => { + stdout += text; + }, + }, + ); + try { + await started; + const served = control.served; + if (served === undefined) throw new Error("Environment was not admitted"); + if (ending === "signal") { + const listeners = process + .listeners("SIGINT") + .filter((listener) => !originalListeners.has(listener)); + expect(listeners).toHaveLength(1); + listeners[0]?.call(process, "SIGINT"); + } else + control.onerror?.( + new StreamableHttpListenError( + Object.assign(new Error("controlled listener failure"), { + code: "EIO", + }), + "serving", + ), + ); + await pending; + const output: unknown = JSON.parse(stdout); + expect(control.closeCalls).toBe(1); + expect(process.exitCode).toBe(ending === "signal" ? 130 : 1); + if (!fails) { + expect(output).toMatchObject( + ending === "signal" + ? { + status: "stopped", + url: served.url, + evidence_id: expect.any(String), + } + : { + code: "resource_constraint", + details: { diagnostics: { listener_phase: "serving" } }, + }, + ); + } else { + expect(JSON.stringify(output)).toContain( + "controlled MCP cleanup failure", + ); + expect(output).not.toHaveProperty("status"); + expect(output).toMatchObject({ + code: "cleanup_incomplete", + details: { resources: [served.url] }, + }); + expect(output).toMatchObject( + ending === "signal" + ? { + details: { + diagnostics: { + shutdown_observation: { + status: "shutdown_requested", + url: served.url, + environment: served.card.name, + tools: served.card.capabilities.tools.length, + evidence_id: expect.any(String), + manifest_evidence_id: expect.any(String), + }, + }, + }, + } + : { + details: { + diagnostics: { + primary_error: { + code: "resource_constraint", + details: { + diagnostics: { + listener_phase: "serving", + listener_failure: { system_code: "EIO" }, + }, + }, + remediation: { action: expect.any(String) }, + }, + }, + }, + }, + ); + expect(control.cleanupFailure?.cleanupIncomplete).toBe(true); + expect(control.cleanupFailure?.cleanupResources).toEqual([served.url]); + await served.close(); + expect(control.closeCalls).toBe(2); + expect(control.cleanupFailure?.cleanupIncomplete).toBe(false); + expect(control.cleanupFailure?.cleanupResources).toEqual([]); + } + expect(process.listeners("SIGINT")).toEqual([...originalListeners]); + } finally { + control.failClose = false; + await control.served?.close(); + control.ready = undefined; + control.served = undefined; + process.exitCode = originalExitCode; + } + }, +);