fix(tbtc/signer): bind the interactive completion marker to the message (re-review)

Codex re-review on the Round2 completion gate:

[P1] The gate trusted aggregated_interactive_attempt_markers keyed by attempt_id
alone, but interactive_aggregate never binds attempt_id to the aggregated package/
message - it only checks the shares aggregate under the session key. So a caller
with any valid aggregate could replay it under a DIFFERENT live attempt's id, set
that marker, and the new gate would then refuse the real Round2 before its nonce is
consumed (an availability/DoS regression the single-member engine never had).

Fix: make the completion marker MESSAGE-BOUND - interactive_aggregated_marker =
"{attempt_id}@{message_digest}". interactive_aggregate writes it from the package it
actually aggregated (all five marker sites: both completion pre-checks, the capacity
guard, the insert, and the persist-rollback). The Round2 gate recomputes it from the
message THIS member opened, so it preempts Round2 only for a genuine same-message
completion; a replayed aggregate carrying a different message sets an unrelated
marker that cannot match. Same-message re-aggregation is still rejected (existing
restart/repeat tests pass unchanged - they re-aggregate the same message).

[P2] On that genuine completion rejection, the unsigned sibling's entry is now dead,
so free it (zeroizing its nonces) instead of letting it hold a live-member cap slot
until the TTL sweep. Only the matching member's entry is removed.

Tests: the multi-seat refusal test now also asserts the marker is bound
(attempt_id@digest, not the bare id) and that the refused sibling's entry is freed.
All 294 lib tests pass; cargo fmt clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: mswilkison

pkg/tbtc/signer/src/engine/interactive.rs

@@ -38,6 +38,16 @@ pub(crate) fn interactive_attempt_consumed(
         || markers.contains(attempt_id)
 }
 
