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

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`.

Split out from #22990 so it can land independently.

## 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

- [x] 261 prover-client tests pass (full `yarn workspace @aztec/prover-client test`).
- [x] `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();
+    }
+  });
+});