fix(agent,cli,scripts): OT-RFC-38 LU-6 — address Codex PR #610 R1

Five correctness bugs raised on PR #610 by Codex round 1, all addressed.

1. memory.ts /catchup fallback now decides per-CG. Previously the
   `totalInserted === 0` gate aggregated across every requested CG,
   so a multi-CG catchup where one CG got triples from standard sync
   would silently skip LU-6 host-catchup for the others. Restructured
   to iterate CGs serially, fan out to peers in parallel within each
   CG, and decide host-catchup per CG. The new response shape exposes
   per-CG breakdown via `perContextGraph` + `hostCatchup.triggeredFor
   ContextGraphIds`, with a backward-compatible flat `results` array
   aggregated per peer.

2. memory.ts /catchup `totalInsertedTriples` now includes the
   host-catchup applied count. Previously a successful LU-6 recovery
   looked like a no-op at the top level (the field only counted
   standard sync inserts). Added a separate `standardInsertedTriples`
   field for callers that want the pre-fallback figure.

3. host-mode-store.ts seqno recovery from the log tail on cold load.
   The log is the source of truth (`appendFile` is durable before
   `persistMeta` runs); a crash in between would let the next append
   reuse a seqno that was already written, which breaks host-catchup
   paging (uses strict-greater-than seqno). `loadMeta()` now scans
   the log tail on first access per CG and takes `max(meta.seqno,
   lastLogSeqno)`. Reconciled value is persisted so subsequent cold
   loads skip the scan. New unit test simulates the stale-meta crash
   and verifies the next append picks the right seqno.

4. dkg-agent.ts reconcileSharedMemoryGossipSubscription — clear
   `swmHostModeSubscribed` when topic-wide `unsubscribe()` wipes the
   host listener. `GossipSubManager.unsubscribe(topic)` removes ALL
   handlers on the topic, not just the member-mode one; on member
   authorisation revocation that would drop the LU-6 host listener
   too, and `reconcileSwmHostModeSubscription()` would early-return
   on the still-set `swmHostModeSubscribed` flag — stranding hosting
   until restart. Clear the flag before the host-mode reconciler
   runs so it re-wires.

5. dkg-agent.ts enableSwmHostModeFor now probes
   `isContextGraphRegisteredOnChain` + `markRegistered` on both
   first-subscribe and idempotent re-entry. The explicit /host-mode/
   subscribe path is the Phase A surface that designates hosting
   from a topic id alone (no CG metadata required); without this
   probe the store would stay on the 6h/1MiB pre-registration limits
   forever, pruning ciphertext from registered CGs much earlier than
   intended. Extracted the probe-and-mark step as
   `maybeMarkRegisteredForHostMode()` and reused it from both call
   sites.

Test harness hardening
- `wait_for_peer_link` helper polls `/api/connections` until a target
  peer shows up, with a 90s budget. Use it before SCENARIO C and D
  sender-key sends to avoid pre-existing `send timeout: backoff
  aborted by overall deadline (20000ms)` flakes after curator restart.
- The flake was not LU-6-related (it pre-dates this PR); the probe
  just makes the suite reliable without changing the protocol-level
  timeout.

Validation

- All 4 late-joiner devnet scenarios PASS on a fresh
  `./scripts/devnet.sh start 6` run. SCENARIO D output now shows
  `totalInsertedTriples: 1` (was `0` before R1), `standardInserted
  Triples: 0`, `hostCatchup.triggeredForContextGraphIds: ["<CG_D>"]`,
  `perContextGraph: [{...perPeer:[...]}]` — all the new visibility
  surfaces working.
- `packages/agent/test/swm/host-mode-store.test.ts` — 9 tests pass
  (8 original + the new seqno-recovery regression).
- `packages/publisher` test suite — 965 passed / 1 skipped.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: 2 people

packages/agent/src/dkg-agent.ts

@@ -8389,8 +8389,24 @@ export class DKGAgent {
     const ctx = createOperationContext('system');
     if (!(await this.canUseSharedMemoryForContextGraph(contextGraphId))) {
       if (isRegistered) {
+        // `gossip.unsubscribe()` drops EVERY handler on the topic,
+        // not just the member-mode one. If this core was already
+        // hosting the curated SWM in HOST MODE (LU-6), losing
+        // member authorisation here would also kill the host
+        // listener, and `swmHostModeSubscribed` would still be set
+        // — making `reconcileSwmHostModeSubscription()` early-
+        // return on the next pass and stranding the hosting state
+        // until restart (Codex PR #610 R1 comment 4).
+        //
+        // We work around the topic-wide unsubscribe by clearing
+        // the host-mode subscribed flag here so the immediate
+        // `reconcileSwmHostModeSubscription()` call below will
+        // re-wire the host listener if host mode is still
+        // applicable.
+        const wasHostModeSubscribed = this.swmHostModeSubscribed.has(contextGraphId);
         this.gossip.unsubscribe(swmTopic);
         this.sharedMemoryGossipRegistered.delete(contextGraphId);
+        if (wasHostModeSubscribed) this.swmHostModeSubscribed.delete(contextGraphId);
         this.log.warn(ctx, `SWM gossip unsubscribed for "${contextGraphId}": local node is no longer authorized`);
       } else {
         this.log.warn(ctx, `SWM gossip subscription denied for "${contextGraphId}": local node is not authorized`);
@@ -8503,14 +8519,7 @@ export class DKGAgent {
       });
     });
 
