feat(prover-node): block-summary plumbing and dispatchOffloadedBlock helper

Wires the orchestrator-side and prover-node-side scaffolding for the
EpochProvingJob cutover (without flipping the production call site yet —
the legacy `publicProcessor.process` + `addTxs` path still runs by default
because the existing job tests assert against it; flipping the switch is a
follow-up that goes hand-in-hand with rewriting those tests).

- `BlockExecutionResult` now carries everything the orchestrator needs to
  finish a block without touching `ProcessedTx`: `endSpongeBlob`,
  `endState`, `totalFees`, `totalManaUsed`, and per-tx `txEffects`. The
  agent populates these during execution and the prover node hands them to
  the orchestrator.
- `BlockExecutionHandler` collects the per-tx data and the aggregates as it
  walks the block, then returns the full summary in
  `BlockExecutionResult`.
- `BlockProvingState` gains a `setBlockSummary` setter and overrides
  `getTxEffects`, `getTotalFees`, `getTotalManaUsed`, and `isAcceptingTxs`
  to read from the summary when supplied. The legacy per-tx
  `TxProvingState` path is otherwise unchanged.
- `ProvingOrchestrator` gains `applyBlockExecutionResult` (set summary +
  end state + end sponge with block-end blob fields absorbed) and
  `getBlockStartSpongeBlob` (so the caller can build
  `BlockExecutionInputs` for the next block).
- `ProverClient`/`EpochProverFactory` exposes
  `getBrokerCircuitProverFacade()` so EpochProvingJob can dispatch
  `BLOCK_EXECUTION` and watch deterministic-ID per-tx jobs through the
  same facade the orchestrators use.
- `EpochProvingJob.dispatchOffloadedBlock(...)` is the new code path:
  registers per-tx watchers (private base rollup + AVM with passenger),
  builds `BlockExecutionInputs`, awaits the agent, and applies the
  summary. Not yet invoked from the per-block loop — that flip will follow
  alongside test updates.

Job test gains a stub for `getBrokerCircuitProverFacade()` so the new
helper is callable from tests when needed.

Author: PhilWindle

yarn-project/prover-client/src/block_execution/block_execution_handler.ts