+// The aggregate completion marker binds attempt_id to the AGGREGATED message digest,
+// so the durable "this attempt is final" record cannot be set for one attempt id via
+// a valid aggregate over a DIFFERENT message - which would otherwise let a replayed
+// aggregate preempt an unrelated live attempt's Round2 (the Round2 completion gate).
+// interactive_aggregate writes it from the package it aggregated; Round2 and the
+// re-aggregate guard recompute it from the message they actually hold.
+pub(crate) fn interactive_aggregated_marker(attempt_id: &str, message_digest_hex: &str) -> String {
+    format!("{attempt_id}@{message_digest_hex}")
+}
+
 pub fn interactive_session_open(
     mut request: InteractiveSessionOpenRequest,
 ) -> Result<InteractiveSessionOpenResult, EngineError> {
@@ -473,15 +483,33 @@ pub fn interactive_round2(
     }
 
     // A completed attempt releases no further shares. Once interactive_aggregate has
-    // produced the attempt's signature (aggregated_interactive_attempt_markers), an
-    // open sibling seat that never signed has NO per-member consumed marker, so the
-    // consumed gate above does not cover it. Gate on the completion marker here too:
-    // re-aggregation is not a recovery path (a lost signature is recovered with a
-    // fresh attempt), so no fresh share may leave for a finished attempt.
-    if session
-        .aggregated_interactive_attempt_markers
-        .contains(&attempt_id)
-    {
+    // produced the attempt's signature, an open sibling seat that never signed has NO
+    // per-member consumed marker, so the consumed gate above does not cover it. Gate
+    // on the completion marker too - but matched to the MESSAGE this member opened
+    // (the marker binds attempt_id to the aggregated message digest), so a replayed
+    // aggregate carrying a different message for this attempt id cannot preempt this
+    // member's live Round2. It fires only for a genuine same-message completion; the
+    // member's entry for that finalized attempt is then dead, so free it (zeroizing
+    // its nonces) rather than holding a live-member slot until the TTL sweep.
+    let member_attempt_message_digest = session
+        .interactive_signing
+        .get(&request.member_identifier)
+        .filter(|entry| entry.attempt_context.attempt_id == attempt_id)
+        .map(|entry| hash_hex(&entry.message_bytes));
+    let attempt_finalized = member_attempt_message_digest
+        .as_deref()
+        .is_some_and(|digest| {
+            session
+                .aggregated_interactive_attempt_markers
+                .contains(&interactive_aggregated_marker(&attempt_id, digest))
+        });
+    if attempt_finalized {
+        if let Some(mut removed) = session
+            .interactive_signing
+            .remove(&request.member_identifier)
+        {
+            zeroize_interactive_round1(&mut removed);
+        }
         return Err(EngineError::InteractiveAttemptAlreadyAggregated {
             session_id: request.session_id.clone(),
             attempt_id,
@@ -675,6 +703,11 @@ pub fn interactive_aggregate(
             "InteractiveAggregate: invalid signing package: {e}"
         ))
     })?;
+    // The completion marker binds attempt_id to THIS aggregated message digest, so a
+    // valid aggregate over a different message cannot finalize this attempt id (and
+    // so cannot, via the Round2 completion gate, preempt an unrelated live attempt).
+    let aggregated_marker =
+        interactive_aggregated_marker(&attempt_id, &hash_hex(signing_package.message().as_slice()));
     let signature_shares =
         decode_signature_share_map("InteractiveAggregate", &request.signature_shares)?;
     let mut taproot_merkle_root_hex = request.taproot_merkle_root_hex.clone();
@@ -705,7 +738,7 @@ pub fn interactive_aggregate(
         // durable so a completed attempt stays rejected across a restart.
         if session
             .aggregated_interactive_attempt_markers
-            .contains(&attempt_id)
+            .contains(&aggregated_marker)
         {
             return Err(EngineError::InteractiveAttemptAlreadyAggregated {
                 session_id: request.session_id.clone(),
@@ -816,7 +849,7 @@ pub fn interactive_aggregate(
     // re-aggregation - the winner already produced the attempt's signature.
     if session
         .aggregated_interactive_attempt_markers
-        .contains(&attempt_id)
+        .contains(&aggregated_marker)
     {
         return Err(EngineError::InteractiveAttemptAlreadyAggregated {
             session_id: request.session_id.clone(),
@@ -825,21 +858,21 @@ pub fn interactive_aggregate(
     }
     ensure_consumed_registry_insert_capacity(
         &session.aggregated_interactive_attempt_markers,
-        &attempt_id,
+        &aggregated_marker,
         "aggregated_interactive_attempt_markers",
         &request.session_id,
     )?;
     session
         .aggregated_interactive_attempt_markers
-        .insert(attempt_id.clone());
+        .insert(aggregated_marker.clone());
     if let Err(persist_error) = persist_engine_state_to_storage(&guard) {
         let session = guard
             .sessions
             .get_mut(&request.session_id)
             .expect("session existed under the held engine lock");
         session
             .aggregated_interactive_attempt_markers
-            .remove(&attempt_id);
+            .remove(&aggregated_marker);
         return Err(persist_error);
     }
     drop(guard);

pkg/tbtc/signer/src/engine/tests.rs

@@ -11808,6 +11808,27 @@ fn interactive_round2_refused_after_aggregate_for_unsigned_sibling() {
     })
     .expect("interactive aggregate completes the attempt");
 
+    // The completion marker is MESSAGE-BOUND (attempt_id@digest), not the bare
+    // attempt_id - so it cannot be set for this attempt id via an aggregate over a
+    // different message (which would otherwise preempt this attempt's live Round2).
+    {
+        let guard = state().expect("state").lock().expect("lock");
+        let session = guard.sessions.get(session_id).expect("session exists");
+        assert!(
+            session
+                .aggregated_interactive_attempt_markers
+                .iter()
+                .any(|marker| marker.starts_with(&format!("{}@", opened.attempt_id))),
+            "completion marker binds attempt_id to the aggregated message digest"
+        );
+        assert!(
+            !session
+                .aggregated_interactive_attempt_markers
+                .contains(&opened.attempt_id),
+            "the bare (unbound) attempt_id marker is not written"
+        );
+    }
+
     // Seat 3 (open, round1'd, never signed) tries to release a share for the now
     // completed attempt, in a subset it WOULD be valid for ({1,3}). Without the
     // completion gate this releases a fresh share; with it the attempt is final.
@@ -11838,6 +11859,17 @@ fn interactive_round2_refused_after_aggregate_for_unsigned_sibling() {
         ),
         "unexpected error: {refused:?}"
     );
+
+    // The refused sibling's now-dead entry is freed (its nonces zeroized) rather
+    // than lingering against the live-member cap until the TTL sweep.
+    {
+        let guard = state().expect("state").lock().expect("lock");
+        let session = guard.sessions.get(session_id).expect("session exists");
+        assert!(
+            !session.interactive_signing.contains_key(&3),
+            "seat 3's dead entry is freed when Round2 is refused for the finalized attempt"
+        );
+    }
 }
 
 #[test]