fix(helper): bind fixture ack to reliable probe

cai

2026-08-12 a640ced9a412100b50b5d9ea0566204b33519bf1

 src/service.rs

@@ -476,11 +476,19 @@
                continue;
            };
            println!("{line}");
            if let Ok(value) = serde_json::from_str::<Value>(&line) {
            if let Some(value) = project_worker_stdout_event(&call_id, &line) {
                apply_worker_event(&call_id, &state, &value);
            }
        }
    });
}

fn project_worker_stdout_event(call_id: &str, line: &str) -> Option<Value> {
    let value = serde_json::from_str::<Value>(line).ok()?;
    (value.get("type").and_then(Value::as_str) == Some("cv_activity")
        && value.get("callId").and_then(Value::as_str) == Some(call_id)
        && value.get("eventName").and_then(Value::as_str).is_some())
    .then_some(value)
}

fn apply_worker_event(call_id: &str, state: &Arc<ServiceState>, value: &Value) {
@@ -1243,12 +1251,90 @@

#[cfg(test)]
mod tests {
    use super::{EventCallbackDispatch, deliver_event_callback_with, run_event_callback_worker};
    use super::{
        EventCallbackDispatch, deliver_event_callback_with, project_worker_stdout_event,
        run_event_callback_worker,
    };
    use crate::{
        CONTROLLED_FIXTURE_GENERATION, CONTROLLED_FIXTURE_PROBE_TTL,
        ControlledFixtureVisibilityEvidence, PendingControlledFixtureProbe,
        controlled_fixture_probe_event, controlled_fixture_visibility_event,
    };
    use anyhow::Result;
    use serde_json::{Value, json};
    use std::{cell::RefCell, rc::Rc, sync::mpsc};

    #[test]
    fn controlled_fixture_activity_requires_production_stdout_envelope() {
        let call_id = "runtime-call";
        let runtime_trace_id = "runtime-trace";
        let trace_hash = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
        let valid = controlled_fixture_probe_event(
            call_id,
            runtime_trace_id,
            "ack_publish_completed",
            "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
            trace_hash,
            1,
            "fixture-01",
            false,
            Some("rejected"),
            Some("wrong_source"),
        );
        let projected = project_worker_stdout_event(call_id, &valid.to_string())
            .expect("valid activity must enter the production stdout projection");
        assert_eq!(projected["eventName"], "controlled_fixture_attribute_probe");
        assert_eq!(projected["extension"]["trace_id_hash"], trace_hash);
        assert_eq!(projected["extension"]["reject_reason"], "wrong_source");

        for invalid in [
            json!({
                "event": "controlled_fixture_attribute_probe",
                "stage": "ack_publish_completed",
                "observed": false,
                "ack_result": "rejected",
                "reject_reason": "wrong_source",
                "call_id_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                "trace_id_hash": trace_hash,
                "generation": 1,
                "sequence_hash": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
            }),
            json!({"type": "not_activity", "callId": call_id, "eventName": "controlled_fixture_attribute_probe"}),
            json!({"type": "cv_activity", "callId": call_id}),
            json!({"type": "cv_activity", "callId": "wrong-call", "eventName": "controlled_fixture_attribute_probe"}),
        ] {
            assert!(project_worker_stdout_event(call_id, &invalid.to_string()).is_none());
        }

        let received_at = std::time::Instant::now();
        let visibility = controlled_fixture_visibility_event(
            call_id,
            runtime_trace_id,
            &PendingControlledFixtureProbe {
                sender: livekit::prelude::ParticipantIdentity("user-1".to_string()),
                call_id_hash: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                    .to_string(),
                call_trace_id_hash: trace_hash.to_string(),
                generation: CONTROLLED_FIXTURE_GENERATION,
                sequence: "fixture-01".to_string(),
                received_at,
                expires_at: received_at + CONTROLLED_FIXTURE_PROBE_TTL,
            },
            &ControlledFixtureVisibilityEvidence {
                first_visible_bucket: "250_500ms",
                visibility_source: "participant_attributes_poll",
                visibility_result: "held_visible",
                binding_matched: true,
            },
        );
        let projected = project_worker_stdout_event(call_id, &visibility.to_string())
            .expect("post-expiry evidence must enter the production stdout projection");
        assert_eq!(projected["extension"]["first_visible_bucket"], "250_500ms");
        assert_eq!(projected["extension"]["binding_matched"], true);
        assert_eq!(projected["extension"]["visibility_result"], "held_visible");
    }

    #[test]
    fn same_session_callback_retries_m7_before_terminal() {
        let (sender, receiver) = mpsc::channel();
        let dispatch = |payload| EventCallbackDispatch {