diff --git a/build.zig b/build.zig
index 02492a0d2..2ecae8d3c 100644
--- a/build.zig
+++ b/build.zig
@@ -214,6 +214,7 @@ pub fn build(b: *Builder) !void {
     zeam_types.addImport("@zeam/utils", zeam_utils);
     zeam_types.addImport("@zeam/metrics", zeam_metrics);
     zeam_types.addImport("@zeam/xmss", zeam_xmss);
+    zeam_types.addImport("@zeam/thread-pool", zeam_thread_pool);
 
     // add zeam-types
     const zeam_configs = b.addModule("@zeam/configs", .{
diff --git a/pkgs/cli/src/node.zig b/pkgs/cli/src/node.zig
index fb7ad898e..84662ac47 100644
--- a/pkgs/cli/src/node.zig
+++ b/pkgs/cli/src/node.zig
@@ -178,6 +178,70 @@ pub const Node = struct {
 
     const Self = @This();
 
+    /// Thread roles configured at startup. Counts are logged once init
+    /// completes so operators can verify `--rayon-threads` / aggregator
+    /// auto-tune and compare against `cpu_count`.
+    const StartupThreadBudget = struct {
+        cpu_count: usize,
+        zig_worker_pool: usize,
+        rayon: usize,
+        aggregate_max_inflight: u32,
+        chain_worker: usize,
+        metrics_server: usize,
+        api_server: usize,
+        rayon_source: []const u8,
+
+        fn estimatedTotal(self: @This()) usize {
+            // main (xev slot-driver) + Io.Threaded + libp2p + zig pool + rayon
+            // + chain-worker + metrics + api + slot watchdog (run()).
+            return 1 + 1 + 1 + self.zig_worker_pool + self.rayon + self.chain_worker +
+                self.metrics_server + self.api_server + 1;
+        }
+    };
+
+    fn logStartupThreadBudget(self: *Self, budget: StartupThreadBudget) void {
+        self.logger.info(
+            "startup thread budget: cpu_count={d} estimated_os_threads≈{d}",
+            .{ budget.cpu_count, budget.estimatedTotal() },
+        );
+        self.logger.info(
+            "  main/xev slot-driver: 1 (this thread runs xev.Loop via clock.run)",
+            .{},
+        );
+        self.logger.info(
+            "  std.Io.Threaded:      1 (blocking I/O for Zig pool + disk)",
+            .{},
+        );
+        self.logger.info(
+            "  libp2p rust bridge:   1 (spawned in network.run)",
+            .{},
+        );
+        self.logger.info(
+            "  zig worker pool:      {d} (aggregate_max_inflight={d})",
+            .{ budget.zig_worker_pool, budget.aggregate_max_inflight },
+        );
+        self.logger.info(
+            "  rayon (XMSS prover):  {d}{s}",
+            .{ budget.rayon, budget.rayon_source },
+        );
+        self.logger.info(
+            "  chain-worker:         {d}{s}",
+            .{ budget.chain_worker, if (budget.chain_worker == 0) " (disabled)" else "" },
+        );
+        self.logger.info(
+            "  metrics HTTP server:  {d}",
+            .{budget.metrics_server},
+        );
+        self.logger.info(
+            "  api HTTP server:      {d}",
+            .{budget.api_server},
+        );
+        self.logger.info(
+            "  slot-driver watchdog: 1 (spawned at run() start)",
+            .{},
+        );
+    }
+
     /// Closes the current database, wipes the on-disk rocksdb directory, and
     /// reopens a fresh database at the same path.
     ///
@@ -420,8 +484,16 @@ pub const Node = struct {
         const cpu_count = std.Thread.getCpuCount() catch 2;
         const reserved_system_threads: usize = 4; // main, p2p, api server, metrics server
         const desired_workers = @max(@as(usize, 1), cpu_count -| reserved_system_threads);
-        const zig_worker_budget = @max(@as(usize, 1), (desired_workers + 1) / 2);
-        const worker_count = @min(zig_worker_budget, @as(usize, ThreadPool.max_thread_count));
+
+        // Aggregators: XMSS recursive prove (rayon) is the per-slot bottleneck. Keep a
+        // small Zig pool for capped in-flight aggregate workers (#907).
+        const worker_count = if (options.is_aggregator) blk: {
+            const aggregator_zig_workers = @max(@as(usize, 2), desired_workers / 4);
+            break :blk @min(@as(usize, ThreadPool.max_thread_count), aggregator_zig_workers);
+        } else blk: {
+            const zig_worker_budget = @max(@as(usize, 1), (desired_workers + 1) / 2);
+            break :blk @min(zig_worker_budget, @as(usize, ThreadPool.max_thread_count));
+        };
         self.thread_pool = try ThreadPool.init(.{
             .allocator = allocator,
             .io = std.Io.Threaded.global_single_threaded.io(),
@@ -447,17 +519,23 @@ pub const Node = struct {
         // pool is initialized lazily on first use.
         const rayon_threads = if (options.rayon_threads) |override|
             @max(@as(usize, 1), @as(usize, override))
+        else if (options.is_aggregator)
+            desired_workers
         else
             @max(@as(usize, 1), desired_workers -| worker_count);
-        self.logger.info(
-            "thread pools: cpu_count={d} zig_workers={d} rayon_threads={d}{s}",
-            .{
-                cpu_count,
-                worker_count,
-                rayon_threads,
-                if (options.rayon_threads != null) " (rayon override via --rayon-threads)" else "",
-            },
-        );
+        // One outer aggregate worker at a time on aggregators. Parallelize the
+        // ~11s STARK work per att_data inside that worker
+        // (`computeAggregatedSignatures` thread pool), not across independent
+        // interval workers. Multiple in-flight outer workers can snapshot the
+        // same gossip sigs before either commits and publish duplicate
+        // aggregates (PR #920 review).
+        const aggregate_max_inflight: u32 = if (options.is_aggregator) 1 else 4;
+        const rayon_source: []const u8 = if (options.rayon_threads != null)
+            " (--rayon-threads override)"
+        else if (options.is_aggregator)
+            " (aggregator auto-tune: full post-system budget)"
+        else
+            " (non-aggregator auto-split)";
         // Operator-typo guard for --rayon-threads (review feedback on #903).
         // Rayon tolerates over-subscription, but values like `--rayon-threads 160`
         // on a 4-vCPU box silently degrade throughput. Warn (don't reject) so the
@@ -474,6 +552,12 @@ pub const Node = struct {
         }
         xmss.setRayonThreads(rayon_threads);
 
+        if (options.is_aggregator) {
+            xmss.setupProver() catch |err| {
+                self.logger.warn("xmss prover setup failed: {any}; aggregation may be unavailable", .{err});
+            };
+        }
+
         // Log the aggregator threshold on startup so operators can see
         // exactly how `--min-aggregation-inputs` was resolved (default vs
         // override). The threshold is enforced by the aggregator-side
@@ -516,6 +600,7 @@ pub const Node = struct {
             .thread_pool = self.thread_pool,
             .chain_worker_enabled = options.chain_worker_enabled,
             .min_aggregation_inputs = options.min_aggregation_inputs,
+            .aggregate_max_inflight = aggregate_max_inflight,
         });
         errdefer self.beam_node.deinit();
 
@@ -563,6 +648,17 @@ pub const Node = struct {
             zeam_metrics.metrics.lean_node_info.set(.{ .name = "zeam", .version = build_options.version }, 1) catch {};
         }
 
+        self.logStartupThreadBudget(.{
+            .cpu_count = cpu_count,
+            .zig_worker_pool = worker_count,
+            .rayon = rayon_threads,
+            .aggregate_max_inflight = aggregate_max_inflight,
+            .chain_worker = if (options.chain_worker_enabled) 1 else 0,
+            .metrics_server = if (options.metrics_enable) 1 else 0,
+            .api_server = if (options.metrics_enable) 1 else 0,
+            .rayon_source = rayon_source,
+        });
+
         self.logger = options.logger_config.logger(.node);
     }
 
diff --git a/pkgs/cli/src/test_driver.zig b/pkgs/cli/src/test_driver.zig
index 37068a194..cb05dd5be 100644
--- a/pkgs/cli/src/test_driver.zig
+++ b/pkgs/cli/src/test_driver.zig
@@ -466,7 +466,7 @@ pub fn initForkChoiceDriver(
     zeam_utils.hashTreeRoot(types.BeamBlock, anchor_block, &anchor_block_root, driver_allocator) catch
         return error.HashFailed;
 
-    var test_thread_pool = try node.testing.initTestThreadPool(driver_allocator);
+    var test_thread_pool = try @import("@zeam/node").testing.initTestThreadPool(driver_allocator);
     errdefer test_thread_pool.deinit();
 
     // Init fork choice (uses anchor_state_ptr for anchorState)
diff --git a/pkgs/node/src/chain.zig b/pkgs/node/src/chain.zig
index 510f7ea7b..915fbe65c 100644
--- a/pkgs/node/src/chain.zig
+++ b/pkgs/node/src/chain.zig
@@ -386,8 +386,9 @@ pub const BeamChain = struct {
     /// In-flight aggregate worker count (issue #907). `submitAggregateOnInterval`
     /// atomically checks against `aggregate_max_inflight` before spawning so the
     /// libxev interval tick never blocks. `aggregateImpl` decrements on exit.
-    /// Concurrent runs are safe by construction — see the three-phase
-    /// snapshot/compute/merge contract on `forkchoice.aggregateUnlocked`.
+    /// Aggregators keep this at 1; per-att_data FFI parallelism lives inside
+    /// `aggregateUnlocked`. Values > 1 rely on commit-time duplicate suppression
+    /// in `forkchoice.aggregateUnlocked` (see PR #920 review).
     aggregate_inflight: std.atomic.Value(u32) = .init(0),
 
     /// Joined at `deinit` so aggregate workers cannot outlive chain state.
diff --git a/pkgs/node/src/forkchoice.zig b/pkgs/node/src/forkchoice.zig
index 45a38b8b2..b89207815 100644
--- a/pkgs/node/src/forkchoice.zig
+++ b/pkgs/node/src/forkchoice.zig
@@ -2106,12 +2106,12 @@ pub const ForkChoice = struct {
     ///      pairs that existed at snapshot time (entries added during
     ///      phase 2 stay so the next aggregator pass can consume them).
     ///
-    /// Concurrent callers are safe by construction: the three-phase contract
-    /// above (snapshot under `signatures_mutex` → lock-free compute on owned
-    /// clones → merge under `signatures_mutex`) ensures two aggregations
-    /// cannot race on the live maps. `submitAggregateOnInterval`'s
-    /// `aggregate_max_inflight` cap is a CPU-budget knob layered on top, not
-    /// a correctness mechanism.
+    /// The three-phase contract (snapshot under `signatures_mutex` → lock-free
+    /// compute on owned clones → merge under `signatures_mutex`) prevents map
+    /// races but not duplicate publishes when two workers snapshot the same
+    /// gossip sigs before either commits. Aggregators keep
+    /// `aggregate_max_inflight` at 1; commit also suppresses merge/publish for
+    /// att_data whose snapshot gossip vids are already gone from the live map.
     fn aggregateUnlocked(self: *Self, state_opt: ?*const types.BeamState, slot_filter: ?[]const types.Slot) ![]types.SignedAggregatedAttestation {
         const state = state_opt orelse return try self.allocator.alloc(types.SignedAggregatedAttestation, 0);
         const agg_timer = zeam_metrics.lean_committee_signatures_aggregation_time_seconds.start();
@@ -2171,6 +2171,7 @@ pub const ForkChoice = struct {
             &snap.new_payloads,
             &snap.known_payloads,
             slot_filter,
+            self.thread_pool,
         );
         observeAggregateBuildPhase("compute_ffi", compute_start_ns);
 
@@ -2253,6 +2254,31 @@ pub const ForkChoice = struct {
             self.signatures_mutex.lock();
             defer self.signatures_mutex.unlock();
 
+            // Drop att_data whose snapshot gossip inputs were already consumed
+            // by an earlier in-flight worker (defense when aggregate_max_inflight > 1).
+            var results_write_idx: usize = 0;
+            var results_idx: usize = 0;
+            while (results_idx < results.items.len) : (results_idx += 1) {
+                const att_data = results.items[results_idx].data;
+                if (shouldSuppressDuplicateAggregateCommit(&snap, att_data, &self.attestation_signatures)) {
+                    results.items[results_idx].deinit();
+                    if (new_payloads_local.fetchRemove(att_data)) |kv| {
+                        var list = kv.value;
+                        deinitAggregatedPayloadsList(self.allocator, &list);
+                    }
+                    self.logger.debug(
+                        "suppress duplicate aggregate commit att_data slot={d} (snapshot gossip already consumed)",
+                        .{att_data.slot},
+                    );
+                } else {
+                    if (results_write_idx != results_idx) {
+                        results.items[results_write_idx] = results.items[results_idx];
+                    }
+                    results_write_idx += 1;
+                }
+            }
+            results.shrinkRetainingCapacity(results_write_idx);
+
             // MERGE (NOT replace). The previous shape did
             //   deinitAggregatedPayloadsMap(allocator, &self.latest_new_aggregated_payloads);
             //   self.latest_new_aggregated_payloads = new_payloads_local;
@@ -3795,6 +3821,95 @@ test "isAggregatorTrivialInput higher thresholds skip more no-children inputs" {
     try std.testing.expect(!isAggregatorTrivialInput(0, 3, 3));
 }
 
+/// Returns true when this worker's snapshot gossip vids for `att_data` were
+/// already consumed by an earlier aggregate commit, so merging/publishing this
+/// pass would duplicate aggregates (PR #920 review).
+fn shouldSuppressDuplicateAggregateCommit(
+    snap: *const AggregateSnapshot,
+    att_data: types.AttestationData,
+    live_sigs: *const types.SignaturesMap,
+) bool {
+    const snap_inner = snap.signatures.get(att_data) orelse return false;
+    if (snap_inner.count() == 0) return false;
+    const live_inner = live_sigs.get(att_data) orelse return true;
+    var it = snap_inner.iterator();
+    while (it.next()) |entry| {
+        if (live_inner.contains(entry.key_ptr.*)) return false;
+    }
+    return true;
+}
+
+test "shouldSuppressDuplicateAggregateCommit: no snap gossip never suppresses" {
+    const allocator = std.testing.allocator;
+    const zero_root = std.mem.zeroes(types.Root);
+    const att_data = types.AttestationData{
+        .slot = 1,
+        .head = .{ .root = zero_root, .slot = 0 },
+        .target = .{ .root = zero_root, .slot = 0 },
+        .source = .{ .root = zero_root, .slot = 0 },
+    };
+    var snap = AggregateSnapshot{
+        .signatures = types.SignaturesMap.init(allocator),
+        .new_payloads = AggregatedPayloadsMap.init(allocator),
+        .known_payloads = AggregatedPayloadsMap.init(allocator),
+    };
+    defer snap.deinit(allocator);
+    var live = types.SignaturesMap.init(allocator);
+    defer live.deinit();
+    try std.testing.expect(!shouldSuppressDuplicateAggregateCommit(&snap, att_data, &live));
+}
+
+test "shouldSuppressDuplicateAggregateCommit: snap gossip still live does not suppress" {
+    const allocator = std.testing.allocator;
+    const zero_root = std.mem.zeroes(types.Root);
+    const att_data = types.AttestationData{
+        .slot = 1,
+        .head = .{ .root = zero_root, .slot = 0 },
+        .target = .{ .root = zero_root, .slot = 0 },
+        .source = .{ .root = zero_root, .slot = 0 },
+    };
+    const stored_sig = types.StoredSignature{ .slot = 1, .signature = std.mem.zeroes(@TypeOf(@as(types.StoredSignature, undefined).signature)) };
+    var snap = AggregateSnapshot{
+        .signatures = types.SignaturesMap.init(allocator),
+        .new_payloads = AggregatedPayloadsMap.init(allocator),
+        .known_payloads = AggregatedPayloadsMap.init(allocator),
+    };
+    defer snap.deinit(allocator);
+    var snap_inner = types.SignaturesMap.InnerMap.init(allocator);
+    try snap_inner.put(0, stored_sig);
+    try snap_inner.put(1, stored_sig);
+    try snap.signatures.put(att_data, snap_inner);
+    var live = types.SignaturesMap.init(allocator);
+    defer live.deinit();
+    try live.addSignature(att_data, 0, stored_sig);
+    try std.testing.expect(!shouldSuppressDuplicateAggregateCommit(&snap, att_data, &live));
+}
+
+test "shouldSuppressDuplicateAggregateCommit: snap gossip fully consumed suppresses" {
+    const allocator = std.testing.allocator;
+    const zero_root = std.mem.zeroes(types.Root);
+    const att_data = types.AttestationData{
+        .slot = 1,
+        .head = .{ .root = zero_root, .slot = 0 },
+        .target = .{ .root = zero_root, .slot = 0 },
+        .source = .{ .root = zero_root, .slot = 0 },
+    };
+    const stored_sig = types.StoredSignature{ .slot = 1, .signature = std.mem.zeroes(@TypeOf(@as(types.StoredSignature, undefined).signature)) };
+    var snap = AggregateSnapshot{
+        .signatures = types.SignaturesMap.init(allocator),
+        .new_payloads = AggregatedPayloadsMap.init(allocator),
+        .known_payloads = AggregatedPayloadsMap.init(allocator),
+    };
+    defer snap.deinit(allocator);
+    var snap_inner = types.SignaturesMap.InnerMap.init(allocator);
+    try snap_inner.put(0, stored_sig);
+    try snap_inner.put(1, stored_sig);
+    try snap.signatures.put(att_data, snap_inner);
+    var live = types.SignaturesMap.init(allocator);
+    defer live.deinit();
+    try std.testing.expect(shouldSuppressDuplicateAggregateCommit(&snap, att_data, &live));
+}
+
 /// Walk the owned aggregator snapshot and drop every `att_data` entry
 /// from `signatures` that is trivial under `min_inputs` (see
 /// `isAggregatorTrivialInput`). The corresponding inner map is freed.
diff --git a/pkgs/node/src/node.zig b/pkgs/node/src/node.zig
index 680b7ec26..b963dd21f 100644
--- a/pkgs/node/src/node.zig
+++ b/pkgs/node/src/node.zig
@@ -62,6 +62,8 @@ const NodeOpts = struct {
     /// aggregation threshold; see
     /// `pkgs/types/src/block.zig:default_min_aggregation_inputs`.
     min_aggregation_inputs: u32 = types.default_min_aggregation_inputs,
+    /// Soft cap on concurrent aggregate workers (`BeamChain.aggregate_inflight`).
+    aggregate_max_inflight: u32 = 4,
 };
 
 pub const BeamNode = struct {
@@ -142,6 +144,7 @@ pub const BeamNode = struct {
                 .is_aggregator = opts.is_aggregator,
                 .thread_pool = opts.thread_pool,
                 .min_aggregation_inputs = opts.min_aggregation_inputs,
+                .aggregate_max_inflight = opts.aggregate_max_inflight,
             },
             network.connected_peers,
         ) catch |init_err| {
diff --git a/pkgs/types/src/block.zig b/pkgs/types/src/block.zig
index 4484c6295..2c0e62a2d 100644
--- a/pkgs/types/src/block.zig
+++ b/pkgs/types/src/block.zig
@@ -5,6 +5,7 @@ const params = @import("@zeam/params");
 const zeam_metrics = @import("@zeam/metrics");
 const xmss = @import("@zeam/xmss");
 const zeam_utils = @import("@zeam/utils");
+const ThreadPool = @import("@zeam/thread-pool").ThreadPool;
 
 const aggregation = @import("./aggregation.zig");
 const attestation = @import("./attestation.zig");
@@ -505,6 +506,9 @@ pub const AggregatedAttestationsResult = struct {
     /// Block proposers, by contrast, MUST aggregate every `att_data`
     /// they choose to include in a block, even if its only input is a
     /// single gossip signature.
+    /// Aggregator entry point: when more than one `att_data` needs recursive
+    /// FFI, run each prove on `thread_pool` after a serial prep phase (issue
+    /// #907 — worker processes ~2 `att_data` per slot window).
     pub fn computeAggregatedSignatures(
         self: *Self,
         validators: *const Validators,
@@ -512,6 +516,7 @@ pub const AggregatedAttestationsResult = struct {
         new_payloads: ?*const AggregatedPayloadsMap,
         known_payloads: ?*const AggregatedPayloadsMap,
         slot_filter: ?[]const Slot,
+        thread_pool: *ThreadPool,
     ) !void {
         const allocator = self.allocator;
 
@@ -534,230 +539,49 @@ pub const AggregatedAttestationsResult = struct {
             }
         }
 
-        // Process each AttestationData
-        var data_it = att_data_set.iterator();
-        while (data_it.next()) |data_entry| {
-            const data = data_entry.key_ptr.*;
-            const epoch: u64 = data.slot;
-            var message_hash: [32]u8 = undefined;
-            try zeam_utils.hashTreeRoot(attestation.AttestationData, data, &message_hash, allocator);
-
-            // Step 2: Greedy child proof selection — new_payloads first, then known_payloads as helpers
-            var selected_children: std.ArrayList(aggregation.AggregatedSignatureProof) = .empty;
-            defer {
-                for (selected_children.items) |*child| {
-                    child.deinit();
-                }
-                selected_children.deinit(allocator);
-            }
+        if (att_data_set.count() == 0) return;
 
-            // We need to know max_validator for bitset sizing — derive from validators count
-            const max_validator = validators.len();
-
-            var covered_by_children = try std.DynamicBitSet.initEmpty(allocator, max_validator);
-            defer covered_by_children.deinit();
+        var att_data_keys = try allocator.alloc(attestation.AttestationData, att_data_set.count());
+        defer allocator.free(att_data_keys);
+        var key_idx: usize = 0;
+        var data_it = att_data_set.keyIterator();
+        while (data_it.next()) |key| {
+            att_data_keys[key_idx] = key.*;
+            key_idx += 1;
+        }
+        std.mem.sort(attestation.AttestationData, att_data_keys, {}, attestationDataLessThan);
 
-            // Dummy empty bitset for gossip_available (not known yet)
-            var empty_available = try std.DynamicBitSet.initEmpty(allocator, max_validator);
-            defer empty_available.deinit();
+        var preps = try allocator.alloc(AggregateAttDataPrep, att_data_keys.len);
+        defer {
+            for (preps) |*prep| prep.deinit(allocator);
+            allocator.free(preps);
+        }
 
-            try extendProofsGreedily(
+        var ffi_prep_count: usize = 0;
+        for (att_data_keys, 0..) |data, i| {
+            preps[i] = try prepareAggregateAttData(
                 allocator,
+                validators,
+                signatures_map,
                 new_payloads,
-                data,
-                &selected_children,
-                &covered_by_children,
-                &empty_available,
-            );
-            try extendProofsGreedily(
-                allocator,
                 known_payloads,
                 data,
-                &selected_children,
-                &covered_by_children,
-                &empty_available,
-            );
-
-            // Step 3: Collect individual gossip signatures not covered by selected children
-            var sigmap_sigs: std.ArrayList(xmss.Signature) = .empty;
-            defer {
-                for (sigmap_sigs.items) |*sig| {
-                    sig.deinit();
-                }
-                sigmap_sigs.deinit(allocator);
-            }
-
-            var sigmap_pks: std.ArrayList(xmss.PublicKey) = .empty;
-            defer {
-                for (sigmap_pks.items) |*pk| {
-                    pk.deinit();
-                }
-                sigmap_pks.deinit(allocator);
-            }
-
-            var sigmap_vids: std.ArrayList(usize) = .empty;
-            defer sigmap_vids.deinit(allocator);
-
-            const inner_map = signatures_map.get(data);
-            if (inner_map) |im| {
-                var vid_it = im.iterator();
-                while (vid_it.next()) |entry| {
-                    const vid: usize = @intCast(entry.key_ptr.*);
-                    const sig_entry = entry.value_ptr.*;
-
-                    // Skip if already covered by children
-                    if (vid < covered_by_children.capacity() and covered_by_children.isSet(vid)) continue;
-
-                    if (std.mem.eql(u8, &sig_entry.signature, &ZERO_SIGBYTES)) continue;
-
-                    var sig = xmss.Signature.fromBytes(&sig_entry.signature) catch continue;
-                    errdefer sig.deinit();
-
-                    if (vid >= validators.len()) {
-                        sig.deinit();
-                        continue;
-                    }
-
-                    const val = validators.get(vid) catch {
-                        sig.deinit();
-                        continue;
-                    };
-                    const pk = xmss.PublicKey.fromBytes(&val.attestation_pubkey) catch {
-                        sig.deinit();
-                        continue;
-                    };
-
-                    try sigmap_sigs.append(allocator, sig);
-                    try sigmap_pks.append(allocator, pk);
-                    try sigmap_vids.append(allocator, vid);
-                }
-            }
-
-            const has_gossip = sigmap_sigs.items.len > 0;
-            const has_children = selected_children.items.len > 0;
-
-            if (!has_gossip and !has_children) continue;
-
-            // Build gossip participants bitfield and handle arrays
-            var xmss_participants: ?attestation.AggregationBits = null;
-            defer if (xmss_participants) |*gp| gp.deinit();
-
-            var pk_handles_buf: ?[]*const xmss.HashSigPublicKey = null;
-            defer if (pk_handles_buf) |buf| allocator.free(buf);
-            var sig_handles_buf: ?[]*const xmss.HashSigSignature = null;
-            defer if (sig_handles_buf) |buf| allocator.free(buf);
-
-            var pk_handles: []*const xmss.HashSigPublicKey = &.{};
-            var sig_handles: []*const xmss.HashSigSignature = &.{};
-
-            if (has_gossip) {
-                var gp = try attestation.AggregationBits.init(allocator);
-                errdefer gp.deinit();
-
-                const pks = try allocator.alloc(*const xmss.HashSigPublicKey, sigmap_sigs.items.len);
-                errdefer allocator.free(pks);
-                const sigs = try allocator.alloc(*const xmss.HashSigSignature, sigmap_sigs.items.len);
-                errdefer allocator.free(sigs);
-
-                for (sigmap_vids.items, 0..) |vid, idx| {
-                    try attestation.aggregationBitsSet(&gp, vid, true);
-                    pks[idx] = sigmap_pks.items[idx].handle;
-                    sigs[idx] = sigmap_sigs.items[idx].handle;
-                }
-
-                xmss_participants = gp;
-                pk_handles_buf = pks;
-                sig_handles_buf = sigs;
-                pk_handles = pks[0..sigmap_sigs.items.len];
-                sig_handles = sigs[0..sigmap_sigs.items.len];
-            }
-
-            // If only 1 child and no gossip, pass through the child directly
-            if (!has_gossip and selected_children.items.len == 1) {
-                const child = &selected_children.items[0];
-
-                var att_bits: attestation.AggregationBits = undefined;
-                try utils.sszClone(allocator, attestation.AggregationBits, child.participants, &att_bits);
-                errdefer att_bits.deinit(); // ownership is for self.attestations
-
-                // Clone the child proof for the result (original will be freed by deferred cleanup)
-                var cloned_child: aggregation.AggregatedSignatureProof = undefined;
-                try utils.sszClone(allocator, aggregation.AggregatedSignatureProof, child.*, &cloned_child);
-                errdefer cloned_child.deinit();
-
-                try self.attestations.append(.{ .aggregation_bits = att_bits, .data = data });
-                try self.attestation_signatures.append(cloned_child);
-                continue;
-            }
-
-            // Recursive aggregation: children + gossip
-            var proof = try aggregation.AggregatedSignatureProof.init(allocator);
-            errdefer proof.deinit();
-
-            // Build per-child pub key arrays for recursive aggregation
-            var child_pk_allocs: std.ArrayList([]*const xmss.HashSigPublicKey) = .empty;
-            defer {
-                for (child_pk_allocs.items) |arr| allocator.free(arr);
-                child_pk_allocs.deinit(allocator);
-            }
-            var child_pk_slices: std.ArrayList([]*const xmss.HashSigPublicKey) = .empty;
-            defer child_pk_slices.deinit(allocator);
-
-            var child_pk_wrappers: std.ArrayList(xmss.PublicKey) = .empty;
-            defer {
-                for (child_pk_wrappers.items) |*pw| pw.deinit();
-                child_pk_wrappers.deinit(allocator);
-            }
-
-            for (selected_children.items) |*child| {
-                var n_participants: usize = 0;
-                for (0..child.participants.len()) |i| {
-                    if (child.participants.get(i) catch false) {
-                        n_participants += 1;
-                    }
-                }
-
-                const cpks = try allocator.alloc(*const xmss.HashSigPublicKey, n_participants);
-                errdefer allocator.free(cpks);
-
-                var cpk_idx: usize = 0;
-                for (0..child.participants.len()) |i| {
-                    if (child.participants.get(i) catch false) {
-                        if (i >= validators.len()) continue;
-                        const val = validators.get(@intCast(i)) catch continue;
-                        const pk = xmss.PublicKey.fromBytes(&val.attestation_pubkey) catch continue;
-                        try child_pk_wrappers.append(allocator, pk);
-                        cpks[cpk_idx] = pk.handle;
-                        cpk_idx += 1;
-                    }
-                }
-
-                try child_pk_allocs.append(allocator, cpks);
-                try child_pk_slices.append(allocator, cpks[0..cpk_idx]);
-            }
-
-            const pq_sig_timer = zeam_metrics.lean_pq_sig_aggregated_signatures_building_time_seconds.start();
-            try aggregation.AggregatedSignatureProof.aggregate(
-                allocator,
-                xmss_participants,
-                selected_children.items,
-                child_pk_slices.items,
-                pk_handles,
-                sig_handles,
-                &message_hash,
-                epoch,
-                &proof,
             );
-            _ = pq_sig_timer.observe();
-            if (xmss_participants) |*gp| gp.deinit();
-            xmss_participants = null;
+            if (preps[i].outcome == .ffi) ffi_prep_count += 1;
+        }
 
-            var att_bits: attestation.AggregationBits = undefined;
-            try utils.sszClone(allocator, attestation.AggregationBits, proof.participants, &att_bits);
-            errdefer att_bits.deinit(); // ownership is for self.attestations
+        try runAggregateAttDataPreps(
+            allocator,
+            preps,
+            thread_pool,
+        );
 
-            try self.attestations.append(.{ .aggregation_bits = att_bits, .data = data });
-            try self.attestation_signatures.append(proof);
+        for (preps) |*prep| {
+            if (prep.outcome != .done) continue;
+            const result = prep.outcome.done;
+            prep.outcome = .skip;
+            try self.attestations.append(result.attestation);
+            try self.attestation_signatures.append(result.signature);
         }
     }
 
@@ -859,6 +683,384 @@ const CompactGroupResult = struct {
     signature: aggregation.AggregatedSignatureProof,
 };
 
+fn attestationDataLessThan(_: void, a: attestation.AttestationData, b: attestation.AttestationData) bool {
+    if (a.slot != b.slot) return a.slot < b.slot;
+    const head_cmp = std.mem.order(u8, &a.head.root, &b.head.root);
+    if (head_cmp != .eq) return head_cmp == .lt;
+    const target_cmp = std.mem.order(u8, &a.target.root, &b.target.root);
+    if (target_cmp != .eq) return target_cmp == .lt;
+    const source_cmp = std.mem.order(u8, &a.source.root, &b.source.root);
+    if (source_cmp != .eq) return source_cmp == .lt;
+    if (a.head.slot != b.head.slot) return a.head.slot < b.head.slot;
+    if (a.target.slot != b.target.slot) return a.target.slot < b.target.slot;
+    return a.source.slot < b.source.slot;
+}
+
+const AggregateAttDataOutcome = union(enum) {
+    skip,
+    ffi: AggregateAttDataFfiArgs,
+    done: CompactGroupResult,
+};
+
+const AggregateAttDataPrep = struct {
+    data: attestation.AttestationData,
+    outcome: AggregateAttDataOutcome,
+
+    fn deinit(self: *AggregateAttDataPrep, allocator: Allocator) void {
+        switch (self.outcome) {
+            .skip => {},
+            .ffi => |*args| args.deinit(allocator),
+            .done => |*result| {
+                result.attestation.deinit();
+                result.signature.deinit();
+            },
+        }
+    }
+};
+
+const AggregateAttDataFfiArgs = struct {
+    message_hash: [32]u8,
+    epoch: u64,
+    xmss_participants: ?attestation.AggregationBits,
+    selected_children: []aggregation.AggregatedSignatureProof,
+    child_pk_slices: []const []*const xmss.HashSigPublicKey,
+    pk_handles: []*const xmss.HashSigPublicKey,
+    sig_handles: []*const xmss.HashSigSignature,
+    child_pk_allocs: [][]*const xmss.HashSigPublicKey,
+    child_pk_wrappers: []xmss.PublicKey,
+    gossip_sig_wrappers: []xmss.Signature,
+    gossip_pk_wrappers: []xmss.PublicKey,
+    pk_handles_buf: ?[]*const xmss.HashSigPublicKey,
+    sig_handles_buf: ?[]*const xmss.HashSigSignature,
+
+    fn deinit(self: *AggregateAttDataFfiArgs, allocator: Allocator) void {
+        for (self.selected_children) |*child| child.deinit();
+        allocator.free(self.selected_children);
+        for (self.child_pk_allocs) |arr| allocator.free(arr);
+        allocator.free(self.child_pk_allocs);
+        for (self.child_pk_wrappers) |*pw| pw.deinit();
+        allocator.free(self.child_pk_wrappers);
+        for (self.gossip_sig_wrappers) |*sig| sig.deinit();
+        allocator.free(self.gossip_sig_wrappers);
+        for (self.gossip_pk_wrappers) |*pk| pk.deinit();
+        allocator.free(self.gossip_pk_wrappers);
+        if (self.pk_handles_buf) |buf| allocator.free(buf);
+        if (self.sig_handles_buf) |buf| allocator.free(buf);
+        if (self.xmss_participants) |*gp| gp.deinit();
+    }
+};
+
+fn prepareAggregateAttData(
+    allocator: Allocator,
+    validators: *const Validators,
+    signatures_map: *const SignaturesMap,
+    new_payloads: ?*const AggregatedPayloadsMap,
+    known_payloads: ?*const AggregatedPayloadsMap,
+    data: attestation.AttestationData,
+) !AggregateAttDataPrep {
+    const epoch: u64 = data.slot;
+    var message_hash: [32]u8 = undefined;
+    try zeam_utils.hashTreeRoot(attestation.AttestationData, data, &message_hash, allocator);
+
+    var selected_children: std.ArrayList(aggregation.AggregatedSignatureProof) = .empty;
+    errdefer {
+        for (selected_children.items) |*child| child.deinit();
+        selected_children.deinit(allocator);
+    }
+
+    const max_validator = validators.len();
+
+    var covered_by_children = try std.DynamicBitSet.initEmpty(allocator, max_validator);
+    defer covered_by_children.deinit();
+
+    var empty_available = try std.DynamicBitSet.initEmpty(allocator, max_validator);
+    defer empty_available.deinit();
+
+    try extendProofsGreedily(allocator, new_payloads, data, &selected_children, &covered_by_children, &empty_available);
+    try extendProofsGreedily(allocator, known_payloads, data, &selected_children, &covered_by_children, &empty_available);
+
+    var sigmap_sigs: std.ArrayList(xmss.Signature) = .empty;
+    errdefer {
+        for (sigmap_sigs.items) |*sig| sig.deinit();
+        sigmap_sigs.deinit(allocator);
+    }
+
+    var sigmap_pks: std.ArrayList(xmss.PublicKey) = .empty;
+    errdefer {
+        for (sigmap_pks.items) |*pk| pk.deinit();
+        sigmap_pks.deinit(allocator);
+    }
+
+    var sigmap_vids: std.ArrayList(usize) = .empty;
+    errdefer sigmap_vids.deinit(allocator);
+
+    if (signatures_map.get(data)) |im| {
+        var vid_it = im.iterator();
+        while (vid_it.next()) |entry| {
+            const vid: usize = @intCast(entry.key_ptr.*);
+            const sig_entry = entry.value_ptr.*;
+
+            if (vid < covered_by_children.capacity() and covered_by_children.isSet(vid)) continue;
+            if (std.mem.eql(u8, &sig_entry.signature, &ZERO_SIGBYTES)) continue;
+
+            var sig = xmss.Signature.fromBytes(&sig_entry.signature) catch continue;
+            errdefer sig.deinit();
+
+            if (vid >= validators.len()) {
+                sig.deinit();
+                continue;
+            }
+
+            const val = validators.get(vid) catch {
+                sig.deinit();
+                continue;
+            };
+            const pk = xmss.PublicKey.fromBytes(&val.attestation_pubkey) catch {
+                sig.deinit();
+                continue;
+            };
+
+            try sigmap_sigs.append(allocator, sig);
+            try sigmap_pks.append(allocator, pk);
+            try sigmap_vids.append(allocator, vid);
+        }
+    }
+
+    const has_gossip = sigmap_sigs.items.len > 0;
+    const has_children = selected_children.items.len > 0;
+
+    if (!has_gossip and !has_children) {
+        return .{ .data = data, .outcome = .skip };
+    }
+
+    if (!has_gossip and selected_children.items.len == 1) {
+        const child = &selected_children.items[0];
+
+        var att_bits: attestation.AggregationBits = undefined;
+        try utils.sszClone(allocator, attestation.AggregationBits, child.participants, &att_bits);
+        errdefer att_bits.deinit();
+
+        var cloned_child: aggregation.AggregatedSignatureProof = undefined;
+        try utils.sszClone(allocator, aggregation.AggregatedSignatureProof, child.*, &cloned_child);
+        errdefer cloned_child.deinit();
+
+        selected_children.items[0].deinit();
+        selected_children.deinit(allocator);
+
+        return .{
+            .data = data,
+            .outcome = .{
+                .done = .{
+                    .attestation = .{ .aggregation_bits = att_bits, .data = data },
+                    .signature = cloned_child,
+                },
+            },
+        };
+    }
+
+    var xmss_participants: ?attestation.AggregationBits = null;
+    var pk_handles_buf: ?[]*const xmss.HashSigPublicKey = null;
+    var sig_handles_buf: ?[]*const xmss.HashSigSignature = null;
+    var pk_handles: []*const xmss.HashSigPublicKey = &.{};
+    var sig_handles: []*const xmss.HashSigSignature = &.{};
+
+    if (has_gossip) {
+        var gp = try attestation.AggregationBits.init(allocator);
+        errdefer gp.deinit();
+
+        const pks = try allocator.alloc(*const xmss.HashSigPublicKey, sigmap_sigs.items.len);
+        errdefer allocator.free(pks);
+        const sigs = try allocator.alloc(*const xmss.HashSigSignature, sigmap_sigs.items.len);
+        errdefer allocator.free(sigs);
+
+        for (sigmap_vids.items, 0..) |vid, idx| {
+            try attestation.aggregationBitsSet(&gp, vid, true);
+            pks[idx] = sigmap_pks.items[idx].handle;
+            sigs[idx] = sigmap_sigs.items[idx].handle;
+        }
+
+        xmss_participants = gp;
+        pk_handles_buf = pks;
+        sig_handles_buf = sigs;
+        pk_handles = pks;
+        sig_handles = sigs;
+    }
+
+    var child_pk_allocs_list: std.ArrayList([]*const xmss.HashSigPublicKey) = .empty;
+    errdefer {
+        for (child_pk_allocs_list.items) |arr| allocator.free(arr);
+        child_pk_allocs_list.deinit(allocator);
+    }
+    var child_pk_slices_list: std.ArrayList([]*const xmss.HashSigPublicKey) = .empty;
+    errdefer child_pk_slices_list.deinit(allocator);
+
+    var child_pk_wrappers_list: std.ArrayList(xmss.PublicKey) = .empty;
+    errdefer {
+        for (child_pk_wrappers_list.items) |*pw| pw.deinit();
+        child_pk_wrappers_list.deinit(allocator);
+    }
+
+    for (selected_children.items) |*child| {
+        var n_participants: usize = 0;
+        for (0..child.participants.len()) |i| {
+            if (child.participants.get(i) catch false) n_participants += 1;
+        }
+
+        const cpks = try allocator.alloc(*const xmss.HashSigPublicKey, n_participants);
+        errdefer allocator.free(cpks);
+
+        var cpk_idx: usize = 0;
+        for (0..child.participants.len()) |i| {
+            if (child.participants.get(i) catch false) {
+                if (i >= validators.len()) continue;
+                const val = validators.get(@intCast(i)) catch continue;
+                const pk = xmss.PublicKey.fromBytes(&val.attestation_pubkey) catch continue;
+                try child_pk_wrappers_list.append(allocator, pk);
+                cpks[cpk_idx] = pk.handle;
+                cpk_idx += 1;
+            }
+        }
+
+        try child_pk_allocs_list.append(allocator, cpks);
+        try child_pk_slices_list.append(allocator, cpks[0..cpk_idx]);
+    }
+
+    const gossip_sigs = try sigmap_sigs.toOwnedSlice(allocator);
+    errdefer {
+        for (gossip_sigs) |*sig| sig.deinit();
+        allocator.free(gossip_sigs);
+    }
+    sigmap_sigs = .empty;
+
+    const gossip_pks = try sigmap_pks.toOwnedSlice(allocator);
+    errdefer {
+        for (gossip_pks) |*pk| pk.deinit();
+        allocator.free(gossip_pks);
+    }
+    sigmap_pks = .empty;
+    sigmap_vids.deinit(allocator);
+
+    const children = try selected_children.toOwnedSlice(allocator);
+    selected_children = .empty;
+
+    const child_pk_allocs = try child_pk_allocs_list.toOwnedSlice(allocator);
+    child_pk_allocs_list = .empty;
+
+    const child_pk_slices = try child_pk_slices_list.toOwnedSlice(allocator);
+    child_pk_slices_list = .empty;
+
+    const child_pk_wrappers = try child_pk_wrappers_list.toOwnedSlice(allocator);
+    child_pk_wrappers_list = .empty;
+
+    return .{
+        .data = data,
+        .outcome = .{
+            .ffi = .{
+                .message_hash = message_hash,
+                .epoch = epoch,
+                .xmss_participants = xmss_participants,
+                .selected_children = children,
+                .child_pk_slices = child_pk_slices,
+                .pk_handles = pk_handles,
+                .sig_handles = sig_handles,
+                .child_pk_allocs = child_pk_allocs,
+                .child_pk_wrappers = child_pk_wrappers,
+                .gossip_sig_wrappers = gossip_sigs,
+                .gossip_pk_wrappers = gossip_pks,
+                .pk_handles_buf = pk_handles_buf,
+                .sig_handles_buf = sig_handles_buf,
+            },
+        },
+    };
+}
+
+fn runAggregateAttDataFfi(
+    allocator: Allocator,
+    data: attestation.AttestationData,
+    args: *AggregateAttDataFfiArgs,
+) !CompactGroupResult {
+    var proof = try aggregation.AggregatedSignatureProof.init(allocator);
+    errdefer proof.deinit();
+
+    const pq_sig_timer = zeam_metrics.lean_pq_sig_aggregated_signatures_building_time_seconds.start();
+    try aggregation.AggregatedSignatureProof.aggregate(
+        allocator,
+        args.xmss_participants,
+        args.selected_children,
+        args.child_pk_slices,
+        args.pk_handles,
+        args.sig_handles,
+        &args.message_hash,
+        args.epoch,
+        &proof,
+    );
+    _ = pq_sig_timer.observe();
+
+    var att_bits: attestation.AggregationBits = undefined;
+    try utils.sszClone(allocator, attestation.AggregationBits, proof.participants, &att_bits);
+    errdefer att_bits.deinit();
+
+    return .{
+        .attestation = .{ .aggregation_bits = att_bits, .data = data },
+        .signature = proof,
+    };
+}
+
+fn runAggregateAttDataPreps(allocator: Allocator, preps: []AggregateAttDataPrep, thread_pool: *ThreadPool) !void {
+    const slots = try allocator.alloc(AggregateAttDataSlot, preps.len);
+    defer allocator.free(slots);
+    for (slots) |*slot| slot.* = .{};
+
+    const Runner = struct {
+        fn runScope(
+            scope: anytype,
+            alloc: Allocator,
+            preps_in: []AggregateAttDataPrep,
+            out_slots: []AggregateAttDataSlot,
+            any_err: *std.atomic.Value(bool),
+        ) Allocator.Error!void {
+            for (preps_in, 0..) |*prep, i| {
+                if (prep.outcome != .ffi) continue;
+                try scope.spawn(runOne, .{ alloc, prep, &out_slots[i], any_err });
+            }
+        }
+
+        fn runOne(
+            alloc: Allocator,
+            prep: *AggregateAttDataPrep,
+            out_slot: *AggregateAttDataSlot,
+            any_err: *std.atomic.Value(bool),
+        ) void {
+            if (any_err.load(.acquire)) return;
+            const result = runAggregateAttDataFfi(alloc, prep.data, &prep.outcome.ffi) catch |err| {
+                out_slot.err = err;
+                any_err.store(true, .release);
+                return;
+            };
+            out_slot.result = result;
+        }
+    };
+
+    var any_err = std.atomic.Value(bool).init(false);
+    try thread_pool.scope(Runner.runScope, .{ allocator, preps, slots, &any_err });
+
+    for (slots) |slot| {
+        if (slot.err) |err| return err;
+    }
+
+    for (preps, 0..) |*prep, i| {
+        if (prep.outcome != .ffi) continue;
+        const result = slots[i].result orelse return error.AggregateParallelMissingResult;
+        prep.outcome.ffi.deinit(allocator);
+        prep.outcome = .{ .done = result };
+    }
+}
+
+const AggregateAttDataSlot = struct {
+    result: ?CompactGroupResult = null,
+    err: ?anyerror = null,
+};
+
 const CompactGroupSlot = struct {
     result: ?CompactGroupResult = null,
     err: ?anyerror = null,
@@ -1322,6 +1524,14 @@ fn testPutSingleChildPayload(allocator: Allocator, payloads: *AggregatedPayloads
     });
 }
 
+fn initTestThreadPool(allocator: Allocator) !*ThreadPool {
+    return ThreadPool.init(.{
+        .allocator = allocator,
+        .io = std.Io.Threaded.global_single_threaded.io(),
+        .thread_count = 4,
+    });
+}
+
 test "computeAggregatedSignatures filters attestation data by slot list" {
     const allocator = std.testing.allocator;
 
@@ -1342,8 +1552,11 @@ test "computeAggregatedSignatures filters attestation data by slot list" {
     var result = try AggregatedAttestationsResult.init(allocator);
     defer result.deinit();
 
+    const thread_pool = try initTestThreadPool(allocator);
+    defer thread_pool.deinit();
+
     const allowed_slots = [_]Slot{10};
-    try result.computeAggregatedSignatures(&validators, &signatures, &payloads, null, allowed_slots[0..]);
+    try result.computeAggregatedSignatures(&validators, &signatures, &payloads, null, allowed_slots[0..], thread_pool);
 
     try std.testing.expectEqual(@as(usize, 1), result.attestations.len());
     const aggregated = try result.attestations.get(0);
@@ -1372,12 +1585,15 @@ test "computeAggregatedSignatures slot filter matches unfiltered for same-slot i
 
     var unfiltered = try AggregatedAttestationsResult.init(allocator);
     defer unfiltered.deinit();
-    try unfiltered.computeAggregatedSignatures(&validators, &signatures_a, &payloads_a, null, null);
+
+    const thread_pool = try initTestThreadPool(allocator);
+    defer thread_pool.deinit();
+    try unfiltered.computeAggregatedSignatures(&validators, &signatures_a, &payloads_a, null, null, thread_pool);
 
     var filtered = try AggregatedAttestationsResult.init(allocator);
     defer filtered.deinit();
     const allowed_slots = [_]Slot{12};
-    try filtered.computeAggregatedSignatures(&validators, &signatures_b, &payloads_b, null, allowed_slots[0..]);
+    try filtered.computeAggregatedSignatures(&validators, &signatures_b, &payloads_b, null, allowed_slots[0..], thread_pool);
 
     try std.testing.expectEqual(unfiltered.attestations.len(), filtered.attestations.len());
     try std.testing.expectEqual(unfiltered.attestation_signatures.len(), filtered.attestation_signatures.len());
@@ -1404,7 +1620,9 @@ test "computeAggregatedSignatures empty slot filter result is clean" {
     defer result.deinit();
 
     const allowed_slots = [_]Slot{14};
-    try result.computeAggregatedSignatures(&validators, &signatures, &payloads, null, allowed_slots[0..]);
+    const thread_pool = try initTestThreadPool(allocator);
+    defer thread_pool.deinit();
+    try result.computeAggregatedSignatures(&validators, &signatures, &payloads, null, allowed_slots[0..], thread_pool);
 
     try std.testing.expectEqual(@as(usize, 0), result.attestations.len());
     try std.testing.expectEqual(@as(usize, 0), result.attestation_signatures.len());