From f709c9ce06731147e010f02b11b438a0eb7b9c99 Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Wed, 12 Aug 2026 13:19:29 +0800
Subject: [PATCH] fix(helper): wrap fixture probe logs as activities
---
src/service.rs | 59 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++--
1 files changed, 57 insertions(+), 2 deletions(-)
diff --git a/src/service.rs b/src/service.rs
index 1e3e0c2..65a8622 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,59 @@
#[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_probe_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());
+ }
+ }
+
+ #[test]
fn same_session_callback_retries_m7_before_terminal() {
let (sender, receiver) = mpsc::channel();
let dispatch = |payload| EventCallbackDispatch {
--
Gitblit v1.9.3