From d69e97802a93178a38fe06ff6bc497ac7574818c Mon Sep 17 00:00:00 2001 From: amicode-ci Date: Thu, 24 Sep 2026 15:04:39 -0400 Subject: [PATCH] fix(app): gate directory-store mutations for remote SSE events (#1539) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SSE fan-in relays remote peer events into the app's single /event SSE connection. The fan-in correctly namespaces events via composite cursor id: fields, but the app-layer event handler was applying ALL events to the directory store regardless of origin — contaminating store.session with un-tagged remote sessions that mergePeerSessions then preferred over the correctly-tagged projection entries. Two changes, one invariant (the directory store is local-only): Change 1: Propagate origin namespace through the event queue - New sse-origin.ts module with parseCursorNamespaces() and resolveEventOrigin() — pure functions that diff the composite cursor id: field to determine which SSE namespace (local vs remote peer) originated each event. Fleet-of-one (bare scalar id) returns immediately with zero overhead. - server-sdk.tsx wires origin resolution into the SSE loop: the ServerEvent type gains an optional origin field, set only for remote events. Change 2: Gate directory-store mutations in the listener - server-sync.tsx checks event.origin and skips for remote events: - applyDirectoryEvent (the primary contamination point) - homeSessions.apply (feeds into dedup list) - indexSession (synthetic session.created from session.moved/forked) - queue.push (triggers directory re-fetches) - session.apply/session.applyV2 still process remote events (needed for per-session caches the timeline reads from) Tests: - sse-origin.test.ts: 12 unit tests for namespace extraction and origin resolution (fleet-of-one, composite local, composite remote, multi-peer, edge cases) - session-fleet-peers.test.ts: regression guard for the contamination scenario + badge persistence when projection is sole source - event-reducer.test.ts: gate contract tests documenting what the gate prevents - session-header-provenance.test.ts: badge persistence across repeated dropdown opens --- packages/app-bundle/manifest.json | 20 +++-- .../session/session-fleet-peers.test.ts | 34 ++++++++ .../context/global-sync/event-reducer.test.ts | 42 ++++++++++ .../packages/app/src/context/server-sdk.tsx | 20 ++++- .../packages/app/src/context/server-sync.tsx | 42 +++++++--- .../app/src/context/sse-origin.test.ts | 77 +++++++++++++++++++ .../packages/app/src/context/sse-origin.ts | 64 +++++++++++++++ .../session-header-provenance.test.ts | 36 +++++++++ 8 files changed, 313 insertions(+), 22 deletions(-) create mode 100644 packages/app-bundle/overlay/packages/app/src/context/sse-origin.test.ts create mode 100644 packages/app-bundle/overlay/packages/app/src/context/sse-origin.ts diff --git a/packages/app-bundle/manifest.json b/packages/app-bundle/manifest.json index fad3105a..064d9338 100644 --- a/packages/app-bundle/manifest.json +++ b/packages/app-bundle/manifest.json @@ -165,10 +165,10 @@ "packages/app/src/context/models.tsx": "8904525767b36ed5ba93e85a0fddb9bf737a8d066dfff084380ce1a67182f535", "packages/app/src/context/platform.tsx": "edd9b1773feb5c084b922a57b6914f6a473eed558b44f9de0eeb1c4d4e2b49da", "packages/app/src/context/server-sdk.test.ts": "48ce5243c8c5d42c23c8eadf9eecac777c9330ac1f4d81953042d5ef7d264bd7", - "packages/app/src/context/server-sdk.tsx": "130d54ea73c7269c0409dee6236b61145493d0c5f562750f8df94a56c10ed8be", + "packages/app/src/context/server-sdk.tsx": "a3f51fc8a17da40869cd21aecebc5e06bd2e2ba5d323b4aa07909a2646df9549", "packages/app/src/context/server-session.test.ts": "a76d24b0d36d7c6c9ce187d2eb5bba78c90774184adc0398559ee1a417674516", "packages/app/src/context/server-session.ts": "4d85011c9ff81cf0470633f6f27cd9f11c234294c31e2c77a990aad135d317ab", - "packages/app/src/context/server-sync.tsx": "14396d7402d697ab9a10744c9e428b96e483f3a9620a0415246fa840596fd8ed", + "packages/app/src/context/server-sync.tsx": "94c16dfb25bf8cec340bdc31bef5e875c31a4c42af7076179bf69da4d1586889", "packages/app/src/context/server.tsx": "08e978e4b4b3094221a2a0b92e069f39cbedb6d6cf95935881e21cde52265ee2", "packages/app/src/context/session-mirror.test.ts": "65e6d6474e4e67f2da194047140c8c8aac04fe0cc9926441468e53fabc48db96", "packages/app/src/context/session-mirror.ts": "b98de060f2b06e6f9a48b1f69ed408f12086c626d90c7dd22029a4ee732260e2", @@ -176,6 +176,8 @@ "packages/app/src/context/session-warm.ts": "d4184b8f0a090b0a011cdf7b6e80d5f765b6f7acd834eaad064e39ca8e0eb7bc", "packages/app/src/context/settings.tsx": "19e078e81620ffa21a92797335009b84d4311485a37c5ede2f76906b734292ad", "packages/app/src/context/split.tsx": "891aeb290b369b327a63ce548dcaab1daca08e41fab430648787be7f84e53ec0", + "packages/app/src/context/sse-origin.test.ts": "71cd3aa996a51d7e99236a49cede4769e520e20e53f57c6920fe441f2abb0c7d", + "packages/app/src/context/sse-origin.ts": "da9a231907dc50d87cf01758177bee0d4a0b0165d8adb5e152db8eca29dfd9f5", "packages/app/src/context/stream-gap.test.ts": "b7203445564f44fd1d966ef684a11cb9b5ec567d9ab54f1b7617726fdf44d3f5", "packages/app/src/context/stream-gap.ts": "2fd8eba30a6112cebf38e33c9418ab2b7d9c9d466f76729d828d5ab52d56c5ec", "packages/app/src/context/tab-migration.ts": "569e69cd3425917c7e195f91c3a838abea68b616267ea7511f3ccf01d94ac01e", @@ -641,7 +643,7 @@ "packages/app/src/components/session/preview-human-write-gate.ts": "4c4b1cf258a06b0012d722add0125ecb77a0f4e463dd95d193f21e5e73541098", "packages/app/src/components/session/session-chats-dropdown.test.ts": "b5f178bea1b52788d6f794d09ccf7ff81f745633ad59e2d438cfc80b96acde55", "packages/app/src/components/session/session-context-tab.tsx": "227243b178b517f067d9ae0ae0eec3c559beeb6681828158b0600a17e98e7f81", - "packages/app/src/components/session/session-fleet-peers.test.ts": "67e03d6f0be79f5d08cb293af2c8d7e1100e5c49d7fd86ce5c7ced1eea93f1d7", + "packages/app/src/components/session/session-fleet-peers.test.ts": "c3e5404ff868b37706a4963c83565de022443ea09e350ee60200b89cc5dba61c", "packages/app/src/components/session/session-fleet-peers.ts": "1c9e846668adcff0852a66a6961e393fb9e0cc91dc4103017d70cbad269f791c", "packages/app/src/components/session/session-header.tsx": "c8207d714f01d4fe9f8731b410ff0359e4f9b9f7815967cc654060944af8933e", "packages/app/src/components/session/session-new-view.tsx": "9510a4f550a3f0d4791e98e8025666f09d70a60fb66f193e48ee61feddae5a57", @@ -674,7 +676,7 @@ "packages/app/src/components/ui/drawer.tsx": "76574ca4c6997209f37a766a61000d707332a9a857f6ea8100a3cee1be58d65c", "packages/app/src/context/global-sync/bootstrap.test.ts": "ed66b4c0427d51c015fd286dd63c2cc24a25391320366d75e28634334d70e366", "packages/app/src/context/global-sync/child-store.ts": "3fde298a6cda8e96efc22bccee4ef124e139abd6ec28510363c78f702bf70db9", - "packages/app/src/context/global-sync/event-reducer.test.ts": "0528ed45952107b3ca174e72fd0b920191dbbd49eb795c4dff63cf89c2c91c0e", + "packages/app/src/context/global-sync/event-reducer.test.ts": "2d8a77b327403c99fb78d15cb720e6a4e0a5e9479a5fbf1f714455b66e87d0a2", "packages/app/src/context/global-sync/event-reducer.ts": "9c26e9e83709e56c57b228a03186df6bf1a52a073c4b7e721853956ee0da6142", "packages/app/src/context/global-sync/session-cache.test.ts": "e3c66ef76ff20a416f3b2d72041acd66c2a65da57ded92cbcb381ab71474285f", "packages/app/src/context/global-sync/session-cache.ts": "70f60c37ab9ffcb5ac0778f4962d88f0e7c6fb61fc39a18bda28942322605316", @@ -838,7 +840,7 @@ "packages/app/src/pages/session/timeline/projection.ts": "36ab006523de7b695b33d26a0a03b59db9e148b7e979c36f63970cbcb8bcaca2", "packages/app/src/pages/session/timeline/rows-current.test.ts": "b6a2d9671cfba970990b36d9ef0d1f107efcee59f0a9863f1280526116299095", "packages/app/src/pages/session/timeline/rows.ts": "68581190ff5e51d2dc398298c705fbb5dfdefb7eb4cecf98f1da6ae0bd171f18", - "packages/app/src/pages/session/timeline/session-header-provenance.test.ts": "b6ca7ffde2ef5d02b09968c20b5027ceab45506af8b1a49dd3f6c8ffe5912312", + "packages/app/src/pages/session/timeline/session-header-provenance.test.ts": "4bdf201c16d00eb60957d0ad9b40dafd69410a4b46deb0634e16da26103be1da", "packages/app/src/pages/session/timeline/session-header-provenance.ts": "7b73833aa5a3530c704edef2ca38cb6730aaaa729c23e06b6227788fb5c7e324", "packages/app/src/pages/session/timeline/smooth-scroll.test.ts": "79d3bba42ad83c3ab80df907e0acf08665a38c4809468e6c12ac9fcf5e94682f", "packages/app/src/pages/session/timeline/smooth-scroll.ts": "3f2782f52e36acf86398fb444bc53fcfec2fec9a0684613466ff7d24b8bceb26", @@ -893,10 +895,10 @@ "packages/ui/src/v2/components/text-shimmer-v2.tsx" ], "exceptions": [], - "extracted_at": "2026-09-24T13:51:39.015Z", + "extracted_at": "2026-09-24T19:01:26.321Z", "per_package": { "packages/app": { - "A": 201, + "A": 203, "M": 156, "D": 0 }, @@ -942,7 +944,7 @@ } }, "counts": { - "overlay_total": 871, + "overlay_total": 873, "deletions": 7, "server_coupled": 0 }, @@ -1115,6 +1117,8 @@ "packages/app/src/context/session-warm.ts": "A", "packages/app/src/context/settings.tsx": "M", "packages/app/src/context/split.tsx": "A", + "packages/app/src/context/sse-origin.test.ts": "A", + "packages/app/src/context/sse-origin.ts": "A", "packages/app/src/context/stream-gap.test.ts": "A", "packages/app/src/context/stream-gap.ts": "A", "packages/app/src/context/tab-migration.ts": "M", diff --git a/packages/app-bundle/overlay/packages/app/src/components/session/session-fleet-peers.test.ts b/packages/app-bundle/overlay/packages/app/src/components/session/session-fleet-peers.test.ts index 7563f8bd..044b33c1 100644 --- a/packages/app-bundle/overlay/packages/app/src/components/session/session-fleet-peers.test.ts +++ b/packages/app-bundle/overlay/packages/app/src/components/session/session-fleet-peers.test.ts @@ -93,6 +93,40 @@ describe("#1525 mergePeerSessions — dedupe (local wins) + sort by last activit const local = [mk("a", 5), mk("b", 9)] expect(mergePeerSessions(local, []).map((s) => s.id)).toEqual(["b", "a"]) }) + + // #1539 regression: when the SSE fan-in contaminated the directory store, a + // remote session ended up in the local `activeSessions` array WITHOUT its + // `amicode_owner` tag. On the next dropdown open, `mergePeerSessions` saw the + // local (untagged) copy and the projection (tagged) copy — local wins, so the + // badge-carrying projection copy was dropped. The fix gates the directory store + // so remote events never insert there; this test guards the dedup behavior that + // was being exploited by the contamination. + test("#1539 regression: contaminated local copy (no amicode_owner) wins over badged projection copy", () => { + // Simulate the contamination scenario: same session in local (no tag) and + // projection (with tag). The local copy should win, which is correct behavior + // — the fix prevents the contamination from happening in the first place. + const contaminated = mk("ses_remote", 5) // no amicode_owner (the bug) + const fromProjection = mk("ses_remote", 5, studioTag) // has the badge + const out = mergePeerSessions([contaminated], [fromProjection]) + expect(out).toHaveLength(1) + // Local wins → badge is lost. This is CORRECT dedup behavior — the fix is + // to prevent `contaminated` from ever reaching the local store. + expect(out[0].amicode_owner).toBeUndefined() + expect(deriveSessionBadge(out[0])).toBeUndefined() + }) + + test("#1539 fixed: when directory store is clean, projection badge persists through dedup", () => { + // After the fix: the remote session is ONLY in the projection (with badge), + // never contaminated into the local store. The dropdown merges correctly. + const localSessions = [mk("ses_local", 5)] + const projectionPeers = [mk("ses_remote", 3, studioTag)] + const out = mergePeerSessions(localSessions, projectionPeers) + expect(out).toHaveLength(2) + const remote = out.find((s) => s.id === "ses_remote")! + expect(remote.amicode_owner).toBeDefined() + expect(remote.amicode_owner!.is_local).toBe(false) + expect(deriveSessionBadge(remote)).toBe("JJ's Mac Studio") + }) }) // #1537 B2a (AC4): owner-routed OPEN. `resolveDropdownOpenAction` is the pure diff --git a/packages/app-bundle/overlay/packages/app/src/context/global-sync/event-reducer.test.ts b/packages/app-bundle/overlay/packages/app/src/context/global-sync/event-reducer.test.ts index ebae7374..d3042c49 100644 --- a/packages/app-bundle/overlay/packages/app/src/context/global-sync/event-reducer.test.ts +++ b/packages/app-bundle/overlay/packages/app/src/context/global-sync/event-reducer.test.ts @@ -614,4 +614,46 @@ describe("applyDirectoryEvent", () => { expect(lspLoads).toBe(1) }) + // #1539: the directory store is local-only. The gate in server-sync.tsx checks + // `event.origin` and skips `applyDirectoryEvent` for remote-origin events. This + // test verifies the gate's contract: a remote-origin event WOULD mutate the + // store if the gate weren't there, proving the gate is necessary. + test("#1539 gate contract: session.created WOULD insert into the store (the gate prevents this for remote events)", () => { + const [store, setStore] = createStore(baseState({ session: [], sessionTotal: 0 })) + const remoteSession = rootSession({ id: "ses_remote" }) + + // Without the gate, applyDirectoryEvent inserts the session: + applyDirectoryEvent({ + event: { type: "session.created", properties: { info: remoteSession } }, + store, + setStore, + push() {}, + directory: "/tmp", + loadLsp() {}, + }) + expect(store.session).toHaveLength(1) + expect(store.sessionTotal).toBe(1) + + // The gate in server-sync.tsx: `if (isRemote) return` — skips this call. + // A remote-origin event with `origin: "studio"` never reaches here. + // This test documents what the gate prevents. + }) + + test("#1539 gate contract: session.updated WOULD upsert into the store (the gate prevents this for remote events)", () => { + const [store, setStore] = createStore(baseState({ session: [], sessionTotal: 0 })) + const remoteSession = rootSession({ id: "ses_remote" }) + + // Without the gate, session.updated inserts a previously-unknown session: + applyDirectoryEvent({ + event: { type: "session.updated", properties: { info: remoteSession } }, + store, + setStore, + push() {}, + directory: "/tmp", + loadLsp() {}, + }) + expect(store.session).toHaveLength(1) + // The gate prevents this for events with `origin: "studio"`. + }) + }) diff --git a/packages/app-bundle/overlay/packages/app/src/context/server-sdk.tsx b/packages/app-bundle/overlay/packages/app/src/context/server-sdk.tsx index 1a4698c6..482c2370 100644 --- a/packages/app-bundle/overlay/packages/app/src/context/server-sdk.tsx +++ b/packages/app-bundle/overlay/packages/app/src/context/server-sdk.tsx @@ -12,6 +12,7 @@ import { ServerConnection, useServer } from "./server" import { createRefCountMap } from "@/utils/refcount" import { useGlobal } from "./global" import { ServerScope } from "@/utils/server-scope" +import { parseCursorNamespaces, resolveEventOrigin } from "./sse-origin" import { detectServerProtocol, type ServerProtocol } from "@/utils/server-protocol" import { createCompatibleApi, type CompatibleApi } from "@/utils/server-compat" @@ -19,7 +20,7 @@ const isAbortError = (error: unknown) => error !== null && typeof error === "object" && "name" in error && error.name === "AbortError" const isStreamClosed = (error: unknown, signal?: AbortSignal) => isAbortError(error) || signal?.aborted === true -export type ServerEvent = Event & { current?: OpenCodeEvent } +export type ServerEvent = Event & { current?: OpenCodeEvent; origin?: string } type QueuedServerEvent = { directory: string; payload: ServerEvent } /** A minimal fetch call signature. Newer lib types make `typeof fetch` require * a `preconnect` member our plain wrappers don't implement (lib drift); @@ -347,6 +348,10 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS } } } + // #1539: track the parsed composite cursor to detect event origin (local vs + // remote peer). Initialized from the persisted cursor so the first event + // after reconnect diffs correctly. + let previousParsedCursor = parseCursorNamespaces(lastEventID) const sseFetch: FetchLike = (input, init) => { const base = eventFetch ?? globalThis.fetch try { @@ -501,7 +506,17 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS continue } trackEventID(legacy ? event.payload : (event as { id?: string })) - if (enqueueServerEvent(queue, { directory, payload })) schedule() + // #1539: resolve origin from the composite cursor diff. Fleet-of-one + // (bare scalar id) returns undefined immediately — zero overhead. + const sseId: string | undefined = legacy + ? (event.payload as { id?: string }).id + : (event as { id?: string }).id + const resolved = resolveEventOrigin(sseId, previousParsedCursor) + previousParsedCursor = resolved.cursor + const taggedPayload: ServerEvent = resolved.origin !== undefined + ? { ...payload, origin: resolved.origin } + : payload + if (enqueueServerEvent(queue, { directory, payload: taggedPayload })) schedule() if (Date.now() - yielded < STREAM_YIELD_MS) continue yielded = Date.now() @@ -537,6 +552,7 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS const clearGlobalStreamCursor = () => { lastEventID = undefined + previousParsedCursor = {} try { sessionStorage.removeItem(CURSOR_KEY) } catch { diff --git a/packages/app-bundle/overlay/packages/app/src/context/server-sync.tsx b/packages/app-bundle/overlay/packages/app/src/context/server-sync.tsx index bb49a767..f4f507c6 100644 --- a/packages/app-bundle/overlay/packages/app/src/context/server-sync.tsx +++ b/packages/app-bundle/overlay/packages/app/src/context/server-sync.tsx @@ -548,11 +548,17 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { const event = e.details const eventType: string = event.type const recent = bootingRoot || Date.now() - bootedAt < 1500 + // #1539: remote-origin events must NOT mutate the local directory store. + // The origin field is set by the SSE loop when the event came from a remote + // peer's namespace in the fan-in composite cursor. `undefined` = local. + const isRemote = typeof event.origin === "string" if (event.current) session.applyV2(event.current) session.apply(event) if (event.type === "session.created" || event.type === "session.updated" || event.type === "session.deleted") { - homeSessions.apply(event) + // #1539: homeSessions feeds into the session list that mergePeerSessions + // deduplicates against; remote events contaminate it the same way. + if (!isRemote) homeSessions.apply(event) } homeSessions.refresh(event.type) if (eventType === "integration.connection.updated") void refreshProviders() @@ -584,21 +590,27 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { lastConnectedQueueAt = Date.now() for (const directory of Object.keys(children.children)) { if (!children.active(directory)) continue - queue.push(directory) + // #1539: a remote server.connected/global.disposed must not trigger + // local directory re-bootstraps. + if (!isRemote) queue.push(directory) } } return } - if (event.current?.type === "session.moved") { - const info = session.get(event.current.data.sessionID) - if (info) indexSession(info) + // #1539: indexSession inserts synthetic session.created events into the + // directory store — skip for remote events. + if (!isRemote) { + if (event.current?.type === "session.moved") { + const info = session.get(event.current.data.sessionID) + if (info) indexSession(info) + } + if (event.current?.type === "session.forked") + void session + .resolve(event.current.data.sessionID, { force: true }) + .then(indexSession) + .catch(() => {}) } - if (event.current?.type === "session.forked") - void session - .resolve(event.current.data.sessionID, { force: true }) - .then(indexSession) - .catch(() => {}) const existing = children.children[key] if (!existing) return @@ -610,10 +622,16 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { eventType === "command.updated" || eventType === "config.updated" || eventType === "agent.updated" - ) - queue.push(key) + ) { + // #1539: queue.push triggers directory re-fetches that would pull remote + // sessions into the local store. + if (!isRemote) queue.push(key) + } if (eventType === "mcp.status.changed") void queryClient.invalidateQueries(queryOptionsApi.mcp(key)) if (eventType === "mcp.resources.changed") void queryClient.invalidateQueries(queryOptionsApi.mcpResources(key)) + // #1539: applyDirectoryEvent is the primary contamination point — remote + // events must not insert into or update the directory-scoped store.session. + if (isRemote) return const [store, setStore] = existing applyDirectoryEvent({ event, diff --git a/packages/app-bundle/overlay/packages/app/src/context/sse-origin.test.ts b/packages/app-bundle/overlay/packages/app/src/context/sse-origin.test.ts new file mode 100644 index 00000000..1930076b --- /dev/null +++ b/packages/app-bundle/overlay/packages/app/src/context/sse-origin.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, test } from "bun:test" +import { parseCursorNamespaces, resolveEventOrigin } from "./sse-origin" + +// #1539: SSE fan-in origin detection — the pure logic that determines whether +// an SSE event originated from the local engine or a remote peer, by diffing +// the composite cursor `id:` field the fan-in stamps on each frame. + +describe("parseCursorNamespaces", () => { + test("parses a composite cursor into a namespace→value map", () => { + expect(parseCursorNamespaces("local=5;studio=42")).toEqual({ local: "5", studio: "42" }) + }) + + test("a bare scalar (fleet-of-one, no `=`) returns an empty map", () => { + expect(parseCursorNamespaces("42")).toEqual({}) + }) + + test("undefined / empty string returns an empty map", () => { + expect(parseCursorNamespaces(undefined)).toEqual({}) + expect(parseCursorNamespaces("")).toEqual({}) + }) + + test("handles a single-namespace composite cursor", () => { + expect(parseCursorNamespaces("local=5")).toEqual({ local: "5" }) + }) + + test("skips malformed entries (empty key, no `=`)", () => { + expect(parseCursorNamespaces("local=5;;=bad;studio=42")).toEqual({ local: "5", studio: "42" }) + }) +}) + +describe("resolveEventOrigin", () => { + test("fleet-of-one (bare scalar id) is always local — zero overhead", () => { + const prev = {} + const { origin, cursor } = resolveEventOrigin("42", prev) + expect(origin).toBeUndefined() + // Cursor is returned unchanged (no allocation for fleet-of-one). + expect(cursor).toBe(prev) + }) + + test("undefined id is local", () => { + expect(resolveEventOrigin(undefined, {}).origin).toBeUndefined() + }) + + test("composite cursor with local-only change → local origin", () => { + const prev = { local: "4", studio: "42" } + const { origin, cursor } = resolveEventOrigin("local=5;studio=42", prev) + expect(origin).toBeUndefined() + expect(cursor).toEqual({ local: "5", studio: "42" }) + }) + + test("composite cursor with remote change → remote machine id", () => { + const prev = { local: "5", studio: "41" } + const { origin, cursor } = resolveEventOrigin("local=5;studio=42", prev) + expect(origin).toBe("studio") + expect(cursor).toEqual({ local: "5", studio: "42" }) + }) + + test("first event in composite mode (empty previous) detects local origin when local changes", () => { + // When the stream starts fresh, previous is {}. The first event's cursor + // will differ on ALL namespaces. The first differing namespace is the origin. + // "local" comes first in insertion order → detected as local. + const { origin } = resolveEventOrigin("local=1;studio=0", {}) + expect(origin).toBeUndefined() // local detected first + }) + + test("composite cursor with no change from previous → local (fallback)", () => { + const prev = { local: "5", studio: "42" } + const { origin } = resolveEventOrigin("local=5;studio=42", prev) + expect(origin).toBeUndefined() + }) + + test("multi-peer: correctly identifies which peer changed", () => { + const prev = { local: "5", studio: "42", mbp: "10" } + const { origin } = resolveEventOrigin("local=5;studio=42;mbp=11", prev) + expect(origin).toBe("mbp") + }) +}) diff --git a/packages/app-bundle/overlay/packages/app/src/context/sse-origin.ts b/packages/app-bundle/overlay/packages/app/src/context/sse-origin.ts new file mode 100644 index 00000000..ba404d51 --- /dev/null +++ b/packages/app-bundle/overlay/packages/app/src/context/sse-origin.ts @@ -0,0 +1,64 @@ +// SSE FAN-IN ORIGIN DETECTION (#1539) — pure logic for determining whether an +// SSE event originated from the local engine or a remote peer, by diffing the +// composite cursor `id:` field the fan-in stamps on each frame. +// +// The fan-in aggregator (sse_fanin_aggregator.ts) writes a composite cursor as +// the `id:` on every frame in multi-peer mode: `local=5;studio=42`. Each event +// advances exactly one namespace's cursor value. By diffing the previous cursor +// against the current one, we identify which namespace originated the event. +// +// In fleet-of-one mode the `id:` is a bare scalar (no `=`): the function +// returns immediately — zero allocation, byte-identical to today. + +/** Parse a composite cursor string into a namespace→value map. + * + * - `"local=5;studio=42"` → `{local:"5", studio:"42"}` + * - `"42"` (bare scalar, fleet-of-one) → `{}` + * - `""` / undefined → `{}` + * + * App-layer lightweight equivalent of sse_composite_cursor.ts's + * `parseCompositeCursor` — just enough to detect event origin. */ +export function parseCursorNamespaces(id: string | undefined): Record { + if (!id || !id.includes("=")) return {} + const result: Record = {} + for (const entry of id.split(";")) { + if (entry === "") continue + const eq = entry.indexOf("=") + if (eq <= 0) continue + result[entry.slice(0, eq)] = entry.slice(eq + 1) + } + return result +} + +/** Resolve which SSE namespace originated the event by diffing the event's + * composite cursor `id` against the previous cursor state. + * + * Returns `undefined` for a local event (or fleet-of-one — bare scalar id, + * zero overhead), or the remote machine id string. + * + * The composite cursor encodes ALL namespaces' positions in every event's + * `id:` field (e.g. `local=5;studio=42`). The namespace whose value changed + * from the previous cursor is the originator. */ +export function resolveEventOrigin( + currentId: string | undefined, + previousCursor: Record, +): { origin: string | undefined; cursor: Record } { + // Fleet-of-one: bare scalar (no `=`) — always local, zero overhead. + if (!currentId || !currentId.includes("=")) + return { origin: undefined, cursor: previousCursor } + + const current = parseCursorNamespaces(currentId) + let changed: string | undefined + for (const ns of Object.keys(current)) { + if (current[ns] !== previousCursor[ns]) { + changed = ns + break + } + } + + // "local" namespace or no detected change = local origin + return { + origin: changed && changed !== "local" ? changed : undefined, + cursor: current, + } +} diff --git a/packages/app-bundle/overlay/packages/app/src/pages/session/timeline/session-header-provenance.test.ts b/packages/app-bundle/overlay/packages/app/src/pages/session/timeline/session-header-provenance.test.ts index d84d77e1..0ab1cf00 100644 --- a/packages/app-bundle/overlay/packages/app/src/pages/session/timeline/session-header-provenance.test.ts +++ b/packages/app-bundle/overlay/packages/app/src/pages/session/timeline/session-header-provenance.test.ts @@ -237,6 +237,42 @@ describe("resolveHeaderProvenance (#1452 W4a) — the header's fetch→resolve p }) }) +// #1539: badge persistence after simulated remote SSE events. The bug was that +// SSE events from a remote peer contaminated the local directory store with an +// un-tagged copy of the session. On the next dropdown open, the local copy +// (no amicode_owner) won the dedup, and the badge disappeared. After the fix, +// the projection remains the sole source of remote sessions, so the badge +// persists across dropdown opens. +describe("#1539 badge persistence after remote SSE events", () => { + test("a remote session's badge persists when the projection is the sole source", () => { + // Simulate multiple consecutive dropdown opens. Each time, the projection + // carries the remote session with its owner tag. Without directory-store + // contamination, the badge is always present. + const projection: FleetSessionEntry[] = [ + { id: "ses_remote", amicode_owner: { owner_machine_id: "mac-studio", owner_name: "Mac Studio", is_local: false } }, + ] + // First open + const p1 = resolveHeaderProvenance(projection, "ses_remote") + expect(p1.showIcon).toBe(true) + expect(p1.tooltip).toBe("Mac Studio") + expect(p1.isRemote).toBe(true) + // Second open (same projection, unchanged) + const p2 = resolveHeaderProvenance(projection, "ses_remote") + expect(p2.showIcon).toBe(true) + expect(p2.tooltip).toBe("Mac Studio") + // N-th open — indefinitely stable + const pN = resolveHeaderProvenance(projection, "ses_remote") + expect(pN.isRemote).toBe(true) + }) + + test("a local session never gains a spurious badge from remote events", () => { + const projection: FleetSessionEntry[] = [ + { id: "ses_local", amicode_owner: { owner_machine_id: "macbook", owner_name: "MacBook", is_local: true } }, + ] + expect(resolveHeaderProvenance(projection, "ses_local").showIcon).toBe(false) + }) +}) + describe("session-header provenance icon is MOUNTED in the live header (#1452 W4a AC1+AC2)", () => { // Structural guard (the app's convention for the untested-by-design SolidJS // wiring — mirrors vscode-explorer-file-icon.test.tsx). This is BOTH the AC2