Aggregate-first follow-up: package single-shard roots and prune empty leaves

Ix/Aiur/Protocol.lean

@@ -25,6 +25,12 @@ opaque toBytes : @& Proof → ByteArray
 @[extern "rs_aiur_proof_of_bytes"]
 opaque ofBytes : @& ByteArray → Proof
 
+/-- Decode an untrusted serialized proof without aborting the process. Cache
+and network boundaries must use this variant; `ofBytes` remains for callers
+whose bytes were produced in-process or already validated. -/
+@[extern "rs_aiur_proof_of_bytes_checked"]
+opaque ofBytesChecked : @& ByteArray → Except String Proof
+
 end Proof
 
 structure CommitmentParameters where

Ix/Cli/BenchCmd.lean

@@ -136,11 +136,13 @@ def workloadOf (testbed : String) : String :=
 /-- The stage qualifiers a pipeline measure may carry ahead of its base
     name — one per pipeline stage, named for what the stage proves
     (`ixvm-`: the IxVM typecheck; `fri-verifier-`: the in-circuit FRI
-    verifier over the previous proof; the KZG stages add their own
-    entries as they land) — plus `pipeline-` for the whole run. Stripped
+    verifier over the previous proof; `join-`: the optional pair-wide
+    aggregate join; the KZG stages add their own entries as they land) —
+    plus `pipeline-` for the whole run. Stripped
     wherever a measure is interpreted by its base name (formatting kind,
     units) or labelled under a heading that already says the stage. -/
-def stagePrefixes : List String := ["ixvm-", "fri-verifier-", "pipeline-"]
+def stagePrefixes : List String :=
+  ["ixvm-", "fri-verifier-", "join-", "pipeline-"]
 
 /-- The stage qualifier `metric` carries, if any. -/
 def stagePrefixOf (metric : String) : Option String :=
@@ -213,7 +215,9 @@ structure BackendSpec where
 
 def backendSpecs : List BackendSpec := [
   -- aiur: the proof-pipeline benchmark (bench-typecheck --recursive) —
-  -- every stage of the pipeline, per constant, plus the total. The ixvm
+  -- every stage of the pipeline, per constant, plus the total. The optional
+  -- `bench-typecheck --join` W0 diagnostic contributes a separate pair row;
+  -- it is not part of the scheduled one-constant CI invocation. The ixvm
   -- stage proves the constant's IxVM typecheck; the fri-verifier stage
   -- executes the in-circuit multi-stark verifier over that fresh proof
   -- and proves THAT execution; the KZG stages will join as stages 3/4
@@ -245,6 +249,9 @@ def backendSpecs : List BackendSpec := [
           "fri-verifier-throughput", "fri-verifier-peak-rss",
           "fri-verifier-proof-size", "fri-verifier-verify-time",
           "fri-verifier-fft-cost"]),
