feat(fast-inbox): validator streaming acceptance conditions, flag off (A-1383)

Implements A-1383: the validator's streaming-Inbox acceptance conditions (AZIP-22 Fast Inbox), behind the shared `streamingInbox` flag (default off). Flag off ⇒ byte-identical behavior. Stacked on #24787 (A-1382).

## The four block-proposal checks (`validator-client/src/streaming_inbox_checks.ts`, wired in `proposal_handler.handleBlockProposal`)
- **Exists**: the referenced bucket resolves in the node's own Inbox view and its consensus rolling hash matches the reference. An unknown bucket is an immediate reject (`bucket_unknown`); the bounded-wait soft path is A-1393. The reference is trusted only as a `bucketSeq` lookup hint — timestamp/counts are read from the locally resolved bucket.
- **Moves forward**: the bucket's cumulative total is at least the parent block's (equal ⇒ empty bundle).
- **Not too new**: the bucket is at least `INBOX_LAG_SECONDS` old at validation time (`dateProvider.now()`, inclusive boundary).
- **Caps**: the per-block count and the running per-checkpoint total fit `MAX_L1_TO_L2_MSGS_PER_BLOCK` / `MAX_L1_TO_L2_MSGS_PER_CHECKPOINT`.

The derived bundle is fed into per-block re-execution (`insertMessagesPerBlock` threaded through `openCheckpoint` → `reexecuteTransactions`); the existing state-ref comparison after re-execution stays the final arbiter.

## Last-block censorship check (`validateCheckpointProposal`)
Before attesting, the minimum-consumption rule runs via the shared `isInboxConsumptionSufficient` predicate: the first unconsumed bucket must be absent, past the cutoff, or a cap-escape. A mandatory unconsumed bucket rejects the checkpoint (no attestation, `inbox_consumption_insufficient`).

## Shared-predicate extraction (`refactor` commit)
The cutoff formula and the minimum-consumption/cap-escape rule moved to `stdlib/messaging/inbox_consumption`, so the sequencer's bucket selection and the validator share one source of truth. The sequencer's `selectStreamingBundle` now calls `getInboxCutoffTimestamp` instead of inlining it — behavior unchanged (selector suite still green).

## Corrected cutoff anchor
The cutoff is `getTimestampForSlot(slot - 1) - INBOX_LAG_SECONDS` (mirroring `ProposeLib.validateInboxConsumption`), **not** `getBuildFrameStart`. The A-1371 §13 cross-layer vectors are pinned in `inbox_consumption.test.ts`.

## Parent-ref derivation
Parent/last-consumed buckets are resolved from L1-to-L2 tree leaf counts (compact post-flip indexing == bucket cumulative total) via a new `getInboxBucketByTotalMsgCount` archiver lookup. A leaf count that does not land on a bucket boundary (a pre-flip padded parent) is rejected/deferred; the flip (A-1384) closes that gap.

## CI must typecheck what can't be verified locally
`tsc -b` cannot complete locally due to pre-existing `noir-protocol-circuits-types` breakage (missing artifacts / stale generated types), which cascades to `archiver`/`validator-client`. stdlib builds clean; no errors in the changed files outside the broken zone. Runnable unit suites are green: `streaming_inbox_checks` (14), `inbox_consumption` (10), `proposal_handler` (34, incl. flag-off unchanged + new streaming/censorship tests), `message_store` (35), stdlib `archiver` interface (RPC schema). CI is the typecheck of record for the unbuildable zone.

Out of scope: bounded-wait (A-1393), the flip (A-1384), reorg rejection (A-1389), the full streaming checkpoint rebuild (A-1385).

Replaces #24788.

Author: spalladino

noir-projects/fnd/noir-protocol-circuits/crates/types/src/constants.nr

@@ -70,7 +70,7 @@ pub global MAX_L1_TO_L2_MSGS_PER_BLOCK: u32 = 1024;
 // Cap on L1-to-L2 messages consumed by a single checkpoint. Semantically today's NUMBER_OF_L1_L2_MESSAGES_PER_ROLLUP.
 pub global MAX_L1_TO_L2_MSGS_PER_CHECKPOINT: u32 = 1024;
 // Minimum bucket age, in seconds, at the start of a checkpoint's build frame for its consumption to be mandatory under
