diff --git a/yarn-project/end-to-end/src/single-node/proving/upload_failed_proof.test.ts b/yarn-project/end-to-end/src/single-node/proving/upload_failed_proof.test.ts
index 81988c55569d..26addfa1b822 100644
--- a/yarn-project/end-to-end/src/single-node/proving/upload_failed_proof.test.ts
+++ b/yarn-project/end-to-end/src/single-node/proving/upload_failed_proof.test.ts
@@ -4,7 +4,7 @@ import type { Logger } from '@aztec/aztec.js/log';
 import { tryRmDir } from '@aztec/foundation/fs';
 import { promiseWithResolvers } from '@aztec/foundation/promise';
 import { sleep } from '@aztec/foundation/sleep';
-import { downloadEpochProvingJob, rerunEpochProvingJob } from '@aztec/prover-node';
+import { downloadEpochProvingJob, rerunCheckpointProvingJob, rerunEpochProvingJob } from '@aztec/prover-node';
 import type { TestProverNode } from '@aztec/prover-node/test';
 
 import { jest } from '@jest/globals';
@@ -61,11 +61,11 @@ describe('single-node/proving/upload_failed_proof', () => {
     await tryRmDir(rerunDownloadDir, logger);
   });
 
-  // Makes the prover's top-tree prove always throw (v5 uses the session's topTreeProveOverride hook;
-  // pre-v5 it patched finalizeEpoch), intercepts tryUploadSessionFailure (pre-v5 tryUploadEpochFailure)
-  // to capture the upload URL, then waits for epoch 1 to start and for the upload to complete. Tears
-  // down the live context, downloads the proving job data, and re-runs it via rerunEpochProvingJob with
-  // fake proofs on a fresh config.
+  // Makes the prover's top-tree prove always throw. Because every checkpoint prover still succeeds, the
+  // session fails on its own account (state 'failed'), which the session manager treats as a genuine,
+  // race-free failure and uploads a post-mortem eagerly via tryUploadEpochFailure(sessionId, checkpoints).
+  // Intercepts that to capture the upload URL, then tears down the live context, downloads the proving
+  // job data, and re-runs it via rerunEpochProvingJob with fake proofs on a fresh config.
   it('uploads failed proving job state and re-runs it on a fresh instance', async () => {
     // Make initial prover node fail to prove, via the session's top-tree-prove hook.
     const proverNode = test.proverNodes[0].getProverNode() as TestProverNode;
@@ -77,19 +77,20 @@ describe('single-node/proving/upload_failed_proof', () => {
       },
     });
 
-    // And track when the epoch failure upload is complete
+    // Track when the epoch failure upload is complete. It fires eagerly when a full session fails with
+    // healthy provers (a top-tree failure here).
     const { promise: epochUploaded, resolve: onEpochUploaded } = promiseWithResolvers<string>();
-    const origTryUploadEpochFailure = proverNode.tryUploadSessionFailure.bind(proverNode);
-    proverNode.tryUploadSessionFailure = async (session: any) => {
-      const url = await origTryUploadEpochFailure(session);
+    const origTryUploadEpochFailure = proverNode.tryUploadEpochFailure.bind(proverNode);
+    proverNode.tryUploadEpochFailure = async (id: any, checkpoints: any) => {
+      const url = await origTryUploadEpochFailure(id, checkpoints);
       if (url !== undefined) {
         onEpochUploaded(url);
       }
       return url;
     };
 
-    // Warp to the start of epoch one so prover node starts proving epoch 0,
-    // and wait for the data to be uploaded to the remote file store
+    // Warp to the start of epoch one so the prover node starts proving, fails at the top tree, and
+    // uploads the failed proving-job data to the remote file store.
     await test.warpToEpochStart(1);
     const epochUploadUrl = await epochUploaded;
 
@@ -122,4 +123,62 @@ describe('single-node/proving/upload_failed_proof', () => {
 
     logger.info(`Test succeeded`);
   });
+
+  // Same shape as above, one level down: forces a single checkpoint's sub-tree to fail (every checkpoint
+  // prover fails, via the test hook), which uploads that checkpoint's proving data on its own. Intercepts
+  // tryUploadCheckpointFailure for the URL, then downloads and re-proves just that checkpoint via
+  // rerunCheckpointProvingJob (no epoch top-tree / L1 submit).
+  it('uploads failed checkpoint proving state and re-proves it on a fresh instance', async () => {
+    const proverNode = test.proverNodes[0].getProverNode() as TestProverNode;
+    proverNode.setCheckpointHooks({
+      checkpointProveOverride: async () => {
+        await sleep(1000);
+        logger.warn(`Triggering error on checkpoint sub-tree prove`);
+        throw new Error(`Fake error while proving checkpoint`);
+      },
+    });
+
+    // The checkpoint post-mortem upload fires eagerly when a CheckpointProver's block proofs reject.
+    const { promise: checkpointUploaded, resolve: onCheckpointUploaded } = promiseWithResolvers<string>();
+    const origTryUploadCheckpointFailure = proverNode.tryUploadCheckpointFailure.bind(proverNode);
+    proverNode.tryUploadCheckpointFailure = async (prover: any) => {
+      const url = await origTryUploadCheckpointFailure(prover);
+      if (url !== undefined) {
+        onCheckpointUploaded(url);
+      }
+      return url;
+    };
+
+    // Warp so the prover node starts registering checkpoints; the first one's sub-tree fails and uploads.
+    await test.warpToEpochStart(1);
+    const checkpointUploadUrl = await checkpointUploaded;
+
+    await test.teardown();
+
+    const rerunDownloadPath = join(rerunDownloadDir, 'data.bin');
+    logger.warn(`Downloading checkpoint proving job data and state`, { rerunDataDir, rerunDownloadPath });
+    await downloadEpochProvingJob(checkpointUploadUrl, logger, {
+      dataDirectory: rerunDataDir,
+      jobDataDownloadPath: rerunDownloadPath,
+    });
+
+    logger.warn(`Rerunning checkpoint proving job from ${rerunDownloadPath}`);
+    await rerunCheckpointProvingJob(
+      rerunDownloadPath,
+      logger,
+      {
+        ...config,
+        realProofs: false,
+        dataStoreMapSizeKb: 1024 * 1024,
+        dataDirectory: rerunDataDir,
+        proverAgentCount: 2,
+        proverId: EthAddress.random(),
+        ...(await getACVMConfig(logger)),
+        ...(await getBBConfig(logger)),
+      },
+      context.genesis,
+    );
+
+    logger.info(`Test succeeded`);
+  });
 });
diff --git a/yarn-project/prover-node/README.md b/yarn-project/prover-node/README.md
index abafeae777ee..d71b75517267 100644
--- a/yarn-project/prover-node/README.md
+++ b/yarn-project/prover-node/README.md
@@ -48,10 +48,12 @@ The prover-node splits responsibility between four classes:
 
 - **`ProverNode`** — owns the long-lived collections, wires the L2BlockStream, and
   translates each chain event into a single method call on the `SessionManager` or
-  `ProofPublishingService`. It also performs the per-event side effects that don't
-  belong on an `EpochSession` (registering new checkpoints with the store, sweeping
-  expired epochs out of the cache and the store, etc.) and runs the failure-upload
-  action when an `EpochSession` exits with `failed`.
+  `ProofPublishingService`. It also performs side effects that don't belong on an
+  `EpochSession`: registering new checkpoints with the store per event, and sweeping expired
+  epochs out of the cache and the store on a periodic ticker (see below). It uploads post-mortem
+  snapshots on failure at two levels: per checkpoint when a `CheckpointProver` fails (its `onFailed`
+  callback), and per session when a full session fails on its own account (`onSessionFailed` /
+  `EpochSession.hasFailed()`).
 - **`CheckpointStore`** — a registry of `CheckpointProver` instances keyed by
   `(checkpointNumber, slot, archiveRoot)`. Each `CheckpointProver` runs its own sub-tree pipeline
   (tx gather → block processing → block-rollup proofs), starting eagerly the moment a
@@ -64,8 +66,12 @@ The prover-node splits responsibility between four classes:
   translated into a `reconcile(trigger)` call, a single idempotent function that walks
   all `EpochSession`s, cancels any whose canonical content has shifted, re-creates them with
   the new content, and opens fresh full `EpochSession`s for any epoch that has become provable.
-  Reconcile runs on a `SerialQueue` (from `@aztec/foundation/queue`), so two concurrent
-  triggers can never interleave on an `await` and race on the `EpochSession` maps.
+  It never builds a session over a **failed** `CheckpointProver` (one whose block proofs rejected —
+  a sub-tree fault or a prune-induced fork fault): a session over it could only fail, so the epoch is
+  cheaply skipped until a prune/re-add installs a fresh prover in its place, at which point a rebuilt
+  session reuses every already-completed sub-proof from the content-addressed broker. Reconcile runs on
+  a `SerialQueue` (from `@aztec/foundation/queue`), so two concurrent triggers can never interleave on an
+  `await` and race on the `EpochSession` maps.
 - **`ProofPublishingService`** — central owner of L1 proof submission. `EpochSession`s hand
   their top-tree proofs to the service as `PublishCandidate`s; the service serialises
   one publish at a time against a freshly-created `ProverNodePublisher`, gates eligibility
@@ -150,9 +156,9 @@ stateDiagram-v2
   awaiting_root --> publishing_proof: epoch proof ready, submit to L1
   publishing_proof --> completed: publish succeeds
   publishing_proof --> superseded: longer same-epoch candidate wins
-  publishing_proof --> failed: L1 submission errored
-  awaiting_checkpoints --> failed: top-tree prove errored
-  awaiting_root --> failed: top-tree prove errored
+  publishing_proof --> failed: L1 submission errored (provers healthy)
+  awaiting_checkpoints --> stopped: a checkpoint prover failed
+  awaiting_root --> failed: top-tree prove errored (provers healthy)
   initialized --> timed_out: deadline
   awaiting_checkpoints --> timed_out: deadline (EpochSession or candidate)
   awaiting_root --> timed_out: deadline (EpochSession or candidate)
@@ -162,9 +168,22 @@ stateDiagram-v2
   superseded --> [*]
   cancelled --> [*]
   timed_out --> [*]
+  stopped --> [*]
   failed --> [*]
 ```
 
+A fault settles the `EpochSession` in one of two terminal states:
+
+- **`stopped`** — a `CheckpointProver` under the session failed (its block proofs rejected). This may
+  be a prune-induced fork fault, so it is *not* a verdict on the epoch and is *not* uploaded. The
+  reconciler drops the session and, guarded by `isFailed()`, does not rebuild over the failed prover;
+  a prune/re-add installs a fresh prover and the epoch is retried.
+- **`failed`** — the session's own work (top-tree prove or L1 submit) failed while *every* checkpoint
+  prover succeeded. Because healthy provers rule out a prune, this is a genuine, race-free failure
+  (`EpochSession.hasFailed()`). The reconciler retains such a full session (so the tick does not
+  re-prove a deterministically-failing epoch) and uploads a post-mortem exactly once. No prune/fault
+  classification is needed — the two states carry it.
+
 The non-terminal states track the window between `start()` and the L1 submission:
 
 - `awaiting-checkpoints` — a `TopTreeJob` is awaiting each checkpoint's sub-tree result
@@ -323,44 +342,57 @@ sequenceDiagram
   PS->>PS: drain reads proven afresh, re-checks eligibility
 ```
 
-### Per-event expiry sweep
+### Periodic expiry sweep
 
 ```mermaid
 sequenceDiagram
-  participant L2 as L2BlockStream
+  participant T as expiryTicker (RunningPromise)
   participant PN as ProverNode
+  participant L2 as L2BlockStream
   participant CC as ChonkCache
   participant CS as CheckpointStore
 
-  L2->>PN: any event
+  T->>PN: checkEpochExpiry() (every poll interval)
   PN->>L2: getSyncedL2SlotNumber()
   PN->>PN: latestEpoch = getEpochAtSlot(latestSlot)
   PN->>PN: newlyExpiredUpTo = latestEpoch - (proofSubmissionEpochs + 1)
   loop for each newly-expired epoch
-    PN->>L2: getCheckpointsData({epoch}) + getBlocks(...)
+    PN->>L2: getBlocks({epoch, onlyCheckpointed})
     PN->>CC: releaseForBlocks(blocks)
     PN->>CS: reapExpired(epoch)
   end
 ```
 
-Expiry runs at the end of every `handleBlockStreamEvent` call (not on any specific
-event type). An epoch `E` is expired once the chain reaches the start of epoch
-`E + proofSubmissionEpochs + 1` — the deadline beyond which an L1 submission for
-`E` would be rejected. A monotonic high-water mark (`lastExpiredEpoch`) makes the
-sweep cheap: it advances per event and never revisits an epoch. It is seeded at
-`start()` from the last fully-proven epoch (computed in `computeStartupState`),
-so on a restart we never re-sweep epochs that already reached L1.
+The sweep is driven solely by the `expiryTicker` (a `RunningPromise` armed in `start()`), which
+calls `checkEpochExpiry()` every poll interval whether or not block-stream events arrive. It is
+deliberately **not** run from `handleBlockStreamEvent` — expiry is a background sweep keyed off the
+archiver's synced slot, not part of event processing — and because a `RunningPromise` never overlaps
+its own runs, no two sweeps can race. An epoch `E` is expired once the chain reaches the start of epoch
+`E + proofSubmissionEpochs + 1` — the deadline beyond which an L1 submission for `E` would be rejected.
+Expiry only releases the chonk cache and reaps `E`'s provers.
+A monotonic high-water mark (`lastExpiredEpoch`) makes the sweep cheap: it only
+advances after a sweep completes and never revisits an epoch. It is seeded at `start()` from the last
+fully-proven epoch (`resolveLastFullyProvenEpoch`), so on a restart we never re-sweep epochs that already
+reached L1.
 
 ### Periodic tick
 
 `SessionManager.start()` arms a `RunningPromise` that fires
 `reconcile({ kind: 'tick' })` every `tickIntervalMs`. The tick picks up epochs that
 became complete by time alone (no fresh checkpoint event) and advances to the
