feat(engine): introduce REST based + SSZ serialized new-payload-with-witness

src/Nethermind/Nethermind.Consensus/Processing/BlockAccessListManager.TxProcessorPool.cs

@@ -20,6 +20,7 @@
 using Nethermind.Int256;
 using Nethermind.Logging;
 using Nethermind.State;
+using Nethermind.Consensus.Stateless;
 
 namespace Nethermind.Consensus.Processing;
 
@@ -86,7 +87,7 @@ static ParallelTxProcessorWithWorldStateManager()
         private readonly ILogManager _logManager;
         private readonly ObjectPool<IReadOnlyTxProcessorSource>? _parentReaderEnvPool;
         private int _processorCount;
-        private readonly bool _witnessMode;
+        private readonly Func<bool>? _isWitnessExecution;
 
         public ParallelTxProcessorWithWorldStateManager(
             IBlockhashProvider blockHashProvider,
@@ -96,13 +97,13 @@ public ParallelTxProcessorWithWorldStateManager(
             PrewarmerEnvFactory? prewarmerEnvFactory,
             PreBlockCaches? preBlockCaches,
             IReadOnlyTxProcessingEnvFactory? readOnlyTxProcessingEnvFactory,
-            bool witnessMode)
+            Func<bool>? isWitnessExecution)
         {
             _blockHashProvider = blockHashProvider;
             _specProvider = specProvider;
             _stateProvider = stateProvider;
             _logManager = logManager;
-            _witnessMode = witnessMode;
+            _isWitnessExecution = isWitnessExecution;
             _parentReaderEnvPool = CreateParentReaderEnvPool(prewarmerEnvFactory, preBlockCaches, readOnlyTxProcessingEnvFactory);
             for (int i = 0; i < ProcessorPoolSize; i++)
             {
@@ -226,7 +227,7 @@ private int ClampBalIndex(uint balIndex)
             => (int)uint.Min(balIndex, (uint)_lastBalIndex);
 
         private TxProcessorWithWorldState NewProcessor()
-            => new(true, _blockHashProvider, _specProvider, _stateProvider, _logManager, _witnessMode);
+            => new(true, _blockHashProvider, _specProvider, _stateProvider, _logManager, _isWitnessExecution);
 
         private TxProcessorWithWorldState RentProcessor()
         {
@@ -333,9 +334,9 @@ public SequentialTxProcessorWithWorldStateManager(
             ISpecProvider specProvider,
             IWorldState stateProvider,
             ILogManager logManager,
-            bool witnessMode)
+            Func<bool>? isWitnessExecution)
         {
-            _txProcessorWithWorldState = new(false, blockHashProvider, specProvider, stateProvider, logManager, witnessMode);
+            _txProcessorWithWorldState = new(false, blockHashProvider, specProvider, stateProvider, logManager, isWitnessExecution);
             _txProcessorWithWorldState.WorldState.SetGeneratingBlockAccessList(new());
         }
 
@@ -377,7 +378,7 @@ public TxProcessorWithWorldState(
             ISpecProvider specProvider,
             IWorldState stateProvider,
             ILogManager logManager,
-            bool witnessMode)
+            Func<bool>? isWitnessExecution)
         {
 
             VirtualMachine virtualMachine = new(blockHashProvider, specProvider, logManager);
@@ -388,12 +389,14 @@ public TxProcessorWithWorldState(
                 worldState = _balWorldState;
             }
             WorldState = new TracedAccessWorldState(worldState, parallel);
-            // Witness mode must record every code access, so it uses the non-caching CodeInfoRepository.
-            // EthereumCodeInfoRepository wraps a CacheCodeInfoRepository whose process-wide static code
-            // cache serves hits without reading through the (traced) WorldState, so cached code accesses
-            // would be missing from the generated witness.
-            ICodeInfoRepository codeInfoRepository = witnessMode
-                ? new CodeInfoRepository(WorldState, new EthereumPrecompileProvider())
+            // On witness-capable managers, route code lookups through CodeInfoRepositoryProxy: while a
+            // witness is being executed (predicate true) it uses the non-caching CodeInfoRepository so
+            // every code access flows through the (traced) WorldState; otherwise it uses the cached repo.
+            // The process-wide static code cache would otherwise serve hits without reading through the
+            // WorldState, dropping those accesses from the witness. Non-witness managers use the cached
+            // repo directly with no per-call indirection.
+            ICodeInfoRepository codeInfoRepository = isWitnessExecution is not null
+                ? new CodeInfoRepositoryProxy(new EthereumCodeInfoRepository(WorldState), WorldState, new EthereumPrecompileProvider(), isWitnessExecution)
                 : new EthereumCodeInfoRepository(WorldState);
             TxProcessor = new(BlobBaseFeeCalculator.Instance, specProvider, WorldState, virtualMachine, codeInfoRepository, logManager, parallel);
             TxProcessorAdapter = new(TxProcessor);

src/Nethermind/Nethermind.Consensus/Processing/BlockAccessListManager.cs

@@ -47,17 +47,17 @@ public partial class BlockAccessListManager(
     PrewarmerEnvFactory? prewarmerEnvFactory = null,
     PreBlockCaches? preBlockCaches = null,
     IReadOnlyTxProcessingEnvFactory? readOnlyTxProcessingEnvFactory = null,
-    bool witnessMode = false)
+    Func<bool>? isWitnessExecution = null)
     : IBlockAccessListManager, IDisposable
 {
     private readonly ILogger _logger = logManager.GetClassLogger<BlockAccessListManager>();
     private BlockExecutionContext? _blockExecutionContext;
     private ITxProcessorWithWorldStateManager? _txProcessorWithWorldStateManager;
     private Task? _balWarmupTask;
     private readonly Lazy<ParallelTxProcessorWithWorldStateManager> _parallelTxProcessorWithWorldStateManager =
-        new(() => new(blockHashProvider, specProvider, stateProvider, logManager, prewarmerEnvFactory, preBlockCaches, readOnlyTxProcessingEnvFactory, witnessMode));
+        new(() => new(blockHashProvider, specProvider, stateProvider, logManager, prewarmerEnvFactory, preBlockCaches, readOnlyTxProcessingEnvFactory, isWitnessExecution));
     private readonly Lazy<SequentialTxProcessorWithWorldStateManager> _sequentialTxProcessorWithWorldStateManager =
-        new(() => new(blockHashProvider, specProvider, stateProvider, logManager, witnessMode));
+        new(() => new(blockHashProvider, specProvider, stateProvider, logManager, isWitnessExecution));
     private const int GasValidationChunkSize = 8;
     private long? _gasRemaining;
     private bool _isBuilding;
@@ -124,11 +124,17 @@ public void PrepareForProcessing(Block suggestedBlock, IReleaseSpec spec, Proces
 
         // Parallel execution needs the decoded BAL body (RLP fixtures only carry the hash)
         // and an active state scope (so we can capture the parent state root for workers).
+        //
+        // Witness execution forces sequential: parallel workers read pre-state through pooled
+        // parent-reader snapshots that bypass the capturing world-state proxy, so their accesses
+        // would be missing from the witness. Evaluated per block via the predicate so regular
+        // blocks on EIP-7928 chains keep parallel execution.
         ParallelExecutionEnabled = Enabled
             && blocksConfig.ParallelExecution
             && !_isBuilding
             && suggestedBlock.BlockAccessList is not null
-            && stateProvider.IsInScope;
+            && stateProvider.IsInScope
+            && isWitnessExecution?.Invoke() is not true;
 
         // BAL-driven read warming: mirrors BlockCachePreWarmer.IsBalReadWarmingEnabled so
         // HintBal honours the same opt-in config as the prewarmer path.

src/Nethermind/Nethermind.Consensus/Stateless/CodeInfoRepositoryProxy.cs

@@ -0,0 +1,49 @@
+// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited
+// SPDX-License-Identifier: LGPL-3.0-only
+
+using System;
+using System.Diagnostics.CodeAnalysis;
+using Nethermind.Core;
+using Nethermind.Core.Specs;
+using Nethermind.Evm;
+using Nethermind.Evm.CodeAnalysis;
+using Nethermind.Evm.State;
+
+namespace Nethermind.Consensus.Stateless;
+
+/// <summary>
+/// Thin <see cref="ICodeInfoRepository"/> decorator that routes every call to a non-caching
+/// <see cref="CodeInfoRepository"/> it owns when <paramref name="useNonCaching"/> returns <c>true</c>,
+/// and to the wrapped cached repository otherwise. The predicate is evaluated per call so the choice
+/// tracks the live witness-execution state rather than being fixed at construction.
+/// </summary>
+/// <remarks>
+/// Witness execution requires every bytecode/code-hash lookup to flow through <see cref="IWorldState"/>
+/// (so a recording world state observes it, or so stateless execution stays isolated to the witness);
+/// the process-wide static code cache used by the inner repository would short-circuit those reads.
+/// The non-caching repository is built inside this decorator (rather than resolved from DI) so no other
+/// DI consumer can pick it up and accidentally bypass the cache for non-witness blocks.
+/// </remarks>
+public sealed class CodeInfoRepositoryProxy(
+    ICodeInfoRepository inner,
+    IWorldState worldState,
+    IPrecompileProvider precompileProvider,
+    Func<bool> useNonCaching) : ICodeInfoRepository
+{
+    private readonly ICodeInfoRepository _inner = inner;
+    private readonly CodeInfoRepository _nonCached = new(worldState, precompileProvider);
+
+    private ICodeInfoRepository Current => useNonCaching() ? _nonCached : _inner;
+
+    public CodeInfo GetCachedCodeInfo(Address codeSource, bool followDelegation, IReleaseSpec vmSpec, out Address? delegationAddress)
+        => Current.GetCachedCodeInfo(codeSource, followDelegation, vmSpec, out delegationAddress);
+
+    public void InsertCode(ReadOnlyMemory<byte> code, Address codeOwner, IReleaseSpec spec)
+        => Current.InsertCode(code, codeOwner, spec);
+
+    public void SetDelegation(Address codeSource, Address authority, IReleaseSpec spec)
+        => Current.SetDelegation(codeSource, authority, spec);
+
+    public bool TryGetDelegation(Address address, IReleaseSpec spec, [NotNullWhen(true)] out Address? delegatedAddress)
+        => Current.TryGetDelegation(address, spec, out delegatedAddress);
+}

src/Nethermind/Nethermind.Consensus/Stateless/StatelessBlockProcessingEnv.cs

@@ -60,7 +60,9 @@ private BlockProcessor GetProcessor()
                 ParallelExecutionBatchRead = false
             },
             new WithdrawalProcessorFactory(logManager),
-            witnessMode: true
+            // Stateless execution must resolve code only from the witness-backed state, never the
+            // shared cache — so it always runs in witness mode (non-caching code reads).
+            isWitnessExecution: static () => true
         );
         BlockProcessor.ParallelBlockValidationTransactionsExecutor txExecutor = new(
             new BlockProcessor.BlockValidationTransactionsExecutor(

src/Nethermind/Nethermind.Consensus/Stateless/WitnessCaptureSession.cs

@@ -0,0 +1,66 @@
+// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited
+// SPDX-License-Identifier: LGPL-3.0-only
+
+using System.Threading;
+
+namespace Nethermind.Consensus.Stateless;
+
+/// <summary>
+/// Per-main-processing-scope arming point for witness capture. Holds nullable pointers to the
+/// recorders that are active during a single <c>ProcessOne</c> call; the decorators
+/// (<see cref="WitnessCapturingWorldStateProxy"/>, <see cref="WitnessCapturingTrieStore"/>,
+/// <see cref="WitnessCapturingHeaderFinder"/>) consult these pointers on every call and forward
+/// straight through to the inner component when null.
+/// </summary>
+/// <remarks>
+/// <para>
+/// Single arm/disarm point in <see cref="WitnessCapturingBlockProcessor"/> replaces the previous
+/// per-decorator <c>TryActivate</c>/<c>Deactivate</c> ceremony: all three decorators become dumb
+/// passthroughs that share one source of truth here.
+/// </para>
+/// <para>
+/// Thread-safety: <see cref="TryArm"/> uses CAS on the world-state pointer so concurrent armers
+/// see at most one winner; <see cref="Disarm"/> clears in reverse so any consumer that still sees
+/// <see cref="WorldStateRecorder"/> set also sees the other recorders set. The main processing
+/// pipeline drives blocks serially so contention is theoretical, but the volatile reads remain
+/// safe under any caller.
+/// </para>
+/// </remarks>
+public sealed class WitnessCaptureSession
+{
+    private WitnessGeneratingWorldState? _worldStateRecorder;
+    private WitnessHeaderRecorder? _headerRecorder;
+    private WitnessTrieStoreRecorder? _trieRecorder;
+
+    public WitnessGeneratingWorldState? WorldStateRecorder => Volatile.Read(ref _worldStateRecorder);
+    public WitnessHeaderRecorder? HeaderRecorder => Volatile.Read(ref _headerRecorder);
+    public WitnessTrieStoreRecorder? TrieRecorder => Volatile.Read(ref _trieRecorder);
+
+    public bool IsActive => WorldStateRecorder is not null;
+
+    /// <summary>
+    /// Atomically installs the three recorders for a single capture pass. Returns <c>false</c>
+    /// when a capture is already in progress on this session.
+    /// </summary>
+    /// <remarks>
+    /// The world-state recorder is the primary slot — the CAS on it gates the operation; the other
+    /// two are written under the post-CAS happens-before, so any reader that observes the
+    /// world-state recorder also observes the header and trie recorders.
+    /// </remarks>
+    public bool TryArm(
+        WitnessGeneratingWorldState worldStateRecorder,
+        WitnessHeaderRecorder headerRecorder,
+        WitnessTrieStoreRecorder trieRecorder)
+    {
+        Volatile.Write(ref _trieRecorder, trieRecorder);
+        Volatile.Write(ref _headerRecorder, headerRecorder);
+        return Interlocked.CompareExchange(ref _worldStateRecorder, worldStateRecorder, null) is null;
+    }
+
+    public void Disarm()
+    {
+        Volatile.Write(ref _worldStateRecorder, null);
+        Volatile.Write(ref _headerRecorder, null);
+        Volatile.Write(ref _trieRecorder, null);
+    }
+}