fix(acpx): stop/start races cannot resurrect a stopped embedded runtime (#120029)

* fix(acpx): stop/start races cannot resurrect a stopped runtime backend The deferred acpx runtime service published its lazily started backend without lifecycle ownership checks: a stop during in-flight activation let the stale service become the active runtime, and stop cleared shared state while startup still used it. Start/stop now use a lifecycle revision; only the current revision may publish a runtime, stop invalidates every deferred proxy before awaiting in-flight startup, and restart waits for the previous stop to settle. * chore: re-fire CI * fix(acpx): fence deferred runtime ownership * refactor(acpx): centralize backend lifecycle ownership * fix(acpx): annotate deferred runtime cycle
2026-08-12 21:53:00 -06:00 · 2026-08-07 01:09:23 -07:00
parent 40ef8c5e68
commit 5a913fa453
4 changed files with 359 additions and 48 deletions
@@ -5,6 +5,11 @@ const { runtimeRegistry } = vi.hoisted(() => ({
  runtimeRegistry: new Map<string, { runtime: unknown }>(),
 }));

+type BackendLifecycle = {
+  publish: (backend: { runtime: unknown }) => void;
+  retract: (runtime: unknown) => void;
+};
+
 const { realRuntime, realServiceStartMock, realServiceStopMock, createRealServiceMock } =
  vi.hoisted(() => {
    const runtime = {
@@ -21,17 +26,29 @@ const { realRuntime, realServiceStartMock, realServiceStopMock, createRealServic
      isHealthy: vi.fn(() => true),
      probeAvailability: vi.fn(async () => {}),
    };
-    const start = vi.fn(async () => {
-      runtimeRegistry.set("acpx", { runtime });
+    const start = vi.fn(async (_ctx: unknown, backendLifecycle?: BackendLifecycle) => {
+      if (backendLifecycle) {
+        backendLifecycle.publish({ runtime });
+      } else {
+        runtimeRegistry.set("acpx", { runtime });
+      }
    });
-    const stop = vi.fn(async () => {
-      runtimeRegistry.delete("acpx");
+    const stop = vi.fn(async (_ctx: unknown, backendLifecycle?: BackendLifecycle) => {
+      if (backendLifecycle) {
+        backendLifecycle.retract(runtime);
+      } else {
+        runtimeRegistry.delete("acpx");
+      }
    });
    return {
      realRuntime: runtime,
      realServiceStartMock: start,
      realServiceStopMock: stop,
-      createRealServiceMock: vi.fn(() => ({ id: "real-acpx-runtime", start, stop })),
+      createRealServiceMock: vi.fn((params: { backendLifecycle?: BackendLifecycle } = {}) => ({
+        id: "real-acpx-runtime",
+        start: (ctx: unknown) => start(ctx, params.backendLifecycle),
+        stop: (ctx: unknown) => stop(ctx, params.backendLifecycle),
+      })),
    };
  });

@@ -61,6 +78,14 @@ function restoreEnv(): void {
  }
 }

+function createDeferred() {
+  let resolve: () => void = () => {};
+  const promise = new Promise<void>((resolvePromise) => {
+    resolve = resolvePromise;
+  });
+  return { promise, resolve };
+}
+
 function createServiceContext() {
  return {
    workspaceDir: "/tmp/openclaw-acpx-register-test",
@@ -121,10 +146,16 @@ describe("acpx register runtime service", () => {
      sessionKey: "agent:codex:acp:test",
    });

-    expect(createRealServiceMock).toHaveBeenCalledWith({
-      pluginConfig: { timeoutSeconds: 10 },
-    });
-    expect(realServiceStartMock).toHaveBeenCalledWith(ctx);
+    expect(createRealServiceMock).toHaveBeenCalledWith(
+      expect.objectContaining({
+        backendLifecycle: expect.objectContaining({
+          publish: expect.any(Function),
+          retract: expect.any(Function),
+        }),
+        pluginConfig: { timeoutSeconds: 10 },
+      }),
+    );
+    expect(realServiceStartMock).toHaveBeenCalledWith(ctx, expect.any(Object));
    expect(runtimeRegistry.get("acpx")?.runtime).toBe(realRuntime);
    expect(ctx.logger.info).toHaveBeenCalledWith("embedded acpx runtime backend registered lazily");

@@ -148,7 +179,144 @@ describe("acpx register runtime service", () => {

    await service.stop?.(ctx as never);

-    expect(realServiceStopMock).toHaveBeenCalledWith(ctx);
+    expect(realServiceStopMock).toHaveBeenCalledWith(ctx, expect.any(Object));
+    expect(runtimeRegistry.get("acpx")).toBeUndefined();
+  });
+
+  it("rejects stale publication after stop invalidates the deferred backend", async () => {
+    delete process.env.OPENCLAW_SKIP_ACPX_RUNTIME;
+    const startEntered = createDeferred();
+    const releasePublication = createDeferred();
+    realServiceStartMock.mockImplementationOnce(async (_ctx, backendLifecycle) => {
+      startEntered.resolve();
+      await releasePublication.promise;
+      backendLifecycle?.publish({ runtime: realRuntime });
+    });
+    const ctx = createServiceContext();
+    const service = createAcpxRuntimeService();
+
+    await service.start(ctx as never);
+    const deferredRuntime = runtimeRegistry.get("acpx")?.runtime as {
+      ensureSession(input: { sessionKey: string; agent: string; mode: string }): Promise<unknown>;
+    };
+    const activation = deferredRuntime.ensureSession({
+      sessionKey: "agent:codex:acp:shutdown-race",
+      agent: "codex",
+      mode: "oneshot",
+    });
+    const activationResult = activation.then(
+      () => null,
+      (error: unknown) => error,
+    );
+    await startEntered.promise;
+
+    const stopping = service.stop?.(ctx as never);
+    await Promise.resolve();
+    expect(realServiceStopMock).not.toHaveBeenCalled();
+
+    releasePublication.resolve();
+    await stopping;
+
+    expect(await activationResult).toEqual(
+      expect.objectContaining({
+        message: "ACPX runtime service stopped during activation",
+      }),
+    );
+    expect(realServiceStopMock).toHaveBeenCalledOnce();
+    expect(realServiceStopMock).toHaveBeenCalledWith(ctx, expect.any(Object));
+    expect(runtimeRegistry.get("acpx")).toBeUndefined();
+    await expect(
+      deferredRuntime.ensureSession({
+        sessionKey: "agent:codex:acp:stale-proxy",
+        agent: "codex",
+        mode: "oneshot",
+      }),
+    ).rejects.toThrow("ACPX runtime service is not started");
+  });
+
+  it("keeps a successor generation registered when old cleanup finishes late", async () => {
+    delete process.env.OPENCLAW_SKIP_ACPX_RUNTIME;
+    const published = createDeferred();
+    const releaseProbe = createDeferred();
+    realServiceStartMock.mockImplementationOnce(async (_ctx, backendLifecycle) => {
+      if (!backendLifecycle) {
+        throw new Error("expected outer backend lifecycle");
+      }
+      backendLifecycle.publish({ runtime: realRuntime });
+      published.resolve();
+      await releaseProbe.promise;
+    });
+    const ctx = createServiceContext();
+    const generationA = createAcpxRuntimeService();
+
+    await generationA.start(ctx as never);
+    const deferredRuntimeA = runtimeRegistry.get("acpx")?.runtime as {
+      ensureSession(input: { sessionKey: string; agent: string; mode: string }): Promise<unknown>;
+    };
+    const activation = deferredRuntimeA.ensureSession({
+      sessionKey: "agent:codex:acp:generation-a",
+      agent: "codex",
+      mode: "oneshot",
+    });
+    const activationResult = activation.then(
+      () => null,
+      (error: unknown) => error,
+    );
+    await published.promise;
+    expect(runtimeRegistry.get("acpx")?.runtime).toBe(realRuntime);
+
+    let concurrentCallSettled = false;
+    const concurrentCallResult = deferredRuntimeA
+      .ensureSession({
+        sessionKey: "agent:codex:acp:generation-a-concurrent",
+        agent: "codex",
+        mode: "oneshot",
+      })
+      .then(
+        () => {
+          concurrentCallSettled = true;
+          return null;
+        },
+        (error: unknown) => {
+          concurrentCallSettled = true;
+          return error;
+        },
+      );
+    await new Promise<void>((resolve) => {
+      setImmediate(resolve);
+    });
+    expect(concurrentCallSettled).toBe(false);
+
+    const stoppingA = generationA.stop?.(ctx as never);
+    expect(runtimeRegistry.get("acpx")).toBeUndefined();
+
+    const generationB = createAcpxRuntimeService();
+    await generationB.start(ctx as never);
+    const deferredRuntimeB = runtimeRegistry.get("acpx")?.runtime;
+    expect(deferredRuntimeB).toBeTruthy();
+    expect(deferredRuntimeB).not.toBe(deferredRuntimeA);
+    expect(deferredRuntimeB).not.toBe(realRuntime);
+    expect(concurrentCallSettled).toBe(false);
+
+    releaseProbe.resolve();
+    await stoppingA;
+
+    expect(await activationResult).toEqual(
+      expect.objectContaining({ message: "ACPX runtime service stopped during activation" }),
+    );
+    expect(await concurrentCallResult).toEqual(
+      expect.objectContaining({ message: "ACPX runtime service stopped during activation" }),
+    );
+    expect(runtimeRegistry.get("acpx")?.runtime).toBe(deferredRuntimeB);
+    await expect(
+      deferredRuntimeA.ensureSession({
+        sessionKey: "agent:codex:acp:stale-generation-a",
+        agent: "codex",
+        mode: "oneshot",
+      }),
+    ).rejects.toThrow("ACPX runtime service is not started");
+
+    await generationB.stop?.(ctx as never);
    expect(runtimeRegistry.get("acpx")).toBeUndefined();
  });

@@ -15,51 +15,101 @@ import { createLazyAcpRuntimeProxy } from "./src/runtime-proxy.js";
 const ACPX_BACKEND_ID = "acpx";

 type RealAcpxServiceModule = typeof import("./src/service.js");
-type CreateAcpxRuntimeServiceParams = NonNullable<
+type InnerAcpxRuntimeServiceParams = NonNullable<
  Parameters<RealAcpxServiceModule["createAcpxRuntimeService"]>[0]
 >;
+type CreateAcpxRuntimeServiceParams = Omit<InnerAcpxRuntimeServiceParams, "backendLifecycle">;

 type DeferredServiceState = {
  ctx: OpenClawPluginServiceContext | null;
+  lifecycleRevision: number;
+  ownedRuntime: AcpRuntime | null;
  params: CreateAcpxRuntimeServiceParams;
  realRuntime: AcpRuntime | null;
  realService: OpenClawPluginService | null;
  startPromise: Promise<AcpRuntime> | null;
+  stopPromise: Promise<void> | null;
 };

 const loadServiceModule = createLazyRuntimeModule(() => import("./src/service.js"));

-async function startRealService(state: DeferredServiceState): Promise<AcpRuntime> {
+function unregisterOwnedRuntime(runtime: AcpRuntime | null): void {
+  if (runtime && getAcpRuntimeBackend(ACPX_BACKEND_ID)?.runtime === runtime) {
+    unregisterAcpRuntimeBackend(ACPX_BACKEND_ID);
+  }
+}
+
+async function startRealService(
+  state: DeferredServiceState,
+  lifecycleRevision: number,
+  deferredRuntime: AcpRuntime,
+): Promise<AcpRuntime> {
+  if (state.lifecycleRevision !== lifecycleRevision || !state.ctx) {
+    throw new Error("ACPX runtime service is not started");
+  }
  if (state.realRuntime) {
    return state.realRuntime;
  }
-  if (!state.ctx) {
-    throw new Error("ACPX runtime service is not started");
+  if (state.startPromise) {
+    return await state.startPromise;
  }
-  state.startPromise ??= (async () => {
+  const ctx = state.ctx;
+  state.startPromise = (async () => {
+    let publishedRuntime: AcpRuntime | null = null;
    const { createAcpxRuntimeService: createAcpxRuntimeServiceLocal } = await loadServiceModule();
-    const service = createAcpxRuntimeServiceLocal(state.params);
+    const service = createAcpxRuntimeServiceLocal({
+      ...state.params,
+      backendLifecycle: {
+        publish(backend) {
+          if (state.lifecycleRevision !== lifecycleRevision || state.ctx !== ctx) {
+            throw new Error("ACPX runtime service stopped during activation");
+          }
+          if (getAcpRuntimeBackend(ACPX_BACKEND_ID)?.runtime !== deferredRuntime) {
+            throw new Error("ACPX runtime service lost registry ownership during activation");
+          }
+          // Publication is a synchronous compare-and-replace: another plugin
+          // generation cannot be adopted between the ownership check and write.
+          registerAcpRuntimeBackend({ id: ACPX_BACKEND_ID, ...backend });
+          publishedRuntime = backend.runtime;
+          state.ownedRuntime = backend.runtime;
+        },
+        retract(runtime) {
+          unregisterOwnedRuntime(runtime);
+        },
+      },
+    });
    state.realService = service;
-    await service.start(state.ctx as OpenClawPluginServiceContext);
-    const backend = getAcpRuntimeBackend(ACPX_BACKEND_ID);
-    if (!backend?.runtime) {
+    await service.start(ctx);
+    if (state.lifecycleRevision !== lifecycleRevision || state.ctx !== ctx) {
+      throw new Error("ACPX runtime service stopped during activation");
+    }
+    if (!publishedRuntime) {
      throw new Error("ACPX runtime service did not register an ACP backend");
    }
-    state.realRuntime = backend.runtime;
-    return state.realRuntime;
+    if (getAcpRuntimeBackend(ACPX_BACKEND_ID)?.runtime !== publishedRuntime) {
+      throw new Error("ACPX runtime service lost registry ownership during activation");
+    }
+    // Registry publication intentionally precedes the startup probe, but callers
+    // must keep sharing the start promise until the inner service is fully ready.
+    state.realRuntime = publishedRuntime;
+    return publishedRuntime;
  })();
  try {
    return await state.startPromise;
  } catch (error) {
-    state.startPromise = null;
-    state.realService = null;
+    if (state.lifecycleRevision === lifecycleRevision) {
+      state.startPromise = null;
+      state.realService = null;
+    }
    throw error;
  }
 }

-function createDeferredRuntime(state: DeferredServiceState): AcpRuntime {
-  const resolveRuntime = () => startRealService(state);
-  return createLazyAcpRuntimeProxy(resolveRuntime);
+function createDeferredRuntime(state: DeferredServiceState, lifecycleRevision: number): AcpRuntime {
+  const deferredRuntime: AcpRuntime = createLazyAcpRuntimeProxy(
+    (): Promise<AcpRuntime> => startRealService(state, lifecycleRevision, deferredRuntime),
+  );
+  return deferredRuntime;
 }

 /** Creates the plugin service that registers ACPX as an ACP runtime backend. */
@@ -68,10 +118,13 @@ export function createAcpxRuntimeService(
 ): OpenClawPluginService {
  const state: DeferredServiceState = {
    ctx: null,
+    lifecycleRevision: 0,
+    ownedRuntime: null,
    params,
    realRuntime: null,
    realService: null,
    startPromise: null,
+    stopPromise: null,
  };

  return {
@@ -81,24 +134,50 @@ export function createAcpxRuntimeService(
        ctx.logger.info("skipping embedded acpx runtime backend (OPENCLAW_SKIP_ACPX_RUNTIME=1)");
        return;
      }
+      if (state.stopPromise) {
+        await state.stopPromise;
+      }

+      state.lifecycleRevision += 1;
+      const lifecycleRevision = state.lifecycleRevision;
      state.ctx = ctx;
+      const deferredRuntime = createDeferredRuntime(state, lifecycleRevision);
+      state.ownedRuntime = deferredRuntime;
      registerAcpRuntimeBackend({
        id: ACPX_BACKEND_ID,
-        runtime: createDeferredRuntime(state),
+        runtime: deferredRuntime,
      });
      ctx.logger.info("embedded acpx runtime backend registered lazily");
    },
    async stop(ctx) {
-      if (state.realService) {
-        await state.realService.stop?.(ctx);
-      } else {
-        unregisterAcpRuntimeBackend(ACPX_BACKEND_ID);
+      if (state.stopPromise) {
+        return await state.stopPromise;
      }
+
+      // Invalidate every deferred proxy before waiting for startup. The in-flight
+      // service still owns cleanup, but it can no longer become the active runtime.
+      state.lifecycleRevision += 1;
      state.ctx = null;
-      state.realRuntime = null;
-      state.realService = null;
-      state.startPromise = null;
+      const ownedRuntime = state.ownedRuntime;
+      unregisterOwnedRuntime(ownedRuntime);
+      const startPromise = state.startPromise;
+      state.stopPromise = (async () => {
+        await startPromise?.catch(() => undefined);
+        try {
+          await state.realService?.stop?.(ctx);
+        } finally {
+          unregisterOwnedRuntime(ownedRuntime);
+          state.ownedRuntime = null;
+          state.realRuntime = null;
+          state.realService = null;
+          state.startPromise = null;
+        }
+      })();
+      try {
+        await state.stopPromise;
+      } finally {
+        state.stopPromise = null;
+      }
    },
  };
 }
@@ -71,12 +71,6 @@ const { acpxRuntimeConstructorMock, createAgentRegistryMock, createFileSessionSt

 vi.mock("../runtime-api.js", () => ({
  getAcpRuntimeBackend: (id: string) => runtimeRegistry.get(id),
-  registerAcpRuntimeBackend: (entry: { id: string; runtime: unknown; healthy?: () => boolean }) => {
-    runtimeRegistry.set(entry.id, entry);
-  },
-  unregisterAcpRuntimeBackend: (id: string) => {
-    runtimeRegistry.delete(id);
-  },
 }));

 vi.mock("./runtime.js", () => ({
@@ -173,10 +167,23 @@ function createOpenKeyedStore(ctx: OpenClawPluginServiceContext) {

 function createAcpxRuntimeService(
  ctx: OpenClawPluginServiceContext,
-  params: Parameters<typeof createRealAcpxRuntimeService>[0] = {},
+  params: Omit<Parameters<typeof createRealAcpxRuntimeService>[0], "backendLifecycle"> & {
+    backendLifecycle?: Parameters<typeof createRealAcpxRuntimeService>[0]["backendLifecycle"];
+  } = {},
 ) {
+  const backendLifecycle = params.backendLifecycle ?? {
+    publish(backend: { runtime: unknown; healthy?: () => boolean }) {
+      runtimeRegistry.set("acpx", backend);
+    },
+    retract(runtime: unknown) {
+      if (runtimeRegistry.get("acpx")?.runtime === runtime) {
+        runtimeRegistry.delete("acpx");
+      }
+    },
+  };
  return createRealAcpxRuntimeService({
    ...params,
+    backendLifecycle,
    openKeyedStore: params.openKeyedStore ?? createOpenKeyedStore(ctx),
  });
 }
@@ -205,6 +212,14 @@ function createMockRuntime(overrides: Record<string, unknown> = {}) {
  };
 }

+function createDeferred() {
+  let resolve: () => void = () => {};
+  const promise = new Promise<void>((resolvePromise) => {
+    resolve = resolvePromise;
+  });
+  return { promise, resolve };
+}
+
 function createStartupTraceRecorder() {
  const measured: string[] = [];
  const details: Array<{
@@ -266,6 +281,49 @@ describe("createAcpxRuntimeService", () => {
    expect(getAcpRuntimeBackend("acpx")).toBeUndefined();
  });

+  it("publishes before probing and retracts the exact runtime through the injected lifecycle", async () => {
+    delete process.env.OPENCLAW_ACPX_RUNTIME_STARTUP_PROBE;
+    const workspaceDir = await makeTempDir();
+    const ctx = createServiceContext(workspaceDir);
+    const probeStarted = createDeferred();
+    const releaseProbe = createDeferred();
+    const events: string[] = [];
+    const runtime = createMockRuntime({
+      probeAvailability: vi.fn(async () => {
+        events.push("probe");
+        probeStarted.resolve();
+        await releaseProbe.promise;
+      }),
+    });
+    const publish = vi.fn((backend: { runtime: unknown; healthy?: () => boolean }) => {
+      events.push("publish");
+      expect(backend.runtime).toBe(runtime);
+      expect(backend.healthy?.()).toBe(true);
+    });
+    const retract = vi.fn((ownedRuntime: unknown) => {
+      events.push("retract");
+      expect(ownedRuntime).toBe(runtime);
+    });
+    const service = createAcpxRuntimeService(ctx, {
+      backendLifecycle: { publish, retract },
+      runtimeFactory: () => runtime as never,
+    });
+
+    const starting = service.start(ctx) as Promise<void>;
+    await probeStarted.promise;
+
+    expect(events).toEqual(["publish", "probe"]);
+    expect(publish).toHaveBeenCalledOnce();
+    expect(getAcpRuntimeBackend("acpx")).toBeUndefined();
+
+    await service.stop?.(ctx);
+    expect(retract).toHaveBeenCalledWith(runtime);
+    expect(events).toEqual(["publish", "probe", "retract"]);
+
+    releaseProbe.resolve();
+    await starting;
+  });
+
  it("skips the startup probe and does not advertise backend health when explicitly disabled", async () => {
    process.env.OPENCLAW_ACPX_RUNTIME_STARTUP_PROBE = "0";
    delete process.env.OPENCLAW_SKIP_ACPX_RUNTIME_PROBE;
@@ -20,7 +20,6 @@ import type {
  OpenClawPluginServiceContext,
  PluginLogger,
 } from "../runtime-api.js";
-import { registerAcpRuntimeBackend, unregisterAcpRuntimeBackend } from "../runtime-api.js";
 import { prepareAcpxCodexAuthConfig } from "./codex-auth-bridge.js";
 import { DEFAULT_ACPX_TIMEOUT_SECONDS } from "./config-schema.js";
 import {
@@ -58,7 +57,6 @@ type AcpxRuntimeLike = AcpRuntime & {
 };
 const ENABLE_STARTUP_PROBE_ENV = "OPENCLAW_ACPX_RUNTIME_STARTUP_PROBE";
 const SKIP_RUNTIME_PROBE_ENV = "OPENCLAW_SKIP_ACPX_RUNTIME_PROBE";
-const ACPX_BACKEND_ID = "acpx";

 type AcpxRuntimeFactoryParams = {
  pluginConfig: ResolvedAcpxPluginConfig;
@@ -68,7 +66,13 @@ type AcpxRuntimeFactoryParams = {
  logger?: PluginLogger;
 };

+type AcpxBackendLifecycle = {
+  publish: (backend: { runtime: AcpRuntime; healthy?: () => boolean }) => void;
+  retract: (runtime: AcpRuntime) => void;
+};
+
 type CreateAcpxRuntimeServiceParams = {
+  backendLifecycle: AcpxBackendLifecycle;
  pluginConfig?: unknown;
  openKeyedStore?: <T>(options: OpenKeyedStoreOptions) => PluginStateKeyedStore<T>;
  runtimeFactory?: (params: AcpxRuntimeFactoryParams) => AcpxRuntimeLike | Promise<AcpxRuntimeLike>;
@@ -324,7 +328,7 @@ async function reapOpenAcpxProcessLeases(params: {

 /** Create the ACPX plugin service that owns runtime registration and cleanup. */
 export function createAcpxRuntimeService(
-  params: CreateAcpxRuntimeServiceParams = {},
+  params: CreateAcpxRuntimeServiceParams,
 ): OpenClawPluginService {
  let runtime: AcpxRuntimeLike | null = null;
  let lifecycleRevision = 0;
@@ -411,11 +415,11 @@ export function createAcpxRuntimeService(
        ["probeAgent", pluginConfig.probeAgent ?? "default"],
      ]);
      await measureAcpxStartup(ctx, "backend.register", () => {
-        registerAcpRuntimeBackend({
-          id: ACPX_BACKEND_ID,
+        const backend = {
          runtime: startedRuntime,
          ...(shouldProbeRuntime ? { healthy: () => runtime?.isHealthy() ?? false } : {}),
-        });
+        };
+        params.backendLifecycle.publish(backend);
        ctx.logger.info(`embedded acpx runtime backend registered (cwd: ${pluginConfig.cwd})`);
      });

@@ -460,7 +464,9 @@ export function createAcpxRuntimeService(
    },
    async stop(_ctx: OpenClawPluginServiceContext): Promise<void> {
      lifecycleRevision += 1;
-      unregisterAcpRuntimeBackend(ACPX_BACKEND_ID);
+      if (runtime) {
+        params.backendLifecycle.retract(runtime);
+      }
      runtime = null;
    },
  };