+       ("Aggregate flat join",
+         ["join-execute-time", "join-prove-time", "join-peak-rss",
+          "join-proof-size", "join-verify-time", "join-fft-cost"]),
        ("Pipeline total",
          ["total-time", "pipeline-throughput", "pipeline-peak-rss"])])],
     metrics := [("execute", ["execute-time", "throughput", "peak-rss",
@@ -269,6 +276,12 @@ def backendSpecs : List BackendSpec := [
                    ("fri-verifier-prove-time", "0.10", "_"),
                    ("ixvm-peak-rss", "0.10", "_"),
                    ("fri-verifier-peak-rss", "0.10", "_"),
+                   ("join-execute-time", "0.10", "_"),
+                   ("join-prove-time", "0.10", "_"),
+                   ("join-peak-rss", "0.10", "_"),
+                   ("join-proof-size", "0.05", "_"),
+                   ("join-verify-time", "0.10", "_"),
+                   ("join-fft-cost", "0.25", "_"),
                    ("pipeline-peak-rss", "0.10", "_"),
                    ("ixvm-proof-size", "0.05", "_"),
                    ("fri-verifier-proof-size", "0.05", "_"),
@@ -418,6 +431,14 @@ def BackendSpec.envNames (b : BackendSpec) : List String :=
       b.scheduledModes.any fun m =>
         !(selectNames env b.name m).isEmpty
 
+/-- Stable pair rows produced by the opt-in aggregate W0 diagnostic. These are
+    registered for dashboard filtering even though the scheduled aiur cell
+    remains one process per constant and therefore does not produce them. Keep
+    the order synchronized with the documented `bench-typecheck --join`
+    invocation: pair-row identity is deliberately order-sensitive. -/
+def aiurJoinBenchmarkNames : Array String :=
+  #["Nat.add_comm + String.append"]
+
 /-- The benchmark row names this backend uploads — the bencher slugs the
     dashboard plots and compare table key on — from its `inputs`: env-keyed
     backends key one row per compiled env; the per-constant backends select
@@ -433,6 +454,8 @@ def BackendSpec.benchmarkNames (b : BackendSpec) (mode : String) :
     for env in b.envNames do
       if b.inputs == .perConstantWithEnv then ns := ns.push env
       ns := ns ++ (selectNames env b.name mode).map (·.name)
+    if b.name == "aiur" && mode == "prove" then
+      return ns ++ aiurJoinBenchmarkNames
     return ns
 
 /-- Default RAM watchdog ceiling (`--ceiling-gb` overrides): see

Ix/Cli/BenchPlots.lean

@@ -70,6 +70,11 @@ def plotTitle (workload measure : String) : String :=
   | "aiur", "fri-verifier-peak-rss"    => "Aiur FRI Verifier Peak RAM Usage"
   | "aiur", "ixvm-proof-size"          => "Aiur IxVM Proof Size"
   | "aiur", "fri-verifier-proof-size"  => "Aiur FRI Verifier Proof Size"
+  | "aiur", "join-prove-time"          => "Aiur Aggregate Join Time"
+  | "aiur", "join-fft-cost"            => "Aiur Aggregate Join FFT Cost"
+  | "aiur", "join-verify-time"         => "Aiur Aggregate Join Verify Time"
+  | "aiur", "join-peak-rss"            => "Aiur Aggregate Join Peak RAM Usage"
+  | "aiur", "join-proof-size"          => "Aiur Aggregate Join Proof Size"
   | "zisk-check-execute", "execute-time" => "Zisk Execute Time"
   | "zisk-check-execute", "throughput"   => "Zisk Execute Throughput"
   | "zisk-check-execute", "peak-rss"     => "Zisk Execute Peak RAM Usage"
@@ -111,6 +116,7 @@ def plotSkips : List (String × String) :=
    ("ix-decompile", "file-size"), ("ix-decompile", "constants"),
    ("aiur", "ixvm-peak-rss"), ("aiur", "ixvm-verify-time"),
    ("aiur", "ixvm-execute-time"), ("aiur", "fri-verifier-execute-time"),
+   ("aiur", "join-execute-time"),
    ("aiur", "ixvm-throughput"), ("aiur", "fri-verifier-throughput")]
 
 /-- Canonical units per measure slug, asserted on every sync: bencher

Ix/Cli/CheckCmd.lean

@@ -471,10 +471,25 @@ where
       let (rightIdx, ops) := go right ops
       (ops.size, ops.push (.join leftIdx rightIdx))
 
+/-- Drop removed shard leaves, contract unary nodes, and rewrite retained shard
+ids to their dense indices in a pruned manifest view. The input tree has
+already passed `validateAggregationTree`, so an out-of-range mapping entry is
+an internal inconsistency rather than untrusted manifest input. -/
+partial def pruneAndRemap (remap : Array (Option Nat)) :
+    AggregationTree → Option AggregationTree
+  | .leaf shard => (remap[shard]?).join.map .leaf
+  | .node left right =>
+    match pruneAndRemap remap left, pruneAndRemap remap right with
+    | some left, some right => some (.node left right)
+    | some tree, none | none, some tree => some tree
+    | none, none => none
+
 end AggregationTree
 
 structure IxesManifestView where
   shards : Array (Array Address)
+  /-- Original manifest shard id for each (possibly pruned) dense array slot. -/
+  shardIds : Array Nat
   aggregationTree : AggregationTree
   deriving BEq, Repr
 
@@ -545,7 +560,7 @@ def parseIxesManifest (bytes : ByteArray) : Except String IxesManifestView :=
     let (buffer, position) ← get
     if position != buffer.size then
       throw s!"ixes: {buffer.size - position} trailing bytes after aggregation tree"
-    pure { shards, aggregationTree := tree }
+    pure { shards, shardIds := Array.range n.toNat, aggregationTree := tree }
   go.run' (bytes, 0)
 
 /-- Backward-compatible shard-only view used by check/prove/verify paths that
@@ -595,11 +610,42 @@ def ownedConstCountsForShards (ixonEnv : Ixon.Env)
     counts := counts.modify shard (· + 1)
   return counts
 
+/-- Remove manifest shards that provably own no environment constants, then
+contract and densely reindex the corresponding aggregation-tree leaves.
+
+Callers must run `shardsCover` on the unpruned view first. That gate establishes
+that every constant is owned exactly once; the zero counts here therefore prove
+that dropping these leaves cannot omit a checked subject. `shardIds` preserves
+the original ids for diagnostics and for matching legacy manifests. -/
+def IxesManifestView.pruneEmpty (view : IxesManifestView)
+    (ixonEnv : Ixon.Env) : Except String (IxesManifestView × Array Nat) := do
+  if view.shards.size != view.shardIds.size then
+    throw "ixes: internal shard/id cardinality mismatch"
+  let counts := ownedConstCountsForShards ixonEnv view.shards
+  let mut remap : Array (Option Nat) := Array.replicate view.shards.size none
+  let mut shards : Array (Array Address) := #[]
+  let mut shardIds : Array Nat := #[]
+  let mut keptCounts : Array Nat := #[]
+  for (count, oldIdx) in counts.mapIdx fun oldIdx count => (count, oldIdx) do
+    if count != 0 then
+      let some blocks := view.shards[oldIdx]?
+        | throw s!"ixes: internal missing shard {oldIdx}"
+      let some originalId := view.shardIds[oldIdx]?
+        | throw s!"ixes: internal missing shard id {oldIdx}"
+      remap := remap.set! oldIdx (some shards.size)
+      shards := shards.push blocks
+      shardIds := shardIds.push originalId
+      keptCounts := keptCounts.push count
+  let some aggregationTree := view.aggregationTree.pruneAndRemap remap
+    | throw "aggregate: manifest has no shard owning an environment constant"
+  pure ({ shards, shardIds, aggregationTree }, keptCounts)
+
 /-- The `CheckEnv` claim digest a shard's proof commits to — reconstructed
     deterministically from the env + the shard's owned blocks. Matches the
     digest `prove --shard K` produced, so a proof can be bound to its shard. -/
 def shardClaimDigest (ixonEnv : Ixon.Env) (blocks : Array Address) : Except String Address := do
-  let (claim, _, _) ← IxVM.ClaimHarness.shardCheckEnvClaim ixonEnv (ownedConstsForBlocks ixonEnv blocks)
+  let (claim, _) ← IxVM.ClaimHarness.shardCheckEnvClaimTrees ixonEnv
+    (ownedConstsForBlocks ixonEnv blocks)
   pure (Address.blake3 (Ix.Claim.ser claim))
 
 /-- Load the `.ixe` env and the `.ixes` shard partition together (each file