RFC-64 M1 1/2: make Edge CG sync explicitly on-demand

packages/agent/src/context-graph-subscription-policy.ts

@@ -0,0 +1,79 @@
+import type {
+  ContextGraphSub,
+  ContextGraphSubInput,
+  ContextGraphSubscriptionRecord,
+  ContextGraphSyncMode,
+} from './dkg-agent-types.js';
+
+export function normalizeLegacyContextGraphSubscriptionInput(
+  previous: ContextGraphSub | undefined,
+  next: ContextGraphSubInput,
+): ContextGraphSub {
+  return {
+    ...next,
+    syncMode: next.syncMode ?? previous?.syncMode ?? 'always-on',
+  };
+}
+
+export function resolveContextGraphSyncMode(input: {
+  existing?: Pick<ContextGraphSub, 'subscribed' | 'syncMode'>;
+  requested?: ContextGraphSyncMode;
+  hasDormantDurableIntent: boolean;
+}): ContextGraphSyncMode {
+  if (
+    input.hasDormantDurableIntent
+    || (input.existing?.subscribed === true && input.existing.syncMode === 'always-on')
+  ) {
+    return 'always-on';
+  }
+  return input.requested ?? input.existing?.syncMode ?? 'always-on';
+}
+
+export type ContextGraphSubscriptionPersistenceProjection =
+  | { action: 'skip'; persistMemberIntent: false }
+  | { action: 'delete'; persistMemberIntent: true }
+  | {
+    action: 'save';
+    persistMemberIntent: boolean;
+    record: ContextGraphSubscriptionRecord;
+  };
+
+/**
+ * Canonical durable projection for live Context Graph subscription state.
+ *
+ * On-demand member intent remains process-local. A Core hosting obligation is
+ * independently durable and therefore projects to a host-only row. Always-on
+ * member intent projects the complete live readiness state.
+ */
+export function projectContextGraphSubscriptionPersistence(input: {
+  contextGraphId: string;
+  subscription: ContextGraphSub | undefined;
+  syncScoped: boolean;
+}): ContextGraphSubscriptionPersistenceProjection {
+  const sub = input.subscription;
+  if (sub?.syncMode === 'on-demand' && sub.coreHosted !== true) {
+    return { action: 'skip', persistMemberIntent: false };
+  }
+  if (!sub?.subscribed && !sub?.coreHosted) {
+    return { action: 'delete', persistMemberIntent: true };
+  }
+
+  const persistMemberIntent = sub.syncMode !== 'on-demand';
+  return {
+    action: 'save',
+    persistMemberIntent,
+    record: {
+      id: input.contextGraphId,
+      name: sub.name,
+      subscribed: persistMemberIntent && sub.subscribed,
+      synced: persistMemberIntent && sub.synced,
+      sharedMemorySynced: persistMemberIntent ? sub.sharedMemorySynced : false,
+      metaSynced: persistMemberIntent ? sub.metaSynced : false,
+      onChainId: sub.onChainId,
+      onChainHash: sub.onChainHash,
+      lastReconciledOrdinal: sub.lastReconciledOrdinal,
+      coreHosted: sub.coreHosted,
+      syncScoped: persistMemberIntent && input.syncScoped,
+    },
+  };
+}

packages/agent/src/dkg-agent-cg-registry.ts