-    // If the CG is already on-chain registered, transition the
-    // store immediately to the larger registered-CG limits.
-    try {
-      const onChainKnown = await this.isContextGraphRegisteredOnChain(contextGraphId);
-      if (onChainKnown) {
-        await this.swmHostModeStore.markRegistered(contextGraphId);
-      }
-    } catch { /* best-effort */ }
+    await this.maybeMarkRegisteredForHostMode(contextGraphId);
 
     this.log.info(
       createOperationContext('system'),
@@ -8900,6 +8909,12 @@ export class DKGAgent {
       return { subscribed: false, alreadySubscribed: false, hostingEnabled: true };
     }
     if (this.swmHostModeSubscribed.has(contextGraphId)) {
+      // Idempotent re-entry: even when the subscription is already
+      // active, re-probe registration state. This handles the
+      // legitimate "CG was unregistered when first subscribed,
+      // operator later registered it on-chain, operator re-calls
+      // /host-mode/subscribe" flow without forcing a daemon restart.
+      await this.maybeMarkRegisteredForHostMode(contextGraphId);
       return { subscribed: false, alreadySubscribed: true, hostingEnabled: true };
     }
     const swmTopic = contextGraphWorkspaceTopic(contextGraphId);
@@ -8914,13 +8929,36 @@ export class DKGAgent {
         );
       });
     });
+    // Codex PR #610 R1 comment 5: a core that only knows the CG by
+    // topic id (the explicit /host-mode/subscribe entrypoint) must
+    // still transition the store to the registered-CG limits as
+    // soon as the on-chain record exists. Without this probe the
+    // store would stay on the 6h/1MiB pre-registration defaults
+    // forever and prune ciphertext from registered CGs much
+    // earlier than intended.
+    await this.maybeMarkRegisteredForHostMode(contextGraphId);
     this.log.info(
       createOperationContext('system'),
       `SWM host-mode subscription explicitly enabled for "${contextGraphId}" via API (role=${this.config.nodeRole ?? 'edge'})`,
     );
     return { subscribed: true, alreadySubscribed: false, hostingEnabled: true };
   }
 
