refactor(prover-client): split orchestrator into sub-tree + top-tree pair (#22996)

Introduces a sub-tree + top-tree orchestrator pair that decomposes the
existing single-class proving orchestrator along the natural
state-coupling boundary — per-checkpoint block-level work vs.
epoch-level top-tree work — while leaving every existing API on the
legacy `EpochProver` / `ProvingOrchestrator` / `EpochProvingState` path
untouched. The prover-node and e2e tests build unchanged; this PR is
purely additive in surface area, with structural refactors on
`ProvingOrchestrator` to share scheduling and top-tree drivers with the
new `TopTreeOrchestrator`.

## What's new

- **`CheckpointSubTreeOrchestrator`**
(`checkpoint-sub-tree-orchestrator.ts`): extends `ProvingOrchestrator`,
single-checkpoint by construction. Drives chonk-verifier / base / merge
/ block-root / block-merge for one checkpoint and resolves a
`SubTreeResult` instead of escalating to the checkpoint root — the
parent's `checkAndEnqueueCheckpointRootRollup` is overridden to
short-circuit. The constructor calls `super.startNewEpoch(epoch, 1,
empty challenges)` to set up a single-checkpoint mini-epoch; the count
and challenges are never read because the override prevents the parent's
finalize / root path from running.

- **`TopTreeOrchestrator`** + **`TopTreeProvingState`**: self-contained
driver from checkpoint-root through epoch-root rollup. Takes
per-checkpoint block-proof promises and pipelines its hint chain against
them. Cancellation surfaces as `TopTreeCancelledError` so callers can
distinguish reorg-driven cancel from a genuine proving failure.

- **`EpochProvingContext`** (`epoch-proving-context.ts`): per-epoch
shared cache for chonk-verifier proofs. Survives sub-tree cancellation
so a tx that gets reorged out and re-appears in a replacement checkpoint
reuses the cached proof.

- **`ProvingScheduler`** (`proving-scheduler.ts`): abstract base owning
the `SerialQueue` deferred-job lifecycle, the `pendingProvingJobs`
controller list, and a unified `deferredProving<S, T>(state, request,
callback, isCancelled?)` submit envelope. The minimal `ProvingStateLike`
contract is just `verifyState()` + `reject(reason)`.

- **`TopTreeProvingScheduler`** (`top-tree-proving-scheduler.ts`):
extends `ProvingScheduler` and holds the checkpoint-merge, padding, and
root-rollup drivers (plus tree-walking helpers) shared by both
orchestrators. Wraps circuit calls via a `wrapCircuitCall` hook
(orchestrator overrides for spans; top-tree leaves identity) and
resolves via an `onRootRollupComplete` hook to bridge the two states'
differing `resolve` signatures. The per-checkpoint root driver stays
subclass-specific because input-building flows differ.

- **`EpochProverFactory` interface on `ProverClient`**: new factory
methods `createEpochProvingContext(epochNumber)`,
`createCheckpointSubTreeOrchestrator(...)`, and
`createTopTreeOrchestrator()`. A single shared
`BrokerCircuitProverFacade` is owned by `ProverClient` and shared across
every orchestrator.

## What changes in existing code

- `ProvingOrchestrator` extends `TopTreeProvingScheduler`; the inline
broker-job submit envelope, queue lifecycle, and the top-tree-section
drivers are inherited. `cancel()` delegates the queue-recreate +
abort-jobs logic to `resetSchedulerState(this.cancelJobsOnStop)`. Three
internal methods (`getOrEnqueueChonkVerifier`,
`checkAndEnqueueBaseRollup`, `checkAndEnqueueCheckpointRootRollup`)
become `protected` so the sub-tree can override them; `provingState` and
`provingPromise` likewise become `protected` so the sub-tree can hook
the parent's failure stream onto `subTreeResult`. No public API change
on `ProvingOrchestrator`.
- `CheckpointProvingState`: gains two read-only accessors used by the
sub-tree's checkpoint-root override — `getSubTreeOutputProofs()` and
`getLastArchiveSiblingPath()`. No state changes.
- `ProverClient` keeps `createEpochProver()` exactly as before (each
call spawns its own `BrokerCircuitProverFacade`); the new factory
methods share a `getFacade()` set up in `start()` and torn down in
`stop()`.

`EpochProver`, `EpochProverManager`, `ServerEpochProver`,
`EpochProvingState`, the integration tests in `orchestrator_*.test.ts`,
`bb_prover_full_rollup.test.ts`, and `stdlib/interfaces/*` are all
unchanged from `merge-train/spartan` — the prover-node and e2e tests
continue to build against the existing `EpochProver` API. Migrating the
prover-node onto the new factories (and the deferred-finalize flow that
goes with optimistic proving) is the follow-up PR.

## Test plan

- 261 prover-client tests pass (full `yarn workspace
@aztec/prover-client test`).
- `yarn build` clean against current merge-train/spartan (modulo the
pre-existing `@aztec/sqlite3mc-wasm` issue inherited from baseline).

Author: PhilWindle

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

@@ -284,6 +284,7 @@ export class TestContext {
 
     return {
       constants,
+      checkpoint,
       header: checkpoint.header,
       blocks,
       l1ToL2Messages,

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

@@ -346,4 +346,17 @@ export class CheckpointProvingState {
       ? [this.blockProofs.getNode(rootLocation)?.provingOutput] // If there's only 1 block, its proof will be stored at the root.
       : this.blockProofs.getChildren(rootLocation).map(c => c?.provingOutput);
   }
+
+  /**
+   * Returns the block-level proof outputs that feed into the checkpoint root rollup.
+   * Used by `CheckpointSubTreeOrchestrator` to surface its sub-tree result.
+   */
+  public getSubTreeOutputProofs() {
+    return this.#getChildProofsForRoot();
+  }
+
+  /** Sibling path of the archive tree captured before any block in this checkpoint landed. */
+  public getLastArchiveSiblingPath() {
+    return this.lastArchiveSiblingPath;
+  }
 }

yarn-project/prover-client/src/orchestrator/checkpoint-sub-tree-orchestrator.test.ts

@@ -0,0 +1,149 @@
+import { FinalBlobBatchingChallenges } from '@aztec/blob-lib';
+import { EpochNumber } from '@aztec/foundation/branded-types';
+import { EthAddress } from '@aztec/foundation/eth-address';
+import { createLogger } from '@aztec/foundation/log';
+
+import { TestContext } from '../mocks/test_context.js';
+import { CheckpointSubTreeOrchestrator } from './checkpoint-sub-tree-orchestrator.js';
+import { EpochProvingContext } from './epoch-proving-context.js';
+
+const logger = createLogger('prover-client:test:checkpoint-sub-tree-orchestrator');
+
+describe('prover/orchestrator/checkpoint-sub-tree', () => {
+  let context: TestContext;
+  let epochContext: EpochProvingContext;
+
+  beforeEach(async () => {
+    context = await TestContext.new(logger);
+    epochContext = new EpochProvingContext(context.prover, EpochNumber(1));
+  });
+
+  afterEach(async () => {
+    epochContext.stop();
+    await context.cleanup();
+  });
+
+  it('resolves the sub-tree result with block-level proofs for a single-block checkpoint', async () => {
+    const numBlocks = 1;
+    const numTxsPerBlock = 1;
+    const { constants, blocks, l1ToL2Messages, previousBlockHeader } = await context.makeCheckpoint(numBlocks, {
+      numTxsPerBlock,
+    });
+
+    const subTree = await CheckpointSubTreeOrchestrator.start(
+      context.worldState,
+      context.prover,
+      EthAddress.ZERO,
+      epochContext,
+      false,
+      10,
+      constants,
+      l1ToL2Messages,
+      numBlocks,
+      previousBlockHeader,
+    );
+    try {
+      const resultPromise = subTree.getSubTreeResult();
+
+      for (const block of blocks) {
+        const { blockNumber, timestamp } = block.header.globalVariables;
+        await subTree.startNewBlock(blockNumber, timestamp, block.txs.length);
+        if (block.txs.length > 0) {
+          await subTree.addTxs(block.txs);
+        }
+        await subTree.setBlockCompleted(blockNumber, block.header);
+      }
+
+      const result = await resultPromise;
+      expect(result.blockProofOutputs).toHaveLength(1);
+      expect(result.blockProofOutputs[0].proof).toBeDefined();
+      expect(result.previousArchiveSiblingPath).toBeDefined();
+    } finally {
+      await subTree.stop();
+    }
+  });
+
+  it('resolves with two block proofs for a two-block checkpoint', async () => {
+    const numBlocks = 2;
+    const numTxsPerBlock = 1;
+    const { constants, blocks, l1ToL2Messages, previousBlockHeader } = await context.makeCheckpoint(numBlocks, {
+      numTxsPerBlock,
+    });
+
+    const subTree = await CheckpointSubTreeOrchestrator.start(
+      context.worldState,
+      context.prover,
+      EthAddress.ZERO,
+      epochContext,
+      false,
+      10,
+      constants,
+      l1ToL2Messages,
+      numBlocks,
+      previousBlockHeader,
+    );
+    try {
+      const resultPromise = subTree.getSubTreeResult();
+
+      for (const block of blocks) {
+        const { blockNumber, timestamp } = block.header.globalVariables;
+        await subTree.startNewBlock(blockNumber, timestamp, block.txs.length);
+        if (block.txs.length > 0) {
+          await subTree.addTxs(block.txs);
+        }
+        await subTree.setBlockCompleted(blockNumber, block.header);
+      }
+
+      const result = await resultPromise;
+      expect(result.blockProofOutputs).toHaveLength(2);
+    } finally {
+      await subTree.stop();
+    }
+  });
+
+  it('throws when startNewEpoch is called explicitly', async () => {
+    const { constants, l1ToL2Messages, previousBlockHeader } = await context.makeCheckpoint(1, { numTxsPerBlock: 0 });
+    const subTree = await CheckpointSubTreeOrchestrator.start(
+      context.worldState,
+      context.prover,
+      EthAddress.ZERO,
+      epochContext,
+      false,
+      10,
+      constants,
+      l1ToL2Messages,
+      1,
+      previousBlockHeader,
+    );
+    try {
+      expect(() => subTree.startNewEpoch(EpochNumber(2), 1, FinalBlobBatchingChallenges.empty())).toThrow(
+        /starts its epoch in the constructor/,
+      );
+    } finally {
+      await subTree.stop();
+    }
+  });
+
+  it('throws when startNewCheckpoint is called explicitly', async () => {
+    const { constants, l1ToL2Messages, previousBlockHeader } = await context.makeCheckpoint(1, { numTxsPerBlock: 0 });
+    const subTree = await CheckpointSubTreeOrchestrator.start(
+      context.worldState,
+      context.prover,
+      EthAddress.ZERO,
+      epochContext,
+      false,
+      10,
+      constants,
+      l1ToL2Messages,
+      1,
+      previousBlockHeader,
+    );
+    try {
+      await expect(subTree.startNewCheckpoint(0, constants, l1ToL2Messages, 1, previousBlockHeader)).rejects.toThrow(
+        /drives its single checkpoint in `start`/,
+      );
+    } finally {
+      await subTree.stop();
+    }
+  });
+});