Fail closed when fixture ACK publish fails

cai

2026-08-12 da9563428e990244ff2145eb6a9cb453f9ef9d22

 src/main.rs

@@ -6,6 +6,7 @@
    borrow::Cow,
    collections::HashSet,
    env, fs,
    future::Future,
    path::{Path, PathBuf},
    sync::{
        Arc,
@@ -209,6 +210,45 @@
        classification.1,
    );
    classification
}

async fn complete_controlled_fixture_ack_publish<F, E>(
    publish: F,
    observed: bool,
    ack_result: &'static str,
    reject_reason: Option<&'static str>,
    call_id: &str,
    trace_id: &str,
    sequence: &str,
    acknowledged_probe_sequences: &mut HashSet<String>,
) -> bool
where
    F: Future<Output = Result<(), E>>,
{
    if publish.await.is_ok() {
        record_controlled_fixture_probe_event(
            "ack_publish_completed",
            call_id,
            trace_id,
            sequence,
            observed,
            Some(ack_result),
            reject_reason,
        );
        acknowledged_probe_sequences.insert(sequence.to_string());
        observed
    } else {
        record_controlled_fixture_probe_event(
            "ack_publish_completed",
            call_id,
            trace_id,
            sequence,
            false,
            Some("publish_failed"),
            Some("ack_publish_failed"),
        );
        false
    }
}

fn controlled_fixture_probe(
@@ -1325,29 +1365,19 @@
        reliable: true,
        destination_identities: vec![probe.sender],
    });
    if publish.await.is_ok() {
        record_controlled_fixture_probe_event(
            "ack_publish_completed",
            call_id,
            trace_id,
            &probe.sequence,
    Some(
        complete_controlled_fixture_ack_publish(
            publish,
            observed,
            Some(ack_result),
            ack_result,
            reject_reason,
        );
        acknowledged_probe_sequences.insert(probe.sequence);
    } else {
        record_controlled_fixture_probe_event(
            "ack_publish_completed",
            call_id,
            trace_id,
            &probe.sequence,
            false,
            Some("publish_failed"),
            Some("ack_publish_failed"),
        );
    }
    Some(result == "observed")
            acknowledged_probe_sequences,
        )
        .await,
    )
}

async fn handle_finished_turn(
@@ -5367,6 +5397,63 @@
    }

    #[tokio::test]
    async fn production_ack_publish_failure_keeps_observer_session_and_audio_closed() {
        let mut acknowledged = HashSet::new();
        let probe_result = complete_controlled_fixture_ack_publish(
            async { Err::<(), ()>(()) },
            true,
            "observed",
            None,
            "call-publish-failure",
            "trace-publish-failure",
            "fixture-01",
            &mut acknowledged,
        )
        .await;
        let mut observer_starts = 0;
        let mut session_starts = 0;
        let mut audio_starts = 0;
        let mut speaking_starts = 0;
        assert!(!start_observer_after_controlled_fixture_probe(
            true,
            Some(probe_result),
            || {
                observer_starts += 1;
                session_starts += 1;
                audio_starts += 1;
                speaking_starts += 1;
            },
        ));
        assert!(!probe_result);
        assert!(acknowledged.is_empty());
        assert_eq!(observer_starts, 0);
        assert_eq!(session_starts, 0);
        assert_eq!(audio_starts, 0);
        assert_eq!(speaking_starts, 0);

        let observed_result = complete_controlled_fixture_ack_publish(
            async { Ok::<(), ()>(()) },
            true,
            "observed",
            None,
            "call-publish-success",
            "trace-publish-success",
            "fixture-02",
            &mut acknowledged,
        )
        .await;
        let mut successful_observer_starts = 0;
        assert!(start_observer_after_controlled_fixture_probe(
            true,
            Some(observed_result),
            || successful_observer_starts += 1,
        ));
        assert!(observed_result);
        assert!(acknowledged.contains("fixture-02"));
        assert_eq!(successful_observer_starts, 1);
    }

    #[tokio::test]
    async fn production_attribute_observation_accepts_server_visibility_within_probe_ttl() {
        let expected = HashMap::from([
            (