fix: recover SSE directory routing in sync pipeline (#830)

packages/ui/src/sync/__tests__/event-pipeline.test.js

@@ -0,0 +1,195 @@
+import { afterEach, describe, expect, it } from 'bun:test';
+import { createEventPipeline } from '../event-pipeline';
+
+const originalDocument = globalThis.document;
+const originalWindow = globalThis.window;
+
+function installDomStubs() {
+  globalThis.document = {
+    visibilityState: 'visible',
+    addEventListener() {},
+    removeEventListener() {},
+  };
+
+  globalThis.window = {
+    addEventListener() {},
+    removeEventListener() {},
+  };
+}
+
+afterEach(() => {
+  globalThis.document = originalDocument;
+  globalThis.window = originalWindow;
+});
+
+function createSdkWithSingleEvent(event, hold) {
+  return {
+    global: {
+      event: async () => ({
+        stream: (async function* () {
+          yield event;
+          await hold;
+        })(),
+      }),
+    },
+  };
+}
+
+describe('createEventPipeline', () => {
+  it('falls back to payload.properties.directory when the SDK event omits top-level directory', async () => {
+    installDomStubs();
+
+    let releaseStream;
+    const hold = new Promise((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const received = [];
+    const sdk = createSdkWithSingleEvent({
+      payload: {
+        type: 'session.status',
+        properties: {
+          directory: 'C:/Users/daveotero/localdev/openchamber',
+          sessionID: 'session-1',
+          status: { type: 'busy' },
+        },
+      },
+    }, hold);
+
+    const delivered = new Promise((resolve) => {
+      const { cleanup } = createEventPipeline({
+        sdk,
+        onEvent: (directory, payload) => {
+          received.push({ directory, payload });
+          cleanup();
+          releaseStream();
+          resolve();
+        },
+      });
+    });
+
+    await delivered;
+
+    expect(received).toHaveLength(1);
+    expect(received[0].directory).toBe('C:/Users/daveotero/localdev/openchamber');
+    expect(received[0].payload.type).toBe('session.status');
+  });
+
+  it('prefers the explicit top-level event directory when present', async () => {
+    installDomStubs();
+
+    let releaseStream;
+    const hold = new Promise((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const received = [];
+    const sdk = createSdkWithSingleEvent({
+      directory: 'C:/top-level',
+      payload: {
+        type: 'session.status',
+        properties: {
+          directory: 'C:/nested',
+          sessionID: 'session-2',
+          status: { type: 'busy' },
+        },
+      },
+    }, hold);
+
+    const delivered = new Promise((resolve) => {
+      const { cleanup } = createEventPipeline({
+        sdk,
+        onEvent: (directory, payload) => {
+          received.push({ directory, payload });
+          cleanup();
+          releaseStream();
+          resolve();
+        },
+      });
+    });
+
+    await delivered;
+
+    expect(received).toHaveLength(1);
+    expect(received[0].directory).toBe('C:/top-level');
+    expect(received[0].payload.type).toBe('session.status');
+  });
+
+  it('uses payload.properties.directory when the top-level directory is an empty string', async () => {
+    installDomStubs();
+
+    let releaseStream;
+    const hold = new Promise((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const received = [];
+    const sdk = createSdkWithSingleEvent({
+      directory: '',
+      payload: {
+        type: 'message.part.updated',
+        properties: {
+          directory: 'C:/fallback-dir',
+          part: {
+            id: 'part-1',
+            type: 'text',
+            messageID: 'message-1',
+          },
+        },
+      },
+    }, hold);
+
+    const delivered = new Promise((resolve) => {
+      const { cleanup } = createEventPipeline({
+        sdk,
+        onEvent: (directory, payload) => {
+          received.push({ directory, payload });
+          cleanup();
+          releaseStream();
+          resolve();
+        },
+      });
+    });
+
+    await delivered;
+
+    expect(received).toHaveLength(1);
+    expect(received[0].directory).toBe('C:/fallback-dir');
+    expect(received[0].payload.type).toBe('message.part.updated');
+  });
+
+  it('keeps truly global events on the global channel when no directory is present anywhere', async () => {
+    installDomStubs();
+
+    let releaseStream;
+    const hold = new Promise((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const received = [];
+    const sdk = createSdkWithSingleEvent({
+      payload: {
+        type: 'server.connected',
+        properties: {},
+      },
+    }, hold);
+
+    const delivered = new Promise((resolve) => {
+      const { cleanup } = createEventPipeline({
+        sdk,
+        onEvent: (directory, payload) => {
+          received.push({ directory, payload });
+          cleanup();
+          releaseStream();
+          resolve();
+        },
+      });
+    });
+
+    await delivered;
+
+    expect(received).toHaveLength(1);
+    expect(received[0].directory).toBe('global');
+    expect(received[0].payload.type).toBe('server.connected');
+  });
+});

packages/ui/src/sync/event-pipeline.ts

@@ -41,6 +41,25 @@ export type EventPipelineInput = {
   onReconnect?: () => void
 }
 
+function resolveEventDirectory(event: unknown, payload: Event): string {
+  const directDirectory =
+    typeof event === "object" && event !== null && typeof (event as { directory?: unknown }).directory === "string"
+      ? (event as { directory: string }).directory
+      : null
+
+  if (directDirectory && directDirectory.length > 0) {
+    return directDirectory
+  }
+
+  const properties =
+    typeof payload.properties === "object" && payload.properties !== null
+      ? (payload.properties as Record<string, unknown>)
+      : null
+  const propertyDirectory = typeof properties?.directory === "string" ? properties.directory : null
+
+  return propertyDirectory && propertyDirectory.length > 0 ? propertyDirectory : "global"
+}
+
 export function createEventPipeline(input: EventPipelineInput) {
   const { sdk, onEvent, onReconnect } = input
   const abort = new AbortController()
@@ -167,11 +186,11 @@ export function createEventPipeline(input: EventPipelineInput) {
         for await (const event of events.stream) {
           resetHeartbeat()
           streamErrorLogged = false
-          const directory = (event as { directory?: string }).directory ?? "global"
           const payload = (event as { payload?: Event }).payload ?? (event as unknown as Event)
           if (!payload || typeof payload !== "object" || typeof (payload as { type?: unknown }).type !== "string") {
             continue
           }
+          const directory = resolveEventDirectory(event, payload)
           const k = key(directory, payload)
           if (k) {
             const i = coalesced.get(k)