fix(gateway): guard stale task observer disposal against replacement …

…gateways
openclaw · steipete · Jul 6, 2026 · Jul 6, 2026 · Jul 6, 2026 · Jul 6, 2026
commit 8b2cbd0cd27c23439afd26135b5a76627b23fc47
diff --git a/src/gateway/server-runtime-subscriptions.test.ts b/src/gateway/server-runtime-subscriptions.test.ts
@@ -233,4 +233,50 @@ describe("startGatewayEventSubscriptions", () => {
     });
     expect(broadcast).not.toHaveBeenCalled();
   });
+
+  it("keeps a replacement gateway's task observer when a stale unsub runs late", async () => {
+    const staleBroadcast = vi.fn<SubscriptionParams["broadcast"]>();
+    const staleSubs = startGatewayEventSubscriptions({
+      ...createParams(),
+      broadcast: staleBroadcast,
+    });
+    await vi.waitFor(() => expect(getTaskRegistryObservers()).not.toBeNull());
+    const staleObservers = getTaskRegistryObservers();
+
+    const replacementBroadcast = vi.fn<SubscriptionParams["broadcast"]>();
+    unsubs = startGatewayEventSubscriptions({
+      ...createParams(),
+      broadcast: replacementBroadcast,
+    });
+    await vi.waitFor(() => {
+      const current = getTaskRegistryObservers();
+      expect(current).not.toBeNull();
+      expect(current).not.toBe(staleObservers);
+    });
+
+    // The stale dispose must not clear the replacement's observer slot.
+    await staleSubs.taskUnsub();
+    staleSubs.agentUnsub();
+    staleSubs.heartbeatUnsub();
+    staleSubs.transcriptUnsub();
+    staleSubs.lifecycleUnsub();
+    expect(getTaskRegistryObservers()).not.toBeNull();
+
+    createTaskRecord({
+      runtime: "cli",
+      requesterSessionKey: "agent:main:main",
+      ownerKey: "agent:main:main",
+      scopeKind: "session",
+      task: "After stale dispose",
+      status: "queued",
+      deliveryStatus: "not_applicable",
+      notifyPolicy: "silent",
+    });
+    expect(replacementBroadcast).toHaveBeenCalledWith("task", expect.anything(), {
+      dropIfSlow: true,
+    });
+    expect(staleBroadcast).not.toHaveBeenCalledWith("task", expect.anything(), {
+      dropIfSlow: true,
+    });
+  });
 });
diff --git a/src/gateway/server-runtime-subscriptions.ts b/src/gateway/server-runtime-subscriptions.ts
@@ -319,43 +319,43 @@ export function startGatewayEventSubscriptions(params: {
   });
 
   let taskObserverDisposed = false;
-  const taskObserverRuntimePromise = import("../tasks/task-registry.store.js").then(
-    ({ configureTaskRegistryRuntime }) => {
-      if (!taskObserverDisposed) {
-        configureTaskRegistryRuntime({
-          observers: {
-            onEvent: (event: TaskRegistryObserverEvent) => {
-              let payload: TaskEventPayload;
-              switch (event.kind) {
-                case "upserted":
-                  payload = { action: "upserted", task: mapTaskSummary(event.task) };
-                  break;
-                case "deleted":
-                  payload = { action: "deleted", taskId: event.taskId };
-                  break;
-                case "restored":
-                  payload = { action: "restored" };
-                  break;
-              }
-              params.broadcast("task", payload, { dropIfSlow: true });
-            },
-          },
-        });
+  const taskObservers = {
+    onEvent: (event: TaskRegistryObserverEvent) => {
+      let payload: TaskEventPayload;
+      switch (event.kind) {
+        case "upserted":
+          payload = { action: "upserted", task: mapTaskSummary(event.task) };
+          break;
+        case "deleted":
+          payload = { action: "deleted", taskId: event.taskId };
+          break;
+        case "restored":
+          payload = { action: "restored" };
+          break;
       }
-      return configureTaskRegistryRuntime;
+      params.broadcast("task", payload, { dropIfSlow: true });
     },
-  );
+  };
+  const taskObserverRuntimePromise = import("../tasks/task-registry.store.js").then((module) => {
+    if (!taskObserverDisposed) {
+      module.configureTaskRegistryRuntime({ observers: taskObservers });
+    }
+    return module;
+  });
   void taskObserverRuntimePromise.catch((error: unknown) => {
     params.log.warn("Task registry observer registration failed", { error });
   });
-  // Return the cleanup promise so shutdown awaits it: the observer slot is a
-  // process-wide singleton, and a replacement gateway registering before this
-  // deferred clear runs would have its observer wiped by the stale cleanup.
+  // The observer slot is a process-wide singleton. Cleanup returns its promise
+  // so shutdown can await it, and only clears the slot when it still holds
+  // this subscription's observer — a replacement gateway may have registered
+  // its own observer before a stale deferred dispose runs.
   const taskUnsub = () => {
     taskObserverDisposed = true;
     return taskObserverRuntimePromise
-      .then((configureTaskRegistryRuntime) => {
-        configureTaskRegistryRuntime({ observers: null });
+      .then((module) => {
+        if (module.getTaskRegistryObservers() === taskObservers) {
+          module.configureTaskRegistryRuntime({ observers: null });
+        }
       })
       .catch(() => undefined);
   };