-// the streaming Inbox censorship cutoff (AZIP-22 Fast Inbox). One L1 slot: validators cannot be required to act on
+// the streaming Inbox censorship cutoff. One L1 slot: validators cannot be required to act on
 // buckets they may not yet have seen. L2 consensus policy consumed by the sequencer and validator (and mirrored by
 // ProposeLib on L1); the protocol circuits do not read it.
 pub global INBOX_LAG_SECONDS: u32 = 12;

yarn-project/archiver/src/errors.ts

@@ -127,6 +127,17 @@ export class InboxBucketNotSyncedError extends Error {
   }
 }
 
+/**
+ * Thrown when a cumulative Inbox message count does not resolve to a bucket boundary this archiver has synced, either
+ * because the count sits inside a bucket or because the bucket is not synced yet.
+ */
+export class InboxBucketBoundaryNotSyncedError extends Error {
+  constructor(public readonly totalMsgCount: bigint) {
+    super(`No synced Inbox bucket ends at cumulative message count ${totalMsgCount}`);
+    this.name = 'InboxBucketBoundaryNotSyncedError';
+  }
+}
+
 /** Thrown when a proposed checkpoint number is stale (already processed). */
 export class ProposedCheckpointStaleError extends Error {
   constructor(

yarn-project/archiver/src/modules/data_source_base.ts

@@ -332,10 +332,18 @@ export abstract class ArchiverDataSourceBase
     return this.stores.messages.getInboxBucket(seq);
   }
 
+  public getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise<InboxBucket | undefined> {
+    return this.stores.messages.getInboxBucketByTotalMsgCount(totalMsgCount);
+  }
+
   public getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise<Fr[]> {
     return this.stores.messages.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive);
   }
 
