refactor(gateway): drop dead session-subscriber registry surface (#124754)

* refactor(gateway): drop dead session-subscriber registry surface

getForConnection and both registry clear() methods had no production
callers — only server-chat-state.test.ts consumed them (verified by
repo-wide grep incl. plugin-sdk). The recency-sorted connToSessionKeys
index existed solely so getForConnection returned keys in subscription
order, an ordering nothing read, rebuilt via toSorted() on every
subscribe/unsubscribe/settle — including the provisional-replay settle
path.

unsubscribeAll now reads the reverse index directly from
connToSessionRecency's keys (same contents, no ordering), removing the
per-call O(k log k) rebuild and the duplicate map. Tests assert through
get()/getApprovals(), the production read paths, per test-audit norms.

Behavior-neutral: subscription state transitions and event-visible
delivery unchanged; chat-state, broadcast, node-subscription,
instance-runtime, agent-events, session-events suites pass.

* test(gateway): fix ETXTBSY flake in workspace-sync candidate bounding

The test rewrote its mock git script between two createGitTransferList
calls. On Linux CI the second spawn can race a lingering fork of the
first child still holding the script's write fd, failing exec with
ETXTBSY. Write the script once; vary the entry count through a data file
the script reads instead.

src/gateway/server-chat-state.test.ts

@@ -168,15 +168,10 @@ describe("createSessionMessageSubscriberRegistry", () => {
     subscribers.subscribe("conn-other", "agent:main:child", { includeApprovals: true });
 
     subscribers.unsubscribeAll("conn-reviewer");
-    expect([...subscribers.getForConnection("conn-reviewer")]).toEqual([]);
     expect([...subscribers.get("agent:main:main")]).toEqual([]);
     expect([...subscribers.getApprovals("agent:main:main")]).toEqual([]);
     expect([...subscribers.get("agent:main:child")]).toEqual(["conn-other"]);
     expect([...subscribers.getApprovals("agent:main:child")]).toEqual(["conn-other"]);
-
-    subscribers.clear();
-    expect([...subscribers.get("agent:main:child")]).toEqual([]);
-    expect([...subscribers.getApprovals("agent:main:child")]).toEqual([]);
   });
 
   it.each(["first", "second"])(
@@ -195,7 +190,6 @@ describe("createSessionMessageSubscriberRegistry", () => {
       }
 
       expect([...subscribers.get("agent:main:main")]).toEqual([]);
-      expect([...subscribers.getForConnection("conn")]).toEqual([]);
     },
   );
 
@@ -215,10 +209,8 @@ describe("createSessionMessageSubscriberRegistry", () => {
         first();
       }
 
-      expect([...subscribers.getForConnection("conn")]).toEqual([
-        "agent:main:other",
-        "agent:main:main",
-      ]);
+      expect([...subscribers.get("agent:main:other")]).toEqual(["conn"]);
+      expect([...subscribers.get("agent:main:main")]).toEqual(["conn"]);
     },
   );
 
@@ -230,10 +222,8 @@ describe("createSessionMessageSubscriberRegistry", () => {
 
     rollback();
 
-    expect([...subscribers.getForConnection("conn")]).toEqual([
-      "agent:main:main",
-      "agent:main:child",
-    ]);
+    expect([...subscribers.get("agent:main:main")]).toEqual(["conn"]);
+    expect([...subscribers.get("agent:main:child")]).toEqual(["conn"]);
   });
 
   it("does not restore a replay invalidated by unsubscribe", () => {
@@ -245,7 +235,6 @@ describe("createSessionMessageSubscriberRegistry", () => {
     subscribers.unsubscribe("conn", "agent:main:main");
     subscription.commit();
 
-    expect([...subscribers.getForConnection("conn")]).toEqual([]);
     expect([...subscribers.get("agent:main:main")]).toEqual([]);
   });
 });

src/gateway/server-chat-state.ts

@@ -338,7 +338,6 @@ export type SessionEventSubscriberRegistry = {
   subscribe: (connId: string) => void;
   unsubscribe: (connId: string) => void;
   getAll: () => ReadonlySet<string>;
-  clear: () => void;
 };
 
 export type SessionMessageSubscriberRegistry = {
@@ -350,10 +349,8 @@ export type SessionMessageSubscriberRegistry = {
   unsubscribe: (connId: string, sessionKey: string) => void;
   unsubscribeAll: (connId: string) => void;
   get: (sessionKey: string) => ReadonlySet<string>;
-  getForConnection: (connId: string) => ReadonlySet<string>;
   getApprovals: (sessionKey: string) => ReadonlySet<string>;
   onChange: (listener: (sessionKey: string) => void) => () => void;
-  clear: () => void;
 };
 
 type SessionMessageSubscription = (() => void) & { commit: () => void };
@@ -391,18 +388,15 @@ export function createSessionEventSubscriberRegistry(): SessionEventSubscriberRe
       connIds.delete(normalized);
     },
     getAll: () => (connIds.size > 0 ? connIds : empty),
