fix(agent): autonomously retry durable finalization inbox

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

@@ -1664,6 +1664,7 @@ export class DKGAgentBase {
 
   /** Drain and checkpoint the inbox before releasing the broader persistence lifetime. */
   protected async closeFinalizationRecoveryStore(): Promise<void> {
+    await this.finalizationHandler?.stopRecoveryWorker();
     const store = this.finalizationRuntime.detachRecoveryStore();
     await store?.close();
   }
@@ -1679,6 +1680,7 @@ export class DKGAgentBase {
         degradedReason: 'not-configured',
         stateCounts: {},
         livePayloadBytes: 0,
+        dueEntries: 0,
       };
     }
     const health = await store.health();
@@ -1692,7 +1694,9 @@ export class DKGAgentBase {
         && canonicalReceiptCapability === 'supported',
       canonicalReceiptCapability,
       ...(
-        canonicalReceiptCapability === 'unsupported' && health.available
+        canonicalReceiptCapability === 'unsupported'
+          && health.available
+          && health.degradedReason === undefined
           ? {
               degradedReason: 'canonical-finalization-receipt-unsupported',
             }

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

@@ -3273,6 +3273,14 @@ export class LifecycleSyncMethods extends DKGAgentBase {
     }, MESSAGE_OUTBOX_TICK_MS);
     if (this.messengerOutboxTimer.unref) this.messengerOutboxTimer.unref();
 
+    // The durable finalization inbox is an executable retry queue, not only a
+    // write-ahead journal. Its lifecycle is independent of chain-cursor
+    // progress so entries received after a watermark advance are still
+    // reconsidered. The worker batches SQLite reads but serializes graph work.
+    if (this.finalizationRuntime.getRecoveryStore()) {
+      this.getOrCreateFinalizationHandler().startRecoveryWorker();
+    }
+
     // Wire V10 Random Sampling prover. Edge nodes no-op. Core nodes with
     // transient identity/RPC startup failures retry in the background so
     // one flaky `getIdentityId()` call does not disable proving until the

packages/agent/src/dkg-agent.ts

@@ -1754,6 +1754,10 @@ export class DKGAgent extends DKGAgentBase {
         );
       }
     }
+    // Stop admission and await the active finalization recovery batch while
+    // chain and graph-store dependencies are still alive. No new retry may
+    // begin after this boundary.
+    await this.finalizationHandler?.stopRecoveryWorker();
     // OT-RFC-64 Gate 1: unregister the public catalog protocols and drain the
     // receiver scheduler (awaiting in-flight durable stage writes) while the
     // router, node, and control-object store are all still live — before

packages/agent/src/finalization-handler.ts

@@ -79,6 +79,9 @@ import {
   type FinalizationRecoveryPreparedMaterialization,
   type FinalizationRecoveryReplayOutcome,
 } from './finalization-recovery.js';
