feat(fast-inbox): make streaming the only node consumption path (A-1384)

Remove the streamingInbox flag and wire streaming as the sole path (AZIP-22 Fast
Inbox):

- Config: drop streamingInbox from foundation env vars, stdlib sequencer/validator
  config + schemas, and the sequencer/validator config plumbing.
- Sequencer: source the parent checkpoint's consumed bucket from the fork's
  L1-to-L2 tree leaf count + getInboxBucketByTotalMsgCount (cross-checkpoint, not
  genesis-only); feed the header/block inHash zero; relax waitForMinTxs for
  message-only blocks; pass the consumed bucket seq as the propose bucketHint.
  Automine selects its single block's bundle the same way.
- Validator: per-block acceptance and checkpoint re-execution always run; the
  checkpoint rebuild derives the consumed message list from the buckets between the
  parent checkpoint's position and the last block.
- World state: appendL1ToL2MessagesToTree and handleL2BlockAndMessages append real
  leaves at compact indices; the synchronizer derives each block's bundle from its
  leaf-index range.
- Lightweight checkpoint builder feeds the checkpoint inHash zero.

Legacy bulk-fetch/insert paths go dead (deleted in FI-18). Verified via jest:
validator proposal_handler (34) + streaming checks (14), sequencer inbox selector
(9), world-state native (56). tsc rides CI.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: spalladino

yarn-project/foundation/src/config/env_var.ts

@@ -382,7 +382,6 @@ export type EnvVar =
   | 'FISHERMAN_MODE'
   | 'MAX_ALLOWED_ETH_CLIENT_DRIFT_SECONDS'
   | 'MAX_BLOCKS_PER_CHECKPOINT'
-  | 'STREAMING_INBOX'
   | 'LEGACY_BLS_CLI'
   | 'DEBUG_FORCE_TX_PROOF_VERIFICATION'
   | 'VALIDATOR_ALLOW_EPHEMERAL_SIGNING_PROTECTION'

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

@@ -11,7 +11,6 @@ import {
   accumulateInboxRollingHash,
   appendL1ToL2MessagesToTree,
   computeCheckpointOutHash,
-  computeInHashFromL1ToL2Messages,
 } from '@aztec/stdlib/messaging';
 import { CheckpointHeader, computeBlockHeadersHash } from '@aztec/stdlib/rollup';
 import { AppendOnlyTreeSnapshot, MerkleTreeId } from '@aztec/stdlib/trees';
