fix(hub): reset message deltas across instances

docs/content/8.references/6.hub-api.md

@@ -15,9 +15,11 @@ What `DevframeHubContext` adds to `DevframeNodeContext`: [Hub](/guide/hub).
 |---|---|---|
 | `ctx.docks` | `register / update / values / activate` | Dock entries (iframes, launchers, custom-render) and groups; `activate(dockId, params?)` sets the active dock ([Cross-iframe dock activation](/guide/hub#cross-iframe-dock-activation)). |
 | `ctx.terminals` | `register / startChildProcess / startPtySession` | Aggregate terminal sessions, streaming output ([Terminals](/add-ons/devframes/terminals#hub-aggregation)). |
-| `ctx.messages` | `add / update / remove / clear` | Server-side toast/notification queue (FIFO, capped at 1000). |
+| `ctx.messages` | `add / update / remove / clear / listSince` | Server-side toast/notification queue (FIFO, capped at 1000). |
 | `ctx.commands` | `register / execute / list` | Hierarchical command palette with keybindings and `when` clauses. |
 
+For incremental message reads, call `ctx.messages.listSince(version, instanceId)` with both values from the previous result. A missing or foreign `instanceId` returns a full snapshot, as does an expired version. When `full` is true, replace the cached list. Otherwise, apply `removedIds` before adding or updating `entries`. Numeric-only callers remain supported through full snapshots.
+
 ## Launcher fields
 
 The optional `launcher` fields that make a `type: 'launcher'` dock entry a live process controller: [Process-control launchers](/guide/hub#process-control-launchers).

packages/hub-ui/src/client/state/messages.test.ts

@@ -38,7 +38,9 @@ async function fixture() {
   await Promise.resolve()
   async function publish(next: DevframeMessagesListDelta) {
     delta = next
+    const previousCalls = call.mock.calls.length
     refresh()
+    await vi.waitUntil(() => call.mock.calls.length > previousCalls)
     await Promise.resolve()
   }
   const message: DevframeMessageEntry = {
@@ -58,6 +60,7 @@ async function fixture() {
   })
   return {
     state,
+    call,
     toasts: useToasts(),
     publish,
     message,
@@ -96,3 +99,34 @@ it('keeps surviving toasts and unread counts when receiving a full snapshot', as
   expect(toasts).toHaveLength(1)
   expect(state.unreadCount).toBe(1)
 })
+
+it('passes the instance ID with the next cursor and replaces stale messages', async () => {
+  const { state, call, publish, message } = await fixture()
+  await publish({
+    entries: [message],
+    removedIds: [],
+    version: 1,
+    instanceId: 'old',
+    full: true,
+  })
+  await publish({
+    entries: [
+      { ...message, id: 'new' },
+    ],
+    removedIds: [],
+    version: 1,
+    instanceId: 'new',
+    full: true,
+  })
+  expect(call).toHaveBeenLastCalledWith('devframes:plugin:messages:list', 1, 'old')
+  expect(state.entries.map(entry => entry.id)).toEqual(['new'])
+  await publish({
+    entries: [],
+    removedIds: ['new'],
+    version: 2,
+    instanceId: 'new',
+    full: false,
+  })
+  expect(call).toHaveBeenLastCalledWith('devframes:plugin:messages:list', 1, 'new')
+  expect(state.entries).toEqual([])
+})

packages/hub-ui/src/client/state/messages.ts

@@ -36,6 +36,7 @@ export function useMessages(context: DocksContext): Reactive<MessagesState> {
   const entryMap = new Map<string, DevframeMessageEntry>()
   let isInitialFetch = true
   let lastVersion: number | null = null
+  let lastInstanceId: string | null = null
 
   async function updateMessages() {
     // The feed is served by `@devframes/plugin-messages` (the hub itself ships
@@ -44,6 +45,7 @@ export function useMessages(context: DocksContext): Reactive<MessagesState> {
     const result = await context.rpc.call(
       'devframes:plugin:messages:list' as any,
       lastVersion,
+      lastInstanceId,
     ) as DevframeMessagesListDelta
     let newCount = 0
 
@@ -74,6 +76,7 @@ export function useMessages(context: DocksContext): Reactive<MessagesState> {
     state.entries = Array.from(entryMap.values())
     state.unreadCount += newCount
     lastVersion = result.version
+    lastInstanceId = result.instanceId ?? null
     isInitialFetch = false
   }
 

packages/hub/src/node/__tests__/host-messages.test.ts

@@ -40,6 +40,33 @@ describe('devframeMessagesHost', () => {
   })
 
   describe('listSince', () => {
+    it.each([1, 2, 3])('replaces a foreign cursor when the new clock is %s', async (count) => {
+      const previous = new DevframeMessagesHost({} as DevframeHubContext)
+      await previous.info('old first', { id: 'old:1' })
+      await previous.info('old second', { id: 'old:2' })
+      const cursor = previous.listSince()
+
+      const current = new DevframeMessagesHost({} as DevframeHubContext)
+      for (let i = 0; i < count; i++)
+        await current.info(`new ${i}`, { id: `new:${i}` })
+
+      const result = current.listSince(cursor.version, cursor.instanceId)
+      expect(result.full).toBe(true)
+      expect(result.entries.map(entry => entry.id)).toEqual(Array.from({ length: count }, (_, i) => `new:${i}`))
+      expect(result.instanceId).not.toBe(cursor.instanceId)
+    })
+
+    it('returns a full snapshot for a legacy numeric-only cursor', async () => {
+      const host = new DevframeMessagesHost({} as DevframeHubContext)
+      await host.info('first', { id: 'first' })
+      const { version } = host.listSince()
+      await host.info('second', { id: 'second' })
+
+      const result = host.listSince(version)
+      expect(result.full).toBe(true)
+      expect(result.entries.map(entry => entry.id)).toEqual(['first', 'second'])
+    })
+
     it('returns a full snapshot without a cursor', async () => {
       const host = new DevframeMessagesHost({} as DevframeHubContext)
       await host.add({ id: 'a', level: 'info', message: 'a' })
@@ -55,12 +82,12 @@ describe('devframeMessagesHost', () => {
     it('returns only entries modified after the cursor', async () => {
       const host = new DevframeMessagesHost({} as DevframeHubContext)
       await host.add({ id: 'a', level: 'info', message: 'a' })
-      const { version } = host.listSince()
+      const { version, instanceId } = host.listSince()
 
       await host.add({ id: 'b', level: 'info', message: 'b' })
       await host.update('a', { message: 'a2' })
 
-      const delta = host.listSince(version)
+      const delta = host.listSince(version, instanceId)
       expect(delta.full).toBe(false)
       expect(delta.entries.map(e => e.id).sort()).toEqual(['a', 'b'])
       expect(delta.removedIds).toEqual([])
@@ -71,11 +98,11 @@ describe('devframeMessagesHost', () => {
       const host = new DevframeMessagesHost({} as DevframeHubContext)
       await host.add({ id: 'a', level: 'info', message: 'a' })
       await host.add({ id: 'b', level: 'info', message: 'b' })
-      const { version } = host.listSince()
+      const { version, instanceId } = host.listSince()
 
       await host.remove('a')
 
-      const delta = host.listSince(version)
+      const delta = host.listSince(version, instanceId)
       expect(delta.full).toBe(false)
       expect(delta.entries).toEqual([])
       expect(delta.removedIds).toEqual(['a'])
@@ -85,21 +112,21 @@ describe('devframeMessagesHost', () => {
       const host = new DevframeMessagesHost({} as DevframeHubContext)
       await host.add({ id: 'a', level: 'info', message: 'a' })
       await host.add({ id: 'b', level: 'info', message: 'b' })
-      const { version } = host.listSince()
+      const { version, instanceId } = host.listSince()
 
       await host.clear()
 
-      const delta = host.listSince(version)
+      const delta = host.listSince(version, instanceId)
       expect(delta.entries).toEqual([])
       expect(delta.removedIds.sort()).toEqual(['a', 'b'])
     })
 
     it('returns an empty delta when nothing changed', async () => {
       const host = new DevframeMessagesHost({} as DevframeHubContext)
       await host.add({ id: 'a', level: 'info', message: 'a' })
-      const { version } = host.listSince()
+      const { version, instanceId } = host.listSince()
 
-      const delta = host.listSince(version)
+      const delta = host.listSince(version, instanceId)
       expect(delta.full).toBe(false)
       expect(delta.entries).toEqual([])
       expect(delta.removedIds).toEqual([])
@@ -109,7 +136,7 @@ describe('devframeMessagesHost', () => {
     it('falls back to a full snapshot when the cursor predates trimmed removals', async () => {
       const host = new DevframeMessagesHost({} as DevframeHubContext)
       await host.add({ id: 'stale', level: 'info', message: 'stale cursor target' })
-      const { version } = host.listSince()
+      const { version, instanceId } = host.listSince()
 
       // Overflow the removal log so records after `version` get trimmed.
       for (let i = 0; i < 1005; i++) {
@@ -118,7 +145,7 @@ describe('devframeMessagesHost', () => {
         await host.remove(id)
       }
 
-      const result = host.listSince(version)
+      const result = host.listSince(version, instanceId)
       expect(result.full).toBe(true)
       expect(result.entries.map(e => e.id)).toEqual(['stale'])
       expect(result.removedIds).toEqual([])
@@ -128,7 +155,7 @@ describe('devframeMessagesHost', () => {
       const host = new DevframeMessagesHost({} as DevframeHubContext)
       await host.add({ id: 'a', level: 'info', message: 'a' })
 
-      const result = host.listSince(9999)
+      const result = host.listSince(9999, host.listSince().instanceId)
       expect(result.full).toBe(true)
       expect(result.entries.map(e => e.id)).toEqual(['a'])
     })

packages/hub/src/node/host-messages.ts

@@ -24,6 +24,7 @@ export class DevframeMessagesHost implements DevframeMessagesHostType {
   readonly removals: Array<{ id: string, time: number }> = []
 
   private _autoDeleteTimers = new Map<string, ReturnType<typeof setTimeout>>()
+  private readonly _instanceId = nanoid()
   private _clock = 0
   /**
    * The tick of the newest removal record dropped from the capped
@@ -159,16 +160,16 @@ export class DevframeMessagesHost implements DevframeMessagesHostType {
     this.events.emit(HUB_EVENTS.bus.messagesCleared)
   }
 
-  listSince(since?: number | null): DevframeMessagesListDelta {
+  listSince(since?: number | null, instanceId?: string | null): DevframeMessagesListDelta {
     const version = this._clock
-    // Fall back to a full snapshot when there is no cursor, when the cursor
-    // predates removal records already trimmed from the capped log, or when
-    // it comes from another host incarnation (ahead of our clock).
-    if (since == null || since < this._removalsTrimmedAt || since > version) {
+    // A numeric cursor alone cannot identify the instance that issued it.
+    // Legacy readers receive full snapshots until they return the instance ID.
+    if (instanceId !== this._instanceId || since == null || since < this._removalsTrimmedAt || since > version) {
       return {
         entries: Array.from(this.entries.values()),
         removedIds: [],
         version,
+        instanceId: this._instanceId,
         full: true,
       }
     }
@@ -188,6 +189,7 @@ export class DevframeMessagesHost implements DevframeMessagesHostType {
       entries,
       removedIds,
       version,
+      instanceId: this._instanceId,
       full: false,
     }
   }

packages/hub/src/types/messages.ts

@@ -187,7 +187,7 @@ export interface DevframeMessagesClient extends DevframeMessagesLevelShortcuts {
 /**
  * A snapshot or delta of the message list, as returned by
  * {@link DevframeMessagesHost.listSince}. Consumers apply `removedIds`
- * first, then upsert `entries`, and pass `version` back as `since` on the
+ * first, then upsert `entries`, and pass `version` and `instanceId` back on the
  * next call.
  */
 export interface DevframeMessagesListDelta {
@@ -197,6 +197,8 @@ export interface DevframeMessagesListDelta {
   removedIds: string[]
   /** The version cursor; pass back as `since` on the next call */
   version: number
+  /** Instance that issued the cursor, when supplied by the messages host. */
+  instanceId?: string
   /**
    * When `true`, `entries` is the complete snapshot and any locally cached
    * list must be reset before applying it.
@@ -232,11 +234,12 @@ export interface DevframeMessagesHost extends DevframeMessagesLevelShortcuts {
   clear: () => Promise<void>
   /**
    * Read the message list incrementally. Pass the `version` from the
-   * previous result as `since` to receive only the entries modified and the
-   * ids removed after that point; pass `null`/`undefined` for the initial
+   * previous result as `since`, together with its `instanceId`, to receive
+   * only the entries modified and the ids removed after that point; pass `null`/`undefined` for the initial
    * full snapshot. When the host can no longer compute a reliable delta for
    * the given cursor (trimmed removal history, or a cursor from another host
    * incarnation), the result carries `full: true` with the complete list.
+   * Calls without an instance ID also receive a full snapshot.
    */
-  listSince: (since?: number | null) => DevframeMessagesListDelta
+  listSince: (since?: number | null, instanceId?: string | null) => DevframeMessagesListDelta
 }

plugins/messages/app/client/state/messages.ts

@@ -31,6 +31,7 @@ export function useMessages(rpc: DevframeRpcClient): Reactive<MessagesState> {
 
   const entryMap = new Map<string, DevframeMessageEntry>()
   let lastVersion: number | null = null
+  let lastInstanceId: string | null = null
 
   // Serialize refreshes so a broadcast landing mid-fetch can't interleave
   // cursor updates; the cursor keeps each pass cheap.
@@ -41,7 +42,7 @@ export function useMessages(rpc: DevframeRpcClient): Reactive<MessagesState> {
       // no-args snapshot; live servers return the full list either way.
       const result = (lastVersion == null
         ? await rpc.call('devframes:plugin:messages:list')
-        : await rpc.call('devframes:plugin:messages:list', lastVersion)) as DevframeMessagesListDelta
+        : await rpc.call('devframes:plugin:messages:list', lastVersion, lastInstanceId)) as DevframeMessagesListDelta
       if (result.full)
         entryMap.clear()
       // Apply removals before upserts, since an id can be evicted and re-added
@@ -52,6 +53,7 @@ export function useMessages(rpc: DevframeRpcClient): Reactive<MessagesState> {
         entryMap.set(entry.id, entry)
       state!.entries = Array.from(entryMap.values())
       lastVersion = result.version
+      lastInstanceId = result.instanceId ?? null
     }).catch(() => {
       // Transport hiccup; the next broadcast or trust flip retries.
     })

plugins/messages/src/node/rpc/functions/list.ts

@@ -4,8 +4,8 @@ import { getMessagesHost } from './_define'
 
 /**
  * Read the message list incrementally. Pass the `version` from the previous
- * result as `since` to receive only what changed; omit it for the initial
- * full snapshot. Apply `removedIds` before upserting `entries`.
+ * result as `since`, together with its `instanceId`, to receive only what
+ * changed. Omit both for the initial full snapshot. Apply `removedIds` before upserting `entries`.
  *
  * `snapshot: true` bakes the full list into static builds, so the panel
  * renders the last captured feed without a live server.
@@ -20,7 +20,7 @@ export const messagesList = defineRpcFunction({
     description: 'List the hub message feed: diagnostics, notifications, and tool output entries with level, category, labels, and positions.',
   },
   setup: ctx => ({
-    handler: async (since?: number | null): Promise<DevframeMessagesListDelta> => {
+    handler: async (since?: number | null, instanceId?: string | null): Promise<DevframeMessagesListDelta> => {
       const host = getMessagesHost(ctx)
       if (!host) {
         return {
@@ -30,7 +30,7 @@ export const messagesList = defineRpcFunction({
           full: true,
         }
       }
-      return host.listSince(since)
+      return host.listSince(since, instanceId)
     },
   }),
 })

plugins/messages/test/dev-server.test.ts

@@ -68,12 +68,26 @@ describe('messages dev-server (hub context)', () => {
     await server.ctx.messages.add({ id: 'later', level: 'warn', message: 'later' })
     await server.ctx.messages.remove('from-server')
 
-    const delta = await rpc.$call('devframes:plugin:messages:list', full.version) as DevframeMessagesListDelta
+    const delta = await rpc.$call('devframes:plugin:messages:list', full.version, full.instanceId) as DevframeMessagesListDelta
     expect(delta.full).toBe(false)
     expect(delta.entries.map(e => e.id)).toEqual(['later'])
     expect(delta.removedIds).toContain('from-server')
   })
 
+  it('returns full snapshots for foreign and legacy cursors over RPC', async () => {
+    const rpc = connect(server)
+    const full = await rpc.$call('devframes:plugin:messages:list') as DevframeMessagesListDelta
+    for (const args of [
+      [full.version],
+      [full.version, 'previous-instance'],
+    ]) {
+      const result = await rpc.$call('devframes:plugin:messages:list', ...args) as DevframeMessagesListDelta
+      expect(result.full).toBe(true)
+      expect(result.entries).toEqual(full.entries)
+      expect(result.instanceId).toBe(full.instanceId)
+    }
+  })
+
   it('add stamps browser origin; update patches; clear empties', async () => {
     const rpc = connect(server)