@@ -1048,9 +1048,10 @@ export class ContextGraphRegistryMethods extends DKGAgentBase {
       // one curator triple per node and `getContextGraphOwner`'s
       // `LIMIT 1` made ownership nondeterministic — any subscriber could
       // win the unordered query and look like the curator.
-      this.subscribeToContextGraph(opts.id);
+      this.subscribeToContextGraph(opts.id, { syncMode: 'always-on' });
       this.setContextGraphSubscription(opts.id, {
         name: opts.name,
+        syncMode: 'always-on',
         subscribed: true,
         synced: true,
         metaSynced: true,
@@ -1108,9 +1109,10 @@ export class ContextGraphRegistryMethods extends DKGAgentBase {
     this.contextGraphMetaProjection.markDirtyFromQuads(quads);
     await gm.ensureNewContextGraph(opts.id);
 
-    this.subscribeToContextGraph(opts.id);
+    this.subscribeToContextGraph(opts.id, { syncMode: 'always-on' });
     this.setContextGraphSubscription(opts.id, {
       name: opts.name,
+      syncMode: 'always-on',
       subscribed: true,
       synced: true,
       metaSynced: true,

packages/agent/src/dkg-agent-context-graph.ts

@@ -696,6 +696,7 @@ export class ContextGraphMethods extends DKGAgentBase {
 
     this.setContextGraphSubscription(opts.id, {
       name: opts.name,
+      syncMode: 'always-on',
       subscribed: !opts.private,
       synced: true,
       metaSynced: true,
@@ -817,7 +818,7 @@ export class ContextGraphMethods extends DKGAgentBase {
     }
 
     if (!opts.private) {
-      this.subscribeToContextGraph(opts.id);
+      this.subscribeToContextGraph(opts.id, { syncMode: 'always-on' });
 
       // Curated CGs: definition lives in _meta, NOT in ONTOLOGY. Do not
       // broadcast to the network — only invited nodes will discover it via
@@ -1483,8 +1484,11 @@ export class ContextGraphMethods extends DKGAgentBase {
       const next = { ...sub, onChainHash: nameHash };
       this.bindSubscriptionOnChainId(id, next, onChainId);
       this.setContextGraphSubscription(id, next, { persist: false });
+      this.subscribeToContextGraph(id, {
+        trackSyncScope: true,
+        syncMode: 'always-on',
+      });
       if (!next.subscribed) {
-        this.subscribeToContextGraph(id, { trackSyncScope: true });
         this.log.info(ctx, `Subscribed to newly registered context graph "${id}"`);
       }
       this.persistContextGraphSubscription(id);

packages/agent/src/dkg-agent-lifecycle.ts

@@ -505,6 +505,7 @@ import {
   type SyncReconcilerProbe,
   type SyncReconcilerBackoff,
 } from './dkg-agent-types.js';
+import { projectContextGraphSubscriptionPersistence } from './context-graph-subscription-policy.js';
 import {
   authoritativeSyncPeerId,
   resolveCuratorSyncPeer,
@@ -2545,6 +2546,7 @@ export class LifecycleSyncMethods extends DKGAgentBase {
             // upsert a minimal stub first.
             if (!this.subscribedContextGraphs.has(localId)) {
               this.setContextGraphSubscription(localId, {
+                syncMode: 'always-on',
                 subscribed: false,
                 synced: false,
                 onChainHash: hashLower,
@@ -2881,6 +2883,7 @@ export class LifecycleSyncMethods extends DKGAgentBase {
             }
             const approvedSubscription: ContextGraphSub = {
               ...this.subscribedContextGraphs.get(contextGraphId),
+              syncMode: 'always-on',
               subscribed: true,
               pendingMeta: true,
               metaSynced: false,
@@ -2949,6 +2952,7 @@ export class LifecycleSyncMethods extends DKGAgentBase {
             // update/finalization) wire up immediately as before.
             this.subscribeToContextGraph(contextGraphId, {
               deferSharedMemoryGossipSubscribe: true,
+              syncMode: 'always-on',
               // The exact approval snapshot was committed above. Scheduling
               // the ordinary background persistence here would reintroduce
               // untracked writes around the compensating transaction.
@@ -3238,7 +3242,7 @@ export class LifecycleSyncMethods extends DKGAgentBase {
 
     // Subscribe to both system context graph GossipSub topics
     for (const systemContextGraph of [SYSTEM_CONTEXT_GRAPHS.AGENTS, SYSTEM_CONTEXT_GRAPHS.ONTOLOGY]) {
-      this.subscribeToContextGraph(systemContextGraph);
+      this.subscribeToContextGraph(systemContextGraph, { syncMode: 'always-on' });
     }
 
     // Connect to bootstrap peers
@@ -7139,9 +7143,10 @@ export class LifecycleSyncMethods extends DKGAgentBase {
       ? this.contextGraphWireId(next.onChainHash)
       : undefined;
     const nextWireId = nextOnChainHash ?? localWireId;
-    const canonicalNext = next.onChainHash === nextOnChainHash
-      ? next
-      : { ...next, onChainHash: nextOnChainHash };
+    const canonicalNext: ContextGraphSub = {
+      ...next,
+      ...(next.onChainHash === nextOnChainHash ? {} : { onChainHash: nextOnChainHash }),
+    };
     if (
       previousWireId !== nextWireId
       && this.wireIdToLocalCgId.get(previousWireId) === contextGraphId
@@ -7153,7 +7158,16 @@ export class LifecycleSyncMethods extends DKGAgentBase {
     if (!canonicalNext.subscribed && !canonicalNext.coreHosted) {
       this.clearVmReconcileStateForContextGraph(contextGraphId);
     }
-    if (options?.persist !== false) {
+    // On-demand member subscriptions deliberately keep their live state and
+    // readiness process-local. A Core's independent hosting obligation is
+    // still durable, though: persistContextGraphSubscription writes a
+    // host-only snapshot without converting the member intent to always-on.
+    const persistence = projectContextGraphSubscriptionPersistence({
+      contextGraphId,
+      subscription: canonicalNext,
+      syncScoped: (this.config.syncContextGraphs ?? []).includes(contextGraphId),
+    });
+    if (options?.persist !== false && persistence.action !== 'skip') {
       if (this.config.contextGraphSubscriptionStore) {
         const revision = this.nextContextGraphSubscriptionPersistRevision(contextGraphId);
         this.persistContextGraphSubscription(
@@ -7164,9 +7178,9 @@ export class LifecycleSyncMethods extends DKGAgentBase {
           },
         );
       }
-      if (canonicalNext.subscribed) {
+      if (persistence.persistMemberIntent && canonicalNext.subscribed) {
         this.persistLocalNodeMembership(contextGraphId);
-      } else {
+      } else if (persistence.persistMemberIntent) {
         this.deleteContextGraphMember(contextGraphId, 'node', this.peerId);
       }
     }
@@ -7263,7 +7277,7 @@ export class LifecycleSyncMethods extends DKGAgentBase {
 
   updateContextGraphSubscriptionRehydrationStatusAfterPersist(this: DKGAgent,
     contextGraphId: string,
-    next?: ContextGraphSub,
+    next?: Pick<ContextGraphSubscriptionRecord, 'subscribed' | 'coreHosted'>,
   ): void {
     const status = this.contextGraphSubscriptionRehydrationStatus;
     if (!status) return;
@@ -7403,12 +7417,24 @@ export class LifecycleSyncMethods extends DKGAgentBase {
       return Promise.resolve();
     }
     const sub = this.subscribedContextGraphs.get(contextGraphId);
+    const persistence = projectContextGraphSubscriptionPersistence({
+      contextGraphId,
+      subscription: sub,
+      syncScoped: (this.config.syncContextGraphs ?? []).includes(contextGraphId),
+    });
+    if (persistence.action === 'skip') {
+      // Some lifecycle paths persist reconciliation watermarks directly
+      // instead of going through setContextGraphSubscription. Preserve the
+      // process-local lifetime at this lowest shared write boundary too.
+      this.clearContextGraphSubscriptionPersistRevisionStateIfIdle(contextGraphId);
+      return Promise.resolve();
+    }
     // Persist member subscriptions AND (Phase D) public CGs this Core hosts —
     // the host-only record MUST survive restart so a Core that was offline
     // during a publish remembers it hosts the CG and fills its gap. Drop the
     // row only when the node neither subscribes to nor hosts the CG.
     this.beginContextGraphSubscriptionPersistRevision(contextGraphId, options?.revision);
-    if (!sub?.subscribed && !sub?.coreHosted) {
+    if (persistence.action === 'delete') {
       return this.enqueueContextGraphSubscriptionPersistWrite(contextGraphId, () => store.delete(contextGraphId))
         .then(() => {
           if (
@@ -7428,26 +7454,14 @@ export class LifecycleSyncMethods extends DKGAgentBase {
           this.finishContextGraphSubscriptionPersistRevision(contextGraphId, options?.revision);
         });
     }
-    const record = {
-      id: contextGraphId,
-      name: sub.name,
-      subscribed: sub.subscribed,
-      synced: sub.synced,
-      sharedMemorySynced: sub.sharedMemorySynced,
-      metaSynced: sub.metaSynced,
-      onChainId: sub.onChainId,
-      onChainHash: sub.onChainHash,
-      lastReconciledOrdinal: sub.lastReconciledOrdinal,
-      coreHosted: sub.coreHosted,
-      syncScoped: (this.config.syncContextGraphs ?? []).includes(contextGraphId),
-    };
+    const record = persistence.record;
     return this.enqueueContextGraphSubscriptionPersistWrite(contextGraphId, () => store.save(record))
       .then(() => {
         if (
           options?.updateRehydrationStatus === true &&
           this.claimContextGraphSubscriptionPersistRevision(contextGraphId, options.revision)
         ) {
-          this.updateContextGraphSubscriptionRehydrationStatusAfterPersist(contextGraphId, sub);
+          this.updateContextGraphSubscriptionRehydrationStatusAfterPersist(contextGraphId, record);
         }
       }).catch((err) => {
         this.log.warn(
@@ -7483,19 +7497,17 @@ export class LifecycleSyncMethods extends DKGAgentBase {
         `Cannot persist context graph "${contextGraphId}": active subscription or host state is missing`,
       );
     }
-    const record = {
-      id: contextGraphId,
-      name: sub.name,
-      subscribed: sub.subscribed,
-      synced: sub.synced,
-      sharedMemorySynced: sub.sharedMemorySynced,
-      metaSynced: sub.metaSynced,
-      onChainId: sub.onChainId,
-      onChainHash: sub.onChainHash,
-      lastReconciledOrdinal: sub.lastReconciledOrdinal,
-      coreHosted: sub.coreHosted,
+    const persistence = projectContextGraphSubscriptionPersistence({
+      contextGraphId,
+      subscription: sub,
       syncScoped: syncScoped ?? (this.config.syncContextGraphs ?? []).includes(contextGraphId),
-    };
+    });
+    if (persistence.action !== 'save' || !persistence.persistMemberIntent) {
+      throw new Error(
+        `Cannot acknowledge join approval for "${contextGraphId}": durable subscription intent is missing`,
+      );
+    }
+    const record = persistence.record;
     // Queue behind any fire-and-forget writes scheduled by subscribe/mark so
     // this final authoritative snapshot is the last write before the ACK.
     await this.enqueueContextGraphSubscriptionPersistWrite(contextGraphId, async () => {
@@ -7935,6 +7947,9 @@ export class LifecycleSyncMethods extends DKGAgentBase {
         const restorePendingMeta = hasJoinApproval && !approvedAgentAuthorized;
         this.setContextGraphSubscription(row.id, {
           name: row.name,
+          // Every row in the durable store predates or represents explicit
+          // restart persistence, so absence of a mode is always-on.
+          syncMode: 'always-on',
           subscribed: row.subscribed,
           synced: restorePendingMeta ? false : row.synced,
           sharedMemorySynced: restorePendingMeta ? false : row.sharedMemorySynced,
@@ -7951,7 +7966,11 @@ export class LifecycleSyncMethods extends DKGAgentBase {
           this.trackSyncContextGraph(row.id);
         }
         if (row.subscribed) {
-          this.subscribeToContextGraph(row.id, { trackSyncScope: false, persist: false });
+          this.subscribeToContextGraph(row.id, {
+            trackSyncScope: false,
+            persist: false,
+            syncMode: 'always-on',
+          });
           this.persistLocalNodeMembership(row.id, 'rehydrated-subscription');
         }
         // Upgrade/self-heal path for late private-CG members whose payload and

packages/agent/src/dkg-agent-publish.ts

@@ -2495,9 +2495,9 @@ export class PublishMethods extends DKGAgentBase {
     this.contextGraphMetaProjection.markDirtyFromQuads(quads);
     await gm.ensureContextGraph(contextGraphId);
     await this.store.flush?.();
-    this.subscribeToContextGraph(contextGraphId);
+    const promotedSub = this.subscribeToContextGraph(contextGraphId, { syncMode: 'always-on' });
     this.setContextGraphSubscription(contextGraphId, {
-      ...existingSub,
+      ...promotedSub,
       name,
       subscribed: true,
       synced: true,

packages/agent/src/dkg-agent-swm-host.ts

@@ -1236,6 +1236,7 @@ export class SwmHostModeMethods extends DKGAgentBase {
     // translate either direction without an extra RPC.
     if (storageCgId !== contextGraphId) {
       const storageSubscription = this.subscribedContextGraphs.get(storageCgId) ?? {
+        syncMode: 'always-on' as const,
         subscribed: false,
         synced: false,
         pendingMeta: true,
@@ -1804,6 +1805,7 @@ export class SwmHostModeMethods extends DKGAgentBase {
     // that did not create or join the CG.
     if (!this.subscribedContextGraphs.has(wireId)) {
       this.setContextGraphSubscription(wireId, {
+        syncMode: 'always-on',
         subscribed: false,
         synced: false,
         onChainHash: wireId,
@@ -2578,7 +2580,13 @@ export class SwmHostModeMethods extends DKGAgentBase {
       existing.coreHosted = true;
       next = existing;
     } else {
-      next = { subscribed: false, synced: false, onChainId: numericStr, coreHosted: true };
+      next = {
+        syncMode: 'always-on',
+        subscribed: false,
+        synced: false,
+        onChainId: numericStr,
+        coreHosted: true,
+      };
     }
     this.setContextGraphSubscription(localCgId, next);
     this.log.info(