@@ -295,7 +294,9 @@ export class LightweightCheckpointBuilder {
     const blobs = await getBlobsPerL1Block(this.blobFields);
     const blobsHash = computeBlobsHashFromBlobs(blobs);
 
-    const inHash = computeInHashFromL1ToL2Messages(this.l1ToL2Messages);
+    // Legacy inHash is dead post-flip; the checkpoint header carries zero (AZIP-22 Fast Inbox). The consensus
+    // rolling hash over the consumed messages is the authoritative Inbox commitment.
+    const inHash = Fr.ZERO;
     const inboxRollingHash = accumulateInboxRollingHash(this.previousInboxRollingHash, this.l1ToL2Messages);
 
     const { slotNumber, coinbase, feeRecipient, gasFees } = this.constants;

yarn-project/sequencer-client/src/config.ts

@@ -68,7 +68,6 @@ export const DefaultSequencerConfig = {
   skipPushProposedBlocksToArchiver: false,
   skipPublishingCheckpointsPercent: 0,
   maxBlocksPerCheckpoint: DEFAULT_MAX_BLOCKS_PER_CHECKPOINT,
-  streamingInbox: false,
 } satisfies ResolvedSequencerConfig;
 
 /**

yarn-project/sequencer-client/src/publisher/sequencer-publisher.ts

@@ -113,6 +113,8 @@ type L1ProcessArgs = {
   attestationsAndSignersSignature: Signature;
   /** The fee asset price modifier in basis points (from oracle) */
   feeAssetPriceModifier: bigint;
+  /** Sequence number of the Inbox bucket the header's rolling hash corresponds to (AZIP-22 Fast Inbox lookup aid). */
+  bucketHint: bigint;
 };
 
 export const Actions = [
@@ -1404,6 +1406,7 @@ export class SequencerPublisher {
     checkpoint: Checkpoint,
     attestationsAndSigners: CommitteeAttestationsAndSigners,
     attestationsAndSignersSignature: Signature,
+    bucketHint: bigint,
     opts: EnqueueProposeCheckpointOpts = {},
   ): Promise<void> {
     const checkpointHeader = checkpoint.header;
@@ -1418,6 +1421,7 @@ export class SequencerPublisher {
       attestationsAndSigners,
       attestationsAndSignersSignature,
       feeAssetPriceModifier: checkpoint.feeAssetPriceModifier,
+      bucketHint,
     };
 
     this.log.verbose(`Enqueuing checkpoint propose transaction`, {
@@ -1588,6 +1592,7 @@ export class SequencerPublisher {
         oracleInput: {
           feeAssetPriceModifier: encodedData.feeAssetPriceModifier,
         },
+        bucketHint: encodedData.bucketHint,
       },
       encodedData.attestationsAndSigners.getPackedAttestations(),
       signers,

yarn-project/sequencer-client/src/sequencer/automine/automine_sequencer.ts

@@ -1,7 +1,9 @@
 import type { Archiver } from '@aztec/archiver';
+import { INBOX_LAG_SECONDS, MAX_L1_TO_L2_MSGS_PER_BLOCK, MAX_L1_TO_L2_MSGS_PER_CHECKPOINT } from '@aztec/constants';
 import type { L1TxUtils } from '@aztec/ethereum/l1-tx-utils';
 import { type EthCheatCodes, RollupCheatCodes } from '@aztec/ethereum/test';
 import { BlockNumber, CheckpointNumber, EpochNumber, SlotNumber } from '@aztec/foundation/branded-types';
+import { Fr } from '@aztec/foundation/curves/bn254';
 import type { EthAddress } from '@aztec/foundation/eth-address';
 import { Signature } from '@aztec/foundation/eth-signature';
 import { type Logger, createLogger } from '@aztec/foundation/log';
@@ -22,8 +24,9 @@ import {
   getTimestampForSlot,
 } from '@aztec/stdlib/epoch-helpers';
 import { InsufficientValidTxsError, type WorldStateSynchronizer } from '@aztec/stdlib/interfaces/server';
-import type { L1ToL2MessageSource } from '@aztec/stdlib/messaging';
+import { type L1ToL2MessageSource, getInboxCutoffTimestamp } from '@aztec/stdlib/messaging';
 import type { CoordinationSignatureContext } from '@aztec/stdlib/p2p';
+import { MerkleTreeId } from '@aztec/stdlib/trees';
 import type { FailedTx, Tx } from '@aztec/stdlib/tx';
 import type {
   BuildBlockInCheckpointResult,
@@ -35,6 +38,7 @@ import type { GlobalVariableBuilder } from '../../global_variable_builder/global
 import type { SequencerPublisherFactory } from '../../publisher/sequencer-publisher-factory.js';
 import type { SequencerPublisher } from '../../publisher/sequencer-publisher.js';
 import type { SequencerConfig } from '../config.js';
+import { selectInboxBucketForBlock } from '../inbox_bucket_selector.js';
 
 /**
  * L1 rollup constants needed by the AutomineSequencer. Same as SequencerRollupConstants
@@ -447,8 +451,6 @@ export class AutomineSequencer {
       SlotNumber(targetSlot),
     );
 
-    const l1ToL2Messages = await this.deps.l1ToL2MessageSource.getL1ToL2Messages(checkpointNumber);
-
     const previousCheckpointOutHashes = await getPreviousCheckpointOutHashes({
       blockSource: this.deps.l2BlockSource,
       epoch: targetEpoch,
@@ -468,15 +470,40 @@ export class AutomineSequencer {
 
     await using fork = await this.deps.worldState.fork(syncedToBlockNumber, { closeDelayMs: 0 });
 
+    // Streaming Inbox (AZIP-22 Fast Inbox): automine builds a single-block checkpoint, so its one block is the
+    // checkpoint's final block; select its bundle from the newest lag-eligible bucket with the last-block censorship
+    // floor. The parent total is the fork's L1-to-L2 leaf count (compact indexing), which resolves the parent bucket.
+    const parentInfo = await fork.getTreeInfo(MerkleTreeId.L1_TO_L2_MESSAGE_TREE);
+    const parentTotalMsgCount = parentInfo.size;
+    const parentBucket = await this.deps.l1ToL2MessageSource.getInboxBucketByTotalMsgCount(parentTotalMsgCount);
+    if (parentBucket === undefined) {
+      this.log.warn(`Automine streaming inbox: parent bucket for total ${parentTotalMsgCount} not synced; skipping`);
+      return undefined;
+    }
+    const selection = await selectInboxBucketForBlock({
+      messageSource: this.deps.l1ToL2MessageSource,
+      now: BigInt(Math.floor(this.deps.dateProvider.now() / 1000)),
+      lagSeconds: BigInt(INBOX_LAG_SECONDS),
+      parent: { seq: parentBucket.seq, totalMsgCount: parentBucket.totalMsgCount },
+      checkpointStartTotalMsgCount: parentTotalMsgCount,
+      perBlockCap: MAX_L1_TO_L2_MSGS_PER_BLOCK,
+      perCheckpointCap: MAX_L1_TO_L2_MSGS_PER_CHECKPOINT,
+      isLastBlock: true,
+      cutoffTimestamp: getInboxCutoffTimestamp(SlotNumber(targetSlot), this.deps.l1Constants, INBOX_LAG_SECONDS),
+    });
+    const streamingBundle = selection.consume ? selection.bundle : [];
+    const bucketHint = selection.consume ? selection.bucket.seq : parentBucket.seq;
+
     const checkpointBuilder = await this.deps.checkpointsBuilder.startCheckpoint(
       checkpointNumber,
       checkpointGlobals,
       feeAssetPriceModifier,
-      l1ToL2Messages,
+      [],
       previousCheckpointOutHashes,
       previousInboxRollingHash,
       fork,
       this.log.getBindings(),
+      true,
     );
 
     // Block building only executes txs, and automine publishes straight to L1, so the proofs are never needed.
@@ -489,6 +516,7 @@ export class AutomineSequencer {
       checkpointGlobals.timestamp,
       allowEmpty,
       checkpointNumber,
+      streamingBundle,
     );
     if (!buildResult) {
       return undefined;
@@ -516,7 +544,12 @@ export class AutomineSequencer {
       feeAssetPriceModifier,
     });
 
-    await this.publisher.enqueueProposeCheckpoint(checkpoint, emptyAttestations, emptyAttestationsSignature);
+    await this.publisher.enqueueProposeCheckpoint(
+      checkpoint,
+      emptyAttestations,
+      emptyAttestationsSignature,
+      bucketHint,
+    );
     // Automine publishes synchronously in the current slot via `sendRequests`. It must NOT use the
     // production `sendRequestsAt` (or `canProposeAt`), which always apply the one-slot pipelining
     // offset — automine is the deliberate non-pipelined exception and builds/publishes in place.
@@ -756,16 +789,19 @@ export class AutomineSequencer {
     timestamp: bigint,
     allowEmpty: boolean,
     checkpointNumber: CheckpointNumber,
+    l1ToL2Messages: Fr[],
   ): Promise<BuildBlockInCheckpointResult | undefined> {
     let buildResult: BuildBlockInCheckpointResult;
     try {
       buildResult = await checkpointBuilder.buildBlock(pendingTxs, nextBlockNumber, timestamp, {
         maxTransactions: this.deps.config.maxTxsPerBlock,
-        // Allow empty for explicit-empty builds; require at least 1 valid tx otherwise.
-        minValidTxs: allowEmpty ? 0 : 1,
+        // Allow empty for explicit-empty builds; a message-only block (non-empty streaming bundle) also builds
+        // with zero txs.
+        minValidTxs: allowEmpty || l1ToL2Messages.length > 0 ? 0 : 1,
         isBuildingProposal: true,
         maxBlocksPerCheckpoint: 1,
         perBlockAllocationMultiplier: 1,
+        l1ToL2Messages,
       });
     } catch (err) {
       // Mirrors production's checkpoint_proposal_job: if every pending tx failed execution and

yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts

@@ -1389,10 +1389,9 @@ describe('CheckpointProposalJob', () => {
     });
   });
 
-  describe('streaming inbox (flag on)', () => {
+  describe('streaming inbox', () => {
     beforeEach(() => {
       job.setTimetable(makeProposerTimetable({ l1Constants, blockDurationMs: 3000 }));
-      job.updateConfig({ streamingInbox: true });
     });
 
     it('streams per-block bucket bundles, advances the reference, and skips the bulk message fetch', async () => {