Fix double unsubscribe removing other bus handlers

apps/x/packages/core/src/application/lib/bus.test.ts

@@ -0,0 +1,40 @@
+import { describe, expect, it } from "vitest";
+import z from "zod";
+import { RunEvent } from "@x/shared/dist/runs.js";
+import { InMemoryBus } from "./bus.js";
+
+function makeEvent(runId: string): z.infer<typeof RunEvent> {
+    return {
+        type: "run-processing-start",
+        runId,
+        subflow: [],
+    };
+}
+
+describe("InMemoryBus", () => {
+    it("keeps other handlers subscribed when a handler is unsubscribed twice", async () => {
+        const bus = new InMemoryBus();
+        const runId = "run-1";
+
+        const receivedByA: z.infer<typeof RunEvent>[] = [];
+        const receivedByB: z.infer<typeof RunEvent>[] = [];
+
+        const unsubscribeA = await bus.subscribe(runId, async (event) => {
+            receivedByA.push(event);
+        });
+        await bus.subscribe(runId, async (event) => {
+            receivedByB.push(event);
+        });
+
+        // Double-unsubscribe of A must be a no-op the second time. The buggy
+        // implementation ran splice(indexOf(A), 1) where indexOf(A) === -1 on
+        // the second call, which removed the last handler (B) instead.
+        unsubscribeA();
+        unsubscribeA();
+
+        await bus.publish(makeEvent(runId));
+
+        expect(receivedByA).toHaveLength(0);
+        expect(receivedByB).toHaveLength(1);
+    });
+});

apps/x/packages/core/src/application/lib/bus.ts

@@ -29,7 +29,10 @@ export class InMemoryBus implements IBus {
         }
         this.subscribers.get(runId)!.push(handler);
         return () => {
-            this.subscribers.get(runId)!.splice(this.subscribers.get(runId)!.indexOf(handler), 1);
+            const handlers = this.subscribers.get(runId);
+            if (!handlers) return;
+            const idx = handlers.indexOf(handler);
+            if (idx >= 0) handlers.splice(idx, 1);
         };
     }
 }