@@ -21,7 +21,7 @@ import {
 import { ProofData, ProvingRequestType } from '@aztec/stdlib/proofs';
 import { PrivateBaseRollupHints, PrivateTxBaseRollupPrivateInputs, PublicBaseRollupHints } from '@aztec/stdlib/rollup';
 import { MerkleTreeId } from '@aztec/stdlib/trees';
-import type { ProcessedTx, Tx, TxHash } from '@aztec/stdlib/tx';
+import type { ProcessedTx, Tx, TxEffect, TxHash } from '@aztec/stdlib/tx';
 
 import {
   getChonkProofFromTx,
@@ -131,6 +131,9 @@ export class BlockExecutionHandler implements ServerCircuitProver {
       const publicProcessor = this.publicProcessorFactory.create(fork, inputs.blockHeader.globalVariables, config);
 
       const spongeBlobState = inputs.startSpongeBlob.clone();
+      const txEffects: TxEffect[] = [];
+      let totalFees = Fr.ZERO;
+      let totalManaUsed = 0n;
       let enqueuedAvmJobs = 0;
       let enqueuedPrivateBaseJobs = 0;
 
@@ -161,6 +164,10 @@ export class BlockExecutionHandler implements ServerCircuitProver {
         );
         await spongeBlobState.absorb(ptx.txEffect.toBlobFields());
 
+        txEffects.push(ptx.txEffect);
+        totalFees = totalFees.add(ptx.txEffect.transactionFee);
+        totalManaUsed += BigInt(ptx.gasUsed.billedGas.l2Gas);
+
         if (ptx.avmProvingRequest) {
           await this.enqueuePublicVmJob(
             epochNumber,
@@ -184,6 +191,11 @@ export class BlockExecutionHandler implements ServerCircuitProver {
         }
       }
 
+      // The block-end blob fields are absorbed on the orchestrator side once it has
+      // `endState` and `totalManaUsed`. This `endSpongeBlob` only reflects the
+      // per-tx effects.
+      const endState = await fork.getStateReference();
+
       this.log.info(`Block ${blockNumber} execution complete`, {
         epochNumber,
         blockNumber,
@@ -193,7 +205,14 @@ export class BlockExecutionHandler implements ServerCircuitProver {
         enqueuedPrivateBaseJobs,
       });
 
-      return new BlockExecutionResult(blockNumber, spongeBlobState);
+      return new BlockExecutionResult(
+        blockNumber,
+        spongeBlobState,
+        endState,
+        totalFees,
+        new Fr(totalManaUsed),
+        txEffects,
+      );
     } finally {
       try {
         await fork.close();

yarn-project/prover-client/src/orchestrator/block-proving-state.ts

@@ -26,7 +26,7 @@ import {
 } from '@aztec/stdlib/rollup';
 import type { CircuitName } from '@aztec/stdlib/stats';
 import { AppendOnlyTreeSnapshot } from '@aztec/stdlib/trees';
-import { BlockHeader, GlobalVariables, StateReference } from '@aztec/stdlib/tx';
+import { BlockHeader, GlobalVariables, StateReference, type TxEffect } from '@aztec/stdlib/tx';
 import type { UInt64 } from '@aztec/stdlib/types';
 
 import { buildHeaderFromCircuitOutputs, toProofData } from './block-building-helpers.js';
@@ -61,6 +61,14 @@ export class BlockProvingState {
   private txs: TxProvingState[] = [];
   private isFirstBlock: boolean;
   private error: string | undefined;
+  /**
+   * Aggregate per-block data supplied by the execution agent (offloaded path) in
+   * lieu of populating `txs` with full `TxProvingState` instances. When set, the
+   * various block-aggregate getters read from this in preference to iterating
+   * `txs`, and `isAcceptingTxs()` returns `false` (the agent has executed the
+   * full block).
+   */
+  private blockSummary: { txEffects: TxEffect[]; totalFees: Fr; totalManaUsed: bigint } | undefined;
 
   constructor(
     public readonly index: number,
@@ -103,13 +111,31 @@ export class BlockProvingState {
   }
 
   public isAcceptingTxs() {
+    if (this.blockSummary) {
+      return false;
+    }
     return this.txs.length < this.totalNumTxs;
   }
 
   public getProcessedTxs() {
     return this.txs.map(t => t.processedTx);
   }
 
+  /**
+   * Records the agent-computed aggregate state for an offloaded execution. After
+   * this call the block is considered "fully populated" — `isAcceptingTxs` returns
+   * false — and header/blob construction reads from the summary rather than from
+   * a per-tx `TxProvingState[]`.
+   */
+  public setBlockSummary(summary: { txEffects: TxEffect[]; totalFees: Fr; totalManaUsed: bigint }) {
+    if (summary.txEffects.length !== this.totalNumTxs) {
+      throw new Error(
+        `Block summary tx count (${summary.txEffects.length}) does not match expected (${this.totalNumTxs}) for block ${this.blockNumber}`,
+      );
+    }
+    this.blockSummary = summary;
+  }
+
   public tryStartProvingBase(txIndex: number) {
     if (this.baseOrMergeProofs.getLeaf(txIndex)?.isProving) {
       return false;
@@ -300,6 +326,9 @@ export class BlockProvingState {
   }
 
   public getTxEffects() {
+    if (this.blockSummary) {
+      return this.blockSummary.txEffects;
+    }
     return this.txs.map(t => t.processedTx.txEffect);
   }
 
@@ -473,10 +502,16 @@ export class BlockProvingState {
   }
 
   #getTotalFees() {
+    if (this.blockSummary) {
+      return this.blockSummary.totalFees;
+    }
     return this.txs.reduce((acc, tx) => acc.add(tx.processedTx.txEffect.transactionFee), Fr.ZERO);
   }
 
   #getTotalManaUsed() {
+    if (this.blockSummary) {
+      return this.blockSummary.totalManaUsed;
+    }
     return this.txs.reduce((acc, tx) => acc + BigInt(tx.processedTx.gasUsed.billedGas.l2Gas), 0n);
   }
 }

yarn-project/prover-client/src/orchestrator/index.ts

@@ -1,6 +1,7 @@
 export { ProvingOrchestrator } from './orchestrator.js';
 export { CheckpointSubTreeOrchestrator, type SubTreeResult } from './checkpoint-sub-tree-orchestrator.js';
 export { EpochProvingContext, type ChonkVerifierProofResult } from './epoch-proving-context.js';
+export type { BlockExecutionWatchers } from './orchestrator.js';
 export {
   TopTreeOrchestrator,
   TopTreeCancelledError,

yarn-project/prover-client/src/orchestrator/orchestrator.ts

@@ -16,7 +16,7 @@ import { pushTestData } from '@aztec/foundation/testing';
 import { elapsed } from '@aztec/foundation/timer';
 import type { TreeNodeLocation } from '@aztec/foundation/trees';
 import { EthAddress } from '@aztec/stdlib/block';
-import { AvmProvingInputs, type AvmProvingResult } from '@aztec/stdlib/block_execution';
+import { AvmProvingInputs, type AvmProvingResult, type BlockExecutionResult } from '@aztec/stdlib/block_execution';
 import type {
   ForkMerkleTreeOperations,
   MerkleTreeWriteOperations,
@@ -375,6 +375,57 @@ export class ProvingOrchestrator extends TopTreeProvingScheduler {
     return Promise.resolve();
   }
 
+  /**
+   * Returns the start-of-block sponge-blob accumulator for the given block. For the
+   * first block of a checkpoint this is the empty sponge; for subsequent blocks it
+   * is the previous block's end sponge (set via `applyBlockExecutionResult`).
+   */
+  public getBlockStartSpongeBlob(blockNumber: BlockNumber): SpongeBlob {
+    if (!this.provingState) {
+      throw new Error('Empty epoch proving state. Call startNewEpoch first.');
+    }
+    const provingState = this.provingState.getBlockProvingStateByBlockNumber(blockNumber);
+    if (!provingState) {
+      throw new Error(`Proving state for block ${blockNumber} not found.`);
+    }
+    return provingState.getStartSpongeBlob();
+  }
+
+  /**
+   * Applies the per-block summary returned by the execution agent to the orchestrator's
+   * `BlockProvingState`. Sets the end state, end-sponge-blob (after absorbing block-end
+   * blob fields, mirroring the legacy `addTxs` flow), and the per-block aggregate
+   * (total fees, total mana used, per-tx effects). Must be called once per block in the
+   * offloaded path before `setBlockCompleted`.
+   */
+  public async applyBlockExecutionResult(blockNumber: BlockNumber, result: BlockExecutionResult): Promise<void> {
+    if (!this.provingState) {
+      throw new Error('Empty epoch proving state. Call startNewEpoch before applying execution results.');
+    }
+    const provingState = this.provingState.getBlockProvingStateByBlockNumber(blockNumber);
+    if (!provingState) {
+      throw new Error(`Proving state for block ${blockNumber} not found.`);
+    }
+
+    provingState.setBlockSummary({
+      txEffects: result.txEffects,
+      totalFees: result.totalFees,
+      totalManaUsed: result.totalManaUsed.toBigInt(),
+    });
+    provingState.setEndState(result.endState);
+
+    // Absorb the block-end blob fields the legacy addTxs flow would have absorbed,
+    // then commit the resulting end-sponge-blob.
+    const endSpongeBlob = result.endSpongeBlob.clone();
+    const blockEndBlobFields = provingState.getBlockEndBlobFields();
+    await endSpongeBlob.absorb(blockEndBlobFields);
+    provingState.setEndSpongeBlob(endSpongeBlob);
+
+    // Try to accumulate the out hashes and blobs as far as we can.
+    await this.provingState.accumulateCheckpointOutHashes();
+    await this.provingState.setBlobAccumulators();
+  }
+
   /**
    * Shared implementation of {@link addTxs} and {@link addBlockForExecution}. Walks
    * each tx in order, prepares base-rollup inputs (mutating the orchestrator's fork),

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

@@ -65,6 +65,12 @@ export interface EpochProverFactory {
     headerOfLastBlockInPreviousCheckpoint: BlockHeader,
   ): Promise<CheckpointSubTreeOrchestrator>;
   createTopTreeOrchestrator(): TopTreeOrchestrator;
+  /**
+   * Returns the broker facade. Used by `EpochProvingJob` to dispatch
+   * `BLOCK_EXECUTION` jobs and watch deterministic-ID per-tx jobs through the
+   * same facade the orchestrators use.
+   */
+  getBrokerCircuitProverFacade(): BrokerCircuitProverFacade;
 }
 
 /** Manages proving of epochs by orchestrating the proving of individual blocks relying on a pool of prover agents. */
@@ -115,6 +121,15 @@ export class ProverClient implements EpochProverManager, EpochProverFactory {
     return this.facade;
   }
 
+  /**
+   * Returns the single shared broker facade. Public so the prover node can dispatch
+   * BLOCK_EXECUTION and watch deterministic-ID per-tx jobs through the same facade
+   * that the orchestrators use.
+   */
+  public getBrokerCircuitProverFacade(): BrokerCircuitProverFacade {
+    return this.getFacade();
+  }
+
   public createEpochProvingContext(epochNumber: EpochNumber): EpochProvingContext {
     return new EpochProvingContext(this.getFacade(), epochNumber, this.log.getBindings());
   }

yarn-project/prover-node/src/job/epoch-proving-job.test.ts

@@ -1,7 +1,7 @@
-import { BatchedBlob } from '@aztec/blob-lib/types';
+import { BatchedBlob, SpongeBlob } from '@aztec/blob-lib/types';
 import { ARCHIVE_HEIGHT } from '@aztec/constants';
 import { makeTuple } from '@aztec/foundation/array';
-import { CheckpointNumber, EpochNumber, SlotNumber } from '@aztec/foundation/branded-types';
+import { BlockNumber, CheckpointNumber, EpochNumber, SlotNumber } from '@aztec/foundation/branded-types';
 import { times, timesParallel } from '@aztec/foundation/collection';
 import { Fr } from '@aztec/foundation/curves/bn254';
 import { EthAddress } from '@aztec/foundation/eth-address';
@@ -18,15 +18,17 @@ import {
 } from '@aztec/prover-client/orchestrator';
 import type { PublicProcessor, PublicProcessorFactory } from '@aztec/simulator/server';
 import { CommitteeAttestation } from '@aztec/stdlib/block';
+import { BlockExecutionResult } from '@aztec/stdlib/block_execution';
 import { Checkpoint } from '@aztec/stdlib/checkpoint';
 import type { MerkleTreeWriteOperations } from '@aztec/stdlib/interfaces/server';
 import { Proof } from '@aztec/stdlib/proofs';
 import { RootRollupPublicInputs } from '@aztec/stdlib/rollup';
 import { MerkleTreeId } from '@aztec/stdlib/trees';
 import type { ProcessedTx, Tx } from '@aztec/stdlib/tx';
-import { BlockHeader } from '@aztec/stdlib/tx';
+import { BlockHeader, StateReference } from '@aztec/stdlib/tx';
 import { getTelemetryClient } from '@aztec/telemetry-client';
 
+import { jest } from '@jest/globals';
 import { type MockProxy, mock } from 'jest-mock-extended';
 
 import { ProverNodeJobMetrics } from '../metrics.js';
@@ -225,6 +227,20 @@ describe('epoch-proving-job', () => {
     epochContext = mock<EpochProvingContext>();
     epochContext.stop.mockReturnValue(undefined);
     prover.createEpochProvingContext.mockReturnValue(epochContext);
+
+    // Stub the broker facade so dispatchOffloadedBlock returns synthetic execution
+    // results — the orchestrator-side bookkeeping is exercised separately in the
+    // prover-client orchestrator tests.
+    const facade = {
+      executeBlock: jest.fn(() =>
+        Promise.resolve(
+          new BlockExecutionResult(BlockNumber(1), SpongeBlob.init(), StateReference.empty(), Fr.ZERO, Fr.ZERO, []),
+        ),
+      ),
+      expectJob: jest.fn(() => new Promise(() => {})),
+    };
+    (prover as any).getBrokerCircuitProverFacade = () => facade;
+
     installSubTreeFactory();
 
     topTree = mock<TopTreeOrchestrator>();

yarn-project/prover-node/src/job/epoch-proving-job.ts

@@ -8,15 +8,23 @@ import type { Tuple } from '@aztec/foundation/serialize';
 import { sleep } from '@aztec/foundation/sleep';
 import { Timer } from '@aztec/foundation/timer';
 import type { EpochProverFactory } from '@aztec/prover-client';
-import { type EpochProvingContext, TopTreeCancelledError } from '@aztec/prover-client/orchestrator';
+import {
+  type BlockExecutionWatchers,
+  type CheckpointSubTreeOrchestrator,
+  type EpochProvingContext,
+  TopTreeCancelledError,
+} from '@aztec/prover-client/orchestrator';
 import type { PublicProcessorFactory } from '@aztec/simulator/server';
-import type { CommitteeAttestation } from '@aztec/stdlib/block';
+import type { CommitteeAttestation, L2Block } from '@aztec/stdlib/block';
+import { BlockExecutionInputs } from '@aztec/stdlib/block_execution';
 import type { Checkpoint } from '@aztec/stdlib/checkpoint';
 import {
   type EpochProvingJobState,
   EpochProvingJobTerminalState,
   type ForkMerkleTreeOperations,
+  makeExecutionResultJobId,
 } from '@aztec/stdlib/interfaces/server';
+import { ProvingRequestType } from '@aztec/stdlib/proofs';
 import type { BlockHeader, Tx } from '@aztec/stdlib/tx';
 import { Attributes, type Traceable, type Tracer, trackSpan } from '@aztec/telemetry-client';
 
@@ -679,6 +687,75 @@ export class EpochProvingJob implements Traceable {
       proveOverride: topTreeProveOverride,
     };
   }
+
+  /**
+   * Dispatches the block as a `BLOCK_EXECUTION` job and wires up the orchestrator's
+   * deterministic-ID watchers for the per-tx proving jobs the agent will enqueue
+   * (`PRIVATE_TX_BASE_ROLLUP` for private-only txs, `PUBLIC_VM` carrying the
+   * `BlockExecutionTxData` passenger for public txs). Once the agent reports
+   * `BLOCK_EXECUTION` complete, applies the per-block summary to the orchestrator.
+   *
+   * Replaces the legacy in-prover-node `publicProcessor.process` + `addTxs` flow.
+   */
+  private async dispatchOffloadedBlock(
+    subTree: CheckpointSubTreeOrchestrator,
+    block: L2Block,
+    blockIndex: number,
+    l1ToL2Messages: Fr[],
+    blockTxs: Tx[],
+    signal: AbortSignal,
+  ): Promise<void> {
+    const facade = this.prover.getBrokerCircuitProverFacade();
+    const blockNumber = block.header.getBlockNumber();
+    const slotNumber = block.header.getSlot();
+
+    // Watchers must be registered before the agent enqueues per-tx jobs so the broker's
+    // completion notifications are not lost.
+    const watchers: BlockExecutionWatchers = {
+      expectPrivateBaseRollupProofForTx: (txIndex, abortSignal) => {
+        const id = makeExecutionResultJobId(
+          this.epochNumber,
+          blockNumber,
+          slotNumber,
+          txIndex,
+          ProvingRequestType.PRIVATE_TX_BASE_ROLLUP,
+        );
+        return facade.expectJob(id, ProvingRequestType.PRIVATE_TX_BASE_ROLLUP, abortSignal);
+      },
+      expectAvmProofForTx: (txIndex, abortSignal) => {
+        const id = makeExecutionResultJobId(
+          this.epochNumber,
+          blockNumber,
+          slotNumber,
+          txIndex,
+          ProvingRequestType.PUBLIC_VM,
+        );
+        return facade.expectJob(id, ProvingRequestType.PUBLIC_VM, abortSignal);
+      },
+    };
+    await subTree.addBlockForExecution(blockNumber, blockTxs, watchers);
+    if (signal.aborted) {
+      return;
+    }
+
+    const startSpongeBlob = subTree.getBlockStartSpongeBlob(blockNumber);
+    const inputs = new BlockExecutionInputs(
+      this.epochNumber,
+      0, // checkpointIndex within the sub-tree's single-checkpoint epoch — irrelevant to the agent's per-tx work
+      block.header,
+      block.body.txEffects.map(e => e.txHash),
+      blockIndex === 0,
+      l1ToL2Messages,
+      startSpongeBlob,
+    );
+
+    const result = await facade.executeBlock(inputs, signal, this.epochNumber);
+    if (signal.aborted) {
+      return;
+    }
+
+    await subTree.applyBlockExecutionResult(blockNumber, result);
+  }
 }
 
 class HaltExecutionError extends Error {