-next unproven epoch once the previous one lands on L1. A monotonic high-water
-mark (`lastTickEpoch`) prevents the tick from re-opening an epoch whose `EpochSession`
-already terminated; the mark advances only after an `EpochSession` actually exists for
-the epoch, so transient blockers (max-pending-jobs reached, archiver still
-indexing) leave the mark in place and the next tick retries.
+next unproven epoch once the previous one lands on L1. The tick is not gated by any per-epoch
+bookkeeping; two things keep it from re-proving a doomed epoch:
+
+- `openFullSessionIfReady` refuses to build a session when any `CheckpointProver` in the set has
+  **failed** (`isFailed()`). So a prover fault (possibly a prune) skips the epoch cheaply each tick —
+  no session, no proving — until a prune/re-add installs a fresh prover.
+- A session that failed on its own account (`hasFailed()` — top-tree/submit failed with every prover
+  healthy) is **retained** by `recreateInvalidSessions` rather than deleted, so the tick's
+  `fullSessions.has(epoch)` check skips it. It is replaced only when its canonical content changes (a
+  re-add), so a deterministically-failing epoch is not re-proved every tick.
+
+Transient blockers (max-pending-jobs reached, archiver still indexing) create no session and no failed
+prover, so the next tick simply tries again.
 
 ## Walkthroughs
 
@@ -467,6 +499,20 @@ by keeping the prover around.  Removing it on prune and rebuilding on re-add is
 correct and simpler; the expensive block-rollup proofs are still reused when content re-appears,
 because the proving broker is content-addressed independently of the prover's lifetime.
 
