protocol拡張の実装

crates/pod/src/controller.rs

@@ -102,6 +102,9 @@ impl PodController {
         // AGENTS.md ingestion during the first turn) can emit user-facing
         // notifications on the same channel.
         pod.attach_notifier(notifier.clone());
+        // Also hand the raw broadcast sender so Pod-internal operations
+        // can emit typed lifecycle `Event`s (currently: compact progress).
+        pod.attach_event_tx(event_tx.clone());
 
         // Start socket server (lives as a background task, cleaned up on drop via RuntimeDir)
         let _socket_server = SocketServer::start(&handle).await?;

crates/pod/src/pod.rs

@@ -27,7 +27,8 @@ use crate::runtime_dir;
 use crate::scope_lock::{self, ScopeAllocationGuard, ScopeLockError};
 use crate::system_prompt::{SystemPromptContext, SystemPromptError, SystemPromptTemplate};
 use crate::usage_tracker::UsageTracker;
-use protocol::{NotificationLevel, NotificationSource};
+use protocol::{Event, NotificationLevel, NotificationSource};
+use tokio::sync::broadcast;
 use async_trait::async_trait;
 use llm_worker::interceptor::PreRequestAction;
 
@@ -120,6 +121,11 @@ pub struct Pod<C: LlmClient, St: Store> {
     /// User-facing notification sink attached by the Controller at
     /// spawn time. `None` in tests / direct `Pod::new` usage.
     notifier: Option<Notifier>,
+    /// Broadcast sender for typed lifecycle `Event`s (compact progress,
+    /// etc.). Attached by the Controller alongside `notifier`. Unlike
+    /// notifications, events sent here are NOT replayed to clients that
+    /// connect after the fact — they are fire-and-forget broadcasts.
+    event_tx: Option<broadcast::Sender<Event>>,
     /// Queue of pending `Method::Notify` notifications awaiting
     /// injection into the next LLM request. Shared with the
     /// PodInterceptor installed in `ensure_interceptor_installed`.
@@ -176,6 +182,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
             tracker: None,
             system_prompt_template: None,
             notifier: None,
+            event_tx: None,
             pending_notifications: NotificationBuffer::new(),
             scope_allocation: None,
             callback_socket: None,
@@ -241,6 +248,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
             tracker: None,
             system_prompt_template: None,
             notifier: None,
+            event_tx: None,
             pending_notifications: NotificationBuffer::new(),
             scope_allocation: None,
             callback_socket: None,
@@ -343,12 +351,30 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         self.notifier = Some(notifier);
     }
 
+    /// Attach the broadcast sender used for typed lifecycle `Event`s.
+    ///
+    /// The Controller wires this alongside [`attach_notifier`] so that
+    /// Pod-internal operations (currently: compaction) can surface
+    /// progress to connected clients.
+    pub fn attach_event_tx(&mut self, event_tx: broadcast::Sender<Event>) {
+        self.event_tx = Some(event_tx);
+    }
+
     fn notify(&self, level: NotificationLevel, source: NotificationSource, message: String) {
         if let Some(n) = self.notifier.as_ref() {
             n.notify(level, source, message);
         }
     }
 
+    /// Broadcast a typed `Event` to connected clients. No-op when no
+    /// `event_tx` is attached (tests / direct `Pod::new` usage) or when
+    /// no clients are currently subscribed.
+    fn send_event(&self, event: Event) {
+        if let Some(tx) = self.event_tx.as_ref() {
+            let _ = tx.send(event);
+        }
+    }
+
     /// Push a `Method::Notify` entry onto the pending buffer.
     ///
     /// The notification will be injected as an `Item::system_message`
@@ -682,12 +708,14 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
                 .map(|s| s.retained_tokens())
                 .unwrap_or(manifest::defaults::COMPACT_RETAINED_TOKENS);
 
+            self.send_event(Event::CompactStart);
             match self.compact(retained).await {
                 Ok(new_session_id) => {
                     info!(
                         new_session_id = %new_session_id,
                         "Compaction succeeded, resuming execution"
                     );
+                    self.send_event(Event::CompactDone { new_session_id });
                     if let Some(ref state) = self.compact_state {
                         state.record_compact_success();
                     }
@@ -695,6 +723,9 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
                 }
                 Err(e) => {
                     warn!(error = %e, "Compaction failed during run");
+                    self.send_event(Event::CompactFailed {
+                        error: e.to_string(),
+                    });
                     self.notify(
                         NotificationLevel::Error,
                         NotificationSource::Compactor,
@@ -723,17 +754,22 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         }
 
         let retained = state.retained_tokens();
+        self.send_event(Event::CompactStart);
         match self.compact(retained).await {
             Ok(new_session_id) => {
                 info!(
                     new_session_id = %new_session_id,
                     "Proactive post-run compaction succeeded"
                 );
+                self.send_event(Event::CompactDone { new_session_id });
                 state.record_compact_success();
                 Ok(())
             }
             Err(e) => {
                 warn!(error = %e, "Proactive post-run compaction failed");
+                self.send_event(Event::CompactFailed {
+                    error: e.to_string(),
+                });
                 self.notify(
                     NotificationLevel::Warn,
                     NotificationSource::Compactor,
@@ -1143,6 +1179,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
             tracker: None,
             system_prompt_template,
             notifier: None,
+            event_tx: None,
             pending_notifications: NotificationBuffer::new(),
             scope_allocation: Some(scope_allocation),
             callback_socket: None,
@@ -1202,6 +1239,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
             tracker: None,
             system_prompt_template,
             notifier: None,
+            event_tx: None,
             pending_notifications: NotificationBuffer::new(),
             scope_allocation: Some(scope_allocation),
             callback_socket: Some(callback_socket),

crates/pod/tests/compact_events_test.rs

@@ -0,0 +1,318 @@
+//! Compact lifecycle `Event` broadcasting.
+//!
+//! Covers three paths:
+//! - `try_post_run_compact` success → `CompactStart + CompactDone`
+//! - `try_post_run_compact` failure → `CompactStart + CompactFailed`
+//! - mid-turn `do_compact_and_resume` success → `CompactStart + CompactDone`
+//!   (driven by `compact_request_threshold` → `PreRequestAction::Yield`)
+
+use std::pin::Pin;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use async_trait::async_trait;
+use futures::Stream;
+use llm_worker::Worker;
+use llm_worker::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
+use llm_worker::llm_client::{ClientError, LlmClient, Request};
+use protocol::Event;
+use session_store::FsStore;
+use tokio::sync::broadcast;
+
+use pod::Pod;
+
+#[derive(Clone)]
+struct MockClient {
+    responses: Arc<Vec<Vec<LlmEvent>>>,
+    call_count: Arc<AtomicUsize>,
+}
+
+impl MockClient {
+    fn new(responses: Vec<Vec<LlmEvent>>) -> Self {
+        Self {
+            responses: Arc::new(responses),
+            call_count: Arc::new(AtomicUsize::new(0)),
+        }
+    }
+}
+
+#[async_trait]
+impl LlmClient for MockClient {
+    fn clone_boxed(&self) -> Box<dyn LlmClient> {
+        Box::new(self.clone())
+    }
+
+    async fn stream(
+        &self,
+        _request: Request,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<LlmEvent, ClientError>> + Send>>, ClientError>
+    {
+        let count = self.call_count.fetch_add(1, Ordering::SeqCst);
+        if count >= self.responses.len() {
+            return Err(ClientError::Config("mock client exhausted".into()));
+        }
+        let events = self.responses[count].clone();
+        let stream = futures::stream::iter(events.into_iter().map(Ok));
+        Ok(Box::pin(stream))
+    }
+}
+
+fn single_text_events(text: &str) -> Vec<LlmEvent> {
+    vec![
+        LlmEvent::text_block_start(0),
+        LlmEvent::text_delta(0, text),
+        LlmEvent::text_block_stop(0, None),
+        LlmEvent::Status(StatusEvent {
+            status: ResponseStatus::Completed,
+        }),
+    ]
+}
+
+/// `single_text_events` + a UsageEvent so the Pod's `usage_history`
+/// picks up a measurement, which is how `pre_llm_request` decides
+/// whether to yield mid-turn.
+fn text_events_with_usage(text: &str, input_tokens: u64) -> Vec<LlmEvent> {
+    vec![
+        LlmEvent::text_block_start(0),
+        LlmEvent::text_delta(0, text),
+        LlmEvent::text_block_stop(0, None),
+        LlmEvent::usage(input_tokens, 1),
+        LlmEvent::Status(StatusEvent {
+            status: ResponseStatus::Completed,
+        }),
+    ]
+}
+
+fn write_summary_tool_use_events(call_id: &str, text: &str) -> Vec<LlmEvent> {
+    let input = serde_json::json!({ "text": text }).to_string();
+    vec![
+        LlmEvent::tool_use_start(0, call_id, "write_summary"),
+        LlmEvent::tool_input_delta(0, input),
+        LlmEvent::tool_use_stop(0),
+        LlmEvent::Status(StatusEvent {
+            status: ResponseStatus::Completed,
+        }),
+    ]
+}
+
+// A low compact_threshold guarantees `try_post_run_compact` will fire
+// the first time we check after a run.
+const POST_RUN_MANIFEST_TOML: &str = r#"
+[pod]
+name = "test-pod"
+pwd = "./"
+
+[model]
+scheme = "anthropic"
+model_id = "test-model"
+
+[worker]
+max_tokens = 100
+
+[compaction]
+compact_threshold = 1
+compact_retained_tokens = 0
+
+[[scope.allow]]
+target = "./"
+permission = "write"
+"#;
+
+// `compact_request_threshold` drives the PodInterceptor's mid-turn yield
+// path. `compact_threshold` is left unset so the post-run check stays inert.
+const MID_TURN_MANIFEST_TOML: &str = r#"
+[pod]
+name = "test-pod"
+pwd = "./"
+
+[model]
+scheme = "anthropic"
+model_id = "test-model"
+
+[worker]
+max_tokens = 100
+
+[compaction]
+compact_request_threshold = 100
+compact_retained_tokens = 0
+
+[[scope.allow]]
+target = "./"
+permission = "write"
+"#;
+
+async fn make_pod_with_manifest(
+    manifest_toml: &str,
+    client: MockClient,
+) -> Pod<MockClient, FsStore> {
+    let manifest = pod::PodManifest::from_toml(manifest_toml).unwrap();
+
+    let store_tmp = tempfile::tempdir().unwrap();
+    let store = FsStore::new(store_tmp.path()).await.unwrap();
+    std::mem::forget(store_tmp);
+
+    let pwd_tmp = tempfile::tempdir().unwrap();
+    let pwd = pwd_tmp.path().to_path_buf();
+    let scope = pod::Scope::writable(&pwd).unwrap();
+    std::mem::forget(pwd_tmp);
+
+    let worker = Worker::new(client);
+    Pod::new(manifest, worker, store, pwd, scope).await.unwrap()
+}
+
+async fn make_pod(client: MockClient) -> Pod<MockClient, FsStore> {
+    make_pod_with_manifest(POST_RUN_MANIFEST_TOML, client).await
+}
+
+/// Drain whatever events are already queued on `rx`. Non-blocking.
+fn drain(rx: &mut broadcast::Receiver<Event>) -> Vec<Event> {
+    let mut out = Vec::new();
+    loop {
+        match rx.try_recv() {
+            Ok(ev) => out.push(ev),
+            Err(_) => break,
+        }
+    }
+    out
+}
+
+#[tokio::test]
+async fn post_run_compact_success_broadcasts_start_and_done() {
+    // Responses: (1) first run returns short text, (2) compact worker
+    // emits write_summary then closes (two LLM calls inside the compact
+    // worker: one for write_summary, one that the compact loop consumes
+    // as the final "I'm done" close response).
+    let client = MockClient::new(vec![
+        single_text_events("hi"),
+        write_summary_tool_use_events("call-1", "summary"),
+        single_text_events("done"),
+    ]);
+    let mut pod = make_pod(client).await;
+
+    let (tx, mut rx) = broadcast::channel::<Event>(64);
+    pod.attach_event_tx(tx);
+
+    pod.run("first").await.unwrap();
+    // Drain run events so only compact events remain in `rx`.
+    let _ = drain(&mut rx);
+
+    pod.try_post_run_compact().await.unwrap();
+
+    let events = drain(&mut rx);
+    let kinds: Vec<&str> = events
+        .iter()
+        .map(|e| match e {
+            Event::CompactStart => "start",
+            Event::CompactDone { .. } => "done",
+            Event::CompactFailed { .. } => "failed",
+            _ => "other",
+        })
+        .collect();
+    assert!(
+        kinds.contains(&"start") && kinds.contains(&"done"),
+        "expected CompactStart + CompactDone in {kinds:?}"
+    );
+    assert!(
+        !kinds.contains(&"failed"),
+        "unexpected CompactFailed in {kinds:?}"
+    );
+
+    // CompactDone carries the new session id.
+    let new_id_in_event = events.iter().find_map(|e| match e {
+        Event::CompactDone { new_session_id } => Some(*new_session_id),
+        _ => None,
+    });
+    assert!(new_id_in_event.is_some(), "CompactDone missing");
+    assert_eq!(new_id_in_event.unwrap(), pod.session_id());
+}
+
+#[tokio::test]
+async fn mid_turn_compact_success_broadcasts_start_and_done() {
+    // Path: `do_compact_and_resume` via PreRequestAction::Yield.
+    //
+    // Sequence of LLM calls the mock will serve:
+    //   [0] first run completes with a UsageEvent(1000 > threshold=100) so
+    //       the next run's pre_llm_request will yield.
+    //   [1] compact worker emits `write_summary` tool call.
+    //   [2] compact worker closes (its final "done" response).
+    //   [3] resume() after compact makes one more LLM call.
+    let client = MockClient::new(vec![
+        text_events_with_usage("a", 1000),
+        write_summary_tool_use_events("call-1", "summary"),
+        single_text_events("done"),
+        single_text_events("b"),
+    ]);
+    let mut pod = make_pod_with_manifest(MID_TURN_MANIFEST_TOML, client).await;
+
+    let (tx, mut rx) = broadcast::channel::<Event>(64);
+    pod.attach_event_tx(tx);
+
+    // First run populates usage_history above the request threshold.
+    pod.run("first").await.unwrap();
+    let _ = drain(&mut rx);
+
+    // Second run: pre_llm_request yields immediately, Worker returns
+    // Yielded, handle_worker_result routes into do_compact_and_resume.
+    pod.run("second").await.unwrap();
+
+    let events = drain(&mut rx);
+    let kinds: Vec<&str> = events
+        .iter()
+        .map(|e| match e {
+            Event::CompactStart => "start",
+            Event::CompactDone { .. } => "done",
+            Event::CompactFailed { .. } => "failed",
+            _ => "other",
+        })
+        .collect();
+    assert!(
+        kinds.contains(&"start") && kinds.contains(&"done"),
+        "expected CompactStart + CompactDone in {kinds:?}"
+    );
+    assert!(
+        !kinds.contains(&"failed"),
+        "unexpected CompactFailed in {kinds:?}"
+    );
+
+    let new_id_in_event = events.iter().find_map(|e| match e {
+        Event::CompactDone { new_session_id } => Some(*new_session_id),
+        _ => None,
+    });
+    assert_eq!(new_id_in_event, Some(pod.session_id()));
+}
+
+#[tokio::test]
+async fn post_run_compact_failure_broadcasts_start_and_failed() {
+    // Only the first run has a response. Compaction will run the
+    // compact worker which immediately exhausts the mock → failure.
+    let client = MockClient::new(vec![single_text_events("hi")]);
+    let mut pod = make_pod(client).await;
+
+    let (tx, mut rx) = broadcast::channel::<Event>(64);
+    pod.attach_event_tx(tx);
+
+    pod.run("first").await.unwrap();
+    let _ = drain(&mut rx);
+
+    // Best-effort: returns Ok(()) even on failure, but emits CompactFailed.
+    pod.try_post_run_compact().await.unwrap();
+
+    let events = drain(&mut rx);
+    let kinds: Vec<&str> = events
+        .iter()
+        .map(|e| match e {
+            Event::CompactStart => "start",
+            Event::CompactDone { .. } => "done",
+            Event::CompactFailed { .. } => "failed",
+            _ => "other",
+        })
+        .collect();
+    assert!(
+        kinds.contains(&"start") && kinds.contains(&"failed"),
+        "expected CompactStart + CompactFailed in {kinds:?}"
+    );
+    assert!(
+        !kinds.contains(&"done"),
+        "unexpected CompactDone in {kinds:?}"
+    );
+}