+  /**
+   * Probe on-chain registration and flip the host-mode store's
+   * per-CG cursor to the registered-CG limits when the CG is
+   * already known to the contracts. Safe to call repeatedly and on
+   * unregistered CGs — both branches early-return without touching
+   * the store.
+   */
+  private async maybeMarkRegisteredForHostMode(contextGraphId: string): Promise<void> {
+    if (!this.swmHostModeStore) return;
+    try {
+      const onChainKnown = await this.isContextGraphRegisteredOnChain(contextGraphId);
+      if (onChainKnown) await this.swmHostModeStore.markRegistered(contextGraphId);
+    } catch { /* best-effort; pre-registration defaults stay in place */ }
+  }
+
   /**
    * Receiver handler for `PROTOCOL_SWM_SHARE_ACK`. Extracted into
    * a named method (mirrors `handleSwmUpdate`'s shape) so the

packages/agent/src/swm/host-mode-store.ts

@@ -352,21 +352,75 @@ export class SwmHostModeStore {
     return meta.registered ? this.registeredLimits : this.unregisteredLimits;
   }
 
+  /**
+   * Load (and cache) the per-CG metadata. On a cold load the seqno
+   * cursor is reconciled against the actual log file: a crash
+   * between `appendFile` (durable) and `persistMeta` (durable) would
+   * otherwise let the next append reuse the same seqno, which would
+   * break host-catchup paging that uses strict-greater-than seqno.
+   *
+   * The log is the source of truth for what was actually persisted;
+   * the meta file is a cache of the highest-known seqno plus the
+   * `registered` flag. After process start we always trust the log
+   * tail's max seqno over the meta file's cursor if the two disagree
+   * — taking `max(metaSeqno, lastLogSeqno)` guarantees we never
+   * recycle a seqno even if the meta write lost a race to the crash.
+   */
   private async loadMeta(contextGraphId: string): Promise<CgMetaState> {
     const cgKey = this.cgKey(contextGraphId);
     const cached = this.metaCache.get(cgKey);
     if (cached) return cached;
     const metaPath = this.metaPath(contextGraphId);
+    let parsed: CgMetaState | undefined;
     try {
       const txt = await fs.readFile(metaPath, 'utf-8');
-      const parsed = JSON.parse(txt) as CgMetaState;
-      this.metaCache.set(cgKey, parsed);
-      return parsed;
+      parsed = JSON.parse(txt) as CgMetaState;
     } catch {
-      const fresh: CgMetaState = { seqno: 0, registered: false, contextGraphId };
-      this.metaCache.set(cgKey, fresh);
-      return fresh;
+      parsed = undefined;
+    }
+    const logSeqno = await this.recoverLastSeqnoFromLog(contextGraphId);
+    const state: CgMetaState = {
+      seqno: Math.max(parsed?.seqno ?? 0, logSeqno),
+      registered: parsed?.registered ?? false,
+      contextGraphId,
+    };
+    // If the log says more than the meta does, persist the
+    // reconciled cursor so subsequent cold loads don't have to
+    // re-scan the log tail.
+    if (parsed && state.seqno !== parsed.seqno) {
+      await fs.writeFile(metaPath, JSON.stringify(state)).catch(() => { /* best-effort */ });
+    }
+    this.metaCache.set(cgKey, state);
+    return state;
+  }
+
+  /**
+   * Scan the per-CG log tail and return the highest seqno actually
+   * persisted on disk. Reads the whole file (the per-CG cap keeps
+   * this bounded — default 1 MiB unregistered, 64 MiB registered)
+   * and walks frame-by-frame. Returns 0 if no log file exists or
+   * the file is empty/corrupt at the head.
+   */
+  private async recoverLastSeqnoFromLog(contextGraphId: string): Promise<number> {
+    const filePath = this.logPath(contextGraphId);
+    if (!(await fileExists(filePath))) return 0;
+    let buf: Buffer;
+    try {
+      buf = await fs.readFile(filePath);
+    } catch {
+      return 0;
+    }
+    let lastSeqno = 0;
+    let offset = 0;
+    while (offset + ENTRY_HEADER_BYTES <= buf.length) {
+      const seqno = Number(buf.readBigUInt64BE(offset + 8));
+      const len = buf.readUInt32BE(offset + 16);
+      const end = offset + ENTRY_HEADER_BYTES + len;
+      if (end > buf.length) break;
+      if (seqno > lastSeqno) lastSeqno = seqno;
+      offset = end;
     }
+    return lastSeqno;
   }
 
   private async persistMeta(contextGraphId: string, meta: CgMetaState): Promise<void> {

packages/agent/test/swm/host-mode-store.test.ts

@@ -125,6 +125,40 @@ describe('SwmHostModeStore', () => {
     await expect(store.append('cg/empty', new Uint8Array(0))).rejects.toThrow(/zero-length/);
   });
 
+  it('recovers seqno from log tail when meta cursor is stale (crash between appendFile and persistMeta)', async () => {
+    const cgId = 'curator/test-7';
+    const limits = { perCgByteCap: 1024 * 1024, ttlMs: 60_000 };
+    const first = new SwmHostModeStore({ dataDir: dir, unregisteredLimits: limits, registeredLimits: limits });
+    await first.append(cgId, new Uint8Array([1]));
+    await first.append(cgId, new Uint8Array([2]));
+    await first.append(cgId, new Uint8Array([3]));
+
+    // Simulate a crash where persistMeta lost a race and the meta
+    // file reports a lower seqno than the log actually contains.
+    // We do this by overwriting the meta file in place with the
+    // stale cursor (seqno=1, but the log has 3 entries).
+    const { promises: fs } = await import('node:fs');
+    const { createHash } = await import('node:crypto');
+    const cgKey = createHash('sha256').update(cgId).digest('base64url');
+    const metaPath = path.join(dir, `${cgKey}.meta`);
+    await fs.writeFile(metaPath, JSON.stringify({ seqno: 1, registered: false, contextGraphId: cgId }));
+
+    // Fresh store instance reads the stale meta + scans the log tail
+    // and reconciles to max(1, 3) = 3. The next append MUST assign
+    // seqno 4, not 2 (which would otherwise collide with the
+    // existing entry).
+    const second = new SwmHostModeStore({ dataDir: dir, unregisteredLimits: limits, registeredLimits: limits });
+    const recovered = await second.getLastSeqno(cgId);
+    expect(recovered).toBe(3);
+    const next = await second.append(cgId, new Uint8Array([4]));
+    expect(next).toBe(4);
+
+    // All four entries are visible via iterate, with unique
+    // strictly-increasing seqnos.
+    const all = await second.iterate(cgId, 0);
+    expect(all.map((e) => e.seqno)).toEqual([1, 2, 3, 4]);
+  });
+
   it('isolates CGs by id', async () => {
     const store = new SwmHostModeStore({
       dataDir: dir,