feat(tbtc/signer): Phase 7.2b-1 InteractiveAggregate completion marker

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

@@ -574,6 +574,15 @@ pub fn interactive_aggregate(
     enforce_provenance_gate()?;
     validate_session_id(&request.session_id)?;
     let attempt_id = canonical_attempt_id(&request.attempt_id);
+    // The completion marker persists attempt_id, and the reload path rejects an
+    // empty key; reject an empty attempt_id here too so a malformed (or
+    // malicious) request cannot write durable state that fails to reload after
+    // a restart.
+    if attempt_id.is_empty() {
+        return Err(EngineError::Validation(
+            "InteractiveAggregate: attempt_id must not be empty".to_string(),
+        ));
+    }
 
     let mut signing_package_bytes = decode_hex_field(
         "InteractiveAggregate",
@@ -612,6 +621,18 @@ pub fn interactive_aggregate(
                 session_id: request.session_id.clone(),
             }
         })?;
+        // Reject a completed attempt: re-aggregation is not a recovery path (a
+        // lost signature is recovered with a fresh attempt), and the marker is
+        // durable so a completed attempt stays rejected across a restart.
+        if session
+            .aggregated_interactive_attempt_markers
+            .contains(&attempt_id)
+        {
+            return Err(EngineError::InteractiveAttemptAlreadyAggregated {
+                session_id: request.session_id.clone(),
+                attempt_id,
+            });
+        }
         if session.dkg_result.is_none() {
             return Err(EngineError::DkgNotReady {
                 session_id: request.session_id.clone(),
@@ -675,6 +696,54 @@ pub fn interactive_aggregate(
     let signature_bytes = signature
         .serialize()
         .map_err(|e| EngineError::Internal(format!("failed to serialize aggregate: {e}")))?;
+    let signature_hex = hex::encode(signature_bytes);
+
+    // Mark the attempt complete before reporting success, so a repeat
+    // InteractiveAggregate is rejected rather than recomputed (Phase 7.2b design
+    // section 6). The engine lock was dropped for the aggregation crypto above;
+    // re-acquire it, re-check the marker (a concurrent aggregate may have
+    // completed first), insert it, and persist before reporting success; on
+    // persist failure roll the marker back and fail closed.
+    let mut guard = state()?
+        .lock()
+        .map_err(|_| EngineError::Internal("engine lock poisoned".to_string()))?;
+    let session = guard.sessions.get_mut(&request.session_id).ok_or_else(|| {
+        EngineError::SessionNotFound {
+            session_id: request.session_id.clone(),
+        }
+    })?;
+    // A concurrent aggregate that raced past the pre-check may have completed
+    // this attempt first; if the marker is now present, reject this call's
+    // re-aggregation - the winner already produced the attempt's signature.
+    if session
+        .aggregated_interactive_attempt_markers
+        .contains(&attempt_id)
+    {
+        return Err(EngineError::InteractiveAttemptAlreadyAggregated {
+            session_id: request.session_id.clone(),
+            attempt_id,
+        });
+    }
+    ensure_consumed_registry_insert_capacity(
+        &session.aggregated_interactive_attempt_markers,
+        &attempt_id,
+        "aggregated_interactive_attempt_markers",
+        &request.session_id,
+    )?;
+    session
+        .aggregated_interactive_attempt_markers
+        .insert(attempt_id.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);
+        return Err(persist_error);
+    }
+    drop(guard);
 
     record_hardening_telemetry(|telemetry| {
         telemetry.interactive_aggregate_success_total = telemetry
@@ -685,7 +754,7 @@ pub fn interactive_aggregate(
     Ok(InteractiveAggregateResult {
         session_id: request.session_id,
         attempt_id,
-        signature_hex: hex::encode(signature_bytes),
+        signature_hex,
     })
 }
 

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

@@ -44,6 +44,11 @@ pub(crate) struct PersistedSessionState {
     // interactive state, including nonces, never persists).
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub(crate) consumed_interactive_attempt_markers: Vec<String>,
+    // Phase 7.2b InteractiveAggregate completion markers (see SessionState).
+    // serde(default) keeps state written before 7.2b loadable: an absent field
+    // deserializes to an empty set.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub(crate) aggregated_interactive_attempt_markers: Vec<String>,
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -1286,6 +1291,26 @@ impl TryFrom<PersistedSessionState> for SessionState {
             consumed_interactive_attempt_markers.len(),
             "consumed_interactive_attempt_markers",
         )?;
+
+        let mut aggregated_interactive_attempt_markers = HashSet::new();
+        for attempt_marker in persisted.aggregated_interactive_attempt_markers {
+            if attempt_marker.is_empty() {
+                return Err(EngineError::Internal(
+                    "persisted aggregated interactive attempt marker must be non-empty".to_string(),
+                ));
+            }
+
+            if !aggregated_interactive_attempt_markers.insert(attempt_marker.clone()) {
+                return Err(EngineError::Internal(format!(
+                    "duplicate persisted aggregated interactive attempt marker [{}]",
+                    attempt_marker
+                )));
+            }
+        }
+        ensure_consumed_registry_persisted_bound(
+            aggregated_interactive_attempt_markers.len(),
+            "aggregated_interactive_attempt_markers",
+        )?;
         if persisted.attempt_transition_records.len()
             > TBTC_SIGNER_MAX_ATTEMPT_TRANSITION_RECORDS_PER_SESSION
         {
@@ -1345,6 +1370,7 @@ impl TryFrom<PersistedSessionState> for SessionState {
             // only the consumption markers survive.
             interactive_signing: None,
             consumed_interactive_attempt_markers,
+            aggregated_interactive_attempt_markers,
         })
     }
 }
@@ -1455,6 +1481,12 @@ impl TryFrom<&SessionState> for PersistedSessionState {
             .cloned()
             .collect::<Vec<_>>();
         consumed_interactive_attempt_markers.sort_unstable();
+        let mut aggregated_interactive_attempt_markers = session_state
+            .aggregated_interactive_attempt_markers
+            .iter()
+            .cloned()
+            .collect::<Vec<_>>();
+        aggregated_interactive_attempt_markers.sort_unstable();
 
         Ok(PersistedSessionState {
             dkg_request_fingerprint: session_state.dkg_request_fingerprint.clone(),
@@ -1479,6 +1511,7 @@ impl TryFrom<&SessionState> for PersistedSessionState {
             refresh_history: session_state.refresh_history.clone(),
             emergency_rekey_event: session_state.emergency_rekey_event.clone(),
             consumed_interactive_attempt_markers,
+            aggregated_interactive_attempt_markers,
         })
     }
 }

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

@@ -111,6 +111,14 @@ pub(crate) struct SessionState {
     pub(crate) emergency_rekey_event: Option<EmergencyRekeyEvent>,
     pub(crate) interactive_signing: Option<InteractiveSigningState>,
     pub(crate) consumed_interactive_attempt_markers: HashSet<String>,
+    // Phase 7.2b InteractiveAggregate completion markers: an attempt whose
+    // aggregate signature has been produced is recorded here so a repeat
+    // InteractiveAggregate is rejected (re-aggregation is not a recovery path;
+    // a lost signature is recovered with a fresh attempt). Durable like the
+    // consumed markers (markers-only durability) and bounded the same way; not
+    // security-load-bearing (the aggregate is deterministic over public data),
+    // but the frozen Phase 7 spec marks the session complete.
+    pub(crate) aggregated_interactive_attempt_markers: HashSet<String>,
 }
 
 #[derive(Default)]

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

@@ -702,6 +702,7 @@ fn persisted_session_state_fixture() -> PersistedSessionState {
         refresh_history: vec![],
         emergency_rekey_event: None,
         consumed_interactive_attempt_markers: vec![],
+        aggregated_interactive_attempt_markers: vec![],
     }
 }
 
@@ -12840,6 +12841,199 @@ fn interactive_aggregate_produces_and_self_verifies_bip340() {
         .expect("interactive aggregate yields a valid BIP-340 signature");
 }
 
+#[test]
+fn interactive_aggregate_rejects_repeat_aggregate_of_completed_attempt() {
+    let _guard = lock_test_state();
+    reset_for_tests();
+
+    let session_id = "interactive-aggregate-repeat";
+    let key_group = "interactive-test-key-group";
+    let message = [0x4eu8; 32];
+    let included = [1u16, 2];
+    let key_packages = ensure_interactive_dkg_session(session_id, key_group);
+
+    let opened = open_interactive_for_test(session_id, key_group, &message, &included, 1, 1, 2)
+        .expect("opens");
+    let round1 = interactive_round1(InteractiveRound1Request {
+        session_id: session_id.to_string(),
+        attempt_id: opened.attempt_id.clone(),
+        member_identifier: 1,
+    })
+    .expect("round 1");
+    let member2 = generate_nonces_and_commitments(GenerateNoncesAndCommitmentsRequest {
+        key_package_identifier: key_packages[&2].identifier.clone(),
+        key_package_hex: key_packages[&2].data_hex.clone(),
+    })
+    .expect("member 2 nonces");
+    let signing_package_hex = interactive_package_for_test(
+        &message,
+        vec![
+            NativeFrostCommitment {
+                identifier: key_packages[&1].identifier.clone(),
+                data_hex: round1.commitments_hex,
+            },
+            member2.commitment.clone(),
+        ],
+    );
+    let round2 = interactive_round2(InteractiveRound2Request {
+        session_id: session_id.to_string(),
+        attempt_id: opened.attempt_id.clone(),
+        member_identifier: 1,
+        signing_package_hex: signing_package_hex.clone(),
+    })
+    .expect("round 2 share");
+    let member2_share = sign_share(SignShareRequest {
+        signing_package_hex: signing_package_hex.clone(),
+        nonces_hex: member2.nonces_hex,
+        key_package_identifier: key_packages[&2].identifier.clone(),
+        key_package_hex: key_packages[&2].data_hex.clone(),
+    })
+    .expect("member 2 share");
+
+    let aggregate_request = InteractiveAggregateRequest {
+        session_id: session_id.to_string(),
+        attempt_id: opened.attempt_id.clone(),
+        signing_package_hex,
+        signature_shares: vec![
+            crate::api::NativeFrostSignatureShare {
+                identifier: key_packages[&1].identifier.clone(),
+                data_hex: round2.signature_share_hex,
+            },
+            member2_share.signature_share,
+        ],
+        taproot_merkle_root_hex: None,
+    };
+
+    // First aggregate completes the attempt.
+    interactive_aggregate(aggregate_request.clone()).expect("first interactive aggregate");
+
+    // Re-aggregating a completed attempt is rejected by the durable completion
+    // marker rather than recomputed (re-aggregation is not a recovery path; a
+    // lost signature is recovered with a fresh attempt). Phase 7.2b design
+    // section 6.
+    let err = interactive_aggregate(aggregate_request)
+        .expect_err("re-aggregating a completed attempt must be rejected");
+    assert!(
+        matches!(
+            err,
+            EngineError::InteractiveAttemptAlreadyAggregated { ref attempt_id, .. }
+                if *attempt_id == opened.attempt_id
+        ),
+        "unexpected error: {err:?}"
+    );
+    assert_eq!(err.code(), "interactive_attempt_already_aggregated");
+    assert_eq!(err.recovery_class(), "recoverable");
+}
+
+#[test]
+fn interactive_aggregate_rejects_empty_attempt_id() {
+    let _guard = lock_test_state();
+    reset_for_tests();
+
+    // An empty attempt_id must be rejected before anything is persisted: the
+    // completion record is keyed by attempt_id and the reload path rejects an
+    // empty key, so persisting one would brick restart. The guard runs before
+    // the signing-package/share decode, so the placeholder inputs are never
+    // reached.
+    let err = interactive_aggregate(InteractiveAggregateRequest {
+        session_id: "interactive-aggregate-empty-attempt".to_string(),
+        attempt_id: String::new(),
+        signing_package_hex: String::new(),
+        signature_shares: vec![],
+        taproot_merkle_root_hex: None,
+    })
+    .expect_err("an empty attempt_id must be rejected");
+    assert!(
+        matches!(err, EngineError::Validation(ref m) if m.contains("attempt_id must not be empty")),
+        "unexpected error: {err:?}"
+    );
+}
+
+#[test]
+fn interactive_aggregate_completion_marker_survives_process_restart() {
+    let _guard = lock_test_state();
+    let state_path = configure_test_state_path("interactive_aggregate_marker_restart");
+    reset_for_tests();
+
+    let session_id = "interactive-aggregate-marker-restart";
+    let key_group = "interactive-test-key-group";
+    let message = [0x4fu8; 32];
+    let included = [1u16, 2];
+    let key_packages = ensure_interactive_dkg_session(session_id, key_group);
+
+    let opened = open_interactive_for_test(session_id, key_group, &message, &included, 1, 1, 2)
+        .expect("opens");
+    let round1 = interactive_round1(InteractiveRound1Request {
+        session_id: session_id.to_string(),
+        attempt_id: opened.attempt_id.clone(),
+        member_identifier: 1,
+    })
+    .expect("round 1");
+    let member2 = generate_nonces_and_commitments(GenerateNoncesAndCommitmentsRequest {
+        key_package_identifier: key_packages[&2].identifier.clone(),
+        key_package_hex: key_packages[&2].data_hex.clone(),
+    })
+    .expect("member 2 nonces");
+    let signing_package_hex = interactive_package_for_test(
+        &message,
+        vec![
+            NativeFrostCommitment {
+                identifier: key_packages[&1].identifier.clone(),
+                data_hex: round1.commitments_hex,
+            },
+            member2.commitment.clone(),
+        ],
+    );
+    let round2 = interactive_round2(InteractiveRound2Request {
+        session_id: session_id.to_string(),
+        attempt_id: opened.attempt_id.clone(),
+        member_identifier: 1,
+        signing_package_hex: signing_package_hex.clone(),
+    })
+    .expect("round 2 share");
+    let member2_share = sign_share(SignShareRequest {
+        signing_package_hex: signing_package_hex.clone(),
+        nonces_hex: member2.nonces_hex,
+        key_package_identifier: key_packages[&2].identifier.clone(),
+        key_package_hex: key_packages[&2].data_hex.clone(),
+    })
+    .expect("member 2 share");
+
+    let aggregate_request = InteractiveAggregateRequest {
+        session_id: session_id.to_string(),
+        attempt_id: opened.attempt_id.clone(),
+        signing_package_hex,
+        signature_shares: vec![
+            crate::api::NativeFrostSignatureShare {
+                identifier: key_packages[&1].identifier.clone(),
+                data_hex: round2.signature_share_hex,
+            },
+            member2_share.signature_share,
+        ],
+        taproot_merkle_root_hex: None,
+    };
+    interactive_aggregate(aggregate_request.clone()).expect("first interactive aggregate");
+
+    // The completion marker is the only durable interactive artifact (live
+    // nonce state is gone after restart by construction). It must survive a
+    // reload so a replayed aggregate is still rejected - this also exercises
+    // the marker's persistence round-trip (serialize + reload validation).
+    simulate_process_restart_for_tests();
+    reload_state_from_storage_for_tests();
+
+    let err = interactive_aggregate(aggregate_request)
+        .expect_err("a completed attempt must stay completed across restart");
+    assert!(
+        matches!(err, EngineError::InteractiveAttemptAlreadyAggregated { .. }),
+        "unexpected error: {err:?}"
+    );
+    assert_eq!(err.code(), "interactive_attempt_already_aggregated");
+
+    reset_for_tests();
+    cleanup_test_state_artifacts(&state_path);
+    clear_state_storage_policy_overrides();
+}
+
 #[test]
 fn interactive_aggregate_rejects_invalid_share_fail_closed() {
     let _guard = lock_test_state();