fix(environment): retain primary facts through server cleanup failure

2026-10-25 22:08:11 -04:00 · 2026-10-24 04:13:07 +08:00
parent 6f8dc9c139
commit 77ebfb5f8a
4 changed files with 346 additions and 32 deletions
@@ -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
@@ -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 {
@@ -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<void>;
+}
+
+// A failed shutdown must retain the same staged HTTP and state-file owners.
+const retainedEnvironmentShutdowns = new Set<EnvironmentShutdownOwner>();
+
 /** 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<void> | undefined;
+  const shutdownFailures = new Set<AnalysisError>();
+  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,
  });
 };
@@ -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<typeof actual.serveEnvironment>
+      ) => {
+        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<void>((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;
+    }
+  },
+);