From caeed9acdc676aa10f788e81edf45892b8cdcc8e Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Tue, 25 Aug 2026 12:48:13 +0800
Subject: [PATCH] fix(helper): bind ASR ingress to fixture generation
---
src/service.rs | 90 ++++++++++++++++++++++++++++++++++++++++++++-
1 files changed, 88 insertions(+), 2 deletions(-)
diff --git a/src/service.rs b/src/service.rs
index 1e3e0c2..2d0d3ab 100644
--- a/src/service.rs
+++ b/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 {
--
Gitblit v1.9.3