+import {
+  FinalizationRecoveryWorker,
+} from './finalization-recovery-worker.js';
 import type {
   FinalizationRecoveryEntry,
   FinalizationRecoveryStore,
@@ -381,6 +384,7 @@ export class FinalizationHandler {
   private readonly negativeSnapshotMemo = new Map<string, NegativeSnapshotMemoEntry>();
   /** Equivalent finalization/reconcile reads share one promise until it settles. */
   private readonly scanSingleFlights = new Map<string, Promise<unknown>>();
+  private readonly recoveryWorker: FinalizationRecoveryWorker;
 
   constructor(
     store: TripleStore,
@@ -448,6 +452,21 @@ export class FinalizationHandler {
       },
       materializer,
     );
+    this.recoveryWorker = new FinalizationRecoveryWorker(
+      (limit) => this.recovery.processDueBatch(limit),
+      {
+        info: (message) => this.log.info(createOperationContext('system'), message),
+        warn: (message) => this.log.warn(createOperationContext('system'), message),
+      },
+    );
+  }
+
+  startRecoveryWorker(): void {
+    this.recoveryWorker.start();
+  }
+
+  stopRecoveryWorker(): Promise<void> {
+    return this.recoveryWorker.stop();
   }
 
   async handleFinalizationMessage(

packages/agent/src/finalization-recovery-sqlite-store.ts

@@ -34,6 +34,17 @@ export type {
   SqliteFinalizationRecoveryStoreOptions,
 } from './finalization-recovery-sqlite-policy.js';
 
+const DUE_FINALIZATION_SQL_PREDICATE = `
+  (
+    state IN ('RECEIVED','VERIFIED','REORGED')
+    OR (
+      state = 'SETTLED'
+      AND (publisher_upgrade_pending = 1 OR next_attempt_at IS NOT NULL)
+    )
+  )
+  AND (next_attempt_at IS NULL OR next_attempt_at <= ?)
+`;
+
 function sameFinalizationRecoveryIdentity(
   existing: FinalizationRecoveryEntry,
   input: FinalizationRecoveryReceiveInput,
@@ -80,6 +91,16 @@ export class SqliteFinalizationRecoveryStore implements FinalizationRecoveryStor
     return this.#closed;
   }
 
+  async get(key: string): Promise<FinalizationRecoveryEntry | undefined> {
+    if (this.#closed || this.#closing) return undefined;
+    await this.#mutationTail;
+    if (this.#closed) return undefined;
+    const row = this.database.prepare(
+      'SELECT * FROM finalization_inbox_v1 WHERE key = ?',
+    ).get(key);
+    return row ? finalizationRecoveryRowToEntry(row) : undefined;
+  }
+
   receive(input: FinalizationRecoveryReceiveInput): Promise<FinalizationRecoveryReceiveResult> {
     if (this.#closed || this.#closing) return Promise.resolve({ status: 'closed' });
     return this.mutate(() => {
@@ -422,6 +443,25 @@ export class SqliteFinalizationRecoveryStore implements FinalizationRecoveryStor
     return entry.nextAttemptAt === undefined || entry.nextAttemptAt <= this.#policy.now();
   }
 
+  async listDue(limit: number): Promise<FinalizationRecoveryEntry[]> {
+    if (this.#closed || this.#closing) return [];
+    if (!Number.isFinite(limit)) return [];
+    const boundedLimit = Math.min(
+      this.#policy.maxEntries,
+      Math.max(0, Math.trunc(limit)),
+    );
+    if (boundedLimit === 0) return [];
+    await this.#mutationTail;
+    if (this.#closed) return [];
+    const now = this.#policy.now();
+    return this.database.prepare(`
+      SELECT * FROM finalization_inbox_v1
+      WHERE ${DUE_FINALIZATION_SQL_PREDICATE}
+      ORDER BY COALESCE(next_attempt_at, updated_at), updated_at, key
+      LIMIT ?
+    `).all(now, boundedLimit).map(finalizationRecoveryRowToEntry);
+  }
+
   async listForKnowledgeAsset(input: {
     chainId: string;
     contextGraphId: string;
@@ -497,17 +537,24 @@ export class SqliteFinalizationRecoveryStore implements FinalizationRecoveryStor
     return this.mutate(() => {
       if (this.#closed) return;
       const now = this.#policy.now();
+      const nextAttemptAt = retryDelayMs === undefined ? null : now + retryDelayMs;
       this.database.prepare(`
         UPDATE finalization_inbox_v1
         SET attempt_count = attempt_count + 1,
             last_error = ?,
-            next_attempt_at = ?,
+            next_attempt_at = CASE
+              WHEN ? IS NULL THEN next_attempt_at
+              WHEN next_attempt_at IS NULL THEN ?
+              ELSE MAX(next_attempt_at, ?)
+            END,
             updated_at = ?
         WHERE key = ? AND generation = ?
           AND state IN ('RECEIVED','VERIFIED','REORGED','SETTLED')
       `).run(
         lastError ?? null,
-        retryDelayMs === undefined ? null : now + retryDelayMs,
+        nextAttemptAt,
+        nextAttemptAt,
+        nextAttemptAt,
         now,
         key,
         generation,
@@ -523,17 +570,31 @@ export class SqliteFinalizationRecoveryStore implements FinalizationRecoveryStor
         degradedReason: 'closing',
         stateCounts: {},
         livePayloadBytes: 0,
+        dueEntries: 0,
       };
     }
     await this.#mutationTail;
     if (this.#closed) {
-      return { available: false, closed: true, degradedReason: 'closed', stateCounts: {}, livePayloadBytes: 0 };
+      return {
+        available: false,
+        closed: true,
+        degradedReason: 'closed',
+        stateCounts: {},
+        livePayloadBytes: 0,
+        dueEntries: 0,
+      };
     }
     const counts = this.database.prepare(
       'SELECT state, COUNT(*) AS count FROM finalization_inbox_v1 GROUP BY state',
     ).all();
     const stateCounts: Partial<Record<FinalizationRecoveryState, number>> = {};
     for (const row of counts) stateCounts[String(row.state) as FinalizationRecoveryState] = Number(row.count);
+    const now = this.#policy.now();
+    const due = this.database.prepare(`
+      SELECT COUNT(*) AS count, MIN(created_at) AS oldest
+      FROM finalization_inbox_v1
+      WHERE ${DUE_FINALIZATION_SQL_PREDICATE}
+    `).get(now) as { count: number | bigint; oldest: number | bigint | null };
     const capacity = readFinalizationRecoveryCapacity(this.database, this.#policy);
     return {
       available: true,
@@ -542,9 +603,13 @@ export class SqliteFinalizationRecoveryStore implements FinalizationRecoveryStor
       ...(capacity.capacityExhausted ? { degradedReason: 'capacity-exhausted' } : {}),
       stateCounts,
       livePayloadBytes: capacity.livePayloadBytes,
+      dueEntries: Number(due.count),
+      ...(due.oldest === null
+        ? {}
+        : { oldestDueAgeMs: Math.max(0, now - Number(due.oldest)) }),
       ...(capacity.oldest === undefined
         ? {}
-        : { oldestPendingAgeMs: Math.max(0, this.#policy.now() - capacity.oldest) }),
+        : { oldestPendingAgeMs: Math.max(0, now - capacity.oldest) }),
     };
   }
 

packages/agent/src/finalization-recovery-store.ts

@@ -80,11 +80,15 @@ export interface FinalizationRecoveryHealth {
   degradedReason?: string;
   stateCounts: Partial<Record<FinalizationRecoveryState, number>>;
   livePayloadBytes: number;
+  dueEntries: number;
+  oldestDueAgeMs?: number;
   oldestPendingAgeMs?: number;
 }
 
 export interface FinalizationRecoveryStore {
   readonly closed: boolean;
+  /** Reloads one entry after waiting on an in-process serialization boundary. */
+  get(key: string): Promise<FinalizationRecoveryEntry | undefined>;
   receive(input: FinalizationRecoveryReceiveInput): Promise<FinalizationRecoveryReceiveResult>;
   recordTrustedPublisher(
     key: string,
@@ -113,6 +117,12 @@ export interface FinalizationRecoveryStore {
   clearSettledRetry(key: string, generation: number): Promise<void>;
   rejectSettled(key: string, generation: number, lastError: string): Promise<boolean>;
   isAttemptDue(entry: FinalizationRecoveryEntry): boolean;
+  /**
+   * Returns a bounded, oldest-first snapshot of entries whose persisted retry
+   * gate is open. Callers must still rely on generation-checked transitions:
+   * live gossip or reconciliation may update an entry after this read.
+   */
+  listDue(limit: number): Promise<FinalizationRecoveryEntry[]>;
   listForKnowledgeAsset(input: {
     chainId: string;
     contextGraphId: string;

packages/agent/src/finalization-recovery-worker.ts

@@ -0,0 +1,100 @@
+export const FINALIZATION_RECOVERY_WORKER_BATCH_SIZE = 16;
+export const FINALIZATION_RECOVERY_WORKER_POLL_INTERVAL_MS = 5_000;
+
+interface FinalizationRecoveryWorkerLog {
+  info(message: string): void;
+  warn(message: string): void;
+}
+
+interface FinalizationRecoveryWorkerOptions {
+  batchSize?: number;
+  pollIntervalMs?: number;
+}
+
+/**
+ * Lifecycle-owned asynchronous retry loop for the durable finalization inbox.
+ *
+ * SQLite reads are batched, while graph materialization remains serial inside
+ * `processDueBatch`. This is an I/O scheduler on the Node.js event loop, not a
+ * worker thread: it never blocks startup and never introduces concurrent
+ * Blazegraph finalization writes.
+ */
+export class FinalizationRecoveryWorker {
+  readonly #batchSize: number;
+  readonly #pollIntervalMs: number;
+  #running = false;
+  #timer: ReturnType<typeof setTimeout> | undefined;
+  #inFlight: Promise<void> | undefined;
+
+  constructor(
+    private readonly processDueBatch: (limit: number) => Promise<number>,
+    private readonly log: FinalizationRecoveryWorkerLog,
+    options: FinalizationRecoveryWorkerOptions = {},
+  ) {
+    this.#batchSize = Math.max(
+      1,
+      Math.trunc(options.batchSize ?? FINALIZATION_RECOVERY_WORKER_BATCH_SIZE),
+    );
+    this.#pollIntervalMs = Math.max(
+      1,
+      Math.trunc(
+        options.pollIntervalMs ?? FINALIZATION_RECOVERY_WORKER_POLL_INTERVAL_MS,
+      ),
+    );
+  }
+
+  get running(): boolean {
+    return this.#running;
+  }
+
+  start(): void {
+    if (this.#running) return;
+    this.#running = true;
+    this.schedule(0);
+  }
+
+  async stop(): Promise<void> {
+    this.#running = false;
+    if (this.#timer) {
+      clearTimeout(this.#timer);
+      this.#timer = undefined;
+    }
+    await this.#inFlight?.catch(() => undefined);
+  }
+
+  private schedule(delayMs: number): void {
+    if (!this.#running || this.#timer) return;
+    this.#timer = setTimeout(() => {
+      this.#timer = undefined;
+      if (!this.#running) return;
+      const run = this.runBatch();
+      this.#inFlight = run;
+      void run.finally(() => {
+        if (this.#inFlight === run) this.#inFlight = undefined;
+      });
+    }, delayMs);
+    this.#timer.unref?.();
+  }
+
+  private async runBatch(): Promise<void> {
+    let selected = 0;
+    try {
+      selected = await this.processDueBatch(this.#batchSize);
+      if (selected > 0) {
+        this.log.info(
+          `Finalization recovery worker inspected ${selected} due inbox entr`
+            + `${selected === 1 ? 'y' : 'ies'}`,
+        );
+      }
+    } catch (error) {
+      this.log.warn(
+        `Finalization recovery worker batch failed: `
+          + `${error instanceof Error ? error.message : String(error)}`,
+      );
+    } finally {
+      if (this.#running) {
+        this.schedule(selected >= this.#batchSize ? 0 : this.#pollIntervalMs);
+      }
+    }
+  }
+}