fix(openclaw): prevent duplicate viewer instances on gateway restart

When OpenClaw calls register() again (deferred reload / gateway restart),
the old viewer HTTP server was not stopped. Each re-registration leaked
a new server on an incrementing port (18799 → 18800 → 18801…).

Root cause: serviceStarted and viewer were closure-scoped inside
register(), so a second call created a fresh closure with no knowledge
of the previous one.

Fix: track active instances at module level, keyed by stateDir.  On
re-registration with the same stateDir, startServiceCore() now awaits
the previous viewer's port release before binding the new one.

Also makes ViewerServer.stop() return a Promise so the caller can
properly wait for the HTTP server to close.

Closes #1471

Author: fancyboi999

apps/memos-local-openclaw/index.ts

@@ -150,6 +150,19 @@ const pluginConfigSchema = {
   },
 };
 
+// ─── Module-level singleton guard ───
+// Prevents duplicate viewers/stores when OpenClaw calls register() multiple
+// times for the SAME stateDir (e.g. deferred reload, gateway restart).
+// Keyed by stateDir so that truly independent instances (different data dirs,
+// as in tests) are not accidentally torn down.
+const activeInstances = new Map<string, {
+  viewer: InstanceType<typeof ViewerServer>;
+  hubServer: InstanceType<typeof HubServer> | null;
+  worker: InstanceType<typeof IngestWorker>;
+  store: InstanceType<typeof SqliteStore>;
+  telemetry: InstanceType<typeof Telemetry>;
+}>();
+
 const memosLocalPlugin = {
   id: "memos-local-openclaw-plugin",
   name: "MemOS Local Memory",
@@ -160,6 +173,19 @@ const memosLocalPlugin = {
   configSchema: pluginConfigSchema,
 
   register(api: OpenClawPluginApi) {
+    // Resolve stateDir early so we can check for a duplicate instance with the
+    // same data directory (deferred reload / gateway restart scenario).
+    const stateDir = process.env.OPENCLAW_STATE_DIR || api.resolvePath("~/.openclaw");
+
+    // Snapshot the previous instance for this stateDir so startServiceCore()
+    // can tear it down asynchronously (awaiting port release) before starting
+    // the new viewer.  Instances with a different stateDir are left untouched.
+    const instanceToReplace = activeInstances.get(stateDir) ?? null;
+    activeInstances.delete(stateDir);
+    if (instanceToReplace) {
+      api.logger.info("memos-local: previous instance detected, will stop before starting new viewer");
+    }
+
     api.registerMemoryCapability({
       promptBuilder: buildMemoryPromptSection,
     });
@@ -286,7 +312,6 @@ const memosLocalPlugin = {
     }
 
     let pluginCfg = (api.pluginConfig ?? {}) as Record<string, unknown>;
-    const stateDir = process.env.OPENCLAW_STATE_DIR || api.resolvePath("~/.openclaw");
 
     // Fallback: read config from file if not provided by OpenClaw
     const configPath = path.join(stateDir, "state", "memos-local", "config.json");
@@ -2378,6 +2403,9 @@ Groups: ${groupNames.length > 0 ? groupNames.join(", ") : "(none)"}`,
       ? new HubServer({ store, log: ctx.log, config: ctx.config, dataDir: stateDir, embedder, defaultHubPort: derivedHubPort })
       : null;
 
+    // Track this instance so the next register() with the same stateDir can tear it down
+    activeInstances.set(stateDir, { viewer, hubServer, worker, store, telemetry });
+
     // ─── Service lifecycle ───
 
     let serviceStarted = false;
@@ -2386,6 +2414,21 @@ Groups: ${groupNames.length > 0 ? groupNames.join(", ") : "(none)"}`,
       if (serviceStarted) return;
       serviceStarted = true;
 
+      // Gracefully shut down the previous instance before binding new ports
+      if (instanceToReplace) {
+        api.logger.info("memos-local: stopping previous instance...");
+        try {
+          await instanceToReplace.viewer.stop();
+          await instanceToReplace.hubServer?.stop();
+          await instanceToReplace.worker.flush().catch(() => {});
+          await instanceToReplace.telemetry.shutdown().catch(() => {});
+          instanceToReplace.store.close();
+        } catch (err) {
+          api.logger.warn(`memos-local: previous instance cleanup error: ${err}`);
+        }
+        api.logger.info("memos-local: previous instance stopped");
+      }
+
       if (hubServer) {
         const hubUrl = await hubServer.start();
         api.logger.info(`memos-local: hub started at ${hubUrl}`);
@@ -2429,10 +2472,11 @@ Groups: ${groupNames.length > 0 ? groupNames.join(", ") : "(none)"}`,
       id: "memos-local-openclaw-plugin",
       start: async () => { await startServiceCore(); },
       stop: async () => {
+        activeInstances.delete(stateDir);
         await worker.flush();
         await telemetry.shutdown();
         await hubServer?.stop();
-        viewer.stop();
+        await viewer.stop();
         store.close();
         api.logger.info("memos-local: stopped");
       },

apps/memos-local-openclaw/src/viewer/server.ts

@@ -211,13 +211,19 @@ export class ViewerServer {
     }
   }
 
-  stop(): void {
+  stop(): Promise<void> {
     this.stopHubHeartbeat();
     this.stopNotifPoll();
     for (const c of this.notifSSEClients) { try { c.end(); } catch {} }
     this.notifSSEClients = [];
-    this.server?.close();
+    if (!this.server) return Promise.resolve();
+    const srv = this.server;
     this.server = null;
+    return new Promise<void>((resolve) => {
+      srv.close(() => resolve());
+      // Force-close idle keep-alive connections so close() doesn't hang
+      srv.closeAllConnections?.();
+    });
   }
 
   getResetToken(): string {

apps/memos-local-openclaw/tests/e2e-duplicate-instance.test.ts

@@ -0,0 +1,146 @@
+/**
+ * Real integration test: verify that calling register() twice stops the
+ * previous viewer instead of leaking a second HTTP server on a new port.
+ *
+ * Spins up actual HTTP servers (ViewerServer) and verifies:
+ *   1. Old instance is torn down before new one starts
+ *   2. New viewer binds to the same port (not port+1)
+ *   3. Only one HTTP server is running after re-registration
+ */
+
+import { describe, it, expect, afterAll } from "vitest";
+import http from "node:http";
+import net from "node:net";
+import os from "node:os";
+import fs from "node:fs";
+import path from "node:path";
+
+const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "memos-dup-test-"));
+fs.mkdirSync(path.join(tmpDir, "memos-local"), { recursive: true });
+fs.mkdirSync(path.join(tmpDir, "workspace", "skills"), { recursive: true });
+fs.mkdirSync(path.join(tmpDir, "skills"), { recursive: true });
+
+/** Find a free port by temporarily binding to port 0 */
+function findFreePort(): Promise<number> {
+  return new Promise((resolve, reject) => {
+    const srv = net.createServer();
+    srv.listen(0, "127.0.0.1", () => {
+      const port = (srv.address() as net.AddressInfo).port;
+      srv.close(() => resolve(port));
+    });
+    srv.on("error", reject);
+  });
+}
+
+/** Check if a port is in use by attempting TCP connect */
+function isPortListening(port: number): Promise<boolean> {
+  return new Promise((resolve) => {
+    const sock = net.createConnection({ host: "127.0.0.1", port }, () => {
+      sock.destroy();
+      resolve(true);
+    });
+    sock.on("error", () => resolve(false));
+    sock.setTimeout(1000, () => { sock.destroy(); resolve(false); });
+  });
+}
+
+function createMockApi(stateDir: string, gatewayPort: number) {
+  const services: Array<{ id: string; start: () => Promise<void>; stop: () => Promise<void> }> = [];
+  const logs: string[] = [];
+  return {
+    api: {
+      id: "memos-local-openclaw-plugin",
+      pluginConfig: {},
+      config: { gateway: { port: gatewayPort } },
+      resolvePath: (p: string) => p.replace("~/.openclaw", stateDir),
+      logger: {
+        info: (msg: string) => logs.push(msg),
+        warn: (msg: string) => logs.push(msg),
+        error: (msg: string) => logs.push(msg),
+        debug: (msg: string) => logs.push(msg),
+      },
+      registerTool: () => {},
+      registerMemoryCapability: () => {},
+      registerService: (svc: any) => { services.push(svc); },
+      registerHook: () => {},
+      on: () => {},
+    },
+    services,
+    logs,
+  };
+}
+
+let lastCleanup: (() => Promise<void>) | null = null;
+
+afterAll(async () => {
+  if (lastCleanup) await lastCleanup();
+  await new Promise((r) => setTimeout(r, 200));
+  fs.rmSync(tmpDir, { recursive: true, force: true });
+});
+
+/** Extract the viewer port from logs like "→ http://127.0.0.1:19799" */
+function extractViewerPort(logs: string[]): number | null {
+  for (const l of logs) {
+    const m = l.match(/→\s*http:\/\/127\.0\.0\.1:(\d+)/);
+    if (m) return parseInt(m[1], 10);
+  }
+  return null;
+}
+
+describe("duplicate instance prevention (real HTTP servers)", () => {
+  it("second register() should stop previous viewer and reuse the same port", async () => {
+    // Pick a random free port to avoid collisions with other processes
+    const freePort = await findFreePort();
+    const gatewayPort = freePort;
+    const expectedViewerPort = gatewayPort + 10;
+
+    // Confirm our ports are actually free
+    expect(await isPortListening(expectedViewerPort)).toBe(false);
+    expect(await isPortListening(expectedViewerPort + 1)).toBe(false);
+
+    const pluginModule = await import("../index");
+    const plugin = pluginModule.default;
+
+    // ─── 1st register + start ───
+    const mock1 = createMockApi(tmpDir, gatewayPort);
+    plugin.register(mock1.api as any);
+    const svc1 = mock1.services[mock1.services.length - 1];
+    await svc1.start();
+
+    const viewerPort1 = extractViewerPort(mock1.logs);
+    expect(viewerPort1).toBe(expectedViewerPort);
+    expect(await isPortListening(expectedViewerPort)).toBe(true);
+
+    // ─── 2nd register (simulates deferred reload / gateway restart) ───
+    const mock2 = createMockApi(tmpDir, gatewayPort);
+    plugin.register(mock2.api as any);
+
+    // Verify the plugin detected the previous instance
+    const detectedMsg = mock2.logs.some((l) => l.includes("previous instance detected"));
+    expect(detectedMsg).toBe(true);
+
+    // Start the 2nd service — cleanup happens inside startServiceCore()
+    const svc2 = mock2.services[mock2.services.length - 1];
+    await svc2.start();
+
+    lastCleanup = async () => { await svc2.stop(); };
+
+    // Verify: stopped previous, then started on the SAME port
+    const stoppedMsg = mock2.logs.some((l) => l.includes("previous instance stopped"));
+    expect(stoppedMsg).toBe(true);
+
+    const viewerPort2 = extractViewerPort(mock2.logs);
+    expect(viewerPort2).toBe(expectedViewerPort);  // Same port, not port+1
+
+    // Verify: only ONE server is running (the expected port), not two
+    expect(await isPortListening(expectedViewerPort)).toBe(true);
+    expect(await isPortListening(expectedViewerPort + 1)).toBe(false);
+
+    // ─── Clean stop ───
+    await svc2.stop();
+    lastCleanup = null;
+    await new Promise((r) => setTimeout(r, 200));
+
+    expect(await isPortListening(expectedViewerPort)).toBe(false);
+  });
+});

apps/memos-local-openclaw/tests/shutdown-lifecycle.test.ts

@@ -83,7 +83,7 @@ describe("shutdown lifecycle", () => {
 
     class MockViewer {
       async start(): Promise<string> { return "http://127.0.0.1:18799"; }
-      stop(): void { events.push("viewer-stop"); }
+      async stop(): Promise<void> { events.push("viewer-stop"); }
       getResetToken(): string { return "token"; }
     }