feat(rwi): carry agent_id/agent_name in recording-end webhook events

Add CallMetaStore::update_agent so the CC session hook can publish the
resolved agent_id/agent_name on call connected. The gateway enrichment
path (enrich_flat_event) then injects these into all owner-scoped events,
including record_end and recording_metadata_available (whose struct had
the fields but were hardcoded to None).

- proto: CallMetaStore::update_agent + unit tests
- gateway: end-to-end enrichment tests
- recording_upload/sipflow_upload: fill RecordingMetadata agent fields
  from the meta store instead of None
- bump submodules: cc (agent publishing), telemetry (Drop impl),
  ivr_editor (latest)

Author: shenjinti

src/addons/cc

@@ -251,6 +251,14 @@ impl CallRecordHook for RecordingUploadHook {
             // (e.g. MQ consumers) are notified that the recording is ready.
             if let Some(ref gw) = self.rwi_gateway {
                 use crate::rwi::proto::RecordingMetadata;
+                // Pull agent context from the RWI meta store (populated on call
+                // connected); fall back to CallRecord.details for agent_name.
+                let (meta_agent_id, meta_agent_name) = gw
+                    .read()
+                    .meta_store
+                    .get_sync(&record.call_id)
+                    .map(|m| (m.agent_id, m.agent_name))
+                    .unwrap_or((None, None));
                 let metadata = RecordingMetadata {
                     filename: record
                         .recorder
@@ -268,8 +276,8 @@ impl CallRecordHook for RecordingUploadHook {
                     callee_name: extract_sip_username(&record.callee),
                     called_phone: extract_sip_username(&record.callee),
                     call_type: record.details.direction.clone(),
-                    agent_id: None,
-                    agent_name: None,
+                    agent_id: meta_agent_id,
+                    agent_name: meta_agent_name.or_else(|| record.details.agent_name.clone()),
                     call_start_time: Some(record.start_time.to_rfc3339()),
                     call_end_time: Some(record.end_time.to_rfc3339()),
                     upload_time: Some(chrono::Utc::now().to_rfc3339()),

src/addons/ivr_editor

@@ -205,6 +205,13 @@ async fn do_upload(
     if let Some(url) = first_uploaded_url.as_ref() {
         if let Some(gw) = rwi_gateway {
             use crate::rwi::proto::RecordingMetadata;
+            // Pull agent context from the RWI meta store when available.
+            let (meta_agent_id, meta_agent_name) = gw
+                .read()
+                .meta_store
+                .get_sync(call_id)
+                .map(|m| (m.agent_id, m.agent_name))
+                .unwrap_or((None, None));
             let metadata = RecordingMetadata {
                 filename: media_key.to_string(),
                 unique_id: call_id.to_string(),
@@ -214,8 +221,8 @@ async fn do_upload(
                 callee_name: None,
                 called_phone: None,
                 call_type: "".to_string(),
-                agent_id: None,
-                agent_name: None,
+                agent_id: meta_agent_id,
+                agent_name: meta_agent_name,
                 call_start_time: Some(start.to_rfc3339()),
                 call_end_time: Some(end.to_rfc3339()),
                 upload_time: Some(Utc::now().to_rfc3339()),

src/addons/telemetry

@@ -797,4 +797,69 @@ mod tests {
         let v = rx.recv().await.unwrap();
         assert!(v.to_string().contains("call_answered"));
     }
+
+    /// `send_to_owner` must enrich the event payload with `agent_id` /
+    /// `agent_name` pulled from the [`CallMetaStore`] (populated on call
+    /// connected). This is the path that carries agent context into the
+    /// `record_end` / `recording_metadata_available` webhook events even
+    /// though those event structs have no agent fields of their own.
+    #[tokio::test]
+    async fn test_send_to_owner_enriches_agent_from_meta_store() {
+        let mut gw = RwiGateway::new();
+        let (tx, mut rx) = broadcast::channel::<EventCacheEntry>(16);
+        gw.set_webhook_tx(tx);
+
+        // Populate agent context as CcCallSessionHook does on call connected.
+        gw.meta_store
+            .update_agent(
+                "call-1",
+                Some("agent-42".to_string()),
+                Some("Alice".to_string()),
+            )
+            .await;
+
+        // RecordEnd has NO agent_id / agent_name fields on the struct itself.
+        gw.send_to_owner(&crate::rwi::RecordEnd {
+            call_id: "call-1".to_string(),
+            url: Some("https://example.com/rec.wav".to_string()),
+            duration_secs: 12,
+            file_size: 1024,
+        });
+
+        let entry = rx.recv().await.expect("webhook must receive record_end");
+        assert_eq!(entry.event.event_type, "record_end");
+        // Agent context injected by enrich_flat_event.
+        assert_eq!(entry.event.payload["agent_id"].as_str(), Some("agent-42"));
+        assert_eq!(entry.event.payload["agent_name"].as_str(), Some("Alice"));
+        // Original RecordEnd fields are preserved.
+        assert_eq!(
+            entry.event.payload["url"].as_str(),
+            Some("https://example.com/rec.wav")
+        );
+        assert_eq!(entry.event.payload["duration_secs"].as_u64(), Some(12));
+    }
+
+    /// When no agent context exists for a call, enrichment must leave the
+    /// payload untouched (no spurious null agent fields).
+    #[tokio::test]
+    async fn test_send_to_owner_no_enrichment_without_meta() {
+        let mut gw = RwiGateway::new();
+        let (tx, mut rx) = broadcast::channel::<EventCacheEntry>(16);
+        gw.set_webhook_tx(tx);
+
+        gw.send_to_owner(&crate::rwi::RecordEnd {
+            call_id: "call-unknown".to_string(),
+            url: None,
+            duration_secs: 0,
+            file_size: 0,
+        });
+
+        let entry = rx.recv().await.unwrap();
+        assert_eq!(entry.event.event_type, "record_end");
+        assert!(
+            entry.event.payload.get("agent_id").is_none(),
+            "no agent fields when meta absent, got: {}",
+            entry.event.payload
+        );
+    }
 }

src/callrecord/recording_upload.rs

@@ -120,6 +120,27 @@ impl CallMetaStore {
     pub async fn remove(&self, call_id: &str) {
         self.store.write().await.remove(call_id);
     }
+
+    /// Update the agent fields on an existing call's metadata.
+    ///
+    /// Called when a call is assigned to an agent (e.g. on call connected) so
+    /// that subsequently emitted events (record_end, recording_metadata_available,
+    /// call_hangup, …) are enriched with `agent_id`/`agent_name` via
+    /// [`crate::rwi::gateway::RwiGateway::enrich_flat_event`].
+    ///
+    /// If no entry exists for `call_id` yet (rare race with session creation),
+    /// a default [`CallMeta`] is inserted carrying only the agent fields.
+    pub async fn update_agent(
+        &self,
+        call_id: &str,
+        agent_id: Option<String>,
+        agent_name: Option<String>,
+    ) {
+        let mut store = self.store.write().await;
+        let meta = store.entry(call_id.to_string()).or_default();
+        meta.agent_id = agent_id;
+        meta.agent_name = agent_name;
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -142,3 +163,96 @@ pub struct RecordingMetadata {
     pub process_flag: Option<String>,
     pub root_call_id: Option<String>,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn update_agent_sets_fields_on_existing_entry() {
+        let store = CallMetaStore::new();
+        store
+            .insert(
+                "call-1".to_string(),
+                CallMeta {
+                    caller: Some("1001".to_string()),
+                    callee: Some("1002".to_string()),
+                    caller_name: Some("alice".to_string()),
+                    ..Default::default()
+                },
+            )
+            .await;
+
+        store
+            .update_agent(
+                "call-1",
+                Some("agent-007".to_string()),
+                Some("James Bond".to_string()),
+            )
+            .await;
+
+        let meta = store.get("call-1").await.expect("meta must exist");
+        assert_eq!(meta.agent_id.as_deref(), Some("agent-007"));
+        assert_eq!(meta.agent_name.as_deref(), Some("James Bond"));
+        // Existing fields must be preserved.
+        assert_eq!(meta.caller.as_deref(), Some("1001"));
+        assert_eq!(meta.caller_name.as_deref(), Some("alice"));
+    }
+
+    #[tokio::test]
+    async fn update_agent_creates_default_entry_when_absent() {
+        let store = CallMetaStore::new();
+        // No prior insert for "call-2".
+        store
+            .update_agent("call-2", Some("agent-x".to_string()), None)
+            .await;
+
+        let meta = store.get("call-2").await.expect("meta created on the fly");
+        assert_eq!(meta.agent_id.as_deref(), Some("agent-x"));
+        assert!(meta.agent_name.is_none());
+    }
+
+    #[tokio::test]
+    async fn update_agent_clears_fields_when_none() {
+        let store = CallMetaStore::new();
+        store
+            .insert(
+                "call-3".to_string(),
+                CallMeta {
+                    agent_id: Some("old".to_string()),
+                    agent_name: Some("old name".to_string()),
+                    ..Default::default()
+                },
+            )
+            .await;
+
+        store.update_agent("call-3", None, None).await;
+
+        let meta = store.get("call-3").await.unwrap();
+        assert!(meta.agent_id.is_none());
+        assert!(meta.agent_name.is_none());
+    }
+
+    #[tokio::test]
+    async fn update_agent_then_eventcallcontext_from_carries_fields() {
+        let store = CallMetaStore::new();
+        store
+            .insert(
+                "call-4".to_string(),
+                CallMeta {
+                    caller: Some("2001".to_string()),
+                    ..Default::default()
+                },
+            )
+            .await;
+        store
+            .update_agent("call-4", Some("agent-9".to_string()), Some("Nine".to_string()))
+            .await;
+
+        let meta = store.get("call-4").await.unwrap();
+        let ctx = EventCallContext::from(meta);
+        assert_eq!(ctx.agent_id.as_deref(), Some("agent-9"));
+        assert_eq!(ctx.agent_name.as_deref(), Some("Nine"));
+        assert_eq!(ctx.caller.as_deref(), Some("2001"));
+    }
+}