diff --git a/yarn-project/archiver/src/archiver-sync.test.ts b/yarn-project/archiver/src/archiver-sync.test.ts
index 978b628b6daa..3e60c0e6fc3f 100644
--- a/yarn-project/archiver/src/archiver-sync.test.ts
+++ b/yarn-project/archiver/src/archiver-sync.test.ts
@@ -27,6 +27,7 @@ import type { L1RollupConstants } from '@aztec/stdlib/epoch-helpers';
 import { CheckpointHeader } from '@aztec/stdlib/rollup';
 import { mockCheckpointAndMessages } from '@aztec/stdlib/testing';
 import { ConsensusTimetable } from '@aztec/stdlib/timetable';
+import { AppendOnlyTreeSnapshot } from '@aztec/stdlib/trees';
 import { BlockHeader } from '@aztec/stdlib/tx';
 import { getTelemetryClient } from '@aztec/telemetry-client';
 
@@ -2165,6 +2166,193 @@ describe('Archiver Sync', () => {
     });
   });
 
+  describe('pruning proposed blocks that consumed rolled back messages', () => {
+    let pruneSpy: jest.Mock;
+
+    // L1 is held here for the whole test: a reorg replaces the head at the same height, so the sync still runs,
+    // while the slot of the proposed blocks is never in the past and neither slot-based prune fires.
+    const l1BlockNumber = 110n;
+
+    beforeEach(() => {
+      pruneSpy = jest.fn();
+      archiver.events.on(L2BlockSourceEvents.L2PruneUncheckpointed, pruneSpy);
+    });
+
+    afterEach(() => {
+      archiver.events.off(L2BlockSourceEvents.L2PruneUncheckpointed, pruneSpy);
+    });
+
+    // Two messages on L1 block 100 (indices 0 and 1) and two on block 102 (indices 2 and 3), so a reorg of block
+    // 102 removes everything from index 2 on.
+    const addMessages = () => {
+      const early = [Fr.random(), Fr.random()];
+      const late = [Fr.random(), Fr.random()];
+      fake.addMessages(CheckpointNumber(1), 100n, early);
+      fake.addMessages(CheckpointNumber(1), 102n, late);
+      return { early, late };
+    };
+
+    // Builds a chain of blocks whose L1-to-L2 tree leaf counts are the given cumulative totals of consumed messages.
+    const makeBlocksConsumingThrough = async (leafCounts: number[], builtAtL1Block = l1BlockNumber) => {
+      const blocks = await fake.makeBlocks(CheckpointNumber(1), {
+        numBlocks: leafCounts.length,
+        txsPerBlock: 1,
+        l1BlockNumber: builtAtL1Block,
+      });
+      blocks.forEach((block, i) => {
+        block.header.state.l1ToL2MessageTree = new AppendOnlyTreeSnapshot(Fr.random(), leafCounts[i]);
+      });
+      return blocks;
+    };
+
+    it('prunes from the first proposed block that consumed a removed message', async () => {
+      const { early } = addMessages();
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+
+      // Blocks 1 and 2 consumed only the two surviving messages (block 2 consumed none of its own), block 3
+      // consumed the two that the reorg removes, and block 4 consumed nothing but builds on block 3.
+      const blocks = await makeBlocksConsumingThrough([2, 2, 4, 4]);
+      for (const block of blocks) {
+        await archiver.addBlock(block);
+      }
+      await archiver.addProposedCheckpoint({
+        checkpointNumber: CheckpointNumber(1),
+        header: CheckpointHeader.empty({ slotNumber: fake.getL2SlotAtL1Block(l1BlockNumber) }),
+        startBlock: BlockNumber(1),
+        blockCount: blocks.length,
+        totalManaUsed: 0n,
+        feeAssetPriceModifier: 0n,
+      });
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(4));
+
+      // L1 block 102 is replaced by one that does not carry its two messages.
+      fake.removeMessagesAfter(2);
+      fake.reorgL1BlocksFrom(102n);
+      await archiver.syncImmediate();
+
+      expect(await getStoredLeaves()).toEqual(asHex(early));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+      // The tips the world state follows move back with the store, which is what unwinds it.
+      expect((await archiver.getL2Tips()).proposed.number).toEqual(BlockNumber(2));
+      // The proposed checkpoint covering the pruned blocks is evicted along with them.
+      expect(await archiverStore.blocks.getLastProposedCheckpoint()).toBeUndefined();
+      expect(pruneSpy).toHaveBeenCalledWith(
+        expect.objectContaining({
+          type: L2BlockSourceEvents.L2PruneUncheckpointed,
+          slotNumber: fake.getL2SlotAtL1Block(l1BlockNumber),
+          blocks: [blocks[2], blocks[3]],
+        }),
+      );
+    });
+
+    it('prunes when the reorg replaces the removed messages at the same indices', async () => {
+      const { early, late } = addMessages();
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+
+      const blocks = await makeBlocksConsumingThrough([2, 4]);
+      for (const block of blocks) {
+        await archiver.addBlock(block);
+      }
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+
+      // The replacement L1 block carries the same two messages in the opposite order: the message count is
+      // unchanged and indices 2 and 3 are filled again, but block 2 consumed the leaves that were there before.
+      fake.reorderMessagesAtL1Block(102n);
+      await archiver.syncImmediate();
+
+      expect(await getStoredLeaves()).toEqual(asHex([...early, late[1], late[0]]));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(1));
+      expect(pruneSpy).toHaveBeenCalledWith(
+        expect.objectContaining({ type: L2BlockSourceEvents.L2PruneUncheckpointed, blocks: [blocks[1]] }),
+      );
+    });
+
+    // The rollback and the prune have to commit together: if the messages were dropped on their own, the next
+    // sync pass would re-download the canonical ones, find the local state consistent with L1, and never come
+    // back to the proposed blocks built on the messages that are gone.
+    it('keeps the messages when pruning the proposed chain fails', async () => {
+      const { early, late } = addMessages();
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+
+      const blocks = await makeBlocksConsumingThrough([2, 4]);
+      for (const block of blocks) {
+        await archiver.addBlock(block);
+      }
+
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+      const pruneFailure = new Error('cannot remove blocks');
+      jest.spyOn(archiverStore.blocks, 'removeBlocksAfter').mockImplementationOnce(() => {
+        throw pruneFailure;
+      });
+
+      fake.removeMessagesAfter(2);
+      fake.reorgL1BlocksFrom(102n);
+      await expect(archiver.syncImmediate()).rejects.toThrow(pruneFailure);
+
+      expect(await getStoredLeaves()).toEqual(asHex([...early, ...late]));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+
+      // With the store rolled back, the next pass sees the same divergence again and completes both halves.
+      await archiver.syncImmediate();
+
+      expect(await getStoredLeaves()).toEqual(asHex(early));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(1));
+    });
+
+    // A rollback below what the published chain already consumed means the message store and a checkpoint L1
+    // accepted disagree about L1, and the message store is not necessarily the right view. Every proposed block
+    // then satisfies the predicate, so acting on it would drop blocks that consumed nothing removed while their
+    // published parent, which did, stays.
+    it('leaves proposed blocks alone when the rollback reaches below the checkpointed tip', async () => {
+      const { early } = addMessages();
+      const checkpointedBlocks = await makeBlocksConsumingThrough([2, 4], 105n);
+      await fake.addCheckpoint(CheckpointNumber(1), {
+        blocks: checkpointedBlocks,
+        l1BlockNumber: 105n,
+        numL1ToL2Messages: 0,
+      });
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+
+      // A proposed block that consumed nothing of its own, sitting on the checkpointed tip.
+      const [proposed] = await fake.makeBlocks(CheckpointNumber(2), { numBlocks: 1, txsPerBlock: 1, l1BlockNumber });
+      proposed.header.state.l1ToL2MessageTree = new AppendOnlyTreeSnapshot(Fr.random(), 4);
+      await archiver.addBlock(proposed);
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(3));
+
+      fake.removeMessagesAfter(2);
+      fake.reorgL1BlocksFrom(102n);
+      await archiver.syncImmediate();
+
+      expect(await getStoredLeaves()).toEqual(asHex(early));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(3));
+      expect(pruneSpy).not.toHaveBeenCalled();
+    });
+
+    it('leaves checkpointed blocks alone when the messages they consumed are rolled back', async () => {
+      const { early } = addMessages();
+      const blocks = await makeBlocksConsumingThrough([2, 4], 105n);
+      await fake.addCheckpoint(CheckpointNumber(1), { blocks, l1BlockNumber: 105n, numL1ToL2Messages: 0 });
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+      expect(await archiver.getCheckpointNumber()).toEqual(CheckpointNumber(1));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+
+      // Only the L1 checkpoint comparison may unwind published state, so the message rollback leaves it in place
+      // even though block 2 consumed a message that is now gone.
+      fake.removeMessagesAfter(2);
+      fake.reorgL1BlocksFrom(102n);
+      await archiver.syncImmediate();
+
+      expect(await getStoredLeaves()).toEqual(asHex(early));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+      expect(pruneSpy).not.toHaveBeenCalled();
+    });
+  });
+
   describe('finalized checkpoint', () => {
     it('reports no finalized blocks before any checkpoint is proven', async () => {
       fake.setL1BlockNumber(100n);
diff --git a/yarn-project/archiver/src/modules/data_store_updater.ts b/yarn-project/archiver/src/modules/data_store_updater.ts
index 5c7f381495ea..2ebe1839cfc7 100644
--- a/yarn-project/archiver/src/modules/data_store_updater.ts
+++ b/yarn-project/archiver/src/modules/data_store_updater.ts
@@ -1,3 +1,4 @@
+import type { L1BlockId } from '@aztec/ethereum/l1-types';
 import { BlockNumber, CheckpointNumber } from '@aztec/foundation/branded-types';
 import { filterAsync } from '@aztec/foundation/collection';
 import { createLogger } from '@aztec/foundation/log';
@@ -57,6 +58,10 @@ export class ArchiverDataStoreUpdater {
    * @param pendingChainValidationStatus - Optional validation status to set.
    * @returns True if the operation is successful.
    */
+  // TODO: a block built before an L1 reorg rolled back the messages it consumed can be pushed here after
+  // `removeProposedBlocksConsumingMessagesFrom` already pruned the chain it belongs to, and lands on the parent that
+  // survived the prune with nothing left to remove it. Closing it needs the builder to carry the message-store state
+  // it built against so this insert can reject a stale one.
   public async addProposedBlock(
     block: L2Block,
     pendingChainValidationStatus?: ValidateCheckpointResult,
@@ -255,22 +260,105 @@ export class ArchiverDataStoreUpdater {
    * @throws Error if any block to be removed is checkpointed.
    */
   public async removeUncheckpointedBlocksAfter(blockNumber: BlockNumber): Promise<L2Block[]> {
-    const result = await this.stores.db.transactionAsync(async () => {
-      // Verify we're only removing uncheckpointed blocks
-      const lastCheckpointedBlockNumber = await this.stores.blocks.getCheckpointedL2BlockNumber();
-      if (blockNumber < lastCheckpointedBlockNumber) {
-        throw new Error(
-          `Cannot remove blocks after ${blockNumber} because checkpointed blocks exist up to ${lastCheckpointedBlockNumber}`,
-        );
-      }
+    const result = await this.stores.db.transactionAsync(() => this.removeUncheckpointedBlocksAfterInTx(blockNumber));
+    await this.l2TipsCache?.refresh();
+    return result;
+  }
 
-      const prunedBlocks = await this.removeBlocksAfter(blockNumber);
-      await this.evictProposedCheckpointsForPrunedBlocks(prunedBlocks);
+  /** Body of {@link removeUncheckpointedBlocksAfter}, for callers that supply the transaction and the tips refresh. */
+  private async removeUncheckpointedBlocksAfterInTx(blockNumber: BlockNumber): Promise<L2Block[]> {
+    // Verify we're only removing uncheckpointed blocks
+    const lastCheckpointedBlockNumber = await this.stores.blocks.getCheckpointedL2BlockNumber();
+    if (blockNumber < lastCheckpointedBlockNumber) {
+      throw new Error(
+        `Cannot remove blocks after ${blockNumber} because checkpointed blocks exist up to ${lastCheckpointedBlockNumber}`,
+      );
+    }
+
+    const prunedBlocks = await this.removeBlocksAfter(blockNumber);
+    await this.evictProposedCheckpointsForPrunedBlocks(prunedBlocks);
 
-      return prunedBlocks;
+    return prunedBlocks;
+  }
+
+  /**
+   * Rewinds the message sync point, dropping every L1-to-L2 message from `firstRemovedIndex` on, and prunes the
+   * proposed blocks that consumed one of them, in a single transaction.
+   *
+   * Splitting the two would make the prune unrecoverable: a crash after the messages were dropped leaves the
+   * proposed chain built on messages the store no longer holds, and the next sync pass re-downloads the canonical
+   * messages, finds the local state consistent with L1, and never comes back to those blocks.
+   *
+   * @returns The pruned blocks.
+   */
+  public async rewindMessagesAndPruneProposedBlocks(
+    messagesSyncPoint: L1BlockId,
+    firstRemovedIndex: bigint,
+  ): Promise<L2Block[]> {
+    const prunedBlocks = await this.stores.db.transactionAsync(async () => {
+      await this.stores.messages.rewindMessagesTo(messagesSyncPoint, firstRemovedIndex);
+      return await this.removeProposedBlocksConsumingMessagesFrom(firstRemovedIndex);
     });
     await this.l2TipsCache?.refresh();
-    return result;
+    return prunedBlocks;
+  }
+
+  /**
+   * Removes the proposed (not yet L1-checkpointed) blocks that consumed an L1-to-L2 message at or after
+   * `firstRemovedIndex`, along with every block after them. A block's L1-to-L2 tree leaf count is the cumulative
+   * count of messages consumed through it, so a leaf count above the index means the block consumed a message the
+   * local chain no longer has, and one equal to it means the block stopped below it.
+   *
+   * The index is where the rollback rewound to, not the first leaf whose value actually changed, so a reorg that
+   * re-mines the same messages in a different L1 block also prunes blocks whose trees are unchanged and which L1
+   * would still accept. That over-pruning is accepted for now: the rollback rewinds before it knows what the
+   * canonical chain re-delivers, so it cannot tell a re-mine from a content change, and a prune from the first
+   * differing leaf has to wait until the rollback becomes content-aware.
+   *
+   * Checkpointed blocks are never touched: a message store that disagrees with a checkpoint L1 accepted means one
+   * of the two views of L1 is mid-reorg and this one is not necessarily the right one, so only the archive
+   * comparison in the checkpoint step may unwind published state. For the same reason nothing is pruned when the
+   * rollback reaches below the checkpointed tip's leaf count, where every proposed descendant would qualify.
+   *
+   * @param firstRemovedIndex - Index of the first L1-to-L2 message that was removed from the store.
+   * @returns The removed blocks.
+   */
+  private async removeProposedBlocksConsumingMessagesFrom(firstRemovedIndex: bigint): Promise<L2Block[]> {
+    const [checkpointedBlockNumber, latestBlockNumber] = await Promise.all([
+      this.stores.blocks.getCheckpointedL2BlockNumber(),
+      this.stores.blocks.getLatestL2BlockNumber(),
+    ]);
+    if (latestBlockNumber <= checkpointedBlockNumber) {
+      return [];
+    }
+
+    const checkpointedTip =
+      checkpointedBlockNumber > 0
+        ? await this.stores.blocks.getBlockData({ number: BlockNumber(checkpointedBlockNumber) })
+        : undefined;
+    const checkpointedTipLeafCount = BigInt(
+      checkpointedTip?.header.state.l1ToL2MessageTree.nextAvailableLeafIndex ?? 0,
+    );
+    if (firstRemovedIndex < checkpointedTipLeafCount) {
+      this.log.warn(
+        `Not pruning proposed blocks: rollback index ${firstRemovedIndex} is below the checkpointed tip's leaf count`,
+        { firstRemovedIndex, checkpointedTipLeafCount, checkpointedBlockNumber },
+      );
+      return [];
+    }
+
+    const proposedBlocks = await this.stores.blocks.getBlocksData({
+      from: BlockNumber(checkpointedBlockNumber + 1),
+      limit: latestBlockNumber - checkpointedBlockNumber,
+    });
+    const firstAffected = proposedBlocks.find(
+      block => BigInt(block.header.state.l1ToL2MessageTree.nextAvailableLeafIndex) > firstRemovedIndex,
+    );
+    if (firstAffected === undefined) {
+      return [];
+    }
+
+    return await this.removeUncheckpointedBlocksAfterInTx(BlockNumber(firstAffected.header.getBlockNumber() - 1));
   }
 
   /**
diff --git a/yarn-project/archiver/src/modules/instrumentation.ts b/yarn-project/archiver/src/modules/instrumentation.ts
index 20eeff25311e..b25dbdf7d1d1 100644
--- a/yarn-project/archiver/src/modules/instrumentation.ts
+++ b/yarn-project/archiver/src/modules/instrumentation.ts
@@ -85,7 +85,7 @@ export class ArchiverInstrumentation {
     this.pruneDuration = meter.createHistogram(Metrics.ARCHIVER_PRUNE_DURATION);
 
     this.pruneCount = createUpDownCounterWithDefault(meter, Metrics.ARCHIVER_PRUNE_COUNT, {
-      [Attributes.PRUNE_TYPE]: ['unproven', 'uncheckpointed', 'l1_conflict', 'orphan', 'l1_mismatch'],
+      [Attributes.PRUNE_TYPE]: ['unproven', 'uncheckpointed', 'l1_conflict', 'orphan', 'l1_mismatch', 'inbox_rollback'],
     });
 
     this.blockProposalTxTargetCount = createUpDownCounterWithDefault(
@@ -161,9 +161,10 @@ export class ArchiverInstrumentation {
    * type distinguishes the cause: 'uncheckpointed' (slot ended without a checkpoint), 'l1_conflict' (proposed blocks
    * conflicting with an L1 checkpoint), 'orphan' (no matching proposed checkpoint arrived before the deadline), or
    * 'l1_mismatch' (the local checkpointed tip diverged from L1 — an L1 reorg or a pruned/missed-proof checkpoint — so
-   * already-checkpointed blocks were rewound).
+   * already-checkpointed blocks were rewound), or 'inbox_rollback' (an L1 reorg orphaned Inbox messages the blocks
+   * had consumed).
    */
-  public recordPrune(pruneType: 'uncheckpointed' | 'l1_conflict' | 'orphan' | 'l1_mismatch') {
+  public recordPrune(pruneType: 'uncheckpointed' | 'l1_conflict' | 'orphan' | 'l1_mismatch' | 'inbox_rollback') {
     this.pruneCount.add(1, { [Attributes.PRUNE_TYPE]: pruneType });
   }
 
diff --git a/yarn-project/archiver/src/modules/l1_synchronizer.ts b/yarn-project/archiver/src/modules/l1_synchronizer.ts
index 07c5d4a0253f..b5d53cf1ef68 100644
--- a/yarn-project/archiver/src/modules/l1_synchronizer.ts
+++ b/yarn-project/archiver/src/modules/l1_synchronizer.ts
@@ -689,17 +689,47 @@ export class ArchiverL1Synchronizer implements Traceable {
   }
 
   /**
-   * Rewinds the message syncpoint to the given L1 block, dropping the messages from `removeFromIndex` on in the same
-   * transaction, so an interruption cannot leave truncated messages behind a syncpoint that would never fetch them.
+   * Rewinds the message syncpoint to the given L1 block, dropping the messages from `removeFromIndex` on and the
+   * locally proposed blocks that consumed one of them, all in the same transaction: an interruption must not leave
+   * truncated messages behind a syncpoint that would never fetch them, nor a proposed chain built on messages the
+   * store no longer holds.
+   *
+   * Runs in the message step of the sync pass, before the checkpoint step, so every consumer that reads the archiver
+   * after a poll sees a local view where messages and blocks agree.
    */
   private async rewindMessagesTo(messagesSyncPoint: L1BlockId, removeFromIndex?: bigint): Promise<L1BlockId> {
-    await this.stores.messages.rewindMessagesTo(messagesSyncPoint, removeFromIndex);
+    let prunedBlocks: L2Block[] = [];
+    if (removeFromIndex === undefined) {
+      await this.stores.messages.rewindMessagesTo(messagesSyncPoint);
+    } else {
+      prunedBlocks = await this.updater.rewindMessagesAndPruneProposedBlocks(messagesSyncPoint, removeFromIndex);
+    }
     this.log.verbose(`Updated messages syncpoint to L1 block ${messagesSyncPoint.l1BlockNumber}`, {
       ...messagesSyncPoint,
     });
+    this.reportPrunedProposedBlocks(prunedBlocks, removeFromIndex);
     return messagesSyncPoint;
   }
 
+  /** Logs, counts and announces the proposed blocks a message rollback dropped. */
+  private reportPrunedProposedBlocks(prunedBlocks: L2Block[], firstRemovedIndex: bigint | undefined): void {
+    if (prunedBlocks.length === 0) {
+      return;
+    }
+
+    const firstPrunedBlock = prunedBlocks[0];
+    this.log.warn(
+      `Pruning ${prunedBlocks.length} proposed blocks from ${firstPrunedBlock.number} built on rolled back messages`,
+      { firstRemovedIndex, firstPrunedBlockNumber: firstPrunedBlock.number, prunedCount: prunedBlocks.length },
+    );
+    this.instrumentation.recordPrune('inbox_rollback');
+    this.events.emit(L2BlockSourceEvents.L2PruneUncheckpointed, {
+      type: L2BlockSourceEvents.L2PruneUncheckpointed,
+      slotNumber: firstPrunedBlock.header.getSlot(),
+      blocks: prunedBlocks,
+    });
+  }
+
   /** Checks if the local consensus rolling hash and message count match the remote Inbox live state. */
   private async localStateMatches(localLastMessage: InboxMessage | undefined, remoteState: InboxContractState) {
     const localMessageCount = await this.stores.messages.getTotalL1ToL2MessageCount();
diff --git a/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts b/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts
new file mode 100644
index 000000000000..278ce82f5f72
--- /dev/null
+++ b/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts
@@ -0,0 +1,209 @@
+import type { Archiver } from '@aztec/archiver';
+import type { TestAztecNodeService } from '@aztec/aztec-node/test';
+import { AztecAddress } from '@aztec/aztec.js/addresses';
+import { Fr } from '@aztec/aztec.js/fields';
+import type { Logger } from '@aztec/aztec.js/log';
+import { isL1ToL2MessageReady, waitForL1ToL2MessageReady } from '@aztec/aztec.js/messaging';
+import type { AztecNode } from '@aztec/aztec.js/node';
+import type { Delayer } from '@aztec/ethereum/l1-tx-utils';
+import type { ChainMonitor } from '@aztec/ethereum/test';
+import type { ExtendedViemWalletClient } from '@aztec/ethereum/types';
+import { CheckpointNumber } from '@aztec/foundation/branded-types';
+import { retryUntil } from '@aztec/foundation/retry';
+import { L2BlockSourceEvents, type L2PruneUncheckpointedEvent } from '@aztec/stdlib/block';
+import { WorldStateSynchronizerError } from '@aztec/world-state';
+
+import 'jest-extended';
+
+import { sendL1ToL2Message } from '../../fixtures/l1_to_l2_messaging.js';
+import type { EndToEndContext } from '../../fixtures/utils.js';
+import { waitForL1ToL2MessageSeen } from '../../shared/wait_for_l1_to_l2_message.js';
+import { L1ReorgsTest, TX_COUNT } from '../l1-reorgs/setup.js';
+import type { SingleNodeTestContext } from '../single_node_test_context.js';
+
+// Single-node + prover-node suite covering what happens when an L1 reorg orphans the Inbox messages that a
+// locally proposed block already consumed. The checkpoint's propose tx is withheld so those blocks are still
+// only proposed when the reorg lands, which is the case the archiver's rollback of the proposed chain exists
+// for; a reorg that also drops a published checkpoint is covered by single-node/l1-reorgs/blocks.parallel.
+//
+// Built on the L1-reorg fixture rather than the cross-chain harness because both the message and the propose
+// tx have to be held back, which needs its delayed L1 clients and its faster reorg cadence.
+describe('single-node/cross-chain/streaming_inbox_reorg', () => {
+  let t: L1ReorgsTest;
+
+  let context: EndToEndContext;
+  let logger: Logger;
+  let node: AztecNode;
+  let archiver: Archiver;
+  let monitor: ChainMonitor;
+  let sequencerDelayer: Delayer;
+
+  let L1_BLOCK_TIME_IN_S: number;
+  let L2_SLOT_DURATION_IN_S: number;
+
+  let test: SingleNodeTestContext;
+
+  let l1Client: ExtendedViemWalletClient;
+
+  // The replacement message is sent from its own account because the reorg below rewinds past the block holding the
+  // first message: that resets the sender's nonce, and a withheld tx signed with the nonce after it could never be
+  // mined into the replacement block.
+  let replacementL1Client: ExtendedViemWalletClient;
+  let replacementL1ClientDelayer: Delayer;
+
+  beforeEach(async () => {
+    t = new L1ReorgsTest();
+    await t.setup();
+    ({ test, context, logger, node, archiver, monitor, sequencerDelayer } = t);
+    ({ L1_BLOCK_TIME_IN_S, L2_SLOT_DURATION_IN_S } = t);
+    ({ client: l1Client } = await test.createL1Client());
+    ({ client: replacementL1Client, delayer: replacementL1ClientDelayer } = await test.createL1Client());
+  });
+
+  afterEach(async () => {
+    await t.teardown();
+  });
+
+  const sendMessage = async (client: ExtendedViemWalletClient) =>
+    sendL1ToL2Message(
+      { recipient: await AztecAddress.random(), content: Fr.random(), secretHash: Fr.random() },
+      { l1ContractAddresses: context.deployL1ContractsValues.l1ContractAddresses, l1Client: client },
+    );
+
+  it('prunes the proposed chain and consumes the replacement message after a reorg', async () => {
+    // Get the chain building multi-block checkpoints before touching the Inbox.
+    await t.sendTransactions(TX_COUNT, 300);
+    await test.waitUntilCheckpointNumber(CheckpointNumber(2), L2_SLOT_DURATION_IN_S * 6);
+
+    // Every proposed-chain prune is recorded, so the one the reorg causes can be identified by the blocks it
+    // carries rather than by the tip having moved for any reason.
+    const prunes: L2PruneUncheckpointedEvent[] = [];
+    archiver.events.on(L2BlockSourceEvents.L2PruneUncheckpointed, event => {
+      prunes.push(event);
+    });
+
+    // Withhold the next checkpoint's propose tx, so the blocks that consume the message below stay in the
+    // proposed chain instead of being published: this is the state the reorg has to find them in.
+    sequencerDelayer.cancelNextTx();
+
+    const msg = await sendMessage(l1Client);
+    logger.warn(`Sent message on L1 block ${msg.txReceipt.blockNumber}`);
+
+    // Readiness means the message sits at a leaf index the tip's L1-to-L2 tree has already grown past, so the
+    // block at the proposed tip (or one before it) is the block that consumed it.
+    await waitForL1ToL2MessageReady(node, msg.msgHash, { timeoutSeconds: L2_SLOT_DURATION_IN_S * 2 });
+    const consumedAtBlockNumber = await archiver.getBlockNumber();
+    logger.warn(`Message consumed by proposed block ${consumedAtBlockNumber}`);
+
+    // Readiness is an archiver-side check, so pin down that world state also applied the consuming block: without
+    // this high-water mark, the rollback assertion after the reorg would also be satisfied by a world state that
+    // simply never got that far.
+    await retryUntil(
+      async () => (await node.getWorldStateSyncStatus()).latestBlockNumber >= consumedAtBlockNumber,
+      'world state synced to the consuming block',
+      L2_SLOT_DURATION_IN_S / 2,
+      0.2,
+    );
+
+    // Wait for the withheld propose before reorging, and check it was the one covering the consuming block: the
+    // cancellation is armed before the message is even sent, so under pipelining it could otherwise have taken
+    // the previous slot's propose and left the consuming block published.
+    await retryUntil(
+      () => sequencerDelayer.getCancelledTxs().length,
+      'sequencer propose tx withheld',
+      L2_SLOT_DURATION_IN_S * 2,
+      0.2,
+    );
+    const tipsBeforeReorg = await archiver.getL2Tips();
+    expect(consumedAtBlockNumber).toBeGreaterThan(tipsBeforeReorg.checkpointed.block.number);
+
+    // Prepare the replacement message but keep its L1 tx out of the chain, so the reorg can mine it in the
+    // block that replaces the orphaned one.
+    replacementL1ClientDelayer.cancelNextTx();
+    const replacementMsgPromise = sendMessage(replacementL1Client);
+    await retryUntil(
+      () => replacementL1ClientDelayer.getCancelledTxs().length,
+      'replacement message tx withheld',
+      L1_BLOCK_TIME_IN_S * 2,
+      0.1,
+    );
+
+    // Replace every L1 block from the one carrying the original message onwards. The message was mined and
+    // then synced, so this always rewrites more than one block.
+    const l1BlockNumber = await monitor.run(true).then(m => m.l1BlockNumber);
+    const reorgDepth = l1BlockNumber - Number(msg.txReceipt.blockNumber) + 1;
+    expect(reorgDepth).toBeGreaterThanOrEqual(2);
+    logger.warn(`Triggering reorg of depth ${reorgDepth} replacing the message with a different one`);
+    await context.cheatCodes.eth.reorgWithReplacement(reorgDepth, [[replacementL1ClientDelayer.getCancelledTxs()[0]]]);
+    const replacementMsg = await replacementMsgPromise;
+    logger.warn(`Reorged-in replacement message on L1 block ${replacementMsg.txReceipt.blockNumber}`);
+
+    // The archiver drops the orphaned message and every proposed block that consumed it within a poll of the
+    // reorg. The bound is what separates this from the end-of-slot prune, which would only catch the same
+    // blocks once the slot they were built for has run out.
+    await retryUntil(
+      () => prunes.some(prune => prune.blocks.some(block => block.number === consumedAtBlockNumber)),
+      'proposed chain pruned back past the consuming block',
+      L2_SLOT_DURATION_IN_S / 2,
+      0.2,
+    );
+    expect(await archiver.getBlockNumber()).toBeLessThan(consumedAtBlockNumber);
+
+    // Published state was not unwound: this path only ever drops proposed blocks, and the checkpoint that would
+    // have published the consuming ones never reached L1.
+    const tipsAfterPrune = await archiver.getL2Tips();
+    expect(tipsAfterPrune.checkpointed.block.number).toBeGreaterThanOrEqual(tipsBeforeReorg.checkpointed.block.number);
+
+    // World state does not consume the archiver's prune event; it rolls back when its own block stream reports the
+    // chain pruned, and until it does the L1-to-L2 tree that the next block is built on still holds the orphaned
+    // message. Same bound as the prune assertion above.
+    await retryUntil(
+      async () => (await node.getWorldStateSyncStatus()).latestBlockNumber < consumedAtBlockNumber,
+      'world state unwound past the consuming block',
+      L2_SLOT_DURATION_IN_S / 2,
+      0.2,
+    );
+
+    // Both sides keep advancing as the chain is rebuilt, so the trees are only comparable at a fixed height: world
+    // state's view at its own synced tip against the archiver's block there. Taking the view by block hash means a
+    // height rebuilt between the two reads is retried rather than compared across two different blocks.
+    const worldState = (context.aztecNodeService as TestAztecNodeService).worldStateSynchronizer;
+    const [worldStateTrees, blockAtWorldStateTip] = await retryUntil(
+      async () => {
+        const worldStateTip = (await node.getWorldStateSyncStatus()).latestBlockNumber;
+        const block = await archiver.getBlockData({ number: worldStateTip });
+        if (block === undefined) {
+          return undefined;
+        }
+        try {
+          const snapshot = await worldState.getVerifiedSnapshot(worldStateTip, block.blockHash);
+          return [await snapshot.getStateReference(), block] as const;
+        } catch (err) {
+          if (err instanceof WorldStateSynchronizerError) {
+            return undefined;
+          }
+          throw err;
+        }
+      },
+      'world state and archiver settled on the same block',
+      L2_SLOT_DURATION_IN_S / 2,
+      0.2,
+    );
+    expect(worldStateTrees.l1ToL2MessageTree).toEqual(blockAtWorldStateTip.header.state.l1ToL2MessageTree);
+
+    // The orphaned message is gone for good, and no block can consume it again.
+    expect(await isL1ToL2MessageReady(node, msg.msgHash)).toBe(false);
+
+    // Its replacement is picked up and consumed by a block of the rebuilt chain, which keeps proposing.
+    await waitForL1ToL2MessageSeen(node, replacementMsg.msgHash, { timeoutSeconds: L1_BLOCK_TIME_IN_S * 4 });
+    await waitForL1ToL2MessageReady(node, replacementMsg.msgHash, { timeoutSeconds: L2_SLOT_DURATION_IN_S * 3 });
+    expect(await isL1ToL2MessageReady(node, msg.msgHash)).toBe(false);
+
+    // Proposing survives the prune: a checkpoint built after the reorg has to reach L1, not just any checkpoint.
+    const checkpointAfterReorg = CheckpointNumber(tipsBeforeReorg.checkpointed.checkpoint.number + 1);
+    await test.waitUntilCheckpointNumber(checkpointAfterReorg, L2_SLOT_DURATION_IN_S * 4);
+    const rebuiltCheckpoints = await archiver.getCheckpoints({ from: checkpointAfterReorg, limit: 10 });
+    expect(rebuiltCheckpoints.some(published => published.checkpoint.blocks.length >= 2)).toBe(true);
+    await test.assertMultipleBlocksPerSlot(2);
+  });
+});
diff --git a/yarn-project/end-to-end/src/single-node/single_node_test_context.ts b/yarn-project/end-to-end/src/single-node/single_node_test_context.ts
index 00285605b4dc..0c5a3f16948f 100644
--- a/yarn-project/end-to-end/src/single-node/single_node_test_context.ts
+++ b/yarn-project/end-to-end/src/single-node/single_node_test_context.ts
@@ -205,6 +205,8 @@ export class SingleNodeTestContext {
   public L1_BLOCK_TIME_IN_S!: number;
   public L2_SLOT_DURATION_IN_S!: number;
 
+  private l1ClientCount = 0;
+
   public static async setup<T extends SingleNodeTestContext>(this: new () => T, opts: SingleNodeTestOpts = {}) {
     const test = new this();
     await test.setup(opts);
@@ -462,8 +464,8 @@ export class SingleNodeTestContext {
     return node;
   }
 
-  protected getNextPrivateKey(): Hex {
-    const key = getPrivateKeyFromIndex(this.nodes.length + this.proverNodes.length + 1);
+  protected getNextPrivateKey(offset = 0): Hex {
+    const key = getPrivateKeyFromIndex(this.nodes.length + this.proverNodes.length + 1 + offset);
     return `0x${key!.toString('hex')}`;
   }
 
@@ -712,11 +714,14 @@ export class SingleNodeTestContext {
     return TestContract.at(instance.address, wallet);
   }
 
-  /** Creates an L1 client using a fresh account with funds from anvil, with a tx delayer already set up. */
+  /**
+   * Creates an L1 client using a fresh account with funds from anvil, with a tx delayer already set up. Each call
+   * gets its own account, so two clients never share a nonce sequence.
+   */
   public async createL1Client() {
     const rawClient = createExtendedL1Client(
       [...this.l1Client.chain.rpcUrls.default.http],
-      privateKeyToAccount(this.getNextPrivateKey()),
+      privateKeyToAccount(this.getNextPrivateKey(this.l1ClientCount++)),
       this.l1Client.chain,
     );
     const delayer = createDelayer(this.context.dateProvider, { ethereumSlotDuration: this.L1_BLOCK_TIME_IN_S }, {});
diff --git a/yarn-project/validator-client/src/proposal_handler.test.ts b/yarn-project/validator-client/src/proposal_handler.test.ts
index 22232ee49a0c..1d15c9625a8d 100644
--- a/yarn-project/validator-client/src/proposal_handler.test.ts
+++ b/yarn-project/validator-client/src/proposal_handler.test.ts
@@ -1,4 +1,5 @@
-import type { Archiver } from '@aztec/archiver';
+import { type Archiver, createArchiverStore } from '@aztec/archiver';
+import { makeInboxMessages } from '@aztec/archiver/test';
 import type { BlobClientInterface } from '@aztec/blob-client/client';
 import { INITIAL_L2_BLOCK_NUM, MAX_BLOCKS_PER_CHECKPOINT } from '@aztec/constants';
 import type { EpochCache } from '@aztec/epoch-cache';
@@ -10,7 +11,7 @@ import { Fr } from '@aztec/foundation/curves/bn254';
 import { TestDateProvider } from '@aztec/foundation/timer';
 import { type FieldsOf, unfreeze } from '@aztec/foundation/types';
 import type { P2P } from '@aztec/p2p';
-import { BlockHash } from '@aztec/stdlib/block';
+import { BlockHash, GENESIS_BLOCK_HEADER_HASH } from '@aztec/stdlib/block';
 import type { BlockData, L2Block, L2BlockSink, L2BlockSource } from '@aztec/stdlib/block';
 import { type Checkpoint, CheckpointReexecutionTracker, type ProposedCheckpointData } from '@aztec/stdlib/checkpoint';
 import type { L1RollupConstants } from '@aztec/stdlib/epoch-helpers';
@@ -232,6 +233,36 @@ describe('ProposalHandler checkpoint validation', () => {
       expect(blockSource.syncImmediate).not.toHaveBeenCalled();
     });
 
+    // The cached verdict is what the attestation path reads, and it rests on blocks this node holds. An archiver
+    // rollback between the two calls p2p makes for one proposal prunes those blocks, so the verdict has to be
+    // re-derived rather than replayed.
+    it('re-validates a cached valid result once the checkpoint blocks are gone', async () => {
+      const proposal = await makeProposal();
+      blockSource.getBlockData.mockResolvedValue({ header: makeBlockHeader() } as BlockData);
+      const validateSpy = jest
+        .spyOn(handler, 'validateCheckpointProposal')
+        .mockResolvedValue({ isValid: true, checkpointNumber: CheckpointNumber(1) });
+
+      expect(await handler.handleCheckpointProposal(proposal, proposalInfo)).toEqual({
+        isValid: true,
+        checkpointNumber: CheckpointNumber(1),
+      });
+
+      // While the blocks are still local the verdict is replayed, so a validation that would now fail is not run.
+      validateSpy.mockResolvedValue({ isValid: false, reason: 'archive_mismatch' });
+      expect(await handler.handleCheckpointProposal(proposal, proposalInfo)).toEqual({
+        isValid: true,
+        checkpointNumber: CheckpointNumber(1),
+      });
+
+      // A rollback pruned the blocks the verdict was based on.
+      blockSource.getBlockData.mockResolvedValue(undefined);
+      expect(await handler.handleCheckpointProposal(proposal, proposalInfo)).toEqual({
+        isValid: false,
+        reason: 'archive_mismatch',
+      });
+    });
+
     it('does not use cache for a different proposal', async () => {
       blockSource.getBlockData.mockResolvedValue(undefined);
 
@@ -1073,6 +1104,56 @@ describe('ProposalHandler checkpoint validation', () => {
       );
     });
 
+    // An L1 reorg that orphans the messages a proposal consumed reaches this node as an archiver rollback, which
+    // re-downloads the bucket at that position from the canonical chain with a different rolling hash. Resolving
+    // the reference by hash is what makes a proposal this node accepted before the rollback fail after it, so no
+    // attestation is signed for a block that L1 will not accept. Driven off a real message store so the rollback,
+    // the bucket resolution and the re-read the handler does after forcing a sync are the production ones.
+    it('rejects a proposal it accepted before its archiver rewound the bucket', async () => {
+      const stores = await createArchiverStore(
+        { archiverStoreMapSizeKb: 1024 * 1024, dataDirectory: undefined, dataStoreMapSizeKb: 1024 * 1024 },
+        GENESIS_BLOCK_HEADER_HASH,
+      );
+      const messages = makeInboxMessages(2);
+      await stores.messages.addL1ToL2MessageBuckets(messages);
+      l1ToL2MessageSource.getInboxBucketByRollingHash.mockImplementation(inboxRollingHash =>
+        stores.messages.getInboxBucketByRollingHash(inboxRollingHash),
+      );
+      l1ToL2MessageSource.getInboxBucketByTotalMsgCount.mockImplementation(total =>
+        stores.messages.getInboxBucketByTotalMsgCount(total),
+      );
+      l1ToL2MessageSource.getL1ToL2MessagesBetweenBuckets.mockImplementation((from, to) =>
+        stores.messages.getL1ToL2MessagesBetweenBuckets(from, to),
+      );
+
+      const consumedBucket = await stores.messages.getInboxBucket(2n);
+      const { proposal, blockHandler } = await setupStreamingProposal(InboxBucketRef.fromBucket(consumedBucket!), {
+        nowMs: 10_000,
+      });
+
+      expect(await blockHandler.handleBlockProposal(proposal, {} as any, false)).toEqual({
+        isValid: true,
+        blockNumber: BlockNumber(INITIAL_L2_BLOCK_NUM),
+      });
+
+      // The reorg drops the second message and mines a different one in its place: bucket 2 keeps its sequence
+      // number and message count, but no bucket commits to the rolling hash the proposal signed any more. The
+      // handler cannot tell that from lag, so it waits out the remaining budget (held short here) and rejects.
+      await stores.messages.rewindMessagesTo({ l1BlockNumber: 0n, l1BlockHash: Buffer32.ZERO }, 1n);
+      await stores.messages.addL1ToL2MessageBuckets(
+        makeInboxMessages(1, { initialIndex: 1n, initialInboxHash: messages[0].inboxRollingHash }),
+      );
+      dateProvider.setTime(38_000);
+
+      expect(await blockHandler.handleBlockProposal(proposal, {} as any, false)).toEqual({
+        isValid: false,
+        blockNumber: BlockNumber(INITIAL_L2_BLOCK_NUM),
+        reason: 'bucket_unknown',
+      });
+
+      await stores.db.close();
+    });
+
     // A bucket the proposer already consumed is on L1 by construction, so a bucket this node cannot resolve is
     // (usually) local archiver lag, not a divergence: the handler forces a sync and re-checks until the
     // attestation deadline instead of dropping the attestation on the spot.
diff --git a/yarn-project/validator-client/src/proposal_handler.ts b/yarn-project/validator-client/src/proposal_handler.ts
index 513d4dc11283..10e0316b7cf7 100644
--- a/yarn-project/validator-client/src/proposal_handler.ts
+++ b/yarn-project/validator-client/src/proposal_handler.ts
@@ -1292,10 +1292,17 @@ export class ProposalHandler {
     const slot = proposal.slotNumber;
     const payloadHash = proposal.getPayloadHash();
 
-    // Check cache: same signed-payload hash means we already validated this exact proposal.
+    // Check cache: same signed-payload hash means we already validated this exact proposal. A valid verdict rests
+    // on blocks this node holds locally, and p2p makes two calls for one proposal (the all-nodes validation, then
+    // the attestation), so an archiver rollback in between can prune those blocks. Re-check that the checkpoint's
+    // last block is still local before reusing a valid verdict, or the attestation outlives what it was based on.
     if (this.lastCheckpointValidationResult && this.lastCheckpointValidationResult.payloadHash === payloadHash) {
-      this.log.debug(`Returning cached validation result for checkpoint proposal at slot ${slot}`, proposalInfo);
-      return this.lastCheckpointValidationResult.result;
+      const cached = this.lastCheckpointValidationResult.result;
+      if (!cached.isValid || (await this.blockSource.getBlockData({ archive: proposal.archive })) !== undefined) {
+        this.log.debug(`Returning cached validation result for checkpoint proposal at slot ${slot}`, proposalInfo);
+        return cached;
+      }
+      this.log.warn(`Re-validating checkpoint proposal at slot ${slot}: its blocks are no longer local`, proposalInfo);
     }
 
     const proposer = proposal.getSender();