-    clear: () => {
-      connIds.clear();
-    },
   };
 }
 
 /** Create the per-session message subscriber registry. */
 export function createSessionMessageSubscriberRegistry(): SessionMessageSubscriberRegistry {
   const sessionToConnIds = new Map<string, Set<string>>();
-  const connToSessionKeys = new Map<string, Set<string>>();
   // The final state after overlapping replays settles to their latest success
   // or the original committed base; failed provisionals cannot leave ghosts.
+  // Its keys double as the per-connection reverse index for unsubscribeAll.
   const connToSessionRecency = new Map<string, Map<string, number>>();
   const provisionalSubscriptions = new Map<string, Map<string, ProvisionalSubscriptionState>>();
   const approvalSessionToConnIds = new Map<string, Set<string>>();
@@ -412,17 +406,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib
   let subscriptionSequence = 0;
 
   const normalize = (value: string): string => value.trim();
-  const rebuildConnectionSessionKeys = (connId: string) => {
-    const recency = connToSessionRecency.get(connId);
-    if (!recency || recency.size === 0) {
-      connToSessionKeys.delete(connId);
-      return;
-    }
-    connToSessionKeys.set(
-      connId,
-      new Set([...recency.entries()].toSorted(([, a], [, b]) => a - b).map(([key]) => key)),
-    );
-  };
   const setMessageSubscription = (connId: string, sessionKey: string, subscribed: boolean) => {
     const connIds = sessionToConnIds.get(sessionKey);
     const wasSubscribed = connIds?.has(connId) === true;
@@ -497,7 +480,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib
       setMessageSubscription(normalizedConnId, normalizedSessionKey, true);
       recency.set(normalizedSessionKey, provisionalRecency);
       connToSessionRecency.set(normalizedConnId, recency);
-      rebuildConnectionSessionKeys(normalizedConnId);
 
       setApprovalSubscription(
         normalizedConnId,
@@ -537,7 +519,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib
         if (recency.size === 0) {
           connToSessionRecency.delete(normalizedConnId);
         }
-        rebuildConnectionSessionKeys(normalizedConnId);
         states.delete(normalizedSessionKey);
         if (states.size === 0) {
           provisionalSubscriptions.delete(normalizedConnId);
@@ -573,7 +554,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib
         if (recency.size === 0) {
           connToSessionRecency.delete(normalizedConnId);
         }
-        rebuildConnectionSessionKeys(normalizedConnId);
       }
       const approvalConnIds = approvalSessionToConnIds.get(normalizedSessionKey);
       if (approvalConnIds) {
@@ -600,14 +580,13 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib
         state.active = false;
       }
       provisionalSubscriptions.delete(normalizedConnId);
-      const sessionKeys = connToSessionKeys.get(normalizedConnId);
+      const sessionKeys = connToSessionRecency.get(normalizedConnId);
       if (!sessionKeys) {
         return;
       }
-      for (const sessionKey of sessionKeys) {
+      for (const sessionKey of sessionKeys.keys()) {
         setMessageSubscription(normalizedConnId, sessionKey, false);
       }
-      connToSessionKeys.delete(normalizedConnId);
       connToSessionRecency.delete(normalizedConnId);
 
       const approvalSessionKeys = connToApprovalSessionKeys.get(normalizedConnId);
@@ -627,13 +606,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib
       }
       return sessionToConnIds.get(normalizedSessionKey) ?? empty;
     },
-    getForConnection: (connId: string) => {
-      const normalizedConnId = normalize(connId);
-      if (!normalizedConnId) {
-        return empty;
-      }
-      return connToSessionKeys.get(normalizedConnId) ?? empty;
-    },
     getApprovals: (sessionKey: string) => {
       const normalizedSessionKey = normalize(sessionKey);
       if (!normalizedSessionKey) {
@@ -645,25 +617,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib
       changeListeners.add(listener);
       return () => changeListeners.delete(listener);
     },
-    clear: () => {
-      const changedSessionKeys = [...sessionToConnIds.keys()].toSorted();
-      sessionToConnIds.clear();
-      connToSessionKeys.clear();
-      connToSessionRecency.clear();
-      for (const states of provisionalSubscriptions.values()) {
-        for (const state of states.values()) {
-          state.active = false;
-        }
-      }
-      provisionalSubscriptions.clear();
-      approvalSessionToConnIds.clear();
-      connToApprovalSessionKeys.clear();
-      for (const sessionKey of changedSessionKeys) {
-        for (const listener of changeListeners) {
-          listener(sessionKey);
-        }
-      }
-    },
   };
   return registry;
 }

src/gateway/worker-environments/workspace-sync-local.test.ts

@@ -268,19 +268,26 @@ describe("runLocalCommandToFile", () => {
     const root = tempDirs.make("openclaw-workspace-candidates-");
     const bin = path.join(root, "bin");
     const mockGit = path.join(bin, "git");
+    const countFile = path.join(bin, "git-entry-count");
     const firstTransfer = `${root}-transfer-accepted`;
     const secondTransfer = `${root}-transfer-rejected`;
+    // Rewriting an executed script between spawns races exec against a forked
+    // child still holding the write fd (ETXTBSY). Write the script once and
+    // vary only a data file it reads.
     const writeMockGit = async (count: number) => {
-      await fs.writeFile(
-        mockGit,
-        `#!/usr/bin/env node
-process.stdout.write("eligible.txt\\0".repeat(${count}));
-`,
-        { mode: 0o755 },
-      );
+      await fs.writeFile(countFile, String(count));
     };
     try {
       await fs.mkdir(bin);
+      await fs.writeFile(
+        mockGit,
+        `#!/usr/bin/env node
+const fs = require("node:fs");
+const count = Number(fs.readFileSync(${JSON.stringify(countFile)}, "utf8"));
+process.stdout.write("eligible.txt\\0".repeat(count));
+`,
+        { mode: 0o755 },
+      );
       await fs.writeFile(path.join(root, "eligible.txt"), "eligible\n");
       vi.stubEnv("PATH", `${bin}${path.delimiter}${process.env.PATH ?? ""}`);
       await writeMockGit(MAX_WORKSPACE_INVENTORY_ENTRIES + 1);