+### How is a genuine failure told apart from a prune, without a racy classification?
+
+Knowledge of an L1 reorg reaches the prover-node on two causally unordered channels: the control
+plane (`chain-pruned` → `onPrune`) and the data plane (world-state unwinds, and an in-flight fork
+read faults inside a `CheckpointProver` mid-proof). If the `EpochSession` reacted to a fault by asking
+"was this a prune or a genuine failure?", every answer would sample control-plane state that lags the
+data-plane unwind — racy by construction. We remove the question by pushing the fact down to where it
+is unambiguous: a `CheckpointProver` records `isFailed()` when *its* block proofs reject. The session
+then reads a clean signal: if any prover failed, the fault might be a prune (`stopped`) — don't upload,
+don't rebuild over it, recover on re-add; if *no* prover failed yet the session's own top-tree/submit
+work failed, that is definitively **not** a prune (`failed`) — a genuine, race-free failure that is
+retained (so a deterministic failure isn't re-proved every tick) and uploaded once. No control-plane
+sampling, no timing race — the two prover-level and session-level facts carry the whole decision.
+
 ## Configuration
 
 | Env var | Description |
@@ -475,7 +521,7 @@ because the proving broker is content-addressed independently of the prover's li
 | `PROVER_NODE_MAX_PENDING_JOBS` | Cap on the number of non-terminal `EpochSession`s (full + partial).  When at limit, reconcile defers opening new full `EpochSession`s; explicit `startProof` calls throw. |
 | `PROVER_NODE_EPOCH_PROVING_DELAY_MS` | Optional sleep at the start of each `EpochSession`, before the TopTreeJob is constructed.  Used in tests to give late events time to land. |
 | `TX_GATHERING_TIMEOUT_MS` | Per-block tx gather deadline used by each `CheckpointProver`. |
-| `PROVER_NODE_FAILED_EPOCH_STORE` | If set, failed `EpochSession`s upload their proving data (every `CheckpointProver`'s txs + register-time data, regardless of sub-tree completion) to this file store. |
+| `PROVER_NODE_FAILED_EPOCH_STORE` | If set, a full session that fails on its own account (top-tree/submit, provers healthy) uploads its proving data (every `CheckpointProver`'s txs + register-time data, regardless of sub-tree completion) to this file store. |
 | `PROVER_NODE_DISABLE_PROOF_PUBLISH` | If true, the publishing service runs `analyzeEpochProofSubmission` (estimates L1 fees) instead of actually submitting. |
 
 ## Failure handling and observability
@@ -493,14 +539,29 @@ Loggers:
 - `prover-node:checkpoint-prover` — sub-tree pipeline (gather, block processing).
 - `prover-client:chonk-cache` — chonk-verifier cache enqueue / release events.
 
-On `failed` exit, `SessionManager.runSession` invokes the `onSessionFailed` callback
-the manager was constructed with. `ProverNode` wires this to `tryUploadSessionFailure`,
-which calls `SessionManager.buildSessionProvingData(session)` to walk every `CheckpointProver`
-referenced by the `EpochSession` and assemble an `EpochProvingJobData` snapshot — including
-every `CheckpointProver`'s txs and register-time data even if its sub-tree never reached
-`isCompleted()`. This snapshot is what `uploadEpochProofFailure` ships to the
-configured file store along with a world-state + archiver backup, so the failure
-can be reproduced offline via `rerunEpochProvingJob`.
+Failed proving data is uploaded to the configured file store (`PROVER_NODE_FAILED_EPOCH_STORE`) at two
+levels, and re-proved offline by the matching downloader/rerunner: an epoch job via
+`downloadEpochProvingJob` + `rerunEpochProvingJob`, and a single checkpoint via the same download plus
+`rerunCheckpointProvingJob` (which rebuilds just that checkpoint's sub-tree prover and awaits its block
+proofs — no epoch top-tree or L1 submit). Both upload levels build an `EpochProvingJobData` snapshot with
+`SessionManager.buildProvingData(...)` — every `CheckpointProver`'s txs and register-time data, even if
+its sub-tree never reached `isCompleted()` — and ship it via `uploadEpochProofFailure` alongside a
+world-state + archiver backup:
+
+- **Per checkpoint.** When a `CheckpointProver`'s block proofs reject for a non-cancel reason (a sub-tree
+  fault or a prune-induced fork fault), it fires its `onFailed` callback and `ProverNode` uploads that one
+  checkpoint via `tryUploadCheckpointFailure(prover)`. This fires for prune-induced faults too — that is
+  intentional; a prune-caused checkpoint snapshot is harmless, and *not* trying to tell prune from genuine
+  failure is exactly what keeps this race-free. A cancelled prover (control-plane prune / shutdown) is not
+  a failure and does not upload.
+- **Per session.** When a full session ends in its own genuine failure (`EpochSession.hasFailed()` —
+  top-tree/submit failed with every prover healthy), the session manager fires `onSessionFailed` and
+  `ProverNode` uploads all the session's checkpoints via `tryUploadEpochFailure(...)`. This is race-free
+  (healthy provers rule out a prune) and fires once, since the failed session is retained and never re-run.
+
+A `stopped` session (a prover under it failed) does not upload at the session level — its failed checkpoint
+already uploaded itself. Each backup is a full world-state + archiver snapshot, so a prune that faults
+several in-flight provers produces several uploads; this is accepted for the diagnostic value.
 
 Metrics emitted by `EpochSession`s:
 
diff --git a/yarn-project/prover-node/src/actions/rerun-epoch-proving-job.ts b/yarn-project/prover-node/src/actions/rerun-epoch-proving-job.ts
index ef80162cd613..25d3b1ee9657 100644
--- a/yarn-project/prover-node/src/actions/rerun-epoch-proving-job.ts
+++ b/yarn-project/prover-node/src/actions/rerun-epoch-proving-job.ts
@@ -21,35 +21,30 @@ import { createWorldState } from '@aztec/world-state';
 import { readFileSync } from 'fs';
 
 import { CheckpointProver } from '../job/checkpoint-prover.js';
-import { deserializeEpochProvingJobData } from '../job/epoch-proving-job-data.js';
+import { type EpochProvingJobData, deserializeEpochProvingJobData } from '../job/epoch-proving-job-data.js';
 import { EpochSession, type SessionSpec } from '../job/epoch-session.js';
 import { ProverNodeJobMetrics } from '../metrics.js';
 
+type RerunConfig = DataStoreConfig &
+  ProverBrokerConfig &
+  ProverClientConfig &
+  Pick<L1ContractsConfig, 'aztecEpochDuration'>;
+
 /**
  * Given a local folder where `downloadEpochProvingJob` was called, creates a new archiver and world state
  * using the state snapshots, and creates a new epoch proving session to prove the downloaded proving job.
  * Proving is done with a local proving broker and agents as specified by the config.
  */
-export async function rerunEpochProvingJob(
-  localPath: string,
-  log: Logger,
-  config: DataStoreConfig & ProverBrokerConfig & ProverClientConfig & Pick<L1ContractsConfig, 'aztecEpochDuration'>,
-  genesis?: GenesisData,
-) {
-  const jobData = deserializeEpochProvingJobData(readFileSync(localPath));
-  log.info(`Loaded proving job data for epoch ${jobData.epochNumber}`);
+export async function rerunEpochProvingJob(localPath: string, log: Logger, config: RerunConfig, genesis?: GenesisData) {
+  const ctx = await createRerunContext(localPath, log, config, genesis);
+  const { jobData, prover, metrics } = ctx;
 
-  const telemetry = getTelemetryClient();
-  const metrics = new ProverNodeJobMetrics(telemetry.getMeter('prover-job'), telemetry.getTracer('prover-job'));
-  const worldState = await createWorldState(config, genesis);
-  const initialBlockHash = await worldState.getInitialHeader().hash();
-  const archiver = await createArchiverStore(config, initialBlockHash);
-  const publicProcessorFactory = new PublicProcessorFactory(
-    createContractDataSource(archiver),
-    undefined,
-    undefined,
-    log.getBindings(),
-  );
+  log.info(`Rerunning epoch proving for epoch ${jobData.epochNumber}`);
+
+  const provers: CheckpointProver[] = [];
+  for (let i = 0; i < jobData.checkpoints.length; i++) {
+    provers.push(await buildCheckpointProver(ctx, i, log));
+  }
 
   // Local rerun never publishes — stub the service so submit() always resolves 'published'
   // and withdraw is a no-op.
@@ -57,51 +52,6 @@ export async function rerunEpochProvingJob(
     submit: () => Promise.resolve('published' as const),
     withdraw: () => {},
   };
-  const broker = await createAndStartProvingBroker(config, telemetry);
-  const prover = await createProverClient(config, worldState, broker, telemetry);
-  const chonkCache = new ChonkCache(log.getBindings());
-
-  const txProvider = makeReplayingTxProvider(jobData.txs);
-
-  log.info(`Rerunning epoch proving for epoch ${jobData.epochNumber}`);
-
-  const provers: CheckpointProver[] = [];
-  for (let i = 0; i < jobData.checkpoints.length; i++) {
-    const checkpoint = jobData.checkpoints[i];
-    const previousBlockHeader =
-      i === 0 ? jobData.previousBlockHeader : jobData.checkpoints[i - 1].blocks.at(-1)!.header;
-    const l1ToL2Messages = jobData.l1ToL2Messages[checkpoint.number] ?? [];
-    const previousArchiveSiblingPath = await getLastSiblingPath(
-      MerkleTreeId.ARCHIVE,
-      worldState.getSnapshot(BlockNumber(checkpoint.blocks[0].number - 1)),
-    );
-    const attestations = checkpoint.number === jobData.checkpoints.at(-1)!.number ? jobData.attestations : [];
-    provers.push(
-      new CheckpointProver(
-        {
-          checkpoint,
-          epochNumber: jobData.epochNumber,
-          attestations,
-          previousBlockHeader,
-          l1ToL2Messages,
-          previousArchiveSiblingPath,
-        },
-        {
-          proverFactory: prover,
-          chonkCache,
-          publicProcessorFactory,
-          dbProvider: worldState,
-          txProvider,
-          dateProvider: new DateProvider(),
-          proverId: prover.getProverId(),
-          metrics,
-          txGatheringTimeoutMs: 120_000,
-          deadline: undefined,
-          log,
-        },
-      ),
-    );
-  }
 
   const l1Constants = { epochDuration: config.aztecEpochDuration };
   const [fromSlot, toSlot] = getSlotRangeForEpoch(jobData.epochNumber, l1Constants);
@@ -125,6 +75,115 @@ export async function rerunEpochProvingJob(
   return finalState;
 }
 
+/**
+ * Re-proves a single downloaded checkpoint proving job (as uploaded by a `CheckpointProver` failure).
+ * Reconstructs just that checkpoint's sub-tree prover from the snapshot and awaits its block proofs — no
+ * epoch top-tree, no L1 submission — so a checkpoint-level failure can be reproduced offline in isolation.
+ * Returns the block-rollup proof outputs on success; throws if the checkpoint fails to prove again.
+ */
+export async function rerunCheckpointProvingJob(
+  localPath: string,
+  log: Logger,
+  config: RerunConfig,
+  genesis?: GenesisData,
+) {
+  const ctx = await createRerunContext(localPath, log, config, genesis);
+  const { jobData } = ctx;
+  const checkpointNumber = jobData.checkpoints[0].number;
+
+  log.info(`Rerunning checkpoint proving for checkpoint ${checkpointNumber} (epoch ${jobData.epochNumber})`);
+
+  const prover = await buildCheckpointProver(ctx, 0, log);
+  try {
+    const blockProofs = await prover.whenBlockProofsReady();
+    log.info(`Completed proving for checkpoint ${checkpointNumber} with ${blockProofs.length} block proof(s)`);
+    return blockProofs;
+  } finally {
+    prover.cancel({ routine: true });
+    await prover.whenDone();
+  }
+}
+
+/** Everything a rerun needs, reconstructed from the downloaded snapshot + job data. */
+type RerunContext = {
+  jobData: EpochProvingJobData;
+  metrics: ProverNodeJobMetrics;
+  worldState: Awaited<ReturnType<typeof createWorldState>>;
+  publicProcessorFactory: PublicProcessorFactory;
+  prover: Awaited<ReturnType<typeof createProverClient>>;
+  chonkCache: ChonkCache;
+  txProvider: ITxProvider;
+};
+
+/**
+ * Rebuilds the offline proving environment from a downloaded job: world state + archiver from the
+ * snapshots, a local proving broker + client, a chonk cache, and a tx provider that replays the job's txs.
+ */
+async function createRerunContext(
+  localPath: string,
+  log: Logger,
+  config: RerunConfig,
+  genesis?: GenesisData,
+): Promise<RerunContext> {
+  const jobData = deserializeEpochProvingJobData(readFileSync(localPath));
+  log.info(`Loaded proving job data for epoch ${jobData.epochNumber}`);
+
+  const telemetry = getTelemetryClient();
+  const metrics = new ProverNodeJobMetrics(telemetry.getMeter('prover-job'), telemetry.getTracer('prover-job'));
+  const worldState = await createWorldState(config, genesis);
+  const initialBlockHash = await worldState.getInitialHeader().hash();
+  const archiver = await createArchiverStore(config, initialBlockHash);
+  const publicProcessorFactory = new PublicProcessorFactory(
+    createContractDataSource(archiver),
+    undefined,
+    undefined,
+    log.getBindings(),
+  );
+  const broker = await createAndStartProvingBroker(config, telemetry);
+  const prover = await createProverClient(config, worldState, broker, telemetry);
+  const chonkCache = new ChonkCache(log.getBindings());
+  const txProvider = makeReplayingTxProvider(jobData.txs);
+
+  return { jobData, metrics, worldState, publicProcessorFactory, prover, chonkCache, txProvider };
+}
+
+/** Reconstructs the `CheckpointProver` for the checkpoint at `index` in the job, ready to prove. */
+async function buildCheckpointProver(ctx: RerunContext, index: number, log: Logger): Promise<CheckpointProver> {
+  const { jobData, worldState, prover, chonkCache, publicProcessorFactory, txProvider, metrics } = ctx;
+  const checkpoint = jobData.checkpoints[index];
+  const previousBlockHeader =
+    index === 0 ? jobData.previousBlockHeader : jobData.checkpoints[index - 1].blocks.at(-1)!.header;
+  const l1ToL2Messages = jobData.l1ToL2Messages[checkpoint.number] ?? [];
+  const previousArchiveSiblingPath = await getLastSiblingPath(
+    MerkleTreeId.ARCHIVE,
+    worldState.getSnapshot(BlockNumber(checkpoint.blocks[0].number - 1)),
+  );
+  const attestations = checkpoint.number === jobData.checkpoints.at(-1)!.number ? jobData.attestations : [];
+  return new CheckpointProver(
+    {
+      checkpoint,
+      epochNumber: jobData.epochNumber,
+      attestations,
+      previousBlockHeader,
+      l1ToL2Messages,
+      previousArchiveSiblingPath,
+    },
+    {
+      proverFactory: prover,
+      chonkCache,
+      publicProcessorFactory,
+      dbProvider: worldState,
+      txProvider,
+      dateProvider: new DateProvider(),
+      proverId: prover.getProverId(),
+      metrics,
+      txGatheringTimeoutMs: 120_000,
+      deadline: undefined,
+      log,
+    },
+  );
+}
+
 /** Build a synthetic ITxProvider that returns the supplied txs map by lookup. */
 function makeReplayingTxProvider(txs: Map<string, Tx>): ITxProvider {
   const lookup = (hashes: TxHash[]) => {
diff --git a/yarn-project/prover-node/src/checkpoint-store.ts b/yarn-project/prover-node/src/checkpoint-store.ts
index c675c1fe1a86..c701805d7414 100644
--- a/yarn-project/prover-node/src/checkpoint-store.ts
+++ b/yarn-project/prover-node/src/checkpoint-store.ts
@@ -4,7 +4,12 @@ import type { L2BlockSource } from '@aztec/stdlib/block';
 import type { Checkpoint } from '@aztec/stdlib/checkpoint';
 import { type L1RollupConstants, getEpochAtSlot, getSlotRangeForEpoch } from '@aztec/stdlib/epoch-helpers';
 
-import { CheckpointProver, type CheckpointProverArgs, type CheckpointProverDeps } from './job/checkpoint-prover.js';
+import {
+  CheckpointProver,
+  type CheckpointProverArgs,
+  type CheckpointProverDeps,
+  type CheckpointProverTestHooks,
+} from './job/checkpoint-prover.js';
 
 /** Register-time data needed to construct a `CheckpointProver` (everything except the checkpoint + epoch). */
 export type RegisterCheckpointData = Omit<CheckpointProverArgs, 'checkpoint' | 'epochNumber'>;
@@ -35,6 +40,8 @@ export class CheckpointStore {
    */
   private readonly pendingTeardowns = new Map<number, Promise<void>>();
   private nextTeardownId = 0;
+  /** Test-only hooks injected into every prover this store constructs. */
+  private testHooks: CheckpointProverTestHooks = {};
   private readonly log: Logger;
 
   constructor(
@@ -100,11 +107,22 @@ export class CheckpointStore {
       }
     }
 
-    const prover = this.proverFactoryFn({ ...data, checkpoint, epochNumber }, { ...this.proverDeps, log: this.log });
+    const prover = this.proverFactoryFn(
+      { ...data, checkpoint, epochNumber },
+      { ...this.proverDeps, checkpointProveOverride: this.testHooks.checkpointProveOverride, log: this.log },
+    );
     this.provers.set(id, prover);
     return prover;
   }
 
+  /**
+   * Installs test-only hooks applied to every prover constructed from now on. Used by the e2e harness to
+   * force a checkpoint sub-tree failure without monkey-patching the prover factory.
+   */
+  public setTestHooks(hooks: CheckpointProverTestHooks): void {
+    this.testHooks = hooks;
+  }
+
   /**
    * Cancels and removes every prover that holds a block above the prune target. A checkpoint is orphaned by a prune to
    * block `targetBlockNumber` iff its last block sits above the target — including a checkpoint whose range straddles
diff --git a/yarn-project/prover-node/src/job/checkpoint-prover.test.ts b/yarn-project/prover-node/src/job/checkpoint-prover.test.ts
index f516b5f33485..288a11c3cb71 100644
--- a/yarn-project/prover-node/src/job/checkpoint-prover.test.ts
+++ b/yarn-project/prover-node/src/job/checkpoint-prover.test.ts
@@ -13,6 +13,7 @@ import { Checkpoint } from '@aztec/stdlib/checkpoint';
 import type { ForkMerkleTreeOperations, ITxProvider } from '@aztec/stdlib/interfaces/server';
 import type { BlockHeader, Tx } from '@aztec/stdlib/tx';
 
+import { jest } from '@jest/globals';
 import { mock } from 'jest-mock-extended';
 
 import { ProverNodeJobMetrics } from '../metrics.js';
@@ -26,6 +27,7 @@ describe('CheckpointProver', () => {
   let publicProcessorFactory: ReturnType<typeof mock<PublicProcessorFactory>>;
   let dbProvider: ReturnType<typeof mock<Pick<ForkMerkleTreeOperations, 'fork'>>>;
   let chonkCache: ReturnType<typeof mock<ChonkCache>>;
+  let onFailed: jest.Mock<(prover: CheckpointProver) => void>;
   let log: Logger;
 
   beforeEach(async () => {
@@ -36,6 +38,7 @@ describe('CheckpointProver', () => {
     publicProcessorFactory = mock<PublicProcessorFactory>();
     dbProvider = mock<Pick<ForkMerkleTreeOperations, 'fork'>>();
     chonkCache = mock<ChonkCache>();
+    onFailed = jest.fn<(prover: CheckpointProver) => void>();
     log = createLogger('test:checkpoint-prover');
 
     // Default: gather rejects fast so the eager pipeline unwinds without hanging. The
@@ -59,6 +62,7 @@ describe('CheckpointProver', () => {
       ),
       txGatheringTimeoutMs: 30_000,
       deadline: undefined,
+      onFailed,
       log,
     };
   });
@@ -95,7 +99,7 @@ describe('CheckpointProver', () => {
       expect(prover.attestations).toEqual([]);
       expect(prover.l1ToL2Messages).toEqual([]);
       expect(prover.isCancelled()).toBe(false);
-      expect(prover.isCompleted()).toBe(false);
+      expect(prover.isFailed()).toBe(false);
       await cleanup(prover);
     });
 
@@ -130,11 +134,15 @@ describe('CheckpointProver', () => {
       await prover.whenDone();
     });
 
-    it('rejects whenBlockProofsReady()', async () => {
+    it('rejects whenBlockProofsReady() but does not mark the prover failed or fire onFailed', async () => {
+      // A cancel (reorg/prune/shutdown) is not a proving failure: isFailed() must stay false and the
+      // onFailed callback must not fire (no post-mortem upload for a cancelled prover).
       const prover = makeProver();
       const blockProofs = prover.whenBlockProofsReady();
       prover.cancel();
       await expect(blockProofs).rejects.toThrow(/cancelled/);
+      expect(prover.isFailed()).toBe(false);
+      expect(onFailed).not.toHaveBeenCalled();
       await prover.whenDone();
     });
 
@@ -274,6 +282,47 @@ describe('CheckpointProver', () => {
     });
   });
 
+  // ---------------- data-plane reorg fork fault ----------------
+
+  describe('data-plane reorg fault', () => {
+    it('rejects whenBlockProofsReady when a world-state fork faults mid-proof', async () => {
+      // Models the data-plane prune race: gather succeeds and the sub-tree starts, but the
+      // world-state synchronizer has already unwound the base block, so forking it faults inside
+      // executeCheckpoint. The fault must reject whenBlockProofsReady() AND mark the prover failed, so
+      // the SessionManager won't build (or rebuild) an EpochSession over it until a re-add replaces it.
+      txProvider.getTxsForBlock.mockReset();
+      txProvider.getTxsForBlock.mockResolvedValue({ txs: [], missingTxs: [] });
+
+      const subTree = {
+        getSubTreeResult: () => new Promise<never>(() => {}),
+        startNewBlock: () => Promise.resolve(),
+        startChonkVerifierCircuits: () => Promise.resolve(),
+        addTxs: () => Promise.resolve(),
+        setBlockCompleted: () => Promise.resolve(),
+        cancel: () => {},
+        stop: () => Promise.resolve(),
+      };
+      proverFactory.createCheckpointSubTreeOrchestrator.mockResolvedValue(subTree as any);
+
+      // The prune-induced fault: the base block was unwound, so forking it rejects. This is the
+      // production signal — `Unable to get meta data for block N` out of world-state.
+      dbProvider.fork.mockRejectedValue(new Error('Unable to get meta data for block 0'));
+
+      const prover = makeProver();
+
+      // blockProofs rejects: the fork error aborts the block loop before completion, so the sub-tree
+      // never yields proofs. (The raw fork error is logged; the promise settles as not-completed.)
+      await expect(prover.whenBlockProofsReady()).rejects.toThrow(/did not complete block processing/);
+      expect(dbProvider.fork).toHaveBeenCalled();
+      expect(prover.isFailed()).toBe(true);
+      // The owner is notified exactly once, with this prover, so it can upload a checkpoint post-mortem.
+      expect(onFailed).toHaveBeenCalledTimes(1);
+      expect(onFailed).toHaveBeenCalledWith(prover);
+
+      await cleanup(prover);
+    });
+  });
+
   // ---------------- helpers ----------------
 
   function makeProver(overrides: Partial<CheckpointProverArgs> = {}): CheckpointProver {
diff --git a/yarn-project/prover-node/src/job/checkpoint-prover.ts b/yarn-project/prover-node/src/job/checkpoint-prover.ts
index 2f39dd15d8df..5a04412bb85a 100644
--- a/yarn-project/prover-node/src/job/checkpoint-prover.ts
+++ b/yarn-project/prover-node/src/job/checkpoint-prover.ts
@@ -38,9 +38,25 @@ export type CheckpointProverDeps = {
   txGatheringTimeoutMs: number;
   /** Public processor deadline. */
   deadline: Date | undefined;
+  /**
+   * Fired once when the prover's block proofs reject for a genuine (non-cancel) reason — a sub-tree
+   * fault or a prune-induced fork fault. Useful for performing post mortem on failures.
+   */
+  onFailed?: (prover: CheckpointProver) => void;
+  /**
+   * Test-only hook: if set, invoked at the start of checkpoint execution instead of proving. Lets e2e
+   * tests force a sub-tree failure (it should throw) to exercise the checkpoint failure/upload path.
+   */
+  checkpointProveOverride?: () => Promise<never>;
   log: Logger;
 };
 
+/** Test-only hooks the store injects into every `CheckpointProver` it constructs. */
+export type CheckpointProverTestHooks = {
+  /** If set, invoked at the start of checkpoint execution instead of proving; should throw to fail. */
+  checkpointProveOverride?: () => Promise<never>;
+};
+
 /** Inputs that fully describe a checkpoint at register time. */
 export type CheckpointProverArgs = {
   checkpoint: Checkpoint;
@@ -89,9 +105,18 @@ export class CheckpointProver {
   /** Resolved by the sub-tree on success, rejected on cancel/failure. */
   private readonly blockProofs: PromiseWithResolvers<SubTreeResult['blockProofOutputs']> = promiseWithResolvers();
 
+  // Three independent lifecycle facts — deliberately not collapsed into one status enum, because several
+  // combinations are legal and relied on: a prover can be `completed` and then `cancelled` (routine
+  // teardown of an already-proven checkpoint), or `completed` and then `failed` (block proving was
+  // enqueued, but the sub-tree subsequently faulted). Only `failed` + `cancelled` is excluded — a cancel
+  // is not a failure (enforced in `failBlockProofs`).
+  /** Block-level proving was fully *enqueued* (a progress marker; the sub-tree may still be proving). */
+  private completed = false;
+  /** Block proofs rejected for a genuine (non-cancel) reason — a sub-tree or prune-induced fork fault. */
+  private failed = false;
+  /** The prover was torn down (prune / reap / shutdown). */
   private cancelled = false;
   private subTree?: CheckpointSubTreeOrchestrator;
-  private completed = false;
   private readonly abortController = new AbortController();
 
   /** Tracks the eager gather+execute task so `cancel()` and `whenDone()` can await its unwind. */
@@ -139,9 +164,14 @@ export class CheckpointProver {
     return this.cancelled;
   }
 
-  /** True once block-level proving has been fully *enqueued* (sub-tree completion may still be pending). */
-  public isCompleted(): boolean {
-    return this.completed;
+  /**
+   * True once this prover's block proofs have rejected for a genuine (non-cancel) reason — a sub-tree
+   * proving fault or a prune-induced world-state fork fault. A failed prover cannot produce its block
+   * proofs, so the reconciler must not build (or rebuild) an EpochSession over it; it is cleared only by
+   * a prune/re-add replacing it with a fresh prover, or by expiry reaping it.
+   */
+  public isFailed(): boolean {
+    return this.failed;
   }
 
   /** AbortSignal that fires on cancel — for callers that want to wire their own tasks. */
@@ -179,10 +209,29 @@ export class CheckpointProver {
       this.deps.log.error(`Error in CheckpointProver ${this.id}`, err, {
         checkpointNumber: this.checkpoint.number,
       });
-      this.blockProofs.reject(err instanceof Error ? err : new Error(String(err)));
+      this.failBlockProofs(err instanceof Error ? err : new Error(String(err)));
     }
   }
 
+  /**
+   * Rejects the block-proof promise and, unless this is a cancellation, records the prover as failed so
+   * the reconciler won't build an EpochSession over it. First rejection wins, so a later duplicate reject
+   * (e.g. the executeCheckpoint `finally`) is a harmless no-op.
+   */
+  private failBlockProofs(err: Error): void {
+    if (!this.cancelled && !this.failed) {
+      this.failed = true;
+      // Notify the owner so it can upload a post-mortem for this checkpoint. Fire-and-forget: the
+      // callback must not block the prover's teardown, and a throw in it must not mask the rejection.
+      try {
+        this.deps.onFailed?.(this);
+      } catch (err) {
+        this.deps.log.error(`Error in CheckpointProver onFailed callback for ${this.id}`, err);
+      }
+    }
+    this.blockProofs.reject(err);
+  }
+
   private async gatherTxs(): Promise<Map<string, Tx>> {
     const deadline = new Date(this.deps.dateProvider.now() + this.deps.txGatheringTimeoutMs);
     const txsByBlock = await Promise.all(
@@ -205,6 +254,11 @@ export class CheckpointProver {
     let subTreeStarted = false;
 
     try {
+      // Test hook: force a sub-tree failure to exercise the checkpoint failure/upload path.
+      if (this.deps.checkpointProveOverride) {
+        await this.deps.checkpointProveOverride();
+      }
+
       for (const [hash, tx] of txs) {
         this.txs.set(hash, tx);
       }
@@ -248,7 +302,7 @@ export class CheckpointProver {
           this.deps.metrics.recordCheckpointProving(checkpointTimer.ms());
           this.blockProofs.resolve(result.blockProofOutputs);
         },
-        err => this.blockProofs.reject(err),
+        err => this.failBlockProofs(err instanceof Error ? err : new Error(String(err))),
       );
       if (signal.aborted) {
         return;
@@ -328,7 +382,7 @@ export class CheckpointProver {
         if (subTreeStarted) {
           await this.teardownSubTree();
         }
-        this.blockProofs.reject(new Error(`Checkpoint ${this.id} did not complete block processing`));
+        this.failBlockProofs(new Error(`Checkpoint ${this.id} did not complete block processing`));
       }
     }
   }
diff --git a/yarn-project/prover-node/src/job/epoch-session.test.ts b/yarn-project/prover-node/src/job/epoch-session.test.ts
index c13701bd4d99..7c6512c2fd02 100644
--- a/yarn-project/prover-node/src/job/epoch-session.test.ts
+++ b/yarn-project/prover-node/src/job/epoch-session.test.ts
@@ -147,11 +147,14 @@ describe('EpochSession', () => {
       expect(state).toBe(expected);
     });
 
-    it('"failed" outcome propagates as a thrown error → state "failed"', async () => {
+    it('"failed" submit outcome propagates as a thrown error → state "failed" (healthy provers)', async () => {
+      // The provers all succeeded (a proof was produced) and only the L1 submission failed — a genuine,
+      // non-prune failure of the session's own work, so it ends in terminal 'failed'.
       publishingService.submit.mockResolvedValue('failed');
       const session = makeSession();
       const state = await session.start();
       expect(state).toBe('failed');
+      expect(session.hasFailed()).toBe(true);
     });
 
     it('"withdrawn" outcome with no prior cancel falls back to "cancelled"', async () => {
@@ -279,10 +282,12 @@ describe('EpochSession', () => {
   // ---------------- checkpoint failure ----------------
 
   describe('checkpoint that fails to prove', () => {
-    it('drives the session to "failed" when a checkpoint\'s blockProofs reject', async () => {
+    it('ends the session in "stopped" (not "failed") when a checkpoint\'s blockProofs reject', async () => {
       // Build a prover whose block-rollup proofs are guaranteed to reject — this mirrors
       // the production path where CheckpointProver.executeCheckpoint catches an internal
-      // error and rejects its blockProofs promise.
+      // error (e.g. a data-plane reorg fork fault) and rejects its blockProofs promise.
+      // The session must NOT declare the epoch failed: it ends in the non-declaring terminal
+      // 'stopped', leaving the reconciler free to rebuild it over current canonical content.
       const failingProver = makeStubProver(cp, { blockProofsError: new Error('block 7 proving failed') });
       const session = new EpochSession(
         makeSpec(),
@@ -294,13 +299,15 @@ describe('EpochSession', () => {
         }),
       );
       const state = await session.start();
-      expect(state).toBe('failed');
+      expect(state).toBe('stopped');
       expect(session.isTerminal()).toBe(true);
+      // A failed prover under the session ⇒ 'stopped', NOT the session's own 'failed' (no upload).
+      expect(session.hasFailed()).toBe(false);
       // Failure happens before submission; the publishing service must never see the candidate.
       expect(publishingService.submit).not.toHaveBeenCalled();
     });
 
-    it('whenDone resolves to "failed" so callers observing the lifecycle agree with the return value', async () => {
+    it('whenDone resolves to "stopped" so callers observing the lifecycle agree with the return value', async () => {
       const failingProver = makeStubProver(cp, { blockProofsError: new Error('boom') });
       const session = new EpochSession(
         makeSpec(),
@@ -310,18 +317,42 @@ describe('EpochSession', () => {
         }),
       );
       const startResult = session.start();
-      await expect(session.whenDone()).resolves.toBe('failed');
-      await expect(startResult).resolves.toBe('failed');
+      await expect(session.whenDone()).resolves.toBe('stopped');
+      await expect(startResult).resolves.toBe('stopped');
+    });
+
+    it('ends the session in "stopped" (not "failed") when a prover was cancelled by a prune (isCancelled, not isFailed)', async () => {
+      // A control-plane prune cancels the prover: its blockProofs reject with "cancelled" and it reports
+      // isCancelled()===true / isFailed()===false. If that rejection reaches start()'s catch before the
+      // reconcile marks the session 'cancelled', it must NOT be classified 'failed' (which would trigger a
+      // spurious full-snapshot upload) — a cancelled prover is prune-ambiguous, so classify 'stopped'.
+      const cancelledProver = makeStubProver(cp, {
+        blockProofsError: new Error('Checkpoint cancelled'),
+        isFailed: false,
+        isCancelled: true,
+      });
+      const session = new EpochSession(
+        makeSpec(),
+        [cancelledProver],
+        makeDeps({
+          hooks: { topTreeProveOverride: () => cancelledProver.whenBlockProofsReady().then(() => synthProof) },
+        }),
+      );
+      const state = await session.start();
+      expect(state).toBe('stopped');
+      expect(session.hasFailed()).toBe(false);
+      expect(publishingService.submit).not.toHaveBeenCalled();
     });
 
-    it('a prove that rejects for any reason ends the session in "failed" without submitting', async () => {
-      // Belt-and-braces: any prove rejection (top-tree internal error, blob computation,
-      // etc.) follows the same path. The session swallows the error and reports 'failed'.
+    it('a top-tree prove that rejects with healthy provers ends the session in "failed" without submitting', async () => {
+      // The provers are healthy but the top-tree (root) prove itself rejects — the session's own work
+      // failed, and since no prover failed this is definitively not a prune, so it ends in 'failed'.
       const session = makeSession({
         hooks: { topTreeProveOverride: () => Promise.reject(new Error('top-tree internal failure')) },
       });
       const state = await session.start();
       expect(state).toBe('failed');
+      expect(session.hasFailed()).toBe(true);
       expect(publishingService.submit).not.toHaveBeenCalled();
     });
   });
@@ -439,7 +470,10 @@ class TestEpochSession extends EpochSession {
  * path where CheckpointProver.executeCheckpoint catches an internal failure and rejects
  * its blockProofs promise.
  */
-function makeStubProver(checkpoint: Checkpoint, opts: { blockProofsError?: Error } = {}): CheckpointProver {
+function makeStubProver(
+  checkpoint: Checkpoint,
+  opts: { blockProofsError?: Error; isFailed?: boolean; isCancelled?: boolean } = {},
+): CheckpointProver {
   const id = CheckpointProver.idFor(checkpoint);
   // By default whenBlockProofsReady never resolves in these tests; the prove override
   // bypasses any path that would actually await it.
@@ -460,8 +494,10 @@ function makeStubProver(checkpoint: Checkpoint, opts: { blockProofsError?: Error
     previousArchiveSiblingPath: makeTuple(ARCHIVE_HEIGHT, () => Fr.ZERO),
     txs: new Map(),
     whenBlockProofsReady: () => blockProofs,
-    isCancelled: () => false,
-    isCompleted: () => false,
+    isCancelled: () => opts.isCancelled ?? false,
+    // A prover configured with a blockProofsError is one whose block proofs rejected — i.e. failed,
+    // unless the caller decouples the two (e.g. to model a cancelled-but-not-failed prune).
+    isFailed: () => opts.isFailed ?? opts.blockProofsError !== undefined,
     cancel: () => {},
     whenDone: () => Promise.resolve(),
     getAbortSignal: () => new AbortController().signal,
diff --git a/yarn-project/prover-node/src/job/epoch-session.ts b/yarn-project/prover-node/src/job/epoch-session.ts
index e174d034953f..19aaf89e2519 100644
--- a/yarn-project/prover-node/src/job/epoch-session.ts
+++ b/yarn-project/prover-node/src/job/epoch-session.ts
@@ -95,7 +95,11 @@ export type EpochSessionDeps = {
  *   initialized → awaiting-checkpoints → awaiting-root → publishing-proof → completed
  *
  * Terminal states map the publishing outcome: `published` → `completed`, `superseded` →
- * `superseded`, `failed` → `failed`, `expired` → `timed-out`, `withdrawn` → `cancelled`.
+ * `superseded`, `expired` → `timed-out`, `withdrawn` → `cancelled`. A fault ends the attempt in one
+ * of two terminal states depending on its cause: `stopped` if a checkpoint prover under it failed
+ * (possibly a prune — the reconciler will rebuild over a fresh prover on re-add), or `failed` if the
+ * session's own top-tree/submit work failed while every prover was healthy (a genuine, non-prune
+ * failure the reconciler retains and uploads — see `hasFailed()`).
  * Additionally, the session-level deadline fires `cancel('deadline')` and transitions
  * to `timed-out` for the pre-submit window (top-tree proving) — the publishing service
  * handles the post-submit window via the candidate's `deadline`.
@@ -187,6 +191,17 @@ export class EpochSession implements Traceable {
     return EpochProvingJobTerminalState.includes(this.state);
   }
 
+  /**
+   * True if the session ended in its own genuine failure — top-tree proving or L1 submission failed
+   * while every checkpoint prover succeeded. Because healthy provers rule out a prune-induced fault,
+   * this is a race-free "the epoch could not be proven" signal: the reconciler retains such a (full)
+   * session rather than re-proving it, and uploads a post-mortem. A `stopped` session (a checkpoint
+   * prover failed under it) is NOT a session failure in this sense.
+   */
+  public hasFailed(): boolean {
+    return this.state === 'failed';
+  }
+
   /** First block this session proves. */
   public getStartBlockNumber(): BlockNumber {
     return BlockNumber(this.checkpoints[0].checkpoint.blocks[0].number);
@@ -213,8 +228,17 @@ export class EpochSession implements Traceable {
         uuid: this.uuid,
         ...this.spec,
       });
+      // Distinguish the two ways an attempt can fault:
+      //  - a checkpoint prover in the set has failed OR was cancelled (a sub-tree fault, a prune-induced
+      //    fork fault, or a control-plane cancel that reached this catch before the reconcile marked the
+      //    session 'cancelled'): end in the non-declaring terminal 'stopped'. This is not the session's own
+      //    failure and may be a prune, so the reconciler does not upload it; a re-add installs a fresh prover.
+      //  - no prover failed or was cancelled, yet top-tree proving or L1 submission failed: this is the
+      //    session's own, genuine failure — and, because every prover is healthy and un-cancelled, it is
+      //    definitively NOT a prune. End in terminal 'failed' so the reconciler retains it (no pointless
+      //    re-prove) and uploads a race-free post-mortem.
       if (!this.isTerminal()) {
-        this.state = 'failed';
+        this.state = this.checkpoints.some(c => c.isFailed() || c.isCancelled()) ? 'stopped' : 'failed';
       }
     } finally {
       clearTimeout(this.deadlineTimeoutHandler);
diff --git a/yarn-project/prover-node/src/prover-node.test.ts b/yarn-project/prover-node/src/prover-node.test.ts
index b8806a612cdb..7c9623b864cb 100644
--- a/yarn-project/prover-node/src/prover-node.test.ts
+++ b/yarn-project/prover-node/src/prover-node.test.ts
@@ -226,7 +226,8 @@ describe('ProverNode', () => {
     expect(publishingService.onChainProven).toHaveBeenCalledWith(BlockNumber(7));
   });
 
-  it('expires elapsed epochs on every block-stream event: releases chonk cache, reaps store', async () => {
+  it('the expiry sweep releases the chonk cache and reaps the store for elapsed epochs', async () => {
+    // The sweep is driven solely by the periodic ticker (callCheckEpochExpiry mimics one tick).
     // Latest synced L2 slot = 4 ⇒ latestEpoch = 4 ⇒ epochs 0..2 are past their submission
     // window (deadline = E+2 with proofSubmissionEpochs=1).
     l2BlockSource.getSyncedL2SlotNumber.mockResolvedValue(SlotNumber(4));
@@ -242,12 +243,7 @@ describe('ProverNode', () => {
 
     const reapSpy = jest.spyOn(proverNode.getCheckpointStore(), 'reapExpired');
 
-    // Any block-stream event is enough to trigger the expiry sweep.
-    await proverNode.handleBlockStreamEvent({
-      type: 'chain-finalized',
-      block: { number: BlockNumber(1), hash: '0x01' },
-      checkpoint: { number: CheckpointNumber(1), hash: '0x01' },
-    });
+    await proverNode.callCheckEpochExpiry();
 
     expect(cache.get(txHash)).toBeUndefined();
     // Three expired epochs ⇒ reapExpired called once per epoch.
@@ -259,20 +255,12 @@ describe('ProverNode', () => {
     l2BlockSource.getCheckpointsData.mockResolvedValue([]);
     const reapSpy = jest.spyOn(proverNode.getCheckpointStore(), 'reapExpired');
 
-    await proverNode.handleBlockStreamEvent({
-      type: 'chain-finalized',
-      block: { number: BlockNumber(1), hash: '0x01' },
-      checkpoint: { number: CheckpointNumber(1), hash: '0x01' },
-    });
+    await proverNode.callCheckEpochExpiry();
     expect(reapSpy.mock.calls.length).toBe(3);
     reapSpy.mockClear();
 
-    // Same latest slot ⇒ nothing new should expire.
-    await proverNode.handleBlockStreamEvent({
-      type: 'chain-finalized',
-      block: { number: BlockNumber(1), hash: '0x01' },
-      checkpoint: { number: CheckpointNumber(1), hash: '0x01' },
-    });
+    // Same latest slot ⇒ nothing new should expire on a subsequent sweep.
+    await proverNode.callCheckEpochExpiry();
     expect(reapSpy).not.toHaveBeenCalled();
   });
 
@@ -280,14 +268,29 @@ describe('ProverNode', () => {
     l2BlockSource.getSyncedL2SlotNumber.mockResolvedValue(undefined);
     const reapSpy = jest.spyOn(proverNode.getCheckpointStore(), 'reapExpired');
 
-    await proverNode.handleBlockStreamEvent({
-      type: 'chain-finalized',
-      block: { number: BlockNumber(1), hash: '0x01' },
-      checkpoint: { number: CheckpointNumber(1), hash: '0x01' },
-    });
+    await proverNode.callCheckEpochExpiry();
     expect(reapSpy).not.toHaveBeenCalled();
   });
 
+  // ---------------- expiry only reaps; the failure upload is not on this path ----------------
+
+  it('expireEpoch reaps the store but never uploads a post-mortem', async () => {
+    // The post-mortem upload fires from the session-failure path (onSessionFailed)
+    setupNotFullyProven();
+    await proverNode.handleBlockStreamEvent(mineCheckpoint(makeCheckpoint(3, 3, 3)));
+    expect(proverNode.getCheckpointStore().listAll().length).toBe(1);
+
+    const uploadSpy = jest.spyOn(proverNode, 'tryUploadEpochFailure');
+    const reapSpy = jest.spyOn(proverNode.getCheckpointStore(), 'reapExpired');
+    l2BlockSource.getBlocks.mockResolvedValue([]);
+
+    l2BlockSource.getSyncedL2SlotNumber.mockResolvedValue(SlotNumber(5));
+    await proverNode.callCheckEpochExpiry();
+
+    expect(uploadSpy).not.toHaveBeenCalled();
+    expect(reapSpy.mock.calls.map(([e]) => Number(e))).toEqual([0, 1, 2, 3]);
+  });
+
   it('propagates a checkpoint registration failure and leaves the tips store unadvanced (A-1041)', async () => {
     setupNotFullyProven();
     // Registration fails: worldState.syncImmediate (inside collectRegisterData) rejects. The
@@ -307,15 +310,20 @@ describe('ProverNode', () => {
     expect(proverNode.getLastProcessedCheckpoint()).toEqual(CheckpointNumber.ZERO);
   });
 
-  it('leaves the tips store unadvanced when a handler propagates an error (A-1041)', async () => {
-    setupNotFullyProven();
-    // Registration succeeds, but the expiry sweep throws — a failure that propagates before the
-    // tips-store update, so the error surfaces to the L2BlockStream and the tips stay put.
-    l2BlockSource.getSyncedL2SlotNumber.mockRejectedValue(new Error('archiver down'));
+  it('leaves the tips store unadvanced when a handler propagates an error', async () => {
+    // The prune handler throws when it cannot resolve the prune target's block data. That failure
+    // propagates before the tips-store update, so the error surfaces to the L2BlockStream and the
+    // tips stay put for a retry on the next poll.
+    l2BlockSource.getBlockData.mockResolvedValue(undefined);
 
-    const event = mineCheckpoint(makeCheckpoint(1, 1, 1));
+    const event: L2BlockStreamEvent = {
+      type: 'chain-pruned',
+      block: { number: BlockNumber(1), hash: '0x01' },
+      checkpointed: makeTipId(1),
+      proven: makeTipId(1),
+    };
 
-    await expect(proverNode.handleBlockStreamEvent(event)).rejects.toThrow('archiver down');
+    await expect(proverNode.handleBlockStreamEvent(event)).rejects.toThrow(/cannot clamp checkpoint cursor/);
 
     // Tips left unadvanced so the L2BlockStream re-emits this event on its next poll.
     expect(await proverNode.getTipsStore().getL2BlockHash(1)).toBeUndefined();
@@ -351,6 +359,61 @@ describe('ProverNode', () => {
     expect(prover.id).toContain(archiveRoot.toString());
   });
 
+  it('uploads a checkpoint post-mortem when a registered checkpoint prover fails', async () => {
+    // The store is wired so a checkpoint prover that fails (here its eager tx-gather rejects, as the
+    // txProvider is unconfigured) routes through to tryUploadCheckpointFailure with that prover.
+    setupNotFullyProven();
+    const uploadSpy = jest.spyOn(proverNode, 'tryUploadCheckpointFailure').mockResolvedValue(undefined);
+
+    await proverNode.handleBlockStreamEvent(mineCheckpoint(makeCheckpoint(3, 3, 3)));
+    const prover = proverNode.getCheckpointStore().listAll()[0];
+    // Wait for the eager pipeline to settle (it fails at gather), then the onFailed hook has fired.
+    await prover.whenDone();
+
+    expect(prover.isFailed()).toBe(true);
+    expect(uploadSpy).toHaveBeenCalledWith(prover);
+  });
+
+  describe('isCheckpointCanonical', () => {
+    it('is true when the archiver holds a block at the checkpoint tip with a matching archive root', async () => {
+      const archiveRoot = Fr.random();
+      const checkpoint = makeCheckpoint(3, 3, 3, archiveRoot);
+      l2BlockSource.getBlock.mockResolvedValue({ archive: { root: archiveRoot } } as unknown as L2Block);
+
+      await expect(proverNode.callIsCheckpointCanonical(checkpoint)).resolves.toBe(true);
+    });
+
+    it('is false when the checkpoint tip block was pruned out (archiver returns nothing)', async () => {
+      const checkpoint = makeCheckpoint(3, 3, 3);
+      l2BlockSource.getBlock.mockResolvedValue(undefined);
+
+      await expect(proverNode.callIsCheckpointCanonical(checkpoint)).resolves.toBe(false);
+    });
+
+    it('is false when the tip block was replaced by a reorg (archive root differs)', async () => {
+      const checkpoint = makeCheckpoint(3, 3, 3, Fr.random());
+      l2BlockSource.getBlock.mockResolvedValue({ archive: { root: Fr.random() } } as unknown as L2Block);
+
+      await expect(proverNode.callIsCheckpointCanonical(checkpoint)).resolves.toBe(false);
+    });
+  });
+
+  it('tryUploadCheckpointFailure skips the upload for a checkpoint pruned out of the canonical chain', async () => {
+    // A prune-induced fork fault reaches onFailed just like a genuine sub-tree failure, but the pruned
+    // checkpoint no longer exists on-chain — the expensive full snapshot must not be produced for it.
+    (proverNode as any).config.proverNodeFailedEpochStore = 'file:///tmp/does-not-matter';
+    const checkpoint = makeCheckpoint(3, 3, 3);
+    l2BlockSource.getBlock.mockResolvedValue(undefined);
+    const failedProver = { id: 'prover-3', checkpoint } as unknown as Parameters<
+      typeof proverNode.tryUploadCheckpointFailure
+    >[0];
+
+    await expect(proverNode.tryUploadCheckpointFailure(failedProver)).resolves.toBeUndefined();
+    expect(l2BlockSource.getBlock).toHaveBeenCalledWith({ number: checkpoint.blocks.at(-1)!.number });
+    // World-state snapshotting is the first thing the real upload path touches; it must never be reached.
+    expect(worldState.getSnapshot).not.toHaveBeenCalled();
+  });
+
   // ---------------- forwarders ----------------
 
   it('startProof forwards to the session manager and returns the job id', async () => {
@@ -377,12 +440,9 @@ describe('ProverNode', () => {
     await expect(proverNode.getJobs()).resolves.toEqual([]);
   });
 
-  // ---------------- handleBlockStreamEvent: chain-proposed is a no-op + still triggers expiry ----------------
+  // ---------------- handleBlockStreamEvent: chain-proposed is a no-op ----------------
 
-  it("'chain-proposed' invokes no event handler but still runs the expiry sweep", async () => {
-    // latestSlot=4 ⇒ epochs 0..2 expire.
-    l2BlockSource.getSyncedL2SlotNumber.mockResolvedValue(SlotNumber(4));
-    l2BlockSource.getCheckpointsData.mockResolvedValue([]);
+  it("'chain-proposed' invokes no event handler but records the tip", async () => {
     const reapSpy = jest.spyOn(proverNode.getCheckpointStore(), 'reapExpired');
 
     await proverNode.handleBlockStreamEvent({
@@ -390,12 +450,11 @@ describe('ProverNode', () => {
       block: { number: BlockNumber(1), hash: '0x01' },
     });
 
-    // No checkpoint, prune, or proven handler should have fired.
+    // No checkpoint, prune, or proven handler should have fired, and expiry is not on the event path.
     expect(sessionManager.onCheckpointAdded).not.toHaveBeenCalled();
     expect(sessionManager.onPrune).not.toHaveBeenCalled();
     expect(publishingService.onChainProven).not.toHaveBeenCalled();
-    // But the expiry sweep ran.
-    expect(reapSpy.mock.calls.map(([e]) => Number(e))).toEqual([0, 1, 2]);
+    expect(reapSpy).not.toHaveBeenCalled();
     // The tips store recorded the proposed tip (it is the walk-back history in tips-only mode).
     expect(await proverNode.getTipsStore().getL2BlockHash(1)).toEqual('0x01');
   });
@@ -407,32 +466,23 @@ describe('ProverNode', () => {
     l2BlockSource.getSyncedL2SlotNumber.mockResolvedValue(SlotNumber(1));
     const reapSpy = jest.spyOn(proverNode.getCheckpointStore(), 'reapExpired');
 
-    await proverNode.handleBlockStreamEvent({
-      type: 'chain-finalized',
-      block: { number: BlockNumber(1), hash: '0x01' },
-      checkpoint: { number: CheckpointNumber(1), hash: '0x01' },
-    });
+    await proverNode.callCheckEpochExpiry();
 
     expect(reapSpy).not.toHaveBeenCalled();
     // High-water mark stays untouched.
     expect(proverNode.getLastExpiredEpoch()).toBeUndefined();
   });
 
-  // ---------------- expireEpoch swallows getCheckpointsData errors ----------------
+  // ---------------- expireEpoch swallows getBlocks errors ----------------
 
-  it('expireEpoch still reaps the store when getCheckpointsData throws', async () => {
-    // Three epochs would expire (latestSlot=4 ⇒ epochs 0..2). getCheckpointsData throws for
-    // every call, but reapExpired must still be invoked for each epoch and the high-water
-    // mark must still advance.
+  it('expireEpoch still reaps the store when the chonk-release block fetch throws', async () => {
+    // Three epochs would expire (latestSlot=4 ⇒ epochs 0..2). getBlocks throws for every call,
+    // but reapExpired must still be invoked for each epoch and the high-water mark must still advance.
     l2BlockSource.getSyncedL2SlotNumber.mockResolvedValue(SlotNumber(4));
-    l2BlockSource.getCheckpointsData.mockRejectedValue(new Error('archiver unavailable'));
+    l2BlockSource.getBlocks.mockRejectedValue(new Error('archiver unavailable'));
     const reapSpy = jest.spyOn(proverNode.getCheckpointStore(), 'reapExpired');
 
-    await proverNode.handleBlockStreamEvent({
-      type: 'chain-finalized',
-      block: { number: BlockNumber(1), hash: '0x01' },
-      checkpoint: { number: CheckpointNumber(1), hash: '0x01' },
-    });
+    await proverNode.callCheckEpochExpiry();
 
     expect(reapSpy.mock.calls.map(([e]) => Number(e))).toEqual([0, 1, 2]);
     expect(proverNode.getLastExpiredEpoch()).toEqual(EpochNumber(2));
@@ -745,6 +795,11 @@ class TestProverNode extends ProverNode {
     return this.resolveLastFullyProvenEpoch();
   }
 
+  /** Drives the expiry sweep directly, as the periodic ticker does in production. */
+  public callCheckEpochExpiry() {
+    return (this as any).checkEpochExpiry();
+  }
+
   public callIsEpochFullyProven(epoch: EpochNumber, l1Constants: { epochDuration: number }) {
     return this.isEpochFullyProven(epoch, l1Constants as any);
   }
@@ -757,6 +812,10 @@ class TestProverNode extends ProverNode {
     return this.isProvenBlockLastOfItsEpoch(provenBlock, provenEpoch, l1Constants as any);
   }
 
+  public callIsCheckpointCanonical(checkpoint: Checkpoint) {
+    return this.isCheckpointCanonical(checkpoint);
+  }
+
   public getLastExpiredEpoch(): EpochNumber | undefined {
     return this.lastExpiredEpoch;
   }
diff --git a/yarn-project/prover-node/src/prover-node.ts b/yarn-project/prover-node/src/prover-node.ts
index 8911ad158515..97eaa5cf5238 100644
--- a/yarn-project/prover-node/src/prover-node.ts
+++ b/yarn-project/prover-node/src/prover-node.ts
@@ -47,7 +47,8 @@ import {
 import { uploadEpochProofFailure } from './actions/upload-epoch-proof-failure.js';
 import { CheckpointStore, type RegisterCheckpointData } from './checkpoint-store.js';
 import type { SpecificProverNodeConfig } from './config.js';
-import type { EpochSession, EpochSessionHooks } from './job/epoch-session.js';
+import type { CheckpointProver, CheckpointProverTestHooks } from './job/checkpoint-prover.js';
+import type { EpochSessionHooks } from './job/epoch-session.js';
 import { ProverNodeJobMetrics, ProverNodeRewardsMetrics } from './metrics.js';
 import { ProofPublishingService } from './proof-publishing-service.js';
 import type { ProverPublisherFactory } from './prover-publisher-factory.js';
@@ -168,6 +169,9 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra
         metrics: this.jobMetrics,
         txGatheringTimeoutMs: this.config.txGatheringTimeoutMs,
         deadline: undefined,
+        // A checkpoint prover that fails (a sub-tree fault or a prune-induced fork fault) uploads a
+        // post-mortem for its own checkpoint, independently of any session. Fire-and-forget.
+        onFailed: prover => void this.tryUploadCheckpointFailure(prover),
       },
       this.log.getBindings(),
     );
@@ -239,11 +243,9 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra
         break;
       }
     }
-    // Expiry is driven by the archiver's latest synced L2 slot
-    await this.checkEpochExpiry();
-    // Advance the local tips store only after the proving-side handling has succeeded. Any
-    // failure above propagates to the L2BlockStream (which logs and stops this poll pass) and
-    // skips this update, so the event is re-emitted on the next poll rather than skipped (A-1041).
+    // Advance the local tips store only after the proving-side handling (registration / prune) has
+    // succeeded. Any failure above propagates to the L2BlockStream (which logs and stops this poll
+    // pass) and skips this update, so the event is re-emitted on the next poll rather than skipped
     await this.tipsStore.handleBlockStreamEvent(event);
   }
 
@@ -458,8 +460,8 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra
   }
 
   /**
-   * Releases chonk-cache entries for every block in the supplied epoch (best-effort) and
-   * reaps every CheckpointProver in the store whose epoch number matches.
+   * Releases chonk-cache entries for every block in the supplied epoch (best-effort) and reaps every
+   * CheckpointProver in the store whose epoch is at or below it.
    */
   private async expireEpoch(epoch: EpochNumber): Promise<void> {
     try {
@@ -515,8 +517,9 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra
     });
     this.blockStream.start();
 
-    // With thin once-per-pass tip events, the expiry sweep no longer fires once per checkpoint; drive it
-    // from a periodic tick so epochs still expire during idle/no-event periods.
+    // The periodic ticker is the sole driver of the expiry sweep: it fires every poll interval whether
+    // or not block-stream events arrive, and RunningPromise never overlaps its own runs, so the sweep's
+    // `lastExpiredEpoch` high-water mark advances — and each epoch's post-mortem uploads — exactly once.
     this.expiryTicker = new RunningPromise(
       () => this.checkEpochExpiry(),
       this.log,
@@ -559,9 +562,10 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra
   }
 
   /**
-   * Constructs the session manager. Extracted so subclasses (test harness) can swap
-   * the implementation. Wired to `tryUploadSessionFailure` so failed sessions get
-   * their proving data uploaded.
+   * Constructs the session manager. Extracted so subclasses (test harness) can swap the
+   * implementation. Wired to upload a post-mortem when a full session ends in its own genuine failure
+   * (`EpochSession.hasFailed()` — top-tree/submit failed with every prover healthy, so definitively not
+   * a prune). A `stopped` session (a prover under it failed) is not uploaded; it recovers on re-add.
    */
   protected createSessionManager(publishingService: ProofPublishingService): SessionManager {
     return new SessionManager({
@@ -578,7 +582,7 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra
         finalizationDelayMs: this.config.proverNodeEpochProvingDelayMs,
       },
       onSessionFailed: async session => {
-        await this.tryUploadSessionFailure(session);
+        await this.tryUploadEpochFailure(session.getId(), session.getCheckpoints());
       },
       bindings: this.log.getBindings(),
     });
@@ -596,15 +600,32 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra
     this.sessionManager.setSessionHooks(hooks);
   }
 
-  /** Uploads failure snapshots when sessions exit with `failed`. Exposed as a method so tests can spy on it. */
-  public async tryUploadSessionFailure(session: EpochSession): Promise<string | undefined> {
-    if (!this.config.proverNodeFailedEpochStore) {
+  /**
+   * Installs checkpoint-prover test hooks (e.g. forcing a sub-tree failure) applied to every
+   * CheckpointProver constructed after this call. For the e2e harness only.
+   */
+  public setCheckpointHooks(hooks: CheckpointProverTestHooks): void {
+    this.checkpointStore.setTestHooks(hooks);
+  }
+
+  /**
+   * Uploads a post-mortem snapshot for an epoch whose full session failed to prove, built from that
+   * session's checkpoint provers. Fired from the session manager's `onSessionFailed` callback (a
+   * genuine, race-free failure). Exposed as a method so tests can spy on it. No-ops if no failed-epoch
+   * store is configured or the checkpoint set is empty.
+   */
+  public async tryUploadEpochFailure(
+    id: string,
+    checkpoints: readonly CheckpointProver[],
+  ): Promise<string | undefined> {
+    if (!this.config.proverNodeFailedEpochStore || checkpoints.length === 0) {
       return undefined;
     }
-    const data = SessionManager.buildSessionProvingData(session);
+    const data = SessionManager.buildProvingData(checkpoints);
     return await uploadEpochProofFailure(
       this.config.proverNodeFailedEpochStore,
-      session.getId(),
+      // The session's own id; `uploadEpochProofFailure` already prefixes the path with the epoch number.
+      id,
       data,
       this.l2BlockSource as Archiver,
       this.worldState,
@@ -613,8 +634,62 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra
     );
   }
 
+  /**
+   * Uploads a post-mortem for a single failed checkpoint prover, built from just that checkpoint's
+   * proving data. Fired (fire-and-forget) from the store's `onFailed` callback for any non-cancel
+   * block-proof failure — a genuine sub-tree fault or a prune-induced fork fault alike. No-ops if no
+   * failed-epoch store is configured, or if the checkpoint is no longer canonical (a prune left nothing
+   * to diagnose). Swallows its own errors so a fire-and-forget caller can't leak.
+   */
+  public async tryUploadCheckpointFailure(prover: CheckpointProver): Promise<string | undefined> {
+    if (!this.config.proverNodeFailedEpochStore) {
+      return undefined;
+    }
+    try {
+      // A prune-induced fork fault and a genuine sub-tree failure are indistinguishable at the moment the
+      // prover rejects (no control-plane cancel has landed yet). But the archiver is the authoritative
+      // committed chain: if this checkpoint was pruned out, its last block is no longer canonical there.
+      // Only upload for a checkpoint that still exists on-chain — a prune leaves nothing to diagnose, and
+      // the snapshot (full world-state + archiver) is expensive to produce and store.
+      if (!(await this.isCheckpointCanonical(prover.checkpoint))) {
+        this.log.debug(`Skipping checkpoint-failure upload for ${prover.id}: no longer canonical (pruned)`, {
+          checkpointNumber: prover.checkpoint.number,
+        });
+        return undefined;
+      }
+      const data = SessionManager.buildProvingData([prover]);
+      return await uploadEpochProofFailure(
+        this.config.proverNodeFailedEpochStore,
+        // The prover's content-addressed id; the epoch number is already in the upload path.
+        prover.id,
+        data,
+        this.l2BlockSource as Archiver,
+        this.worldState,
+        assertRequired(pick(this.config, 'l1ChainId', 'rollupVersion', 'dataDirectory')),
+        this.log,
+      );
+    } catch (err) {
+      this.log.error(`Error uploading checkpoint failure for ${prover.id}`, err);
+      return undefined;
+    }
+  }
+
   // ---------------- helpers ----------------
 
+  /**
+   * True if the checkpoint still exists on the canonical chain: the archiver holds a block at its last
+   * block's height whose archive root matches. A prune (fork fault) leaves the block missing or replaced,
+   * so this returns false. Protected for direct unit-test access.
+   */
+  protected async isCheckpointCanonical(checkpoint: Checkpoint): Promise<boolean> {
+    const lastBlock = checkpoint.blocks.at(-1);
+    if (!lastBlock) {
+      return false;
+    }
+    const onChain = await this.l2BlockSource.getBlock({ number: lastBlock.number });
+    return !!onChain && onChain.archive.root.equals(checkpoint.archive.root);
+  }
+
   @memoize
   private getL1Constants(): Promise<L1RollupConstants> {
     return this.l2BlockSource.getL1Constants();
diff --git a/yarn-project/prover-node/src/session-manager.test.ts b/yarn-project/prover-node/src/session-manager.test.ts
index dde4d6c793d8..c03041499feb 100644
--- a/yarn-project/prover-node/src/session-manager.test.ts
+++ b/yarn-project/prover-node/src/session-manager.test.ts
@@ -245,57 +245,147 @@ describe('SessionManager', () => {
     expect(stubs.length).toBe(1);
   });
 
-  it('onTick does not retry an epoch whose session already terminated', async () => {
-    // The tick attempts each epoch at most once; a failed proving attempt must not be
-    // resubmitted by a later tick (only a new checkpoint event reopens it). Without the
-    // high-water mark the reaped session would be reopened, resubmitting the proof.
+  it('skips opening a full session while a checkpoint prover in the set has failed', async () => {
+    // A checkpoint prover fault (a sub-tree fault or a prune-induced fork fault) marks the checkpoint prover failed.
+    // A session over it can never produce its block proofs, so openFullSessionIfReady skips it — on both
+    // the tick and checkpoint triggers, cheaply and with no bookkeeping. It stays skipped until a
+    // prune/re-add replaces the failed checkpoint (see the recovery tests below) or the epoch expires.
     mockNextUnprovenSlot(2, 6);
-    const provers = [proverForCheckpoint(1, 6)];
     l2BlockSource.isEpochComplete.mockResolvedValue(true);
     l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]);
-    store.listInSlotRange.mockReturnValue(provers);
+    store.listInSlotRange.mockReturnValue([failedProverForCheckpoint(1, 6)]);
+
+    await manager.onTick();
+    await manager.onTick();
+    await manager.onCheckpointAdded(EpochNumber(3));
+
+    expect(stubs.length).toBe(0);
+    expect(manager.getFullSession(EpochNumber(3))).toBeUndefined();
+  });
+
+  it('retains a full session that failed on its own account, uploads once, and does not re-prove it', async () => {
+    // A session-level failure (top-tree prove or L1 submit) with every checkpoint prover healthy ends the
+    // session in 'failed'. Because healthy checkpoint provers rule out a prune, this is a race free failure:
+    // it uploads a post-mortem exactly once, and the failed session is retained so the tick does not
+    // re-prove a deterministically-failing epoch.
+    mockNextUnprovenSlot(2, 6);
+    l2BlockSource.isEpochComplete.mockResolvedValue(true);
+    l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]);
+    store.listInSlotRange.mockReturnValue([proverForCheckpoint(1, 6)]);
+
+    await manager.onTick();
+    expect(stubs.length).toBe(1);
+    const failed = stubs[0];
+    failed.terminate('failed');
+    await flushSessionCompletion();
+
+    // Uploaded exactly once, for this session.
+    expect(sessionFailures).toHaveLength(1);
+    expect(sessionFailures[0]).toBe(failed as unknown as EpochSession);
 
+    // Retained (not deleted) and never re-proved by the tick.
+    await manager.onTick();
     await manager.onTick();
     expect(stubs.length).toBe(1);
+    expect(manager.getFullSession(EpochNumber(3))).toBe(failed as unknown as EpochSession);
+    expect(sessionFailures).toHaveLength(1); // still just the one upload
+  });
+
+  it('replaces a retained failed session when its canonical content changes (re-add)', async () => {
+    // The one way a retained failed session is retried: its content changes. A re-add over new content
+    // replaces the marker with a fresh session so the epoch can be proven over the new provers.
+    mockNextUnprovenSlot(2, 6);
+    l2BlockSource.isEpochComplete.mockResolvedValue(true);
+    l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]);
+    store.listInSlotRange.mockReturnValue([proverForCheckpoint(1, 6)]);
 
-    // Session fails. Proven height has not advanced, so the next tick reaps the failed
-    // session via recreateInvalidSessions (always called in reconcile) but the
-    // lastTickEpoch high-water mark prevents resubmission.
+    await manager.onTick();
     stubs[0].terminate('failed');
+    await flushSessionCompletion();
+
+    // Content changes (a re-add at a different slot in the epoch) → the retained session is replaced.
+    store.listInSlotRange.mockReturnValue([proverForCheckpoint(2, 7)]);
     await manager.onTick();
-    expect(manager.getFullSession(EpochNumber(3))).toBeUndefined();
+
+    const replacement = manager.getFullSession(EpochNumber(3)) as unknown as StubSession | undefined;
+    expect(replacement).toBeDefined();
+    expect(replacement).not.toBe(stubs[0]);
+    expect(replacement!.isTerminal()).toBe(false);
+    expect(replacement!.provers.map(p => p.id)).toEqual([proverForCheckpoint(2, 7).id]);
+    expect(stubs.length).toBe(2);
+  });
+
+  it('does not upload when a session stops because a checkpoint prover failed', async () => {
+    // A 'stopped' session (a prover under it failed — possibly a prune) is not the session's own failure,
+    // so no post-mortem is uploaded (contrast with a 'failed' session, which uploads).
+    mockNextUnprovenSlot(2, 6);
+    l2BlockSource.isEpochComplete.mockResolvedValue(true);
+    l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]);
+    store.listInSlotRange.mockReturnValue([proverForCheckpoint(1, 6)]);
+
+    await manager.onTick();
+    expect(stubs.length).toBe(1);
+    stubs[0].terminate('stopped');
+    await flushSessionCompletion();
+
+    expect(sessionFailures).toEqual([]);
+  });
+
+  it('does not churn sessions after a stopped session while the failed prover remains canonical', async () => {
+    const epoch = EpochNumber(3);
+    mockNextUnprovenSlot(2, 6);
+    l2BlockSource.isEpochComplete.mockResolvedValue(true);
+    l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]);
+    store.listInSlotRange.mockReturnValue([proverForCheckpoint(1, 6)]);
+
+    await manager.onTick();
+    expect(stubs.length).toBe(1);
+
+    stubs[0].terminate('stopped');
+    await flushSessionCompletion();
+
+    // The same canonical prover is now sticky-failed. Reconcile should delete the terminal stopped
+    // session, but the failed-prover guard must prevent constructing/proving/deleting a new one on
+    // repeated triggers.
+    store.listInSlotRange.mockReturnValue([failedProverForCheckpoint(1, 6)]);
+
+    await manager.onTick();
+    expect(manager.getFullSession(epoch)).toBeUndefined();
+    expect(stubs.length).toBe(1);
+
+    await manager.onTick();
+    await manager.onCheckpointAdded(epoch);
+    await manager.onPrune([epoch]);
+
+    expect(manager.getFullSession(epoch)).toBeUndefined();
     expect(stubs.length).toBe(1);
+    expect(sessionFailures).toEqual([]);
   });
 
-  it('onTick does not re-attempt a failed epoch, but a checkpoint re-add recovers it', async () => {
-    // Pins the invariant: a genuinely-failed epoch is never re-attempted by the periodic tick
-    // (gated by lastTickEpoch), yet the SAME epoch is recovered when a chain event fires, because
-    // checkpoint/prune triggers reach openFullSessionIfReady ungated. This is how a failed proof is
-    // not resubmitted on a loop while a pruned-then-re-added epoch still gets reproven.
+  it('onTick does not reopen an epoch once the proven height advances past it', async () => {
+    // Once the epoch is proven, the proven tip advances so nextUnprovenEpoch moves on and the tick no
+    // longer selects the proven epoch.
     mockNextUnprovenSlot(2, 6); // proven tip block 2 → first unproven slot 6 → epoch 3
     const provers = [proverForCheckpoint(1, 6)];
     l2BlockSource.isEpochComplete.mockResolvedValue(true);
     l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]);
     store.listInSlotRange.mockReturnValue(provers);
 
-    // Tick opens the session; it fails. lastTickEpoch is now pinned at epoch 3.
     await manager.onTick();
     expect(stubs.length).toBe(1);
-    stubs[0].terminate('failed');
+    stubs[0].terminate('completed');
     await flushSessionCompletion();
 
-    // Further ticks must NOT re-attempt epoch 3 — the gate holds, no new session is constructed.
+    // Proven height jumps past epoch 3's blocks: the next unproven block is now in a later epoch.
+    mockNextUnprovenSlot(8, 8); // epoch 4
+    l2BlockSource.getCheckpoints.mockResolvedValue([]); // epoch 4 has no canonical content yet
+    store.listInSlotRange.mockReturnValue([]);
+
     await manager.onTick();
     await manager.onTick();
-    expect(stubs.length).toBe(1);
+    // No new session for epoch 3 — it is proven; nothing opened for the empty epoch 4 either.
     expect(manager.getFullSession(EpochNumber(3))).toBeUndefined();
-
-    // A checkpoint (re-)added to the epoch reaches openFullSessionIfReady ungated → fresh session.
-    await manager.onCheckpointAdded(EpochNumber(3));
-    const recreated = manager.getFullSession(EpochNumber(3)) as unknown as StubSession | undefined;
-    expect(recreated).toBeDefined();
-    expect(recreated!.isTerminal()).toBe(false);
-    expect(stubs.length).toBe(2);
+    expect(stubs.length).toBe(1);
   });
 
   it('onTick keeps retrying the same epoch while a transient blocker prevents opening', async () => {
@@ -389,54 +479,66 @@ describe('SessionManager', () => {
     expect(stubs.length).toBe(2);
   });
 
-  it('reopens an epoch whose session failed once its checkpoints are re-added', async () => {
-    // A prune can fault a checkpoint mid-proof before it is reconciled, driving the session to
-    // `failed`. That terminal state must not strand the epoch — a re-add of its checkpoints reopens a
-    // fresh, live session (via the checkpoint trigger, ungated by the tick high-water mark).
+  it('data-plane fault then identical-content re-add: opens over the fresh prover and completes', async () => {
+    // A checkpoint prover faults mid-proof from a data-plane reorg — its blockProofs reject, so it is
+    // marked failed and no session is built over it. A prune+re-add then replaces it with a FRESH prover
+    // of identical content (same content-addressed id); the next open builds a session that completes.
+    // Opening over the same content id is what lets the content-addressed broker reuse the already-
+    // completed sub-proofs — see checkpoint-store.test.ts for the reuse itself.
     const epoch = EpochNumber(3);
-    const prover = proverForCheckpoint(1, 6);
-    await openCanonicalFullSession(epoch, [prover]);
-    const original = stubs[0];
+    const failed = failedProverForCheckpoint(1, 6);
+    l2BlockSource.isEpochComplete.mockResolvedValue(true);
+    l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]);
+    store.listInSlotRange.mockReturnValue([failed]);
 
-    original.terminate('failed');
-    await flushSessionCompletion();
-    expect(original.state).toBe('failed');
+    await manager.onCheckpointAdded(epoch);
+    expect(stubs.length).toBe(0); // failed prover ⇒ no session
 
-    await openCanonicalFullSession(epoch, [prover]);
-    const recreated = manager.getFullSession(epoch) as unknown as StubSession | undefined;
-    expect(recreated).toBeDefined();
-    expect(recreated).not.toBe(original);
-    expect(recreated!.uuid).not.toBe(original.uuid);
-    expect(recreated!.isTerminal()).toBe(false);
-    expect(stubs.length).toBe(2);
-  });
+    // Re-add replaces the failed prover with a fresh, healthy prover of identical content id.
+    const fresh = proverForCheckpoint(1, 6);
+    expect(fresh.id).toBe(failed.id);
+    store.listInSlotRange.mockReturnValue([fresh]);
+    await manager.onCheckpointAdded(epoch);
 
-  it('uploads a post-mortem for a genuine proving failure (canonical content unchanged)', async () => {
-    const epoch = EpochNumber(3);
-    const prover = proverForCheckpoint(1, 6);
-    await openCanonicalFullSession(epoch, [prover]);
-    const session = manager.getFullSession(epoch);
-    // The session's checkpoints still match canonical → the failure was genuine, not a prune.
-    store.listInSlotRange.mockReturnValue([prover]);
+    const session = manager.getFullSession(epoch) as unknown as StubSession | undefined;
+    expect(session).toBeDefined();
+    expect(session!.provers.map(p => p.id)).toEqual([fresh.id]);
+    expect(session!.isTerminal()).toBe(false);
+    expect(stubs.length).toBe(1);
 
-    stubs[0].terminate('failed');
+    session!.terminate('completed');
     await flushSessionCompletion();
-
-    expect(sessionFailures).toEqual([session]);
+    expect(session!.state).toBe('completed');
   });
 
-  it('skips the failure upload when the failure coincides with a canonical content change', async () => {
+  it('data-plane fault then different-content re-add: opens over the new prover and completes', async () => {
+    // Same data-plane fault, but the reorg replaces the epoch's content: the re-added checkpoint has a
+    // different content-addressed id (reflected on both the archiver and the store). The epoch opens
+    // over the NEW prover (nothing to reuse) and completes.
     const epoch = EpochNumber(3);
-    const prover = proverForCheckpoint(1, 6);
-    await openCanonicalFullSession(epoch, [prover]);
-    // A prune removed the checkpoint from the store, so the session's content no longer matches
-    // canonical — the failure was caused by the content changing under it, not a proving fault.
-    store.listInSlotRange.mockReturnValue([]);
+    const failed = failedProverForCheckpoint(1, 6);
+    l2BlockSource.isEpochComplete.mockResolvedValue(true);
+    l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]);
+    store.listInSlotRange.mockReturnValue([failed]);
 
-    stubs[0].terminate('failed');
-    await flushSessionCompletion();
+    await manager.onCheckpointAdded(epoch);
+    expect(stubs.length).toBe(0);
 
-    expect(sessionFailures).toEqual([]);
+    const fresh = proverForCheckpoint(2, 7);
+    expect(fresh.id).not.toBe(failed.id);
+    l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(2, 7)]);
+    store.listInSlotRange.mockReturnValue([fresh]);
+    await manager.onCheckpointAdded(epoch);
+
+    const session = manager.getFullSession(epoch) as unknown as StubSession | undefined;
+    expect(session).toBeDefined();
+    expect(session!.provers.map(p => p.id)).toEqual([fresh.id]);
+    expect(session!.isTerminal()).toBe(false);
+    expect(stubs.length).toBe(1);
+
+    session!.terminate('completed');
+    await flushSessionCompletion();
+    expect(session!.state).toBe('completed');
   });
 
   it('drops terminal sessions on the next reconcile', async () => {
@@ -542,7 +644,7 @@ describe('SessionManager', () => {
     const canonical = [proverForCheckpoint(1, 14)];
     await openCanonicalFullSession(epoch, canonical);
     const terminalFull = stubs[0];
-    terminalFull.terminate('failed');
+    terminalFull.terminate('stopped');
     expect(terminalFull.isTerminal()).toBe(true);
 
     store.listForEpoch.mockResolvedValue(canonical);
@@ -571,7 +673,7 @@ describe('SessionManager', () => {
     const firstPromise = awaitNextStub();
     const firstStart = manager.startProof(epoch);
     const firstPartial = await firstPromise;
-    firstPartial.terminate('failed');
+    firstPartial.terminate('stopped');
     await firstStart;
     expect(firstPartial.isTerminal()).toBe(true);
 
@@ -627,6 +729,20 @@ describe('SessionManager', () => {
     await expect(manager.startProof(EpochNumber(7))).rejects.toThrow(/No blocks found/);
   });
 
+  it('startProof does not construct a partial session over a failed canonical prover', async () => {
+    const epoch = EpochNumber(7);
+    const failed = failedProverForCheckpoint(1, 14);
+    store.listForEpoch.mockResolvedValue([failed]);
+    store.listInSlotRange.mockReturnValue([failed]);
+
+    await expect(manager.startProof(epoch)).rejects.toThrow(/Failed to schedule partial proof/);
+    await expect(manager.startProof(epoch)).rejects.toThrow(/Failed to schedule partial proof/);
+
+    expect(stubs).toHaveLength(0);
+    expect(manager.getJobs()).toEqual([]);
+    expect(sessionFailures).toEqual([]);
+  });
+
   it('startProof refuses to re-prove an epoch the proven chain already encompasses', async () => {
     const epoch = EpochNumber(7);
     // proverForCheckpoint builds a checkpoint whose single block number equals the checkpoint
@@ -838,8 +954,10 @@ type StubSession = {
   getId(): string;
   getState(): EpochProvingJobState;
   getEpochNumber(): EpochNumber;
+  getKind(): SessionSpec['kind'];
   getCheckpoints(): readonly CheckpointProver[];
   isTerminal(): boolean;
+  hasFailed(): boolean;
   cancel(reason?: string, opts?: { abortJobs?: boolean }): Promise<void>;
   start(): Promise<EpochProvingJobState>;
   whenDone(): Promise<EpochProvingJobState>;
@@ -876,6 +994,9 @@ function makeStubSession(spec: SessionSpec, provers: readonly CheckpointProver[]
     getEpochNumber() {
       return this.spec.epochNumber;
     },
+    getKind() {
+      return this.spec.kind;
+    },
     getCheckpoints() {
       return this.provers;
     },
@@ -890,6 +1011,9 @@ function makeStubSession(spec: SessionSpec, provers: readonly CheckpointProver[]
       ];
       return terminal.includes(this.state);
     },
+    hasFailed() {
+      return this.state === 'failed';
+    },
     async cancel(reason?: string, opts?: { abortJobs?: boolean }) {
       this.cancelReasons.push(reason ?? 'cancelled');
       this.cancelAbortJobs.push(opts?.abortJobs ?? true);
@@ -924,16 +1048,22 @@ function makeCheckpointContent(number: number, slot: number) {
   } as any;
 }
 
-function proverForCheckpoint(number: number, slot: number): CheckpointProver {
+function proverForCheckpoint(number: number, slot: number, failed = false): CheckpointProver {
   const checkpoint = makeCheckpointContent(number, slot);
   return {
     id: CheckpointProver.idFor(checkpoint),
     checkpoint,
     slotNumber: SlotNumber(slot),
     isCancelled: () => false,
+    isFailed: () => failed,
   } as unknown as CheckpointProver;
 }
 
+/** A checkpoint prover whose block proofs have failed (a sub-tree/fork fault). Same content id as the healthy one. */
+function failedProverForCheckpoint(number: number, slot: number): CheckpointProver {
+  return proverForCheckpoint(number, slot, true);
+}
+
 /** Archiver-side PublishedCheckpoint stub whose content matches `proverForCheckpoint(number, slot)`. */
 function archiverCp(number: number, slot: number) {
   return { checkpoint: makeCheckpointContent(number, slot) } as any;
diff --git a/yarn-project/prover-node/src/session-manager.ts b/yarn-project/prover-node/src/session-manager.ts
index e6721f921951..3a42a59e74dc 100644
--- a/yarn-project/prover-node/src/session-manager.ts
+++ b/yarn-project/prover-node/src/session-manager.ts
@@ -60,8 +60,9 @@ export type SessionManagerDeps = {
   dateProvider: DateProvider;
   config: SessionManagerConfig;
   /**
-   * Optional callback fired when a session terminates with `failed`. The session manager
-   * doesn't own the failure-upload action; it just notifies the owner.
+   * Fired once when a full session ends in its own genuine failure (`EpochSession.hasFailed()` — top-tree
+   * or submit failed with every prover healthy). The owner uploads a post-mortem here. Not fired for a
+   * `stopped` session (a prover under it failed — possibly a prune), which is recovered on re-add instead.
    */
   onSessionFailed?: (session: EpochSession) => Promise<void>;
   bindings?: LoggerBindings;
@@ -88,15 +89,6 @@ export class SessionManager {
   private readonly reconcileQueue = new SerialQueue();
   /** Cached L1 constants, populated on first read. */
   private cachedL1Constants: L1RollupConstants | undefined;
-  /**
-   * Highest epoch for which the periodic tick has successfully created a full session.
-   * Monotonic high-water mark: once the tick observes a session for epoch X, it stops
-   * trying to open one — even if that session subsequently fails (only a new checkpoint
-   * event reopens it). Crucially, the mark only advances when a session actually exists
-   * post-open, so transient blockers (atMaxSessionLimit, archiver still indexing) leave
-   * the mark in place and the next tick retries.
-   */
-  private lastTickEpoch: EpochNumber | undefined;
   /** Test-only hooks applied to every session this manager constructs. */
   private sessionHooks: EpochSessionHooks | undefined;
   /** Periodic tick that nudges reconcile to pick up newly-complete epochs. Started by `start()`. */
@@ -248,17 +240,6 @@ export class SessionManager {
       await this.openFullSessionIfReady(epoch);
     }
 
-    // Advance the tick high-water mark only once a session actually exists for the epoch.
-    // `openFullSessionIfReady` can early-return without creating one (atMaxSessionLimit,
-    // archiver still indexing, etc.); in those cases we want the next tick to try again
-    // rather than skip the epoch forever.
-    if (trigger.kind === 'tick' && implicatedEpochs.length === 1) {
-      const epoch = implicatedEpochs[0];
-      if (this.fullSessions.has(epoch)) {
-        this.lastTickEpoch = epoch;
-      }
-    }
-
     if (trigger.kind === 'start-proof') {
       this.openPartialSession(trigger.spec);
     }
@@ -266,15 +247,30 @@ export class SessionManager {
 
   private recreateInvalidSessions(): void {
     for (const [key, session] of Array.from(this.fullSessions.entries())) {
+      const canonical = this.checkpointsForSpec(session.getSpec());
+      const contentChanged = !this.checkpointsMatch(session.getCheckpoints(), canonical);
+
       if (session.isTerminal()) {
+        // A full session that failed on its own account is retained as a "do not re-prove" marker while
+        // its content is unchanged — this is what stops the tick re-proving a deterministically-failing
+        // epoch. When the content changes (a re-add), it is replaced so the epoch retries over the new
+        // provers. Any other terminal full session is simply dropped.
+        if (session.hasFailed() && !contentChanged) {
+          continue;
+        }
         this.fullSessions.delete(key);
+        if (contentChanged && this.canBuildOver(canonical)) {
+          const newSession = this.constructSession(session.getSpec(), canonical);
+          this.fullSessions.set(key, newSession);
+          void this.runSession(newSession);
+        }
         continue;
       }
-      const canonical = this.checkpointsForSpec(session.getSpec());
-      if (!this.checkpointsMatch(session.getCheckpoints(), canonical)) {
+
+      if (contentChanged) {
         this.fireAndForgetCancel(session, 'canonical content changed');
         this.fullSessions.delete(key);
-        if (canonical.length > 0) {
+        if (this.canBuildOver(canonical)) {
           const newSession = this.constructSession(session.getSpec(), canonical);
           this.fullSessions.set(key, newSession);
           void this.runSession(newSession);
@@ -290,7 +286,7 @@ export class SessionManager {
       if (!this.checkpointsMatch(session.getCheckpoints(), canonical)) {
         this.fireAndForgetCancel(session, 'canonical content changed');
         this.partialSessions.delete(key);
-        if (canonical.length > 0) {
+        if (this.canBuildOver(canonical)) {
           const newSession = this.constructSession(session.getSpec(), canonical);
           this.partialSessions.set(key, newSession);
           void this.runSession(newSession);
@@ -299,9 +295,15 @@ export class SessionManager {
     }
   }
 
+  /** A session may be built over a checkpoint set only when it is non-empty and contains no failed prover. */
+  private canBuildOver(canonical: readonly CheckpointProver[]): boolean {
+    return canonical.length > 0 && !this.hasFailedProver(canonical);
+  }
+
   private async openFullSessionIfReady(epoch: EpochNumber): Promise<void> {
-    // `recreateInvalidSessions` runs at the top of every reconcile and deletes terminal sessions
-    // before this is called, so a session present here is live and already covers the epoch.
+    // A session present here already covers the epoch: either live, or a retained genuinely-failed
+    // session kept by `recreateInvalidSessions` as a "do not re-prove" marker. Either way, don't open
+    // another — the retained-failed one is replaced only when its canonical content changes.
     if (this.fullSessions.has(epoch)) {
       return;
     }
@@ -326,6 +328,13 @@ export class SessionManager {
       });
       return;
     }
+    if (this.hasFailedProver(canonical)) {
+      // A checkpoint prover in the set has failed (a sub-tree fault or a prune-induced fork fault), so a
+      // session over it would fail immediately. Don't re-create it every tick — it recovers when a
+      // prune/re-add replaces the failed prover with a fresh one, or fails for good at expiry.
+      this.log.debug(`Skipping full-session open for epoch ${epoch}: a checkpoint prover has failed`, { epoch });
+      return;
+    }
     const spec: SessionSpec = { kind: 'full', epochNumber: epoch, fromSlot, toSlot };
     const session = this.constructSession(spec, canonical);
     this.fullSessions.set(epoch, session);
@@ -334,7 +343,7 @@ export class SessionManager {
 
   private openPartialSession(spec: SessionSpec): void {
     const canonical = this.deps.checkpointStore.listInSlotRange(spec.fromSlot, spec.toSlot);
-    if (canonical.length === 0) {
+    if (canonical.length === 0 || this.hasFailedProver(canonical)) {
       return;
     }
     // Reuse a live partial session for this epoch whose checkpoint set already matches the
@@ -406,33 +415,30 @@ export class SessionManager {
     }
     const state = await session.start();
     this.log.info(`Session ${session.getId()} exited with state ${state}`);
-    if (state === 'failed' && this.deps.onSessionFailed) {
-      // Best-effort suppression of the spurious post-mortem upload a prune produces: if the session's
-      // checkpoints no longer match the store's current set, the failure was caused by the content
-      // changing under it, not a genuine proving fault, so skip the upload. This is inherently racy —
-      // the store lags the world-state unwind, so a fault observed before the prune is reconciled here
-      // still uploads. The epoch is recovered regardless by recreating the session on re-add.
-      if (!this.checkpointsMatch(session.getCheckpoints(), this.checkpointsForSpec(session.getSpec()))) {
-        this.log.info(`Skipping failure upload for session ${session.getId()}: canonical content changed`, {
-          ...session.getSpec(),
-        });
-        return;
-      }
+
+    // A full session that failed on its own account (top-tree/submit failed with every prover healthy)
+    // is a genuine, race-free failure: upload its post-mortem once. `recreateInvalidSessions` retains
+    // the terminal session so this fires exactly once (it is never re-run over the same content). A
+    // `stopped` session (a prover under it failed) is not uploaded — it may be a prune, and recovers on
+    // re-add.
+    if (session.getKind() === 'full' && session.hasFailed() && this.deps.onSessionFailed) {
       try {
         await this.deps.onSessionFailed(session);
       } catch (err) {
-        this.log.error(`Error in onSessionFailed callback for ${session.getSpec().epochNumber}`, err);
+        this.log.error(`Error in onSessionFailed callback for epoch ${session.getEpochNumber()}`, err);
       }
     }
   }
 
   /**
-   * Builds the EpochProvingJobData snapshot for failure upload. Includes every checkpoint
-   * referenced by the session, regardless of whether sub-tree proving completed —
-   * partial state is still useful for post-mortem analysis.
+   * Builds the EpochProvingJobData snapshot for a post-mortem failure upload from a set of
+   * checkpoint provers. Includes every checkpoint regardless of whether sub-tree proving
+   * completed — partial state is still useful for post-mortem analysis.
    */
-  public static buildSessionProvingData(session: EpochSession): EpochProvingJobData {
-    const checkpoints = session.getCheckpoints();
+  public static buildProvingData(checkpoints: readonly CheckpointProver[]): EpochProvingJobData {
+    if (checkpoints.length === 0) {
+      throw new Error('Cannot build proving data from an empty checkpoint set');
+    }
     const txs = new Map();
     const l1ToL2Messages: Record<number, Fr[]> = {};
     for (const c of checkpoints) {
@@ -442,7 +448,7 @@ export class SessionManager {
       l1ToL2Messages[c.checkpoint.number] = c.l1ToL2Messages;
     }
     return {
-      epochNumber: session.getSpec().epochNumber,
+      epochNumber: checkpoints[0].epochNumber,
       checkpoints: checkpoints.map(c => c.checkpoint),
       txs,
       l1ToL2Messages,
@@ -465,20 +471,12 @@ export class SessionManager {
   /**
    * Maps a reconcile trigger to the epochs whose full session should be (re)opened.
    *
-   * This is where the "don't retry a genuinely-failed epoch, but do recover a pruned one" invariant
-   * lives — enforced by which triggers are gated by `lastTickEpoch`:
-   *
-   * - The periodic `tick` IS gated: once a tick has opened a session for an epoch, `lastTickEpoch`
-   *   advances to it and later ticks skip it (`epoch <= lastTickEpoch`). So a failed attempt is never
-   *   resubmitted on a loop by the tick.
-   * - `checkpoint` and `prune` are deliberately NOT gated. They only fire when the epoch's canonical
-   *   content actually changes — a checkpoint arrives, or a reorg prunes/replaces one — which is
-   *   exactly when re-attempting is correct.
-   *
-   * A genuine proving failure produces no content change, hence no checkpoint/prune event, so only
-   * the gated tick could reopen it — and it won't. A prune + re-add fires ungated events, so the
-   * epoch is reopened through this path (and `openFullSessionIfReady` rebuilds over the fresh
-   * provers). See the "onTick does not retry ... but recovers ... re-added" test.
+   * The periodic `tick` returns the next unproven epoch every time; it does not track prior attempts.
+   * `openFullSessionIfReady` is what keeps this from re-proving a doomed epoch: it refuses to build a
+   * session when any checkpoint prover in the set has failed, so a stuck epoch is cheaply skipped each
+   * tick rather than re-proved. `checkpoint` and `prune` fire when an epoch's canonical content changes
+   * (a checkpoint arrives, or a reorg prunes/replaces one) — which is what installs a fresh prover in
+   * place of a failed one, letting the next open succeed and recovering a pruned-then-re-added epoch.
    */
   private async epochsForTrigger(trigger: ReconcileTrigger): Promise<EpochNumber[]> {
     switch (trigger.kind) {
@@ -488,10 +486,7 @@ export class SessionManager {
         return trigger.affectedEpochs;
       case 'tick': {
         const epoch = await this.nextUnprovenEpoch();
-        if (epoch === undefined || (this.lastTickEpoch !== undefined && epoch <= this.lastTickEpoch)) {
-          return [];
-        }
-        return [epoch];
+        return epoch === undefined ? [] : [epoch];
       }
       case 'start-proof':
         return [];
@@ -518,6 +513,15 @@ export class SessionManager {
     return this.deps.checkpointStore.listInSlotRange(spec.fromSlot, spec.toSlot);
   }
 
+  /**
+   * True if any prover in the set has failed. The epoch cannot be proven over a failed prover (it can
+   * never produce its block proofs), so a session must not be built or rebuilt over it until a prune/re-add
+   * has replaced it with a fresh prover.
+   */
+  private hasFailedProver(checkpoints: readonly CheckpointProver[]): boolean {
+    return checkpoints.some(c => c.isFailed());
+  }
+
   private fireAndForgetCancel(session: EpochSession, reason: string): void {
     void session.cancel(reason).catch(err => this.log.warn(`Error cancelling session ${session.getId()}`, err));
   }