fix: bridge worker log events to protocol

crates/worker-runtime/src/http_server.rs

@@ -1339,34 +1339,6 @@ mod ws_tests {
         ));
     }
 
-    #[tokio::test]
-    async fn protocol_ws_echoes_accepted_run_as_user_message() {
-        let (_runtime, _worker_ref, url) = spawn_runtime_server().await;
-        let (mut stream, _) = connect_async(&url).await.unwrap();
-        let _ = next_frame(&mut stream).await;
-
-        stream
-            .send(Message::Text(
-                serde_json::to_string(&protocol::Method::Run {
-                    input: vec![protocol::Segment::text("hello from protocol")],
-                })
-                .unwrap()
-                .into(),
-            ))
-            .await
-            .unwrap();
-
-        match next_frame(&mut stream).await {
-            protocol::Event::UserMessage { segments } => {
-                assert_eq!(
-                    protocol::Segment::flatten_to_text(&segments),
-                    "hello from protocol"
-                );
-            }
-            event => panic!("expected user message echo, got {event:?}"),
-        }
-    }
-
     #[tokio::test]
     async fn protocol_ws_reports_malformed_cursor_and_method_frame() {
         let (_runtime, _worker_ref, url) = spawn_runtime_server().await;

crates/worker-runtime/src/runtime.rs

@@ -576,8 +576,6 @@ impl Runtime {
             return Ok(vec![Event::Completions { kind, entries }]);
         }
 
-        let observation_payload = protocol_method_observation_event(&method);
-
         let (backend, handle) = {
             let mut state = self.lock()?;
             state.ensure_running()?;
@@ -624,9 +622,6 @@ impl Runtime {
         }
 
         self.record_execution_result(worker_ref, dispatch_result)?;
-        if let Some(payload) = observation_payload {
-            self.record_protocol_method_observation(worker_ref, payload)?;
-        }
         Ok(Vec::new())
     }
 
@@ -963,7 +958,12 @@ impl Runtime {
         worker_ref: &WorkerRef,
         input: WorkerInput,
     ) -> Result<(), RuntimeError> {
-        self.record_protocol_method_observation(worker_ref, input_protocol_event(&input))
+        let mut state = self.lock()?;
+        state.ensure_worker_ref(worker_ref)?;
+        let event =
+            state.push_worker_observation_event(worker_ref.clone(), input_protocol_event(&input));
+        state.persist_worker_observation_event(&event)?;
+        Ok(())
     }
 
     #[cfg(not(feature = "ws-server"))]
@@ -975,28 +975,6 @@ impl Runtime {
         Ok(())
     }
 
-    #[cfg(feature = "ws-server")]
-    fn record_protocol_method_observation(
-        &self,
-        worker_ref: &WorkerRef,
-        payload: Event,
-    ) -> Result<(), RuntimeError> {
-        let mut state = self.lock()?;
-        state.ensure_worker_ref(worker_ref)?;
-        let event = state.push_worker_observation_event(worker_ref.clone(), payload);
-        state.persist_worker_observation_event(&event)?;
-        Ok(())
-    }
-
-    #[cfg(not(feature = "ws-server"))]
-    fn record_protocol_method_observation(
-        &self,
-        _worker_ref: &WorkerRef,
-        _payload: Event,
-    ) -> Result<(), RuntimeError> {
-        Ok(())
-    }
-
     fn transition_worker(
         &self,
         worker_ref: &WorkerRef,
@@ -1858,48 +1836,6 @@ fn validate_worker_input(input: &WorkerInput) -> Result<(), RuntimeError> {
     Ok(())
 }
 
-#[cfg(feature = "ws-server")]
-fn protocol_method_observation_event(method: &Method) -> Option<Event> {
-    match method {
-        Method::Run { input } => Some(Event::UserMessage {
-            segments: input.clone(),
-        }),
-        Method::Notify { message, .. } => Some(Event::SystemItem {
-            item: serde_json::json!({
-                "kind": "embedded_worker_system_input",
-                "content": message,
-            }),
-        }),
-        Method::RegisterPeer { name } => Some(Event::SystemItem {
-            item: serde_json::json!({
-                "kind": "embedded_worker_command_input",
-                "command": "register_peer",
-                "content": name,
-            }),
-        }),
-        Method::Compact => Some(Event::SystemItem {
-            item: serde_json::json!({
-                "kind": "embedded_worker_command_input",
-                "command": "compact",
-                "content": "",
-            }),
-        }),
-        Method::ListRewindTargets => Some(Event::SystemItem {
-            item: serde_json::json!({
-                "kind": "embedded_worker_command_input",
-                "command": "list_rewind_targets",
-                "content": "",
-            }),
-        }),
-        _ => None,
-    }
-}
-
-#[cfg(not(feature = "ws-server"))]
-fn protocol_method_observation_event(_method: &Method) -> Option<Event> {
-    None
-}
-
 #[cfg(feature = "ws-server")]
 fn input_protocol_event(input: &WorkerInput) -> protocol::Event {
     match input.kind {
@@ -1941,7 +1877,6 @@ mod tests {
         WorkerExecutionRestoreRequest, WorkerExecutionRunState,
     };
     use std::collections::BTreeMap;
-    use std::sync::atomic::{AtomicU64, Ordering};
     use std::sync::{Arc, Mutex};
 
     fn task_request(_objective: &str) -> CreateWorkerRequest {

crates/worker-runtime/src/worker_backend.rs

@@ -30,9 +30,9 @@ use crate::working_directory::{
 };
 use async_trait::async_trait;
 use manifest::paths;
-use protocol::{Method, Segment, WorkerStatus};
+use protocol::{Event, Method, Segment, WorkerStatus};
 use session_store::FsStore;
-use session_store::{CombinedStore, FsWorkerStore};
+use session_store::{CombinedStore, FsWorkerStore, LogEntry};
 use tokio::runtime::Runtime;
 #[cfg(feature = "ws-server")]
 use tokio::sync::broadcast;
@@ -667,19 +667,35 @@ where
         #[cfg(feature = "ws-server")]
         {
             let mut events = handle.subscribe();
+            let (_entries, mut entry_events) = handle.sink.subscribe_with_snapshot();
             let bridge_handle = handle.clone();
             let bridge_busy = busy.clone();
             if let Err(message) = self.spawn_on_adapter_runtime(async move {
                 loop {
-                    match events.recv().await {
-                        Ok(event) => {
-                            let _ = bridge_context.publish_protocol_event(event);
-                            if bridge_handle.shared_state.get_status() == WorkerStatus::Idle {
-                                bridge_busy.store(false, Ordering::SeqCst);
+                    tokio::select! {
+                        event = events.recv() => {
+                            match event {
+                                Ok(event) => {
+                                    let _ = bridge_context.publish_protocol_event(event);
+                                    if bridge_handle.shared_state.get_status() == WorkerStatus::Idle {
+                                        bridge_busy.store(false, Ordering::SeqCst);
+                                    }
+                                }
+                                Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                                Err(broadcast::error::RecvError::Closed) => break,
+                            }
+                        }
+                        entry = entry_events.recv() => {
+                            match entry {
+                                Ok(entry) => {
+                                    if let Some(event) = live_log_entry_event(entry) {
+                                        let _ = bridge_context.publish_protocol_event(event);
+                                    }
+                                }
+                                Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                                Err(broadcast::error::RecvError::Closed) => break,
                             }
                         }
-                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
-                        Err(broadcast::error::RecvError::Closed) => break,
                     }
                 }
             }) {
@@ -722,6 +738,22 @@ impl<F> Drop for WorkerRuntimeExecutionBackend<F> {
     }
 }
 
+fn live_log_entry_event(entry: LogEntry) -> Option<Event> {
+    match entry {
+        LogEntry::SegmentStart { .. } => {
+            let value = serde_json::to_value(&entry).expect("LogEntry is Serialize");
+            Some(Event::SegmentRotated { entry: value })
+        }
+        LogEntry::UserInput { segments, .. } => Some(Event::UserMessage { segments }),
+        LogEntry::SystemItem { item, .. } => {
+            let value = serde_json::to_value(&item).expect("SystemItem is Serialize");
+            Some(Event::SystemItem { item: value })
+        }
+        LogEntry::Invoke { trigger, .. } => Some(Event::InvokeStart { kind: trigger }),
+        _ => None,
+    }
+}
+
 fn method_starts_turn(method: &Method) -> bool {
     matches!(
         method,
@@ -1208,6 +1240,21 @@ mod tests {
     use llm_engine::llm_client::{ClientError, LlmClient, Request};
     use manifest::{Scope, WorkerManifest};
 
+    #[test]
+    fn runtime_bridge_maps_live_user_input_log_entry_to_user_message() {
+        let segments = vec![Segment::text("hello through normal bridge")];
+        let event = live_log_entry_event(LogEntry::UserInput {
+            ts: session_store::segment_log::now_millis(),
+            segments: segments.clone(),
+        })
+        .expect("UserInput must be live-relevant");
+
+        match event {
+            Event::UserMessage { segments: echoed } => assert_eq!(echoed, segments),
+            other => panic!("expected UserMessage, got {other:?}"),
+        }
+    }
+
     #[derive(Clone)]
     struct MockClient {
         responses: Arc<Vec<Vec<LlmEvent>>>,