+  public getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise<Fr[]> {
+    return this.stores.messages.getL1ToL2MessagesBetweenLeafCounts(startLeafCount, endLeafCount);
+  }
+
   private async getPublishedCheckpointFromCheckpointData(checkpoint: CheckpointData): Promise<PublishedCheckpoint> {
     const blocksForCheckpoint = await this.stores.blocks.getBlocksForCheckpoint(checkpoint.checkpointNumber);
     if (!blocksForCheckpoint) {

yarn-project/archiver/src/store/message_store.test.ts

@@ -8,7 +8,11 @@ import { Checkpoint, type PublishedCheckpoint } from '@aztec/stdlib/checkpoint';
 import { updateInboxRollingHash } from '@aztec/stdlib/messaging';
 import '@aztec/stdlib/testing/jest';
 
-import { InboxBucketNotSyncedError, L1ToL2MessagesNotReadyError } from '../errors.js';
+import {
+  InboxBucketBoundaryNotSyncedError,
+  InboxBucketNotSyncedError,
+  L1ToL2MessagesNotReadyError,
+} from '../errors.js';
 import { type InboxMessage, updateRollingHash } from '../structs/inbox_message.js';
 import {
   makeInboxMessage,
@@ -401,6 +405,28 @@ describe('MessageStore', () => {
       expect(await messageStore.getInboxBucket(4n)).toBeUndefined();
     });
 
+    it('resolves a bucket by its cumulative message total', async () => {
+      const msgs = makeBucketedMessages(threeBucketSpec);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+
+      // Each bucket boundary (cumulative totals 3, 5, 6) resolves to its bucket.
+      expect((await messageStore.getInboxBucketByTotalMsgCount(3n))?.seq).toEqual(1n);
+      expect((await messageStore.getInboxBucketByTotalMsgCount(5n))?.seq).toEqual(2n);
+      expect((await messageStore.getInboxBucketByTotalMsgCount(6n))?.seq).toEqual(3n);
+      // A total inside a bucket (not on a boundary) does not resolve.
+      expect(await messageStore.getInboxBucketByTotalMsgCount(4n)).toBeUndefined();
+      // A total past the last synced bucket does not resolve.
+      expect(await messageStore.getInboxBucketByTotalMsgCount(7n)).toBeUndefined();
+    });
+
+    it('synthesizes the genesis sentinel bucket (sequence 0, total 0) which is never ingested', async () => {
+      // With real messages present but no ingested sequence-0 snapshot, both lookups still resolve genesis.
+      await messageStore.addL1ToL2MessageBuckets(makeBucketedMessages(threeBucketSpec));
+
+      expect(await messageStore.getInboxBucket(0n)).toMatchObject({ seq: 0n, totalMsgCount: 0n, msgCount: 0 });
+      expect(await messageStore.getInboxBucketByTotalMsgCount(0n)).toMatchObject({ seq: 0n, totalMsgCount: 0n });
+    });
+
     it('rejects a bucket delivered without the messages already stored for it', async () => {
       const msgs = makeBucketedMessages(threeBucketSpec);
       await messageStore.addL1ToL2MessageBuckets(msgs.slice(0, 2));
@@ -492,6 +518,40 @@ describe('MessageStore', () => {
       await expect(messageStore.getL1ToL2MessagesBetweenBuckets(4n, 3n)).rejects.toThrow(/Invalid Inbox bucket range/);
     });
 
+    it('returns messages between cumulative leaf counts', async () => {
+      const msgs = makeBucketedMessages(threeBucketSpec);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+      const leaves = msgs.map(m => m.leaf);
+
+      // Bucket boundaries sit at cumulative counts 0 (genesis), 3, 5 and 6.
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 6n)).toEqual(leaves);
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 3n)).toEqual(leaves.slice(0, 3));
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(3n, 5n)).toEqual(leaves.slice(3, 5));
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(5n, 6n)).toEqual(leaves.slice(5));
+      // An empty range consumes nothing, at a bucket boundary or at genesis.
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(5n, 5n)).toEqual([]);
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 0n)).toEqual([]);
+    });
+
+    it('throws when a leaf count does not land on a synced bucket boundary', async () => {
+      const msgs = makeBucketedMessages(threeBucketSpec);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+
+      // Counts inside a bucket and past the last synced bucket both fail rather than returning a partial range.
+      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 4n)).rejects.toThrow(
+        InboxBucketBoundaryNotSyncedError,
+      );
+      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(4n, 6n)).rejects.toThrow(
+        InboxBucketBoundaryNotSyncedError,
+      );
+      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(3n, 9n)).rejects.toThrow(
+        InboxBucketBoundaryNotSyncedError,
+      );
+      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(5n, 3n)).rejects.toThrow(
+        /Invalid Inbox leaf count range/,
+      );
+    });
+
     it('rewinds buckets when messages are removed', async () => {
       const msgs = makeBucketedMessages(threeBucketSpec);
       await messageStore.addL1ToL2MessageBuckets(msgs);

yarn-project/archiver/src/store/message_store.ts

@@ -15,7 +15,11 @@ import {
 } from '@aztec/kv-store';
 import { type InboxBucket, InboxLeaf, updateInboxRollingHash } from '@aztec/stdlib/messaging';
 
-import { InboxBucketNotSyncedError, L1ToL2MessagesNotReadyError } from '../errors.js';
+import {
+  InboxBucketBoundaryNotSyncedError,
+  InboxBucketNotSyncedError,
+  L1ToL2MessagesNotReadyError,
+} from '../errors.js';
 import {
   type InboxMessage,
   deserializeInboxMessage,
@@ -85,6 +89,19 @@ function groupMessagesByBucket(messages: InboxMessage[]): IncomingBucket[] {
   return buckets;
 }
 
+// The genesis sentinel bucket: sequence 0 with a zero rolling hash and no messages, mirroring the
+// on-chain Inbox's base case. The archiver never ingests a snapshot for it (no message is absorbed into sequence 0), so
+// it is synthesized on read. Consumers use its sequence number and zero total; its deploy-time timestamp is not tracked
+// here and is unused.
+const GENESIS_INBOX_BUCKET: InboxBucket = {
+  seq: 0n,
+  inboxRollingHash: Fr.ZERO,
+  totalMsgCount: 0n,
+  timestamp: 0n,
+  msgCount: 0,
+  lastMessageIndex: 0n,
+};
+
 export class MessageStoreError extends Error {
   constructor(
     message: string,
@@ -487,17 +504,65 @@ export class MessageStore {
   }
 
   /**
-   * Returns the Inbox bucket with the given sequence number, or undefined if it has not been synced (AZIP-22 Fast
-   * Inbox).
+   * Returns the Inbox bucket with the given sequence number, or undefined if it has not been synced. Sequence 0 is
+   * the genesis sentinel: the on-chain Inbox reserves it as the "consumed nothing" base case
+   * and never absorbs a message into it, so the archiver ingests no snapshot for it; it is synthesized here (rolling
+   * hash 0, total 0) so streaming consumers can resolve a genesis parent or an empty checkpoint's last-consumed bucket.
    */
   public async getInboxBucket(seq: bigint): Promise<InboxBucket | undefined> {
     const snapshot = await this.getBucketSnapshotBySeq(seq);
-    return snapshot && this.toInboxBucket(seq, snapshot);
+    if (snapshot !== undefined) {
+      return this.toInboxBucket(seq, snapshot);
+    }
+    return seq === 0n ? GENESIS_INBOX_BUCKET : undefined;
+  }
+
+  /**
+   * Returns the Inbox bucket whose cumulative message total equals `totalMsgCount`, or undefined if no synced bucket
+   * sits on that boundary. Sequence 0 (total 0) is the genesis sentinel base case; otherwise the
+   * message at global index `totalMsgCount - 1` is the last message of the bucket with that cumulative total, so its
+   * `bucketSeq` resolves the bucket. A total that does not land on a bucket boundary returns undefined.
+   */
+  public async getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise<InboxBucket | undefined> {
+    if (totalMsgCount === 0n) {
+      return this.getInboxBucket(0n);
+    }
+    const buffer = await this.#l1ToL2Messages.getAsync(this.indexToKey(totalMsgCount - 1n));
+    if (buffer === undefined) {
+      return undefined;
+    }
+    const bucket = await this.getInboxBucket(deserializeInboxMessage(buffer).bucketSeq);
+    return bucket !== undefined && bucket.totalMsgCount === totalMsgCount ? bucket : undefined;
+  }
+
+  /**
+   * Returns the message leaves in the cumulative Inbox message-count range `[startLeafCount, endLeafCount)`, in
+   * insertion order. The bounds are compact L1-to-L2 tree leaf counts, which every block header
+   * carries, so consumers can ask for the messages a block or checkpoint consumed without resolving buckets
+   * themselves. Both bounds must land on a bucket boundary this archiver has synced; it throws otherwise, since a
+   * caller asking for a range always expects the messages in it.
+   */
+  public async getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise<Fr[]> {
+    if (startLeafCount > endLeafCount) {
+      throw new Error(`Invalid Inbox leaf count range [${startLeafCount}, ${endLeafCount})`);
+    }
+    const startBucket = await this.getBucketAtBoundary(startLeafCount);
+    const endBucket = await this.getBucketAtBoundary(endLeafCount);
+    return this.getL1ToL2MessagesBetweenBuckets(startBucket.seq, endBucket.seq);
+  }
+
+  /** Resolves the bucket ending at the given cumulative message count, failing loudly if there is none. */
+  private async getBucketAtBoundary(totalMsgCount: bigint): Promise<InboxBucket> {
+    const bucket = await this.getInboxBucketByTotalMsgCount(totalMsgCount);
+    if (bucket === undefined) {
+      throw new InboxBucketBoundaryNotSyncedError(totalMsgCount);
+    }
+    return bucket;
   }
 
   /**
    * Returns the latest Inbox bucket opened at or before the given L1 timestamp, or undefined if every synced bucket
-   * was opened strictly after it (AZIP-22 Fast Inbox).
+   * was opened strictly after it.
    */
   public async getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise<InboxBucket | undefined> {
     // Bucket timestamps are non-decreasing in sequence number, so the bucket we want is the highest sequence indexed
@@ -519,8 +584,8 @@ export class MessageStore {
   }
 
   /**
-   * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion order
-   * (AZIP-22 Fast Inbox). Both bounds must name buckets this archiver has synced, so that an empty result means the
+   * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion order.
+   * Both bounds must name buckets this archiver has synced, so that an empty result means the
    * range holds no messages rather than hiding an unsynced bound; callers route the
    * `InboxBucketNotSyncedError` to their own catch-up handling. Sequence 0 is the genesis base case and always
    * resolves: the range then starts at the first message of the Inbox.

yarn-project/archiver/src/structs/inbox_message.ts

@@ -12,9 +12,9 @@ export type InboxMessage = {
   l1BlockHash: Buffer32;
   /** Legacy 128-bit keccak rolling hash of all messages inserted up to and including this one. */
   rollingHash: Buffer16;
-  /** Consensus rolling hash (truncated sha256 chain) of all messages up to and including this one (AZIP-22 Fast Inbox). */
+  /** Consensus rolling hash (truncated sha256 chain) of all messages up to and including this one. */
   inboxRollingHash: Fr;
-  /** Sequence number of the Inbox bucket this message was absorbed into (AZIP-22 Fast Inbox). */
+  /** Sequence number of the Inbox bucket this message was absorbed into. */
   bucketSeq: bigint;
   /** L1 block timestamp at which this message's bucket was opened; the bucket's recency key, in seconds. */
   bucketTimestamp: bigint;

yarn-project/archiver/src/test/fake_l1_state.ts

@@ -143,7 +143,7 @@ export class FakeL1State {
   private checkpoints: CheckpointData[] = [];
   private messages: MessageData[] = [];
   private messagesRollingHash: Buffer16 = Buffer16.ZERO;
-  // Consensus rolling-hash and bucket-ring state, mirroring the on-chain Inbox (AZIP-22 Fast Inbox).
+  // Consensus rolling-hash and bucket-ring state, mirroring the on-chain Inbox.
   private messagesConsensusRollingHash: Fr = Fr.ZERO;
   private currentBucketSeq: bigint = 0n;
   private currentBucketTimestamp: bigint = 0n;

yarn-project/archiver/src/test/mock_archiver.ts

@@ -37,9 +37,17 @@ export class MockArchiver extends MockL2BlockSource implements L2BlockSource, L1
     return this.messageSource.getInboxBucket(seq);
   }
 
+  getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise<InboxBucket | undefined> {
+    return this.messageSource.getInboxBucketByTotalMsgCount(totalMsgCount);
+  }
+
   getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise<Fr[]> {
     return this.messageSource.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive);
   }
+
+  getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise<Fr[]> {
+    return this.messageSource.getL1ToL2MessagesBetweenLeafCounts(startLeafCount, endLeafCount);
+  }
 }
 
 /**

yarn-project/archiver/src/test/mock_l1_to_l2_message_source.ts

@@ -38,6 +38,13 @@ export class MockL1ToL2MessageSource implements L1ToL2MessageSource {
     return Promise.resolve(this.buckets.get(seq));
   }
 
+  getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise<InboxBucket | undefined> {
+    if (totalMsgCount === 0n) {
+      return Promise.resolve(this.buckets.get(0n));
+    }
+    return Promise.resolve([...this.buckets.values()].find(bucket => bucket.totalMsgCount === totalMsgCount));
+  }
+
   getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise<InboxBucket | undefined> {
     const atOrBefore = [...this.buckets.values()]
       .filter(bucket => bucket.timestamp <= timestamp)
@@ -52,6 +59,15 @@ export class MockL1ToL2MessageSource implements L1ToL2MessageSource {
     return Promise.resolve(seqs.flatMap(seq => this.messagesPerBucket.get(seq) ?? []));
   }
 
+  async getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise<Fr[]> {
+    const startBucket = await this.getInboxBucketByTotalMsgCount(startLeafCount);
+    const endBucket = await this.getInboxBucketByTotalMsgCount(endLeafCount);
+    if (startBucket === undefined || endBucket === undefined) {
+      throw new Error(`No mocked Inbox bucket boundary at ${startLeafCount} or ${endLeafCount}`);
+    }
+    return this.getL1ToL2MessagesBetweenBuckets(startBucket.seq, endBucket.seq);
+  }
+
   getBlockNumber() {
     return Promise.resolve(BlockNumber(this.blockNumber));
   }

yarn-project/prover-client/src/light/lightweight_checkpoint_builder.ts

@@ -208,7 +208,7 @@ export class LightweightCheckpointBuilder {
       timings.insertSideEffects = msInsertSideEffects;
     }
 
-    // Streaming Inbox (AZIP-22 Fast Inbox): insert this block's L1-to-L2 message bundle before reading the end state,
+    // Streaming Inbox: insert this block's L1-to-L2 message bundle before reading the end state,
     // so the block header's L1-to-L2 tree snapshot reflects it. First-in-checkpoint bundles are padded to
     // NUMBER_OF_L1_L2_MESSAGES_PER_ROLLUP (matching the legacy per-checkpoint insertion and the world-state
     // synchronizer); non-first bundles are appended compactly. The logical (unpadded) messages are accumulated only