refactor: Podのメインループのリファクタリング

Cargo.lock

@@ -3645,6 +3645,7 @@ version = "0.1.0"
 dependencies = [
  "client",
  "crossterm 0.28.1",
+ "llm-worker",
  "manifest",
  "pod-registry",
  "protocol",

TODO.md

@@ -8,7 +8,6 @@
 - Pod: 任意ターンからの Fork（複数ターン巻き戻しを汎用化） → [tickets/pod-session-fork.md](tickets/pod-session-fork.md)
 - Pod: 子→親の TurnEnded/Errored callback を親由来ターンのみに絞る → [tickets/pod-parent-turn-callback.md](tickets/pod-parent-turn-callback.md)
 - Pod: セッションログをバックエンドにした Pod 単位の永続化 → [tickets/pod-persistent-state.md](tickets/pod-persistent-state.md)
-- Pod: 状態と socket 配信を session log 正本に統合 → [tickets/pod-state-from-session-log.md](tickets/pod-state-from-session-log.md)
 - 永続化層のセマンティック整理 → [tickets/persistence-semantics.md](tickets/persistence-semantics.md)
 - Exchange / Turn / Call セマンティクス整理 → [tickets/exchange-turn-call-semantics.md](tickets/exchange-turn-call-semantics.md)
 - llm-worker のエラー耐性

crates/llm-worker/src/worker.rs

@@ -1013,13 +1013,16 @@ impl<C: LlmClient, S: WorkerState> Worker<C, S> {
             }
             self.turn_count += 1;
 
-            // Collect and commit assistant items
+            // Collect and commit assistant items. Routed through
+            // `extend_history_with_callbacks` so observers (e.g. the
+            // Pod-side per-item session-log committer) see each item
+            // as it lands.
             let reasoning_items = self.reasoning_item_collector.take_collected();
             let text_blocks = self.text_block_collector.take_collected();
             let tool_calls = self.tool_call_collector.take_collected();
             let assistant_items =
                 self.build_assistant_items(&reasoning_items, &text_blocks, &tool_calls);
-            self.history.extend(assistant_items);
+            self.extend_history_with_callbacks(assistant_items);
 
             if tool_calls.is_empty() {
                 match self.interceptor.on_turn_end(&self.history).await {
@@ -1134,14 +1137,18 @@ impl<C: LlmClient, S: WorkerState> Worker<C, S> {
                 Ok(Some(WorkerResult::Paused))
             }
             Ok(ToolExecutionResult::Completed(results)) => {
-                for result in results {
+                // Route per-result pushes through the callback path so
-                    self.history.push(Item::tool_result_item(
+                // observers (e.g. the Pod-side per-item session-log
+                // committer) see each tool result as it lands.
+                let items = results.into_iter().map(|result| {
+                    Item::tool_result_item(
                         &result.tool_use_id,
                         &result.summary,
                         result.content,
                         result.is_error,
-                    ));
+                    )
-                }
+                });
+                self.extend_history_with_callbacks(items);
                 Ok(None)
             }
             Err(err) => {

crates/pod/examples/pod_protocol.rs

@@ -104,10 +104,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         "\n[shared_state] final: {}",
         handle.shared_state.status_json()
     );
-    println!(
+    println!("[session log] {} entries", handle.sink.len());
-        "[history] {} bytes",
-        handle.shared_state.history_json().len()
-    );
 
     drop(handle);
     let _ = listener.await;

crates/pod/src/controller.rs

@@ -3,15 +3,19 @@ use std::sync::Arc;
 
 use llm_worker::WorkerError;
 use llm_worker::llm_client::client::LlmClient;
-use llm_worker::llm_client::types::{Item, Role};
 use session_store::Store;
 use tokio::sync::{broadcast, mpsc, oneshot};
 
+use llm_worker::Item;
+use session_store::LogEntry;
+use session_store::session_log;
+
 use crate::ipc::alerter::Alerter;
 use crate::ipc::notify_buffer::NotifyBuffer;
 use crate::ipc::server::SocketServer;
-use crate::pod::{Pod, PodError, PodRunResult};
+use crate::pod::{LogCommand, LogDrainHandle, Pod, PodError, PodRunResult};
 use crate::runtime::dir::RuntimeDir;
+use crate::session_log_sink::SessionLogSink;
 use crate::shared_state::PodSharedState;
 use crate::spawn::comm_tools::{
     list_pods_tool, read_pod_output_tool, send_to_pod_tool, stop_pod_tool,
@@ -22,16 +26,6 @@ use protocol::{
     AlertLevel, AlertSource, ErrorCode, Event, Method, PodStatus, RunResult, Segment, TurnResult,
 };
 
-fn is_system_message_item(item: &Item) -> bool {
-    matches!(
-        item,
-        Item::Message {
-            role: Role::System,
-            ..
-        }
-    )
-}
-
 // ---------------------------------------------------------------------------
 // PodHandle — client-facing, Clone-able
 // ---------------------------------------------------------------------------
@@ -43,6 +37,10 @@ pub struct PodHandle {
     pub shared_state: Arc<PodSharedState>,
     pub runtime_dir: Arc<RuntimeDir>,
     pub alerter: Alerter,
+    /// Session-log mirror + broadcast handle. The IPC server snapshots
+    /// it on every new connection (Event::Snapshot) and forwards
+    /// subsequent commits (Event::Entry) on the receiver.
+    pub sink: SessionLogSink,
 }
 
 impl PodHandle {
@@ -86,11 +84,11 @@ async fn finish_controller_run<C, St>(
     C: LlmClient + Clone + 'static,
     St: Store + Clone + 'static,
 {
-    let items = pod.worker().history().to_vec();
+    // history / user_segments are no longer mirrored on PodSharedState —
-    shared_state.update_history(items);
+    // clients reconstruct them from `Event::Snapshot` + live
-    shared_state.set_user_segments(pod.user_segments().to_vec());
+    // `Event::Entry` deliveries driven by the session-log sink. We
+    // only flip the status and kick post-run memory jobs here.
     set_controller_status(shared_state, runtime_dir, event_tx, new_status).await;
-    let _ = runtime_dir.write_history(shared_state).await;
     pod.spawn_post_run_memory_jobs();
 }
 
@@ -167,8 +165,23 @@ impl PodController {
         }])
         .map_err(std::io::Error::other)?;
 
+        // === 1.5. Per-item history-commit drain task ===
+        //
+        // Worker callbacks fire `on_history_append` for each assistant
+        // item / tool result / hook-injected item that lands in
+        // history. The drain task picks them up off an unbounded mpsc
+        // and commits each as a typed `LogEntry` through the sink,
+        // serialised against the same `session_head` lock the Pod uses
+        // for its own commits. This gives mid-turn snapshot visibility:
+        // a late-attaching client sees in-flight tool calls + completed
+        // assistant blocks without waiting for the turn-end persist.
+        let (log_cmd_tx, log_cmd_rx) = mpsc::unbounded_channel::<LogCommand>();
+        let drain_ctx = pod.log_drain_handle();
+        let _drain_task = tokio::spawn(run_log_drain(log_cmd_rx, drain_ctx));
+        pod.attach_log_cmd_tx(log_cmd_tx.clone());
+
         // === 2. Worker event bridge wiring ===
-        wire_event_bridges_on_worker(&mut pod, &event_tx, &alerter);
+        wire_event_bridges_on_worker(&mut pod, &event_tx, &alerter, log_cmd_tx);
 
         // === 3. Tool registration (builtin / memory / spawn-orchestration) ===
         let fs_for_view = register_pod_tools(
@@ -193,8 +206,6 @@ impl PodController {
             manifest_toml.clone(),
             greeting,
         ));
-        shared_state.update_history(pod.worker().history().to_vec());
-        shared_state.set_user_segments(pod.user_segments().to_vec());
         shared_state.set_fs_view(crate::fs_view::PodFsView::new(fs_for_view));
         shared_state.set_workflows(
             pod.workflow_completions()
@@ -210,7 +221,6 @@ impl PodController {
         );
         runtime_dir.write_manifest(&manifest_toml).await?;
         runtime_dir.write_status(&shared_state).await?;
-        runtime_dir.write_history(&shared_state).await?;
 
         let handle = PodHandle {
             method_tx,
@@ -218,6 +228,7 @@ impl PodController {
             shared_state: shared_state.clone(),
             runtime_dir: runtime_dir.clone(),
             alerter: alerter.clone(),
+            sink: pod.sink(),
         };
 
         let socket_server = SocketServer::start(&handle).await?;
@@ -251,16 +262,30 @@ impl PodController {
 /// Wire the per-event broadcast bridges on the Pod's Worker. Each callback
 /// re-publishes a worker-level signal as a `protocol::Event` on `event_tx`
 /// so subscribers (TUI, socket clients) get a single typed stream.
+///
+/// Also wires `on_history_append` into the per-item drain channel so
+/// every history append observed by the worker becomes a typed
+/// `LogEntry` commit (via the drain task).
 fn wire_event_bridges_on_worker<C, St>(
     pod: &mut Pod<C, St>,
     event_tx: &broadcast::Sender<Event>,
     alerter: &Alerter,
+    log_cmd_tx: mpsc::UnboundedSender<LogCommand>,
 ) where
     C: LlmClient + Clone + 'static,
     St: Store + Clone + 'static,
 {
     let worker = pod.worker_mut();
 
+    // Per-history-append → drain channel. Sends are infallible-by-design
+    // here (UnboundedSender never blocks); a closed receiver just means
+    // the controller is shutting down, in which case dropping the item
+    // is acceptable.
+    let drain_tx = log_cmd_tx.clone();
+    worker.on_history_append(move |item| {
+        let _ = drain_tx.send(LogCommand::Item(item.clone()));
+    });
+
     let tx = event_tx.clone();
     worker.on_turn_start(move |turn| {
         let _ = tx.send(Event::TurnStart { turn });
@@ -365,13 +390,80 @@ fn wire_event_bridges_on_worker<C, St>(
         alerter_for_worker.alert(AlertLevel::Warn, AlertSource::Worker, message.to_owned());
     });
 
-    let tx = event_tx.clone();
+    // History-append broadcasts (previously `Event::SystemMessage`)
-    worker.on_history_append(move |item| {
+    // have been removed: every persistent history item is now committed
-        if is_system_message_item(item) {
+    // through the session-log sink as a typed `LogEntry`, and clients
-            let value = serde_json::to_value(item).expect("Item is Serialize");
+    // see it via `Event::Snapshot` + live `Event::Entry`. The
-            let _ = tx.send(Event::SystemMessage { item: value });
+    // per-item commit channel is wired at the top of this function.
+}
+
+/// Drain task: consumes `LogCommand::Item` and `LogCommand::Flush`
+/// off the channel and commits each item as a typed `LogEntry` through
+/// the supplied store + sink. Lives as long as the controller; exits
+/// when the sender is dropped (controller shutdown).
+async fn run_log_drain<St>(
+    mut rx: mpsc::UnboundedReceiver<LogCommand>,
+    ctx: LogDrainHandle<St>,
+) where
+    St: session_store::Store + Clone + Send + 'static,
+{
+    while let Some(cmd) = rx.recv().await {
+        match cmd {
+            LogCommand::Item(item) => {
+                let Some(entry) = classify_history_item(item) else {
+                    continue;
+                };
+                let mut head = ctx.session_head.lock().await;
+                match session_store::append_entry_with_hash(
+                    &ctx.store,
+                    head.session_id,
+                    &mut head.head_hash,
+                    entry.clone(),
+                )
+                .await
+                {
+                    Ok(_) => {
+                        // Publish under the same critical section view
+                        // a `subscribe_with_snapshot` would observe.
+                        ctx.sink.publish(entry);
+                    }
+                    Err(e) => {
+                        tracing::warn!(error = %e, "drain: append_entry failed; entry dropped");
+                    }
+                }
+            }
+            LogCommand::Flush(ack) => {
+                let _ = ack.send(());
+            }
         }
-    });
+    }
+}
+
+/// Map a single worker-history `Item` to its corresponding `LogEntry`
+/// classification. `None` is the skip signal for `user_message` items —
+/// those are committed via `LogEntry::UserInput` by `Pod::run` at
+/// submit time and would otherwise produce a duplicate entry here.
+fn classify_history_item(item: Item) -> Option<LogEntry> {
+    let ts = session_log::now_millis();
+    if item.is_user_message() {
+        return None;
+    }
+    if item.is_tool_result() {
+        return Some(LogEntry::ToolResults {
+            ts,
+            items: vec![session_store::LoggedItem::from(&item)],
+        });
+    }
+    if item.is_assistant_message() || item.is_tool_call() || item.is_reasoning() {
+        return Some(LogEntry::AssistantItems {
+            ts,
+            items: vec![session_store::LoggedItem::from(&item)],
+        });
+    }
+    Some(LogEntry::HookInjectedItems {
+        ts,
+        items: vec![session_store::LoggedItem::from(&item)],
+    })
 }
 
 /// Register the builtin file-manipulation tools, optional memory tools,
@@ -656,10 +748,9 @@ async fn controller_loop<C, St>(
                 break;
             }
 
-            // GetHistory / ListCompletions are handled at the socket
+            // ListCompletions is handled at the socket layer (direct
-            // layer (direct response). If they reach the controller,
+            // response). If it reaches the controller, ignore it.
-            // ignore them.
+            Method::ListCompletions { .. } => {}
-            Method::GetHistory | Method::ListCompletions { .. } => {}
 
             Method::PodEvent(event) => {
                 // Echo the received event to all subscribers so every
@@ -820,7 +911,7 @@ where
                         // drain it at its next pre_llm_request.
                         notify_buffer.push(message);
                     }
-                    Some(Method::GetHistory | Method::ListCompletions { .. }) => {}
+                    Some(Method::ListCompletions { .. }) => {}
                     Some(Method::PodEvent(event)) => {
                         let _ = event_tx.send(Event::PodEvent(event.clone()));
                         // mpsc is consume-once, so we cannot defer this

crates/pod/src/ipc/server.rs

@@ -62,6 +62,13 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: PodHandle) {
     let mut reader = JsonLineReader::new(reader);
     let mut writer = JsonLineWriter::new(writer);
 
+    // Atomically subscribe to the session-log mirror first. The
+    // returned (snapshot, rx) pair partitions the entry timeline:
+    // entries committed before this call appear in `entries`, every
+    // entry after lands on `entry_rx`. Doing this before the alert
+    // snapshot keeps both ordering pairs internally consistent.
+    let (entries_snapshot, mut entry_rx) = handle.sink.subscribe_with_snapshot();
+
     // Atomically subscribe and snapshot buffered alerts so that
     // warnings emitted before this client connected are replayed
     // exactly once — they appear in the snapshot, and any alert
@@ -73,8 +80,41 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: PodHandle) {
         }
     }
 
+    // Send the typed snapshot up front so late attachers can
+    // reconstruct view state without an extra round trip.
+    let snapshot_event = Event::Snapshot {
+        entries: entries_snapshot
+            .into_iter()
+            .map(|e| serde_json::to_value(&e).expect("LogEntry is Serialize"))
+            .collect(),
+        greeting: handle.shared_state.greeting.clone(),
+        status: handle.shared_state.get_status(),
+    };
+    if writer.write(&snapshot_event).await.is_err() {
+        return;
+    }
+
     loop {
         tokio::select! {
+            // Live session-log entries → this client as Event::Entry.
+            entry = entry_rx.recv() => {
+                match entry {
+                    Ok(entry) => {
+                        let value = serde_json::to_value(&entry)
+                            .expect("LogEntry is Serialize");
+                        if writer.write(&Event::Entry { entry: value }).await.is_err() {
+                            break;
+                        }
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
+                        // Slow client fell behind the broadcast buffer.
+                        // Drop the connection so the next reconnect
+                        // re-seeds the prefix via subscribe_with_snapshot.
+                        break;
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+                }
+            }
             // Broadcast events → this client
             event = rx.recv() => {
                 match event {
@@ -129,57 +169,6 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: PodHandle) {
                             break;
                         }
                     }
-                    Ok(Some(Method::GetHistory)) => {
-                        let items = handle.shared_state.history();
-                        let segments_per_user = handle.shared_state.user_segments();
-                        // Embed `segments` on user-message JSON values so
-                        // the TUI can re-render typed atoms on restore.
-                        // Alignment: segments are recorded only for
-                        // submissions made during the live session, never
-                        // for seed history loaded via `SessionStart.history`
-                        // (post-compaction). The seed user_messages always
-                        // come first in worker history, so the last
-                        // `segments_per_user.len()` user_messages are the
-                        // ones that map 1:1 to the segments list.
-                        let total_user_msgs =
-                            items.iter().filter(|i| i.is_user_message()).count();
-                        let skip = total_user_msgs.saturating_sub(segments_per_user.len());
-                        let mut user_idx = 0usize;
-                        let values = items
-                            .iter()
-                            .map(|item| {
-                                let mut value =
-                                    serde_json::to_value(item).expect("Item is Serialize");
-                                if item.is_user_message() {
-                                    if user_idx >= skip {
-                                        let seg_idx = user_idx - skip;
-                                        if let Some(obj) = value.as_object_mut() {
-                                            let segs = serde_json::to_value(
-                                                &segments_per_user[seg_idx],
-                                            )
-                                            .expect("Segment is Serialize");
-                                            obj.insert("segments".into(), segs);
-                                        }
-                                    }
-                                    user_idx += 1;
-                                }
-                                value
-                            })
-                            .collect();
-                        let greeting = handle.shared_state.greeting.clone();
-                        let status = handle.shared_state.get_status();
-                        if writer
-                            .write(&Event::History {
-                                items: values,
-                                greeting,
-                                status,
-                            })
-                            .await
-                            .is_err()
-                        {
-                            break;
-                        }
-                    }
                     Ok(Some(method)) => {
                         let _ = handle.send(method).await;
                     }

crates/pod/src/lib.rs

@@ -5,6 +5,7 @@ pub mod hook;
 pub mod ipc;
 pub mod prompt;
 pub mod runtime;
+pub mod session_log_sink;
 pub mod shared_state;
 pub mod spawn;
 pub mod workflow;
@@ -30,4 +31,5 @@ pub use prompt::system::{SystemPromptContext, SystemPromptError, SystemPromptTem
 pub use protocol::{ErrorCode, Event, Method, PodStatus, TurnResult};
 pub use provider::{ProviderError, build_client};
 pub use runtime::dir::RuntimeDir;
+pub use session_log_sink::{SessionLogSink, SessionLogWriter};
 pub use shared_state::PodSharedState;

crates/pod/src/pod.rs

@@ -7,10 +7,40 @@ use llm_worker::Item;
 use llm_worker::llm_client::RequestConfig;
 use llm_worker::llm_client::client::LlmClient;
 use llm_worker::state::Mutable;
-use llm_worker::{Role, ToolOutputLimits, UsageRecord, Worker, WorkerError, WorkerResult};
+use llm_worker::{ToolOutputLimits, UsageRecord, Worker, WorkerError, WorkerResult};
-use session_store::{EntryHash, PodScopeSnapshot, SessionId, SessionStartState, Store, StoreError};
+use session_store::{
+    EntryHash, HashedEntry, LogEntry, PodScopeSnapshot, SessionId, Store, StoreError, session_log,
+    to_logged,
+};
 use tracing::{info, warn};
 
+use crate::session_log_sink::SessionLogSink;
+
+/// Command sent to the per-Pod history-drain task.
+///
+/// `Item` carries one worker-history append observed via
+/// `Worker::on_history_append`; the drain classifies it into a
+/// `LogEntry::AssistantItems` / `LogEntry::ToolResults` /
+/// `LogEntry::HookInjectedItems` and commits it through the sink.
+/// `Flush(ack)` is the barrier used by `persist_turn` to ensure every
+/// in-flight item is committed before the trailing `TurnEnd` entry
+/// lands.
+#[derive(Debug)]
+pub enum LogCommand {
+    Item(Item),
+    Flush(tokio::sync::oneshot::Sender<()>),
+}
+
+/// State shared between Pod and the controller-spawned history-drain
+/// task: store + session-head lock + broadcast sink. All three are
+/// `Clone`able (the latter two as `Arc` clones, the store per its
+/// `Clone` impl) so handing a copy to the drain task is cheap.
+pub struct LogDrainHandle<St> {
+    pub store: St,
+    pub session_head: Arc<AsyncMutex<SessionHead>>,
+    pub sink: SessionLogSink,
+}
+
 use manifest::{
     Permission, PodManifest, PodManifestConfig, ResolveError, Scope, ScopeConfig, ScopeError,
     ScopeRule, SharedScope, WorkerManifest,
@@ -38,9 +68,9 @@ use protocol::{AlertLevel, AlertSource, Event, Segment};
 use tokio::sync::broadcast;
 use tokio::task::JoinHandle;
 
-struct SessionHead {
+pub struct SessionHead {
-    session_id: SessionId,
+    pub session_id: SessionId,
-    head_hash: Option<EntryHash>,
+    pub head_hash: Option<EntryHash>,
 }
 
 /// Pre-LLM-request hook that records `history.len()` at send time into a
@@ -190,9 +220,22 @@ pub struct Pod<C: LlmClient, St: Store> {
     /// the K-th `Item::user_message` in `worker.history()` (modulo seed
     /// history loaded via `SessionStart.history`, whose original segments
     /// are not preserved). Populated from log on `restore_from_manifest`,
-    /// appended after `save_user_input` on each `run`. Mirrored to
+    /// appended after `save_user_input` on each `run`. Pre-`Event::Snapshot`
-    /// `PodSharedState` by the controller for `Event::History` use.
+    /// this fed `PodSharedState.user_segments`; the new wire format
+    /// carries typed atoms via `LogEntry::UserInput { segments }` so
+    /// this remains purely an in-memory tracker for compact alignment.
     user_segments: Vec<Vec<Segment>>,
+    /// Pod-side session-log mirror + broadcast sink. Populated alongside
+    /// every successful `session_store::append_entry` write so connected
+    /// clients see a `(snapshot, live)` stream consistent with what's
+    /// on disk.
+    sink: SessionLogSink,
+    /// Sender into the controller-spawned history-drain task.
+    /// `None` when no controller has wired one (tests, low-level Pod
+    /// usage). The drain task is the source of mid-turn `AssistantItems`
+    /// / `ToolResults` / `HookInjectedItems` commits, fed by the
+    /// `Worker::on_history_append` callback.
+    log_cmd_tx: Option<tokio::sync::mpsc::UnboundedSender<LogCommand>>,
 }
 
 impl<C: LlmClient + 'static, St: Store + 'static> Pod<C, St> {
@@ -249,6 +292,22 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Pod<C, St> {
             extract_pointer: self.extract_pointer.clone(),
             memory_task: None,
             user_segments: self.user_segments.clone(),
+            // The memory-task clone never appends to the session log
+            // (it only reads `worker.history()`), so a fresh sink is
+            // fine — nothing observes its broadcast.
+            sink: SessionLogSink::new(),
+            log_cmd_tx: None,
+        }
+    }
+
+    /// Build a `LogDrainHandle` carrying everything the controller's
+    /// drain task needs: store handle, the shared session-head lock,
+    /// and the broadcast sink. All three are cheap clones.
+    pub fn log_drain_handle(&self) -> LogDrainHandle<St> {
+        LogDrainHandle {
+            store: self.store.clone(),
+            session_head: self.session_head.clone(),
+            sink: self.sink.clone(),
         }
     }
 
@@ -332,6 +391,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
             extract_pointer: Arc::new(Mutex::new(None)),
             memory_task: None,
             user_segments: Vec::new(),
+            sink: SessionLogSink::new(),
+            log_cmd_tx: None,
         };
         pod.apply_permissions_from_manifest();
         pod.apply_prune_from_manifest();
@@ -426,8 +487,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
     /// can restore the narrowed scope instead of reclaiming delegated
     /// writes.
     pub async fn persist_scope_snapshot(&mut self) -> Result<(), StoreError> {
-        let mut head = self.session_head.lock().await;
+        if self.session_head.lock().await.head_hash.is_none() {
-        if head.head_hash.is_none() {
             return Ok(());
         }
         let snapshot = {
@@ -437,8 +497,50 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
                 deny: scope.deny_rules(),
             }
         };
-        session_store::save_pod_scope(&self.store, head.session_id, &mut head.head_hash, &snapshot)
+        let payload = serde_json::to_value(&snapshot).expect("PodScopeSnapshot is Serialize");
-            .await
+        self.commit_entry(LogEntry::Extension {
+            ts: session_log::now_millis(),
+            domain: session_store::POD_SCOPE_EXTENSION_DOMAIN.into(),
+            payload,
+        })
+        .await
+        .map(|_| ())
+    }
+
+    /// Append `entry` to the session log AND publish it through the
+    /// broadcast sink. Holds the session-head async lock across the
+    /// disk write and the sink publish so subscribers see a gap-free
+    /// `(snapshot, live)` stream consistent with what's on disk.
+    pub(crate) async fn commit_entry(
+        &self,
+        entry: LogEntry,
+    ) -> Result<EntryHash, StoreError> {
+        let mut head = self.session_head.lock().await;
+        let hash = session_store::append_entry_with_hash(
+            &self.store,
+            head.session_id,
+            &mut head.head_hash,
+            entry.clone(),
+        )
+        .await?;
+        self.sink.publish(entry);
+        Ok(hash)
+    }
+
+    /// Cloneable sink handle. Exposed to the controller so the IPC
+    /// layer can `subscribe_with_snapshot` and stream entries to
+    /// clients without consulting any other state.
+    pub fn sink(&self) -> SessionLogSink {
+        self.sink.clone()
+    }
+
+    /// Wire a history-drain task. The controller calls this once per
+    /// Pod after the drain task is spawned; the matching mpsc receiver
+    /// drives per-item commits of assistant items / tool results /
+    /// hook-injected items committed by the worker via
+    /// `Worker::on_history_append`.
+    pub fn attach_log_cmd_tx(&mut self, tx: tokio::sync::mpsc::UnboundedSender<LogCommand>) {
+        self.log_cmd_tx = Some(tx);
     }
 
     /// Cloneable callback handed to dynamic-scope tools. It cannot append
@@ -459,13 +561,12 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
             .expect("pending_scope_snapshot poisoned")
             .take();
         if let Some(snapshot) = snapshot {
-            let mut head = self.session_head.lock().await;
+            let payload = serde_json::to_value(&snapshot).expect("PodScopeSnapshot is Serialize");
-            session_store::save_pod_scope(
+            self.commit_entry(LogEntry::Extension {
-                &self.store,
+                ts: session_log::now_millis(),
-                head.session_id,
+                domain: session_store::POD_SCOPE_EXTENSION_DOMAIN.into(),
-                &mut head.head_hash,
+                payload,
-                &snapshot,
+            })
-            )
             .await?;
         }
         Ok(())
@@ -629,15 +730,13 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
     /// (the entry is dropped) and a `Warn` alert + `tracing::warn!` are
     /// emitted so the failure isn't completely silent.
     async fn try_record_metric(&mut self, metric: &session_metrics::Metric) {
-        let mut head = self.session_head.lock().await;
+        let payload = serde_json::to_value(metric).expect("Metric is Serialize");
-        if let Err(err) = session_metrics::record_metric(
+        let entry = LogEntry::Extension {
-            &self.store,
+            ts: session_log::now_millis(),
-            head.session_id,
+            domain: session_metrics::DOMAIN.into(),
-            &mut head.head_hash,
+            payload,
-            metric,
+        };
-        )
+        if let Err(err) = self.commit_entry(entry).await {
-        .await
-        {
             warn!(name = %metric.name, error = %err, "failed to record session metric; dropping");
             self.alert(
                 AlertLevel::Warn,
@@ -656,20 +755,6 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         }
     }
 
-    fn broadcast_system_message_item(&self, item: &Item) {
-        if !matches!(
-            item,
-            Item::Message {
-                role: Role::System,
-                ..
-            }
-        ) {
-            return;
-        }
-        let value = serde_json::to_value(item).expect("Item is Serialize");
-        self.send_event(Event::SystemMessage { item: value });
-    }
-
     /// Push a `Method::Notify` (or rendered `Method::PodEvent`) entry
     /// onto the pending buffer.
     ///
@@ -975,17 +1060,12 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         // Persist the user input as typed segments before the worker
         // pushes its flattened copy into history. save_delta deliberately
         // skips the resulting `is_user_message()` item to avoid double-write.
-        {
+        self.session_id = self.session_head.lock().await.session_id;
-            let mut head = self.session_head.lock().await;
+        self.commit_entry(LogEntry::UserInput {
-            self.session_id = head.session_id;
+            ts: session_log::now_millis(),
-            session_store::save_user_input(
+            segments: input.clone(),
-                &self.store,
+        })
-                head.session_id,
+        .await?;
-                &mut head.head_hash,
-                input.clone(),
-            )
-            .await?;
-        }
         self.user_segments.push(input.clone());
 
         // Resolve `@<path>` refs, `#<slug>` Knowledge refs, and `/<slug>`
@@ -1330,34 +1410,64 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
     /// another writer has advanced the store head behind our back.
     async fn ensure_session_head(&mut self) -> Result<(), PodError> {
         let w = self.worker.as_ref().unwrap();
-        let state = SessionStartState {
+        let prev_session_id;
-            system_prompt: w.get_system_prompt(),
+        let initial_state = {
-            config: w.request_config(),
+            let head = self.session_head.lock().await;
-            history: w.history(),
+            prev_session_id = head.session_id;
+            head.head_hash.is_none()
         };
-        let mut head = self.session_head.lock().await;
+        if initial_state {
-        if head.head_hash.is_none() {
+            let initial = LogEntry::SessionStart {
-            let hash =
+                ts: session_log::now_millis(),
-                session_store::create_session_with_id(&self.store, head.session_id, state).await?;
+                system_prompt: w.get_system_prompt().map(String::from),
-            head.head_hash = Some(hash);
+                config: w.request_config().clone(),
-            drop(head);
+                history: to_logged(w.history()),
+                forked_from: None,
+                compacted_from: None,
+            };
+            self.commit_entry(initial).await?;
             self.persist_scope_snapshot().await?;
             return Ok(());
         }
-        let prev_session_id = head.session_id;
+        // Check store head + auto-fork if it drifted.
-        let mut session_id = head.session_id;
+        let store_head = self
-        let mut head_hash = head.head_hash.clone();
+            .store
-        session_store::ensure_head_or_fork(&self.store, &mut session_id, &mut head_hash, state)
+            .read_head_hash(prev_session_id)
-            .await?;
+            .await
-        head.session_id = session_id;
+            .map_err(PodError::from)?;
-        head.head_hash = head_hash;
+        let mut head = self.session_head.lock().await;
-        self.session_id = session_id;
+        if store_head == head.head_hash {
-        // ensure_head_or_fork mints a fresh session_id when it auto-
+            return Ok(());
-        // forks. Sync that to pods.json so a concurrent
+        }
-        // restore_from_manifest can't see "no live writer" for the new
+        // Fork: mint a fresh session and switch to it. The new
-        // session and grab it.
+        // SessionStart entry replaces the mirror and is broadcast
-        if session_id != prev_session_id && self.scope_allocation.is_some() {
+        // through the sink so existing subscribers reset their view.
-            pod_registry::update_session(&self.manifest.pod.name, session_id)?;
+        let fork_id = session_store::new_session_id();
+        let entry = LogEntry::SessionStart {
+            ts: session_log::now_millis(),
+            system_prompt: w.get_system_prompt().map(String::from),
+            config: w.request_config().clone(),
+            history: to_logged(w.history()),
+            forked_from: None,
+            compacted_from: None,
+        };
+        let hash = session_log::compute_hash(None, &entry);
+        let hashed = HashedEntry {
+            hash: hash.clone(),
+            prev_hash: None,
+            entry: entry.clone(),
+        };
+        self.store
+            .create_session(fork_id, &[hashed])
+            .await
+            .map_err(PodError::from)?;
+        head.session_id = fork_id;
+        head.head_hash = Some(hash);
+        self.session_id = fork_id;
+        self.sink.reset_with_initial(entry);
+        drop(head);
+        if self.scope_allocation.is_some() {
+            pod_registry::update_session(&self.manifest.pod.name, fork_id)?;
         }
         Ok(())
     }
@@ -1493,28 +1603,84 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         history_before: usize,
         result: &Result<WorkerResult, WorkerError>,
     ) -> Result<(), StoreError> {
-        // Use direct field access for split borrows (worker immutable,
+        // Per-item commits for AssistantItems / ToolResults /
-        // head_hash mutable).
+        // HookInjectedItems already landed mid-turn through the
-        let w = self.worker.as_ref().unwrap();
+        // controller-spawned drain task, fed by
-        let new_items = &w.history()[history_before..];
+        // `Worker::on_history_append`. Drain the queue here so every
-        let mut head = self.session_head.lock().await;
+        // in-flight item has actually been committed before the
-        self.session_id = head.session_id;
+        // trailing `TurnEnd` entry. When no drain is wired (low-level
-        session_store::save_delta(&self.store, head.session_id, &mut head.head_hash, new_items)
+        // tests / direct `Pod::new` usage) we fall back to a synchronous
-            .await?;
+        // pass that replicates the legacy `save_delta` classification —
+        // those code paths don't fire `on_history_append`, so the items
+        // would otherwise be lost.
+        let _ = history_before; // referenced only by the fallback below.
+        self.session_id = self.session_head.lock().await.session_id;
+        if let Some(tx) = self.log_cmd_tx.as_ref() {
+            let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
+            if tx.send(LogCommand::Flush(ack_tx)).is_ok() {
+                let _ = ack_rx.await;
+            }
+        } else {
+            // Fallback path for tests / Pod::new: classify and commit
+            // the post-`history_before` slice inline, matching the old
+            // `save_delta` shape.
+            let new_items: Vec<Item> = self.worker.as_ref().unwrap().history()[history_before..]
+                .iter()
+                .cloned()
+                .collect();
+            let ts = session_log::now_millis();
+            let mut i = 0;
+            while i < new_items.len() {
+                let item = &new_items[i];
+                if item.is_user_message() {
+                    i += 1;
+                } else if item.is_tool_result() {
+                    let start = i;
+                    while i < new_items.len() && new_items[i].is_tool_result() {
+                        i += 1;
+                    }
+                    let items = new_items[start..i]
+                        .iter()
+                        .map(session_store::LoggedItem::from)
+                        .collect();
+                    self.commit_entry(LogEntry::ToolResults { ts, items }).await?;
+                } else if item.is_assistant_message()
+                    || item.is_tool_call()
+                    || item.is_reasoning()
+                {
+                    let start = i;
+                    while i < new_items.len()
+                        && (new_items[i].is_assistant_message()
+                            || new_items[i].is_tool_call()
+                            || new_items[i].is_reasoning())
+                    {
+                        i += 1;
+                    }
+                    let items = new_items[start..i]
+                        .iter()
+                        .map(session_store::LoggedItem::from)
+                        .collect();
+                    self.commit_entry(LogEntry::AssistantItems { ts, items })
+                        .await?;
+                } else {
+                    self.commit_entry(LogEntry::HookInjectedItems {
+                        ts,
+                        items: vec![session_store::LoggedItem::from(&new_items[i])],
+                    })
+                    .await?;
+                    i += 1;
+                }
+            }
+        }
 
-        drop(head);
         self.flush_pending_scope_snapshot().await?;
 
         let turn_count = self.worker.as_ref().unwrap().turn_count();
-        let mut head = self.session_head.lock().await;
+        self.commit_entry(LogEntry::TurnEnd {
-        session_store::save_turn_end(
+            ts: session_log::now_millis(),
-            &self.store,
-            head.session_id,
-            &mut head.head_hash,
             turn_count,
-        )
+        })
         .await?;
-        drop(head);
 
         // Flush any sync-buffered metrics from this run first
         // (currently `prune.fire` / `prune.skip` from the prune observer).
@@ -1547,19 +1713,15 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
                 record,
                 correlation_id,
             } = recorded;
-            let mut head = self.session_head.lock().await;
+            self.commit_entry(LogEntry::LlmUsage {
-            session_store::save_usage(
+                ts: session_log::now_millis(),
-                &self.store,
+                history_len: record.history_len,
-                head.session_id,
+                input_total_tokens: record.input_total_tokens,
-                &mut head.head_hash,
+                cache_read_tokens: record.cache_read_tokens,
-                record.history_len,
+                cache_write_tokens: record.cache_write_tokens,
-                record.input_total_tokens,
+                output_tokens: record.output_tokens,
-                record.cache_read_tokens,
+            })
-                record.cache_write_tokens,
-                record.output_tokens,
-            )
             .await?;
-            drop(head);
             if let Some(id) = correlation_id {
                 let metric = session_metrics::Metric::now("prune.post_request")
                     .with_correlation_id(&id)
@@ -1577,25 +1739,19 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         let interrupted = self.worker.as_ref().unwrap().last_run_interrupted();
         match result {
             Ok(r) => {
-                let mut head = self.session_head.lock().await;
+                self.commit_entry(LogEntry::RunCompleted {
-                session_store::save_run_completed(
+                    ts: session_log::now_millis(),
-                    &self.store,
-                    head.session_id,
-                    &mut head.head_hash,
-                    r.clone(),
                     interrupted,
-                )
+                    result: r.clone(),
+                })
                 .await?;
             }
             Err(e) => {
-                let mut head = self.session_head.lock().await;
+                self.commit_entry(LogEntry::RunErrored {
-                session_store::save_run_errored(
+                    ts: session_log::now_millis(),
-                    &self.store,
-                    head.session_id,
-                    &mut head.head_hash,
-                    e.to_string(),
                     interrupted,
-                )
+                    message: e.to_string(),
+                })
                 .await?;
             }
         }
@@ -1824,34 +1980,46 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
             task_snapshot_text.clone(),
         ));
 
-        // Persist as a new compacted session.
+        // Build the SessionStart entry for the new compacted session,
-        let mut head = self.session_head.lock().await;
+        // then atomically rotate to it: create on disk, swap head, reset
-        let old_session_id = head.session_id;
+        // the broadcast sink so existing subscribers see the new
-        let old_head_hash = head
+        // `SessionStart { compacted_from }` and reset their view.
-            .head_hash
+        let new_session_id = session_store::new_session_id();
-            .clone()
+        let session_start = {
-            .expect("head_hash should be set after at least one entry");
+            let mut head = self.session_head.lock().await;
-
+            let old_session_id = head.session_id;
-        let w = self.worker.as_ref().unwrap();
+            let old_head_hash = head
-        let state = SessionStartState {
+                .head_hash
-            system_prompt: w.get_system_prompt(),
+                .clone()
-            config: w.request_config(),
+                .expect("head_hash should be set after at least one entry");
-            history: &new_history,
+            let w = self.worker.as_ref().unwrap();
+            let entry = LogEntry::SessionStart {
+                ts: session_log::now_millis(),
+                system_prompt: w.get_system_prompt().map(String::from),
+                config: w.request_config().clone(),
+                history: to_logged(&new_history),
+                forked_from: None,
+                compacted_from: Some(session_store::SessionOrigin {
+                    session_id: old_session_id,
+                    at_hash: old_head_hash,
+                }),
+            };
+            let hash = session_log::compute_hash(None, &entry);
+            let hashed = HashedEntry {
+                hash: hash.clone(),
+                prev_hash: None,
+                entry: entry.clone(),
+            };
+            self.store.create_session(new_session_id, &[hashed]).await?;
+            head.session_id = new_session_id;
+            head.head_hash = Some(hash);
+            self.session_id = new_session_id;
+            entry
         };
-        let (new_session_id, new_head_hash) = session_store::create_compacted_session(
+        // Broadcast the SessionStart through the sink. This atomically
-            &self.store,
+        // resets the mirror to `[SessionStart]` so any subscriber
-            state,
+        // querying after this point sees the post-compaction prefix.
-            old_session_id,
+        self.sink.reset_with_initial(session_start);
-            old_head_hash,
-        )
-        .await?;
-
-        // Swap in the new session state. usage_history belongs to the old
-        // session — the new compacted session starts with no measurements
-        // until its first LLM call.
-        self.session_id = new_session_id;
-        head.session_id = new_session_id;
-        head.head_hash = Some(new_head_hash);
         // Keep pods.json pointing at the live session_id. Without this
         // a concurrent `restore_from_manifest(new_session_id)` would
         // see no live writer and grab the session this Pod just moved
@@ -1861,7 +2029,6 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         if self.scope_allocation.is_some() {
             pod_registry::update_session(&self.manifest.pod.name, new_session_id)?;
         }
-        drop(head);
         // Align user_segments with the post-compaction history. Items
         // before `retain_from` (now folded into the summary) lose their
         // segments; only the user_messages surviving in retained_items
@@ -1873,9 +2040,11 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         }
 
         self.worker.as_mut().unwrap().set_history(new_history);
-        for item in &compact_introduced_system_messages {
+        // Compaction-introduced system messages are part of the new
-            self.broadcast_system_message_item(item);
+        // SessionStart's history (broadcast above) — clients derive
-        }
+        // their blocks from `SessionStart.history`. No per-item
+        // broadcast is required.
+        let _ = &compact_introduced_system_messages;
         let worker = self.worker.as_mut().unwrap();
         // Anchor the prompt cache at the summary item so that Anthropic
         // can place a durable `cache_control` breakpoint there — our
@@ -2139,18 +2308,13 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         };
         let payload_value = serde_json::to_value(&pointer_payload)
             .expect("ExtractPointerPayload is always JSON-serializable");
-        {
+        self.commit_entry(LogEntry::Extension {
-            let mut head = self.session_head.lock().await;
+            ts: session_log::now_millis(),
-            session_store::save_extension(
+            domain: extract::EXTRACT_DOMAIN.into(),
-                &self.store,
+            payload: payload_value,
-                head.session_id,
+        })
-                &mut head.head_hash,
+        .await?;
-                extract::EXTRACT_DOMAIN,
+        self.session_id = self.session_head.lock().await.session_id;
-                payload_value,
-            )
-            .await?;
-            self.session_id = head.session_id;
-        }
 
         *self
             .extract_pointer
@@ -2488,6 +2652,8 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
             extract_pointer: Arc::new(Mutex::new(None)),
             memory_task: None,
             user_segments: Vec::new(),
+            sink: SessionLogSink::new(),
+            log_cmd_tx: None,
         };
         pod.apply_permissions_from_manifest();
         pod.apply_prune_from_manifest();
@@ -2559,6 +2725,8 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
             extract_pointer: Arc::new(Mutex::new(None)),
             memory_task: None,
             user_segments: Vec::new(),
+            sink: SessionLogSink::new(),
+            log_cmd_tx: None,
         };
         pod.apply_permissions_from_manifest();
         pod.apply_prune_from_manifest();
@@ -2590,10 +2758,16 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
         store: St,
         loader: PromptLoader,
     ) -> Result<Self, PodError> {
-        let state = session_store::restore(&store, session_id).await?;
+        // Read raw entries once so we can both reconstruct state and
+        // seed the broadcast sink's mirror with the same prefix that
+        // sits on disk.
+        let raw_entries = store.read_all(session_id).await?;
+        let state = session_store::collect_state(&raw_entries);
         if state.head_hash.is_none() {
             return Err(PodError::SessionEmpty { session_id });
         }
+        let mirror_entries: Vec<LogEntry> =
+            raw_entries.iter().map(|e| e.entry.clone()).collect();
         let scope_snapshot = state
             .pod_scope
             .clone()
@@ -2696,6 +2870,11 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
             extract_pointer: Arc::new(Mutex::new(extract_pointer)),
             memory_task: None,
             user_segments: state.user_segments,
+            // Seed the mirror with the entries we just replayed so a
+            // late-attaching client sees the full prefix without an
+            // extra round trip.
+            sink: SessionLogSink::with_initial(mirror_entries),
+            log_cmd_tx: None,
         };
         pod.apply_permissions_from_manifest();
         pod.apply_prune_from_manifest();

crates/pod/src/runtime/dir.rs

@@ -73,12 +73,6 @@ impl RuntimeDir {
         atomic_write(&self.path.join("manifest.toml"), toml.as_bytes()).await
     }
 
-    /// Write history.json atomically.
-    pub async fn write_history(&self, state: &PodSharedState) -> Result<(), io::Error> {
-        let content = state.history_json();
-        atomic_write(&self.path.join("history.json"), content.as_bytes()).await
-    }
-
     /// Write `spawned_pods.json` atomically. The entries are the full
     /// set of spawned children known to this Pod — callers pass the
     /// replacement list, no incremental merge.
@@ -223,18 +217,6 @@ mod tests {
         assert_eq!(parsed[0].pod_name, "child");
     }
 
-    #[tokio::test]
-    async fn write_history_creates_file() {
-        let tmp = tempfile::tempdir().unwrap();
-        let rt = RuntimeDir::create(tmp.path(), "my-pod").await.unwrap();
-        let state = test_state();
-
-        rt.write_history(&state).await.unwrap();
-
-        let content = std::fs::read_to_string(rt.path().join("history.json")).unwrap();
-        assert_eq!(content, "[]");
-    }
-
     #[tokio::test]
     async fn socket_path() {
         let tmp = tempfile::tempdir().unwrap();

crates/pod/src/session_log_sink.rs

@@ -0,0 +1,474 @@
+//! Pod-side session-log mirror + broadcast.
+//!
+//! Owns the in-memory `Vec<LogEntry>` mirror that backs `Event::Snapshot`
+//! delivery to newly connected clients and the
+//! `broadcast::Sender<LogEntry>` that fans out per-entry commits to
+//! existing subscribers. Disk writes remain the responsibility of the
+//! Pod (which still owns the `Store` handle); the sink stays focused on
+//! the wire-side fan-out.
+//!
+//! Atomicity contract (see ticket `tickets/pod-state-from-session-log.md`):
+//!
+//! 1. Pod writes the entry to disk via the `Store`.
+//! 2. Pod calls [`SessionLogSink::publish`] which acquires the mirror
+//!    mutex, pushes the entry, and fires `broadcast::send` — all under
+//!    the same critical section.
+//!
+//! [`SessionLogSink::subscribe_with_snapshot`] takes the same mutex,
+//! so the `(snapshot, receiver)` pair returned to a connecting client
+//! splits the entry sequence cleanly: every entry shows up in exactly
+//! one of `snapshot` or on `receiver`.
+//!
+//! Disk-write failures short-circuit before `publish`, so a failed
+//! entry never appears in the mirror or on the broadcast.
+
+use std::sync::{Arc, Mutex as StdMutex};
+
+use session_store::{
+    EntryHash, HashedEntry, LogEntry, SessionId, SessionStartState, Store, StoreError, session_log,
+};
+use tokio::sync::{Mutex as AsyncMutex, MutexGuard, broadcast};
+
+/// Broadcast capacity for the live receiver. Slow subscribers that
+/// fall behind will see `RecvError::Lagged` and are expected to drop
+/// the connection so that the next reconnect's `subscribe_with_snapshot`
+/// re-seeds the prefix.
+const BROADCAST_CAPACITY: usize = 256;
+
+/// In-memory mirror + broadcast fan-out for the active session log.
+///
+/// Clone is cheap (`Arc` clone) — the Pod hands one to the IPC layer
+/// for read-only `subscribe_with_snapshot` access and keeps one for
+/// its own write path.
+#[derive(Clone)]
+pub struct SessionLogSink {
+    inner: Arc<SinkInner>,
+}
+
+struct SinkInner {
+    /// Full session log mirror in commit order. Reset on session swap
+    /// (compaction / fork) via [`SessionLogSink::reset_with_initial`].
+    mirror: StdMutex<Vec<LogEntry>>,
+    /// Broadcast channel for live entry updates. The same `Sender`
+    /// survives session swaps so existing subscribers keep their
+    /// receiver — they observe the swap as a freshly broadcast
+    /// `LogEntry::SessionStart` and reset their view accordingly.
+    broadcast_tx: broadcast::Sender<LogEntry>,
+}
+
+impl SessionLogSink {
+    /// Create a fresh sink with an empty mirror. Used before any entry
+    /// has been written (deferred SessionStart) or as a placeholder in
+    /// tests.
+    pub fn new() -> Self {
+        let (broadcast_tx, _) = broadcast::channel(BROADCAST_CAPACITY);
+        Self {
+            inner: Arc::new(SinkInner {
+                mirror: StdMutex::new(Vec::new()),
+                broadcast_tx,
+            }),
+        }
+    }
+
+    /// Create a sink seeded with a prefix of entries already on disk.
+    /// Used by restore / fork-at-restore code paths that materialise
+    /// the existing log before the sink starts taking new commits.
+    pub fn with_initial(entries: Vec<LogEntry>) -> Self {
+        let (broadcast_tx, _) = broadcast::channel(BROADCAST_CAPACITY);
+        Self {
+            inner: Arc::new(SinkInner {
+                mirror: StdMutex::new(entries),
+                broadcast_tx,
+            }),
+        }
+    }
+
+    /// Push `entry` to the mirror and broadcast it.
+    ///
+    /// MUST be called only after the Pod has successfully persisted the
+    /// entry to the underlying `Store` — disk write is the gate. Failed
+    /// disk writes must not call `publish`.
+    pub fn publish(&self, entry: LogEntry) {
+        let mut mirror = self
+            .inner
+            .mirror
+            .lock()
+            .expect("session log mirror mutex poisoned");
+        mirror.push(entry.clone());
+        // SendError means there are zero subscribers; harmless. We hold
+        // the mirror lock across `send` so that `subscribe_with_snapshot`
+        // cannot observe an inconsistent (snapshot, receiver) pair.
+        let _ = self.inner.broadcast_tx.send(entry);
+    }
+
+    /// Atomically swap the mirror to `[initial]` and broadcast the new
+    /// session-start entry. Used during compaction / fork: the new
+    /// `LogEntry::SessionStart` is the first entry of the replacement
+    /// session, and existing subscribers transition by replaying it
+    /// like any other live entry.
+    ///
+    /// Existing snapshot prefixes seen by old subscribers stay valid
+    /// for the prior session; the new `SessionStart` on the broadcast
+    /// is the signal to reset their derived view.
+    pub fn reset_with_initial(&self, initial: LogEntry) {
+        let mut mirror = self
+            .inner
+            .mirror
+            .lock()
+            .expect("session log mirror mutex poisoned");
+        mirror.clear();
+        mirror.push(initial.clone());
+        let _ = self.inner.broadcast_tx.send(initial);
+    }
+
+    /// Replace the mirror with the supplied prefix without broadcasting.
+    ///
+    /// Used by restore paths that load a session's complete log into
+    /// the mirror before any subscriber is connected. Callers that need
+    /// to notify existing subscribers should use [`reset_with_initial`].
+    pub fn replace_silent(&self, entries: Vec<LogEntry>) {
+        let mut mirror = self
+            .inner
+            .mirror
+            .lock()
+            .expect("session log mirror mutex poisoned");
+        *mirror = entries;
+    }
+
+    /// Atomically read the current mirror and subscribe to subsequent
+    /// commits. The returned snapshot and receiver split the entry
+    /// timeline into a duplicate-free, gap-free prefix/suffix pair.
+    pub fn subscribe_with_snapshot(&self) -> (Vec<LogEntry>, broadcast::Receiver<LogEntry>) {
+        let mirror = self
+            .inner
+            .mirror
+            .lock()
+            .expect("session log mirror mutex poisoned");
+        let snapshot = mirror.clone();
+        let rx = self.inner.broadcast_tx.subscribe();
+        (snapshot, rx)
+    }
+
+    /// Current entry count. Useful for tests / diagnostics.
+    pub fn len(&self) -> usize {
+        self.inner
+            .mirror
+            .lock()
+            .expect("session log mirror mutex poisoned")
+            .len()
+    }
+
+    /// Whether the mirror is empty.
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+}
+
+impl Default for SessionLogSink {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Active session head for the Pod's persistent log: session id +
+/// last-committed entry hash. Replaces the previous `SessionHead`
+/// struct local to `Pod`; bundled here so the writer can hand a
+/// cloneable handle to background tasks (e.g. the per-item drain
+/// task spawned by the controller).
+#[derive(Debug, Clone)]
+pub struct SessionHeadState {
+    pub session_id: SessionId,
+    pub head_hash: Option<EntryHash>,
+}
+
+/// Pod-side session-log writer.
+///
+/// Bundles the (1) persistent store, (2) the in-memory session-head
+/// state (id + hash), and (3) the broadcast sink. `append_entry`
+/// chains the hash on disk, advances the head, then publishes the
+/// entry through the sink — under a single async mutex so two writers
+/// cannot interleave the chain.
+///
+/// `Clone` is a cheap `Arc` clone. The Pod keeps one writer for its
+/// inline commits (UserInput, TurnEnd, Usage, RunCompleted/Errored,
+/// scope snapshots, metrics) and hands clones to background tasks
+/// (e.g. the controller's per-item history drain task).
+pub struct SessionLogWriter<St> {
+    inner: Arc<WriterInner<St>>,
+}
+
+struct WriterInner<St> {
+    store: St,
+    head: AsyncMutex<SessionHeadState>,
+    sink: SessionLogSink,
+}
+
+impl<St> Clone for SessionLogWriter<St> {
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl<St> SessionLogWriter<St>
+where
+    St: Store + Clone,
+{
+    /// Create a writer for a fresh Pod with no entries on disk yet.
+    /// `head_hash` is `None` until the first `append_entry` (typically
+    /// the deferred `SessionStart` written by `ensure_session_head`).
+    pub fn new(store: St, session_id: SessionId) -> Self {
+        Self {
+            inner: Arc::new(WriterInner {
+                store,
+                head: AsyncMutex::new(SessionHeadState {
+                    session_id,
+                    head_hash: None,
+                }),
+                sink: SessionLogSink::new(),
+            }),
+        }
+    }
+
+    /// Create a writer seeded with a session already on disk. The
+    /// mirror is populated with `mirror` (typically loaded via
+    /// `Store::read_all`), and `head_hash` should be the hash of the
+    /// last entry.
+    pub fn restored(
+        store: St,
+        session_id: SessionId,
+        head_hash: Option<EntryHash>,
+        mirror: Vec<LogEntry>,
+    ) -> Self {
+        Self {
+            inner: Arc::new(WriterInner {
+                store,
+                head: AsyncMutex::new(SessionHeadState {
+                    session_id,
+                    head_hash,
+                }),
+                sink: SessionLogSink::with_initial(mirror),
+            }),
+        }
+    }
+
+    /// Append `entry` to the log: disk write → in-memory mirror push →
+    /// broadcast — atomic w.r.t. `subscribe_with_snapshot` callers.
+    pub async fn append_entry(&self, entry: LogEntry) -> Result<EntryHash, StoreError> {
+        let mut head = self.inner.head.lock().await;
+        let hash = session_store::append_entry_with_hash(
+            &self.inner.store,
+            head.session_id,
+            &mut head.head_hash,
+            entry.clone(),
+        )
+        .await?;
+        self.inner.sink.publish(entry);
+        Ok(hash)
+    }
+
+    /// Atomically swap to a new compacted session.
+    ///
+    /// Creates the new session on disk with `initial` as its
+    /// `SessionStart`, advances the head, and resets the sink mirror
+    /// to `[initial]` while broadcasting the entry. Existing
+    /// subscribers observe the swap as a freshly broadcast
+    /// `SessionStart` (with `compacted_from` set), which is their
+    /// signal to reset their derived view.
+    pub async fn swap_session(
+        &self,
+        new_session_id: SessionId,
+        initial: LogEntry,
+    ) -> Result<EntryHash, StoreError> {
+        let hash = session_log::compute_hash(None, &initial);
+        let hashed = HashedEntry {
+            hash: hash.clone(),
+            prev_hash: None,
+            entry: initial.clone(),
+        };
+        self.inner
+            .store
+            .create_session(new_session_id, &[hashed])
+            .await?;
+        let mut head = self.inner.head.lock().await;
+        head.session_id = new_session_id;
+        head.head_hash = Some(hash.clone());
+        self.inner.sink.reset_with_initial(initial);
+        Ok(hash)
+    }
+
+    /// If the store's head no longer matches our cached head, mint a
+    /// fresh session that forks from the current state and switch to
+    /// it. Returns `true` when a fork happened.
+    pub async fn ensure_head_or_fork(
+        &self,
+        state: SessionStartState<'_>,
+    ) -> Result<bool, StoreError> {
+        let mut head = self.inner.head.lock().await;
+        let store_head = self.inner.store.read_head_hash(head.session_id).await?;
+        if store_head == head.head_hash {
+            return Ok(false);
+        }
+        let fork_id = session_store::new_session_id();
+        let entry = LogEntry::SessionStart {
+            ts: session_log::now_millis(),
+            system_prompt: state.system_prompt.map(String::from),
+            config: state.config.clone(),
+            history: session_store::to_logged(state.history),
+            forked_from: None,
+            compacted_from: None,
+        };
+        let hash = session_log::compute_hash(None, &entry);
+        let hashed = HashedEntry {
+            hash: hash.clone(),
+            prev_hash: None,
+            entry: entry.clone(),
+        };
+        self.inner.store.create_session(fork_id, &[hashed]).await?;
+        head.session_id = fork_id;
+        head.head_hash = Some(hash);
+        self.inner.sink.reset_with_initial(entry);
+        Ok(true)
+    }
+
+    /// Cloneable handle to the broadcast sink. Used by the IPC layer
+    /// for `subscribe_with_snapshot` and by tests that just want the
+    /// non-write side.
+    pub fn sink(&self) -> SessionLogSink {
+        self.inner.sink.clone()
+    }
+
+    /// Underlying store handle. Direct access is preserved for callers
+    /// that read state (`read_all`, `read_head_hash`) without going
+    /// through the writer's hash chain.
+    pub fn store(&self) -> &St {
+        &self.inner.store
+    }
+
+    /// Cheap snapshot of the current session id.
+    pub async fn current_session_id(&self) -> SessionId {
+        self.inner.head.lock().await.session_id
+    }
+
+    /// Cheap snapshot of the current head hash.
+    pub async fn current_head_hash(&self) -> Option<EntryHash> {
+        self.inner.head.lock().await.head_hash.clone()
+    }
+
+    /// Direct lock on the head. Used by paths that need to coordinate
+    /// custom writes with the hash chain (currently
+    /// `session_metrics::record_metric`).
+    pub async fn lock_head(&self) -> MutexGuard<'_, SessionHeadState> {
+        self.inner.head.lock().await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use llm_worker::llm_client::RequestConfig;
+    use session_store::session_log::now_millis;
+
+    fn session_start() -> LogEntry {
+        LogEntry::SessionStart {
+            ts: now_millis(),
+            system_prompt: None,
+            config: RequestConfig::default(),
+            history: vec![],
+            forked_from: None,
+            compacted_from: None,
+        }
+    }
+
+    fn turn_end(n: usize) -> LogEntry {
+        LogEntry::TurnEnd {
+            ts: now_millis(),
+            turn_count: n,
+        }
+    }
+
+    #[test]
+    fn publish_then_subscribe_returns_history_in_snapshot() {
+        let sink = SessionLogSink::new();
+        sink.publish(session_start());
+        sink.publish(turn_end(1));
+
+        let (snapshot, mut rx) = sink.subscribe_with_snapshot();
+        assert_eq!(snapshot.len(), 2);
+        assert!(matches!(snapshot[0], LogEntry::SessionStart { .. }));
+        assert!(matches!(snapshot[1], LogEntry::TurnEnd { turn_count: 1, .. }));
+        assert!(rx.try_recv().is_err());
+    }
+
+    #[test]
+    fn subscribe_then_publish_delivers_live_entries() {
+        let sink = SessionLogSink::new();
+        sink.publish(session_start());
+
+        let (snapshot, mut rx) = sink.subscribe_with_snapshot();
+        assert_eq!(snapshot.len(), 1);
+
+        sink.publish(turn_end(1));
+        match rx.try_recv() {
+            Ok(LogEntry::TurnEnd { turn_count: 1, .. }) => {}
+            other => panic!("unexpected: {other:?}"),
+        }
+        assert!(rx.try_recv().is_err());
+    }
+
+    #[test]
+    fn snapshot_and_live_never_overlap() {
+        let sink = SessionLogSink::new();
+        sink.publish(session_start());
+        let (snapshot, mut rx) = sink.subscribe_with_snapshot();
+        sink.publish(turn_end(1));
+
+        assert_eq!(snapshot.len(), 1);
+        match rx.try_recv() {
+            Ok(LogEntry::TurnEnd { turn_count: 1, .. }) => {}
+            other => panic!("unexpected: {other:?}"),
+        }
+        assert!(rx.try_recv().is_err());
+    }
+
+    #[test]
+    fn reset_with_initial_clears_and_broadcasts() {
+        let sink = SessionLogSink::new();
+        sink.publish(session_start());
+        sink.publish(turn_end(1));
+
+        let (_pre_snapshot, mut rx) = sink.subscribe_with_snapshot();
+        sink.reset_with_initial(session_start());
+
+        match rx.try_recv() {
+            Ok(LogEntry::SessionStart { .. }) => {}
+            other => panic!("expected SessionStart broadcast, got {other:?}"),
+        }
+
+        let (post_snapshot, _) = sink.subscribe_with_snapshot();
+        assert_eq!(post_snapshot.len(), 1);
+        assert!(matches!(post_snapshot[0], LogEntry::SessionStart { .. }));
+    }
+
+    #[test]
+    fn replace_silent_does_not_broadcast() {
+        let sink = SessionLogSink::new();
+        sink.publish(session_start());
+        let (_pre_snapshot, mut rx) = sink.subscribe_with_snapshot();
+
+        sink.replace_silent(vec![session_start(), turn_end(1)]);
+
+        // No broadcast fired.
+        assert!(rx.try_recv().is_err());
+        let (post_snapshot, _) = sink.subscribe_with_snapshot();
+        assert_eq!(post_snapshot.len(), 2);
+    }
+
+    #[test]
+    fn with_initial_seeds_the_mirror() {
+        let sink = SessionLogSink::with_initial(vec![session_start(), turn_end(1)]);
+        let (snapshot, _) = sink.subscribe_with_snapshot();
+        assert_eq!(snapshot.len(), 2);
+    }
+}

crates/pod/src/shared_state.rs

@@ -1,7 +1,6 @@
 use std::sync::{OnceLock, RwLock};
 
-use llm_worker::llm_client::types::Item;
+use protocol::PodStatus;
-use protocol::{PodStatus, Segment};
 use serde_json::json;
 use session_store::SessionId;
 
@@ -19,21 +18,20 @@ pub struct KnowledgeCandidate {
 
 /// Shared state between PodController and runtime directory.
 ///
-/// Controller updates this in-memory; RuntimeDir writes it to disk.
+/// Controller updates this in-memory; RuntimeDir writes the status
-/// Wrapped in `Arc` for sharing.
+/// snapshot to disk. Wrapped in `Arc` for sharing.
+///
+/// History and typed user-segment mirrors used to live here so the
+/// IPC layer could answer `Method::GetHistory`. Those reads now go
+/// directly through the session-log sink (`Event::Snapshot` +
+/// `Event::Entry`), so this struct holds only status, identity,
+/// greeting, and completion lookup hubs.
 pub struct PodSharedState {
     pub pod_name: String,
     pub session_id: SessionId,
     pub manifest_toml: String,
     pub greeting: protocol::Greeting,
     pub status: RwLock<PodStatus>,
-    pub history: RwLock<Vec<Item>>,
-    /// Typed user submissions in submit order. The K-th entry corresponds
-    /// to the K-th `Item::user_message` in `history` (modulo seed history
-    /// loaded from a pre-compaction `SessionStart.history`, whose original
-    /// segments are not preserved). Surfaced via `Event::History` so
-    /// clients can re-render typed atoms on session restore.
-    pub user_segments: RwLock<Vec<Vec<Segment>>>,
     /// Pod-from-the-inside view of the filesystem. Set once in
     /// `PodController::start` after the `ScopedFs` is materialised, and
     /// read from the IPC server layer to answer `ListCompletions`
@@ -58,8 +56,6 @@ impl PodSharedState {
             manifest_toml,
             greeting,
             status: RwLock::new(PodStatus::Idle),
-            history: RwLock::new(Vec::new()),
-            user_segments: RwLock::new(Vec::new()),
             fs_view: OnceLock::new(),
             workflows: OnceLock::new(),
             knowledge: OnceLock::new(),
@@ -112,25 +108,6 @@ impl PodSharedState {
             .unwrap_or_default()
     }
 
-    pub fn user_segments(&self) -> Vec<Vec<Segment>> {
-        self.user_segments
-            .read()
-            .map(|s| s.clone())
-            .unwrap_or_default()
-    }
-
-    pub fn set_user_segments(&self, segments: Vec<Vec<Segment>>) {
-        if let Ok(mut s) = self.user_segments.write() {
-            *s = segments;
-        }
-    }
-
-    pub fn push_user_segments(&self, segments: Vec<Segment>) {
-        if let Ok(mut s) = self.user_segments.write() {
-            s.push(segments);
-        }
-    }
-
     pub fn set_status(&self, status: PodStatus) {
         if let Ok(mut s) = self.status.write() {
             *s = status;
@@ -141,16 +118,6 @@ impl PodSharedState {
         self.status.read().map(|s| *s).unwrap_or(PodStatus::Idle)
     }
 
-    pub fn history(&self) -> Vec<Item> {
-        self.history.read().map(|h| h.clone()).unwrap_or_default()
-    }
-
-    pub fn update_history(&self, items: Vec<Item>) {
-        if let Ok(mut h) = self.history.write() {
-            *h = items;
-        }
-    }
-
     /// Serialize status as JSON.
     pub fn status_json(&self) -> String {
         let status = self.get_status();
@@ -161,21 +128,11 @@ impl PodSharedState {
         })
         .to_string()
     }
-
-    /// Serialize history as JSON.
-    pub fn history_json(&self) -> String {
-        if let Ok(h) = self.history.read() {
-            serde_json::to_string(&*h).unwrap_or_else(|_| "[]".into())
-        } else {
-            "[]".into()
-        }
-    }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use llm_worker::llm_client::types::{ContentPart, Item, Role};
 
     fn test_state() -> PodSharedState {
         PodSharedState::new(
@@ -231,29 +188,6 @@ mod tests {
         assert_eq!(parsed["state"], "running");
     }
 
-    #[test]
-    fn history_json_empty_initially() {
-        let state = test_state();
-        assert_eq!(state.history_json(), "[]");
-    }
-
-    #[test]
-    fn history_json_after_update() {
-        let state = test_state();
-        let items = vec![Item::Message {
-            id: None,
-            role: Role::Assistant,
-            content: vec![ContentPart::Text {
-                text: "Hello".into(),
-            }],
-            status: None,
-        }];
-        state.update_history(items);
-        let json = state.history_json();
-        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
-        assert!(parsed.is_array());
-        assert_eq!(parsed[0]["role"], "assistant");
-    }
 
     #[test]
     fn knowledge_completions_empty_when_unset() {

crates/pod/src/spawn/comm_tools.rs

@@ -19,6 +19,7 @@ use llm_worker::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
 use protocol::stream::{JsonLineReader, JsonLineWriter};
 use protocol::{ErrorCode, Event, Method};
 use serde::Deserialize;
+use session_store::LogEntry;
 use tokio::net::UnixStream;
 
 use crate::runtime::dir::SpawnedPodRecord;
@@ -385,33 +386,31 @@ async fn send_run_and_confirm(socket: &Path, input: String) -> Result<(), SendRu
     }
 }
 
-/// Connect and ask the Pod for its conversation history. Skips
+/// Connect to a Pod's socket and read the connect-time `Event::Snapshot`.
-/// pre-History events (such as buffered alerts replayed to new
+///
-/// clients). Returns the raw JSON items as `serde_json::Value` since
+/// Pods deliver the session-log mirror as the first non-Alert event on
-/// the pod crate already round-trips via `Value` on the wire.
+/// every new connection, so consuming it is sufficient — no explicit
+/// `GetHistory` method round trip. Returns the entries as raw JSON
+/// values; callers deserialize as `session_store::LogEntry` if they
+/// need typed access.
 async fn fetch_history(socket: &Path) -> std::io::Result<Vec<serde_json::Value>> {
     let stream = tokio::time::timeout(SOCKET_OP_TIMEOUT, UnixStream::connect(socket))
         .await
         .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "connect timed out"))??;
-    let (r, w) = stream.into_split();
+    let (r, _w) = stream.into_split();
-    let mut writer = JsonLineWriter::new(w);
     let mut reader = JsonLineReader::new(r);
 
-    tokio::time::timeout(SOCKET_OP_TIMEOUT, writer.write(&Method::GetHistory))
-        .await
-        .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "write timed out"))??;
-
     loop {
         let event = tokio::time::timeout(SOCKET_OP_TIMEOUT, reader.next::<Event>())
             .await
             .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "read timed out"))??;
         match event {
-            Some(Event::History { items, .. }) => return Ok(items),
+            Some(Event::Snapshot { entries, .. }) => return Ok(entries),
             Some(_) => continue,
             None => {
                 return Err(std::io::Error::new(
                     std::io::ErrorKind::UnexpectedEof,
-                    "pod closed connection before History event",
+                    "pod closed connection before Snapshot event",
                 ));
             }
         }
@@ -426,24 +425,36 @@ async fn is_reachable(socket: &Path) -> bool {
         .unwrap_or(false)
 }
 
-fn extract_assistant_text(items: &[serde_json::Value]) -> String {
+fn extract_assistant_text(entries: &[serde_json::Value]) -> String {
     let mut out = String::new();
-    for value in items {
+    for value in entries {
-        let Ok(item) = serde_json::from_value::<Item>(value.clone()) else {
+        // The wire payload is the JSON form of `session_store::LogEntry`.
+        // Walk Assistant items inside each entry that can carry them:
+        // post-compaction `SessionStart.history` (seed) and per-LLM-call
+        // `AssistantItems` deltas.
+        let Ok(entry) = serde_json::from_value::<LogEntry>(value.clone()) else {
             continue;
         };
-        if let Item::Message {
+        let logged_items = match entry {
-            role: Role::Assistant,
+            LogEntry::SessionStart { history, .. } => history,
-            content,
+            LogEntry::AssistantItems { items, .. } => items,
-            ..
+            _ => continue,
-        } = item
+        };
-        {
+        for logged in logged_items {
-            for part in content {
+            let item: Item = logged.into();
-                if let ContentPart::Text { text } = part {
+            if let Item::Message {
-                    if !out.is_empty() {
+                role: Role::Assistant,
-                        out.push_str("\n\n");
+                content,
+                ..
+            } = item
+            {
+                for part in content {
+                    if let ContentPart::Text { text } = part {
+                        if !out.is_empty() {
+                            out.push_str("\n\n");
+                        }
+                        out.push_str(&text);
                     }
-                    out.push_str(&text);
                 }
             }
         }

crates/pod/tests/compact_events_test.rs

@@ -177,17 +177,39 @@ fn drain(rx: &mut broadcast::Receiver<Event>) -> Vec<Event> {
     out
 }
 
-fn system_event_text(event: &Event) -> Option<&str> {
+/// Collect every system-message text that the post-compaction
-    match event {
+/// `SessionStart.history` carries, by reading the sink mirror directly.
-        Event::SystemMessage { item } => item["content"]
+fn system_texts_in_sink_session_start(pod: &pod::Pod<impl llm_worker::llm_client::client::LlmClient + Clone + 'static, impl session_store::Store + Clone + 'static>) -> Vec<String> {
-            .as_array()
+    let (entries, _rx) = pod.sink().subscribe_with_snapshot();
-            .and_then(|parts| parts.iter().filter_map(|p| p["text"].as_str()).next()),
+    for entry in entries.into_iter().rev() {
-        _ => None,
+        if let session_store::LogEntry::SessionStart { history, .. } = entry {
+            return history
+                .into_iter()
+                .filter_map(|logged| {
+                    let item: Item = logged.into();
+                    match item {
+                        Item::Message {
+                            role: llm_worker::Role::System,
+                            content,
+                            ..
+                        } => Some(
+                            content
+                                .iter()
+                                .map(|p| p.as_text().to_owned())
+                                .collect::<Vec<_>>()
+                                .join(""),
+                        ),
+                        _ => None,
+                    }
+                })
+                .collect();
+        }
     }
+    Vec::new()
 }
 
 #[tokio::test]
-async fn compact_broadcasts_only_new_system_messages_not_retained_ones() {
+async fn compact_emits_session_start_carrying_summary_and_task_snapshot() {
     let client = MockClient::new(vec![
         single_text_events("hi"),
         write_summary_tool_use_events("call-1", "summary"),
@@ -195,18 +217,16 @@ async fn compact_broadcasts_only_new_system_messages_not_retained_ones() {
     ]);
     let mut pod = make_pod(client).await;
 
-    let (tx, mut rx) = broadcast::channel::<Event>(64);
+    let (tx, _rx_keep) = broadcast::channel::<Event>(64);
     pod.attach_event_tx(tx);
 
     pod.run_text("first").await.unwrap();
-    let retained_message = Item::system_message("[Retained system]\nold");
-    pod.worker_mut().push_item(retained_message);
-    let _ = drain(&mut rx);
-
     pod.compact(10_000).await.unwrap();
 
-    let events = drain(&mut rx);
+    let system_texts = system_texts_in_sink_session_start(&pod);
-    let system_texts: Vec<&str> = events.iter().filter_map(system_event_text).collect();
+    // The post-compaction `SessionStart.history` carries the new system
+    // messages introduced by the compactor. Clients re-seed their view
+    // from this entry alone, so it is the load-bearing payload.
     assert!(
         system_texts
             .iter()
@@ -219,12 +239,6 @@ async fn compact_broadcasts_only_new_system_messages_not_retained_ones() {
             .any(|text| text.starts_with("[Session TaskStore snapshot]")),
         "task snapshot system message missing from {system_texts:?}"
     );
-    assert!(
-        !system_texts
-            .iter()
-            .any(|text| text.starts_with("[Retained system]")),
-        "retained system message should not be rebroadcast: {system_texts:?}"
-    );
 }
 
 #[tokio::test]

crates/pod/tests/controller_test.rs

@@ -6,12 +6,40 @@ use async_trait::async_trait;
 use futures::{Stream, StreamExt};
 use llm_worker::Worker;
 use llm_worker::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
+use llm_worker::llm_client::types::Item;
 use llm_worker::llm_client::{ClientError, LlmClient, Request};
 use llm_worker::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
-use session_store::FsStore;
+use session_store::{FsStore, LogEntry};
 
 use pod::{Event, Method, Pod, PodController, PodHandle, PodManifest, PodStatus};
 
+/// Reconstruct a worker-history-like `Vec<Item>` from the live session
+/// log mirror held by the Pod's broadcast sink. Replaces the previous
+/// `PodSharedState.history()` test helper now that the mirror lives in
+/// the sink.
+fn history_from_sink(handle: &PodHandle) -> Vec<Item> {
+    let (entries, _rx) = handle.sink.subscribe_with_snapshot();
+    let mut items = Vec::new();
+    for entry in entries {
+        match entry {
+            LogEntry::SessionStart { history, .. } => {
+                items.extend(history.into_iter().map(Item::from));
+            }
+            LogEntry::UserInput { segments, .. } => {
+                let text = protocol::Segment::flatten_to_text(&segments);
+                items.push(Item::user_message(text));
+            }
+            LogEntry::AssistantItems { items: i, .. }
+            | LogEntry::ToolResults { items: i, .. }
+            | LogEntry::HookInjectedItems { items: i, .. } => {
+                items.extend(i.into_iter().map(Item::from));
+            }
+            _ => {}
+        }
+    }
+    items
+}
+
 // ---------------------------------------------------------------------------
 // Mock LLM Client
 // ---------------------------------------------------------------------------
@@ -218,8 +246,61 @@ async fn run_end_returns_to_idle_without_busy_status() {
     assert_eq!(handle.shared_state.get_status(), PodStatus::Idle);
 }
 
+/// Mid-turn re-attach: a client connecting while the worker is still
+/// running observes the in-flight `UserInput` entry in the connect-time
+/// `Event::Snapshot`. This is the load-bearing property of the new
+/// session-log-driven IPC: a late attacher reconstructs the running
+/// view without needing the prior client's diff.
 #[tokio::test]
-async fn attach_history_includes_current_status() {
+async fn snapshot_includes_user_input_for_in_flight_turn() {
+    let client = MockClient::sequential(vec![MockResponse::Hang(simple_text_events())]);
+    let pod = make_pod(client).await;
+    let handle = spawn_controller(pod).await;
+
+    handle
+        .send(Method::run_text("hello in-flight"))
+        .await
+        .unwrap();
+    wait_for_status(&handle, PodStatus::Running).await;
+
+    let stream = tokio::net::UnixStream::connect(handle.runtime_dir.socket_path())
+        .await
+        .unwrap();
+    let (reader, _writer) = stream.into_split();
+    let mut reader = protocol::stream::JsonLineReader::new(reader);
+
+    loop {
+        let event = reader.next::<Event>().await.unwrap().unwrap();
+        match event {
+            Event::Snapshot { entries, .. } => {
+                // Walk the entries, find a `LogEntry::UserInput` and
+                // confirm its segments flatten to our submitted text.
+                let mut found = false;
+                for value in entries {
+                    let entry: session_store::LogEntry =
+                        serde_json::from_value(value).expect("LogEntry deserialise");
+                    if let session_store::LogEntry::UserInput { segments, .. } = entry {
+                        let text = protocol::Segment::flatten_to_text(&segments);
+                        if text == "hello in-flight" {
+                            found = true;
+                            break;
+                        }
+                    }
+                }
+                assert!(
+                    found,
+                    "snapshot must carry the in-flight UserInput entry"
+                );
+                return;
+            }
+            Event::Alert(_) => continue,
+            other => panic!("expected Snapshot first, got {other:?}"),
+        }
+    }
+}
+
+#[tokio::test]
+async fn attach_snapshot_includes_current_status() {
     let client = MockClient::sequential(vec![MockResponse::Hang(simple_text_events())]);
     let pod = make_pod(client).await;
     let handle = spawn_controller(pod).await;
@@ -230,15 +311,20 @@ async fn attach_history_includes_current_status() {
     let stream = tokio::net::UnixStream::connect(handle.runtime_dir.socket_path())
         .await
         .unwrap();
-    let (reader, writer) = stream.into_split();
+    let (reader, _writer) = stream.into_split();
     let mut reader = protocol::stream::JsonLineReader::new(reader);
-    let mut writer = protocol::stream::JsonLineWriter::new(writer);
-    writer.write(&Method::GetHistory).await.unwrap();
 
-    let event = reader.next::<Event>().await.unwrap().unwrap();
+    // First event after connect is the snapshot — it carries the current status.
-    match event {
+    loop {
-        Event::History { status, .. } => assert_eq!(status, PodStatus::Running),
+        let event = reader.next::<Event>().await.unwrap().unwrap();
-        other => panic!("expected History, got {other:?}"),
+        match event {
+            Event::Snapshot { status, .. } => {
+                assert_eq!(status, PodStatus::Running);
+                return;
+            }
+            Event::Alert(_) => continue,
+            other => panic!("expected Snapshot, got {other:?}"),
+        }
     }
 }
 
@@ -275,11 +361,11 @@ async fn run_populates_history() {
 
     tokio::time::sleep(std::time::Duration::from_millis(100)).await;
 
-    let history = handle.shared_state.history_json();
+    let history = history_from_sink(&handle);
-    assert_ne!(history, "[]");
+    assert!(
-    let parsed: serde_json::Value = serde_json::from_str(&history).unwrap();
+        history.len() >= 2,
-    assert!(parsed.is_array());
+        "history must include user + assistant items, got {history:?}"
-    assert!(parsed.as_array().unwrap().len() >= 2); // user + assistant
+    );
 }
 
 #[tokio::test]
@@ -712,7 +798,7 @@ async fn notify_while_idle_auto_starts_turn_and_injects_system_message() {
     // The notification must also be persisted into the Worker history
     // (and therefore eventually into history.json), per
     // tickets/notify-history-persist.md.
-    let history = handle.shared_state.history();
+    let history = history_from_sink(&handle);
     let notify_in_history = history.iter().any(|i| {
         i.as_text()
             .is_some_and(|t| t.contains("[Notification]") && t.contains("turn finished"))
@@ -796,7 +882,7 @@ async fn pod_event_turn_ended_while_idle_auto_starts_turn_and_injects_system_mes
 
     // Same item must be present in worker.history (persisted lane),
     // not just the per-request clone — see tickets/notify-history-persist.md.
-    let history = handle.shared_state.history();
+    let history = history_from_sink(&handle);
     let event_in_history = history.iter().any(|i| {
         i.as_text().is_some_and(|t| {
             t.contains("[Notification]") && t.contains("child") && t.contains("finished a turn")
@@ -1149,22 +1235,42 @@ async fn pause_then_resume_transitions_and_preserves_history_consistency() {
     // History consistency: exactly [user "hello", assistant
     // "resumed output"]. No artifacts from the aborted stream
     // (partial text is not committed), no orphan tool_use.
-    let history_json = handle.shared_state.history_json();
+    let history = history_from_sink(&handle);
-    let items: Vec<serde_json::Value> = serde_json::from_str(&history_json).unwrap();
+    let roles: Vec<&str> = history
-    let roles: Vec<&str> = items.iter().filter_map(|i| i["role"].as_str()).collect();
+        .iter()
+        .filter_map(|i| match i {
+            Item::Message { role, .. } => match role {
+                llm_worker::Role::User => Some("user"),
+                llm_worker::Role::Assistant => Some("assistant"),
+                llm_worker::Role::System => Some("system"),
+            },
+            _ => None,
+        })
+        .collect();
     assert_eq!(
         roles,
         vec!["user", "assistant"],
-        "history = user + assistant only; got {items:?}"
+        "history = user + assistant only; got {history:?}"
     );
-    let assistant_text = items[1]["content"]
+    let assistant_text = history
-        .as_array()
-        .and_then(|parts| parts.iter().filter_map(|p| p["text"].as_str()).next())
-        .unwrap_or("");
-    assert_eq!(assistant_text, "resumed output");
-    let has_tool_call = items
         .iter()
-        .any(|i| i["type"].as_str() == Some("tool_call"));
+        .find_map(|i| match i {
+            Item::Message {
+                role: llm_worker::Role::Assistant,
+                content,
+                ..
+            } => Some(
+                content
+                    .iter()
+                    .map(|p: &llm_worker::ContentPart| p.as_text().to_owned())
+                    .collect::<Vec<_>>()
+                    .join(""),
+            ),
+            _ => None,
+        })
+        .unwrap_or_default();
+    assert_eq!(assistant_text, "resumed output");
+    let has_tool_call = history.iter().any(|i| i.is_tool_call());
     assert!(!has_tool_call, "no orphan tool_call in history");
 }
 

crates/pod/tests/pod_comm_tools_test.rs

@@ -144,44 +144,41 @@ fn accept_method_and_respond(
     })
 }
 
-/// Pretend to be a spawned Pod that responds to `GetHistory` with a
+/// Pretend to be a spawned Pod whose connect-time snapshot carries a
-/// fixed set of items. Accepts connections until the first one that
+/// fixed set of assistant items. Sends `Event::Snapshot` immediately on
-/// delivers a `GetHistory` method; earlier probes (empty accepts) and
+/// every accept — the real Pod does the same, so `ReadPodOutput`'s
-/// non-history methods are ignored. Returns nothing — tests await the
+/// `fetch_history` just consumes the first non-Alert event.
-/// handle only to keep the listener alive until shutdown.
 fn serve_history(listener: UnixListener, items: Vec<Item>) -> JoinHandle<()> {
     tokio::spawn(async move {
         loop {
             let Ok((stream, _)) = listener.accept().await else {
                 return;
             };
-            let (r, w) = stream.into_split();
+            let (_r, w) = stream.into_split();
-            let mut reader = JsonLineReader::new(r);
             let mut writer = JsonLineWriter::new(w);
-            match reader.next::<Method>().await {
+            // Wrap the assistant items in a single
-                Ok(Some(Method::GetHistory)) => {
+            // `LogEntry::AssistantItems` entry — that's the only kind
-                    let values: Vec<serde_json::Value> = items
+            // that contributes assistant text via `extract_assistant_text`.
-                        .iter()
+            let logged: Vec<session_store::LoggedItem> =
-                        .map(|i| serde_json::to_value(i).unwrap())
+                items.iter().map(session_store::LoggedItem::from).collect();
-                        .collect();
+            let entry = session_store::LogEntry::AssistantItems {
-                    let event = Event::History {
+                ts: 0,
-                        items: values,
+                items: logged,
-                        greeting: Greeting {
+            };
-                            pod_name: "child".into(),
+            let entry_value = serde_json::to_value(&entry).unwrap();
-                            cwd: "/tmp".into(),
+            let event = Event::Snapshot {
-                            provider: "anthropic".into(),
+                entries: vec![entry_value],
-                            model: "x".into(),
+                greeting: Greeting {
-                            scope_summary: String::new(),
+                    pod_name: "child".into(),
-                            tools: Vec::new(),
+                    cwd: "/tmp".into(),
-                        },
+                    provider: "anthropic".into(),
-                        status: protocol::PodStatus::Idle,
+                    model: "x".into(),
-                    };
+                    scope_summary: String::new(),
-                    let _ = writer.write(&event).await;
+                    tools: Vec::new(),
-                }
+                },
-                Ok(Some(_)) | Ok(None) | Err(_) => {
+                status: protocol::PodStatus::Idle,
-                    // Ignore: loop accepts another connection.
+            };
-                }
+            let _ = writer.write(&event).await;
-            }
         }
     })
 }

crates/protocol/src/lib.rs

@@ -32,14 +32,13 @@ pub enum Method {
     /// synthetic tool result before the new user message is appended).
     Pause,
     Shutdown,
-    GetHistory,
     /// Request a list of completion candidates from the Pod.
     ///
     /// Reply is sent on the same socket as `Event::Completions` (not
-    /// broadcast). Same shape as `GetHistory` / `Event::History`:
+    /// broadcast). The IPC server handles this directly and writes
-    /// the IPC server handles this directly and writes the response
+    /// the response straight back to the requesting socket. Empty
-    /// straight back to the requesting socket. Empty results for
+    /// results for resolvers that are not yet wired up
-    /// resolvers that are not yet wired up (Knowledge / Workflow).
+    /// (Knowledge / Workflow).
     ListCompletions {
         kind: CompletionKind,
         prefix: String,
@@ -224,15 +223,6 @@ pub enum Event {
     Notify {
         message: String,
     },
-    /// Persisted `role:system` history item that should be rendered by
-    /// clients through the same path used for `Event::History` replay.
-    ///
-    /// The payload is the serialized history item, not an ad-hoc display
-    /// DTO, so live subscribers and late subscribers have the same source
-    /// of truth: worker history / history.json.
-    SystemMessage {
-        item: serde_json::Value,
-    },
     /// Echo of `Method::PodEvent` received by this Pod. Same rationale
     /// as `Notify`: subscribers render the event as a log element,
     /// while a rendered summary is independently injected into the LLM
@@ -312,16 +302,34 @@ pub enum Event {
         code: ErrorCode,
         message: String,
     },
-    History {
+    /// Sent exactly once at the start of every client connection.
-        items: Vec<serde_json::Value>,
+    ///
+    /// `entries` is the session-log mirror at subscribe time, serialised
+    /// as the JSON form of `session_store::LogEntry`. Late attachers
+    /// reconstruct view state by replaying entries through their own
+    /// `LogEntry → block` mapping, then continue applying live
+    /// `Event::Entry` updates received after the snapshot.
+    ///
+    /// `greeting` and `status` accompany the snapshot so clients render
+    /// pod identity and current controller state without an extra round
+    /// trip.
+    Snapshot {
+        entries: Vec<serde_json::Value>,
         greeting: Greeting,
-        /// Current Pod controller status at the moment the history snapshot
-        /// was taken. This lets late-attaching clients render and route
-        /// controls from the real controller state instead of inferring from
-        /// replayed history.
         #[serde(default)]
         status: PodStatus,
     },
+    /// A single session-log entry committed atomically with the disk
+    /// write. Streamed as the suffix following the connect-time
+    /// `Snapshot`; the prefix/suffix boundary is gap-free and
+    /// duplicate-free per `SessionLogSink` semantics.
+    ///
+    /// Payload is the JSON form of `session_store::LogEntry`. Clients
+    /// deserialize as needed to render typed atoms (e.g.
+    /// `UserInput.segments`).
+    Entry {
+        entry: serde_json::Value,
+    },
     /// Current Pod controller status. Broadcast on every controller-level
     /// transition and included in `History` snapshots for late attach.
     Status {
@@ -418,8 +426,8 @@ pub struct CompletionEntry {
 /// Pod self-description rendered by the TUI when a session starts empty.
 ///
 /// Built once in the Pod controller from the resolved manifest and
-/// transmitted alongside `Event::History` so clients don't need their
+/// transmitted alongside `Event::Snapshot` so clients don't need
-/// own view of the manifest.
+/// their own view of the manifest.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Greeting {
     pub pod_name: String,
@@ -674,13 +682,6 @@ mod tests {
         assert_eq!(serialized, json);
     }
 
-    #[test]
-    fn method_get_history() {
-        let json = r#"{"method":"get_history"}"#;
-        let method: Method = serde_json::from_str(json).unwrap();
-        assert!(matches!(method, Method::GetHistory));
-    }
-
     #[test]
     fn method_list_completions_roundtrip() {
         let method = Method::ListCompletions {
@@ -734,9 +735,9 @@ mod tests {
     }
 
     #[test]
-    fn event_history_format() {
+    fn event_snapshot_format() {
-        let event = Event::History {
+        let event = Event::Snapshot {
-            items: vec![serde_json::json!({"type": "message", "role": "user"})],
+            entries: vec![serde_json::json!({"kind": "user_input", "ts": 1, "segments": []})],
             greeting: Greeting {
                 pod_name: "test".into(),
                 cwd: "/tmp".into(),
@@ -749,14 +750,30 @@ mod tests {
         };
         let json = serde_json::to_string(&event).unwrap();
         let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
-        assert_eq!(parsed["event"], "history");
+        assert_eq!(parsed["event"], "snapshot");
-        assert!(parsed["data"]["items"].is_array());
+        assert!(parsed["data"]["entries"].is_array());
-        assert_eq!(parsed["data"]["items"][0]["role"], "user");
+        assert_eq!(parsed["data"]["entries"][0]["kind"], "user_input");
         assert_eq!(parsed["data"]["greeting"]["pod_name"], "test");
         assert_eq!(parsed["data"]["greeting"]["tools"][0], "Read");
         assert_eq!(parsed["data"]["status"], "paused");
     }
 
+    #[test]
+    fn event_entry_roundtrip() {
+        let event = Event::Entry {
+            entry: serde_json::json!({"kind": "assistant_items", "ts": 42, "items": []}),
+        };
+        let json = serde_json::to_string(&event).unwrap();
+        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
+        assert_eq!(parsed["event"], "entry");
+        assert_eq!(parsed["data"]["entry"]["kind"], "assistant_items");
+        let decoded: Event = serde_json::from_str(&json).unwrap();
+        match decoded {
+            Event::Entry { entry } => assert_eq!(entry["kind"], "assistant_items"),
+            other => panic!("expected Entry, got {other:?}"),
+        }
+    }
+
     #[test]
     fn event_status_format() {
         let event = Event::Status {
@@ -777,12 +794,12 @@ mod tests {
     }
 
     #[test]
-    fn event_history_legacy_without_status_defaults_to_idle() {
+    fn event_snapshot_legacy_without_status_defaults_to_idle() {
-        let json = r#"{"event":"history","data":{"items":[],"greeting":{"pod_name":"test","cwd":"/tmp","provider":"anthropic","model":"claude","scope_summary":"","tools":[]}}}"#;
+        let json = r#"{"event":"snapshot","data":{"entries":[],"greeting":{"pod_name":"test","cwd":"/tmp","provider":"anthropic","model":"claude","scope_summary":"","tools":[]}}}"#;
         let decoded: Event = serde_json::from_str(json).unwrap();
         match decoded {
-            Event::History { status, .. } => assert_eq!(status, PodStatus::Idle),
+            Event::Snapshot { status, .. } => assert_eq!(status, PodStatus::Idle),
-            other => panic!("expected History, got {other:?}"),
+            other => panic!("expected Snapshot, got {other:?}"),
         }
     }
 

crates/session-store/src/lib.rs

@@ -39,10 +39,10 @@ pub use llm_worker::UsageRecord;
 pub use llm_worker::llm_client::types::{ContentPart, Item, Role};
 pub use logged_item::{LoggedContentPart, LoggedItem, LoggedRole, from_logged, to_logged};
 pub use session::{
-    SessionStartState, create_compacted_session, create_session, create_session_with_id,
+    SessionStartState, append_entry, append_entry_with_hash, create_compacted_session,
-    ensure_head_or_fork, fork, fork_at, restore, save_config_changed, save_delta, save_extension,
+    create_session, create_session_with_id, ensure_head_or_fork, fork, fork_at, restore,
-    save_pod_scope, save_run_completed, save_run_errored, save_turn_end, save_usage,
+    save_config_changed, save_delta, save_extension, save_pod_scope, save_run_completed,
-    save_user_input,
+    save_run_errored, save_turn_end, save_usage, save_user_input,
 };
 pub use session_log::{
     EntryHash, HashedEntry, LogEntry, POD_SCOPE_EXTENSION_DOMAIN, PodScopeSnapshot, RestoredState,

crates/session-store/src/session.rs

@@ -457,14 +457,31 @@ pub async fn fork_at(
     Ok(fork_id)
 }
 
-// ── Private helper ──────────────────────────────────────────────────────
+/// Append a single `LogEntry`, chaining the hash and updating `head_hash`.
-
+///
-async fn append_entry(
+/// Lower-level dual of the `save_*` convenience wrappers in this module.
+/// Use when the caller already builds the typed entry itself (e.g. when
+/// it needs the same value for an in-memory mirror + broadcast).
+pub async fn append_entry(
     store: &impl Store,
     session_id: SessionId,
     head_hash: &mut Option<EntryHash>,
     entry: LogEntry,
 ) -> Result<(), StoreError> {
+    append_entry_with_hash(store, session_id, head_hash, entry).await?;
+    Ok(())
+}
+
+/// Same as [`append_entry`] but returns the freshly computed entry hash.
+///
+/// Used by paths that need the hash for downstream broadcast or mirror
+/// updates (e.g. the Pod's `SessionLogSink`).
+pub async fn append_entry_with_hash(
+    store: &impl Store,
+    session_id: SessionId,
+    head_hash: &mut Option<EntryHash>,
+    entry: LogEntry,
+) -> Result<EntryHash, StoreError> {
     let hash = session_log::compute_hash(head_hash.as_ref(), &entry);
     let hashed_entry = HashedEntry {
         hash: hash.clone(),
@@ -472,6 +489,6 @@ async fn append_entry(
         entry,
     };
     store.append(session_id, &hashed_entry).await?;
-    *head_hash = Some(hash);
+    *head_hash = Some(hash.clone());
-    Ok(())
+    Ok(hash)
 }

crates/tui/Cargo.toml

@@ -19,6 +19,7 @@ session-store = { workspace = true }
 pod-registry = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 pulldown-cmark = { version = "0.13.3", default-features = false }
+llm-worker.workspace = true
 
 [dev-dependencies]
 tools = { workspace = true }

crates/tui/src/app.rs

@@ -491,9 +491,8 @@ impl App {
                 self.blocks.push(Block::PodEvent { event });
                 self.assistant_streaming = false;
             }
-            Event::SystemMessage { item } => {
+            Event::Entry { entry } => {
-                self.push_history_item(&item);
+                self.apply_log_entry(&entry);
-                self.assistant_streaming = false;
             }
             Event::TurnStart { .. } => {
                 self.set_pod_status(PodStatus::Running);
@@ -726,12 +725,12 @@ impl App {
                     message: alert.message,
                 });
             }
-            Event::History {
+            Event::Snapshot {
-                items,
+                entries,
                 greeting,
                 status,
             } => {
-                self.restore_history(&items, greeting);
+                self.restore_snapshot(&entries, greeting);
                 self.set_pod_status(status);
             }
             Event::Status { status } => {
@@ -905,10 +904,13 @@ impl App {
         self.input.move_down();
     }
 
-    fn restore_history(&mut self, items: &[serde_json::Value], greeting: protocol::Greeting) {
+    /// Reset the block list and replay a connect-time `Event::Snapshot`.
-        // Fresh session: greeting + any replayed items. Append-only — we
+    ///
-        // don't try to merge with already-displayed live events because
+    /// Walks the session-log entries in commit order, expanding each
-        // `History` only fires on an empty live state.
+    /// LogEntry variant into the same blocks live events would have
+    /// produced. Followed by `Event::Entry` updates for anything
+    /// committed after the snapshot.
+    fn restore_snapshot(&mut self, entries: &[serde_json::Value], greeting: protocol::Greeting) {
         self.turn_index = 0;
         self.blocks.clear();
         self.cache = FileCache::new();
@@ -917,14 +919,86 @@ impl App {
         self.blocks.push(Block::Greeting(greeting));
         self.assistant_streaming = false;
 
-        for item in items {
+        for entry in entries {
-            self.push_history_item(item);
+            self.apply_log_entry_raw(entry);
         }
 
-        // Any tool_call entries that never got paired with a
+        self.mark_orphan_tool_calls_incomplete_pass();
-        // tool_result (truncated or racing mid-turn on the server side)
+    }
-        // stay as Executing up to this point. Surface them as
+
-        // Incomplete so the replay matches live semantics.
+    /// Apply a single live `Event::Entry`.
+    ///
+    /// `SessionStart` entries that arrive live (compaction / fork)
+    /// reset the block list to a freshly seeded view, matching what a
+    /// reconnect's `Event::Snapshot` would produce.
+    fn apply_log_entry(&mut self, entry: &serde_json::Value) {
+        if entry.get("kind").and_then(|k| k.as_str()) == Some("session_start") {
+            // Compaction / fork on the server side. Reset our derived
+            // view but keep the greeting (identity hasn't changed).
+            let greeting = self
+                .blocks
+                .iter()
+                .find_map(|b| match b {
+                    Block::Greeting(g) => Some(g.clone()),
+                    _ => None,
+                });
+            self.turn_index = 0;
+            self.blocks.clear();
+            self.cache = FileCache::new();
+            self.task_store = TaskStore::new();
+            self.task_pane_scroll = 0;
+            if let Some(g) = greeting {
+                self.blocks.push(Block::Greeting(g));
+            }
+        }
+        self.apply_log_entry_raw(entry);
+        self.assistant_streaming = false;
+    }
+
+    /// Walk a single `LogEntry` JSON value and translate it into blocks
+    /// the live event path would have produced. Shared between
+    /// `restore_snapshot` (replay path) and `apply_log_entry` (live
+    /// path).
+    fn apply_log_entry_raw(&mut self, value: &serde_json::Value) {
+        let Ok(entry) = serde_json::from_value::<session_store::LogEntry>(value.clone()) else {
+            return;
+        };
+        match entry {
+            session_store::LogEntry::SessionStart { history, .. } => {
+                for logged in history {
+                    let item: llm_worker::Item = logged.into();
+                    let item_value = serde_json::to_value(&item).expect("Item is Serialize");
+                    self.push_history_item(&item_value);
+                }
+            }
+            session_store::LogEntry::UserInput { segments, .. } => {
+                self.turn_index += 1;
+                self.blocks.push(Block::TurnHeader {
+                    turn: self.turn_index,
+                });
+                if !segments.is_empty() {
+                    self.blocks.push(Block::UserMessage { segments });
+                }
+            }
+            session_store::LogEntry::AssistantItems { items, .. }
+            | session_store::LogEntry::ToolResults { items, .. }
+            | session_store::LogEntry::HookInjectedItems { items, .. } => {
+                for logged in items {
+                    let item: llm_worker::Item = logged.into();
+                    let item_value = serde_json::to_value(&item).expect("Item is Serialize");
+                    self.push_history_item(&item_value);
+                }
+            }
+            // Non-history-bearing variants don't affect the block view.
+            _ => {}
+        }
+    }
+
+    /// Sweep all current tool-call blocks: any that never resolved into
+    /// a Done / Error state get marked Incomplete. Called after a
+    /// snapshot replay so dangling in-flight tool calls in the seed
+    /// log match live semantics.
+    fn mark_orphan_tool_calls_incomplete_pass(&mut self) {
         for b in self.blocks.iter_mut() {
             if let Block::ToolCall(tc) = b
                 && matches!(
@@ -1325,18 +1399,22 @@ mod completion_flow_tests {
     }
 
     #[test]
-    fn history_restore_renders_system_message_block() {
+    fn snapshot_renders_system_message_block_from_session_start() {
         let mut app = App::new("test".into());
-        app.handle_pod_event(Event::History {
+        let session_start = session_store::LogEntry::SessionStart {
+            ts: 1,
+            system_prompt: None,
+            config: Default::default(),
+            history: vec![session_store::LoggedItem::from(
+                &llm_worker::Item::system_message("[File: src/main.rs]\nfn main() {}"),
+            )],
+            forked_from: None,
+            compacted_from: None,
+        };
+        let session_start_value = serde_json::to_value(&session_start).unwrap();
+        app.handle_pod_event(Event::Snapshot {
             greeting: test_greeting(),
-            items: vec![serde_json::json!({
+            entries: vec![session_start_value],
-                "type": "message",
-                "role": "system",
-                "content": [{
-                    "type": "text",
-                    "text": "[File: src/main.rs]\nfn main() {}",
-                }],
-            })],
             status: PodStatus::Running,
         });
 
@@ -1349,18 +1427,18 @@ mod completion_flow_tests {
     }
 
     #[test]
-    fn live_system_message_event_uses_history_item_path() {
+    fn live_entry_routes_system_message_via_hook_injected_items() {
         let mut app = App::new("test".into());
-        app.handle_pod_event(Event::SystemMessage {
+        let entry = serde_json::json!({
-            item: serde_json::json!({
+            "kind": "hook_injected_items",
-                "type": "message",
+            "ts": 1,
+            "items": [{
+                "kind": "message",
                 "role": "system",
-                "content": [{
+                "content": [{ "kind": "text", "text": "[Workflow /build]\nRun the build" }],
-                    "type": "text",
+            }],
-                    "text": "[Workflow /build]\nRun the build",
-                }],
-            }),
         });
+        app.handle_pod_event(Event::Entry { entry });
 
         assert!(matches!(
             app.blocks.as_slice(),
@@ -1504,11 +1582,15 @@ mod completion_flow_tests {
             ```json\n{\n  \"tasks\": [\n    {\n      \"taskid\": 4,\n      \
             \"status\": \"inprogress\",\n      \"subject\": \"from snapshot\",\n      \
             \"description\": \"d\"\n    }\n  ]\n}\n```\n";
-        app.handle_pod_event(Event::SystemMessage {
+        app.handle_pod_event(Event::Entry {
-            item: serde_json::json!({
+            entry: serde_json::json!({
-                "type": "message",
+                "kind": "hook_injected_items",
-                "role": "system",
+                "ts": 1,
-                "content": [{ "type": "text", "text": snapshot }],
+                "items": [{
+                    "kind": "message",
+                    "role": "system",
+                    "content": [{ "kind": "text", "text": snapshot }],
+                }],
             }),
         });
 
@@ -1519,10 +1601,10 @@ mod completion_flow_tests {
     }
 
     #[test]
-    fn history_replay_reconstructs_task_store() {
+    fn snapshot_reconstructs_task_store() {
         let mut app = App::new("test".into());
-        // Live tool call before history lands — restore_history must
+        // Live tool call before the snapshot lands — restore must wipe
-        // wipe this so it doesn't double-count after replay.
+        // this so it doesn't double-count after replay.
         app.handle_pod_event(Event::ToolCallStart {
             id: "live".into(),
             name: "TaskCreate".into(),
@@ -1533,28 +1615,33 @@ mod completion_flow_tests {
             arguments: r#"{"subject":"live","description":""}"#.into(),
         });
 
-        app.handle_pod_event(Event::History {
+        let assistant_items_entry = serde_json::json!({
-            greeting: test_greeting(),
+            "kind": "assistant_items",
-            items: vec![
+            "ts": 1,
-                serde_json::json!({
+            "items": [
-                    "type": "tool_call",
+                {
+                    "kind": "tool_call",
                     "call_id": "c1",
                     "name": "TaskCreate",
                     "arguments": r#"{"subject":"a","description":"A"}"#,
-                }),
+                },
-                serde_json::json!({
+                {
-                    "type": "tool_call",
+                    "kind": "tool_call",
                     "call_id": "c2",
                     "name": "TaskCreate",
                     "arguments": r#"{"subject":"b","description":"B"}"#,
-                }),
+                },
-                serde_json::json!({
+                {
-                    "type": "tool_call",
+                    "kind": "tool_call",
                     "call_id": "u1",
                     "name": "TaskUpdate",
                     "arguments": r#"{"taskid":2,"status":"inprogress"}"#,
-                }),
+                },
             ],
+        });
+        app.handle_pod_event(Event::Snapshot {
+            greeting: test_greeting(),
+            entries: vec![assistant_items_entry],
             status: PodStatus::Running,
         });
 

crates/tui/src/main.rs

@@ -256,9 +256,10 @@ async fn run(
     let mut app = App::new(pod_name);
 
     match PodClient::connect(socket_path).await {
-        Ok(mut client) => {
+        Ok(client) => {
             app.connected = true;
-            let _ = client.send(&Method::GetHistory).await;
+            // The Pod sends `Event::Snapshot` automatically on connect;
+            // no explicit method call is required to fetch history.
             run_loop(terminal, &mut app, client).await?;
         }
         Err(e) => {

tickets/pod-state-from-session-log.md

@@ -1,55 +0,0 @@
-# Pod: 状態と socket 配信を session log 正本に統合する
-
-## 背景
-
-Pod の状態は現在 3 つの形で同居している:
-
-1. **session log** (`crates/session-store/src/session_log.rs:LogEntry`) — append-only の typed 正本。 `UserInput { segments }` / `AssistantItems` / `ToolResults` / `HookInjectedItems` / `TurnEnd` / `SessionStart` で会話全体を表現できる。
-2. **worker.history** (`Vec<Item>`) — LLM に投げるために flatten / 加工された派生 view。 user_message は `Vec<Segment>` を flatten した String になっている。
-3. **PodSharedState** の `history` + `user_segments` — worker.history を ipc 層に渡すための中継ミラー。 typed segments は parallel 配列で別途保持。
-
-`Method::GetHistory` (`crates/pod/src/ipc/server.rs:132-182`) は (3) の中継から組み立てており、 平坦化された user_message に segments を後付けする overlay + skip-align ロジックが必要になっている。 broadcast (`event_tx`) はライブイベントだけを流し、 接続時 snapshot は別経路 + 別 Event 型 (`Event::History`) で返るため、 再アタッチ時に snapshot ↔ live が連続しない (`tickets/pod-socket-state-view.md` の問題)。
-
-派生方向が逆転している: 正本は session log なのに (3) は (2) を経由した二次派生になっており、 (1) が既に持ってる typed 情報を flatten/復元で往復する歪んだ構造を生んでいる。 また `Method::GetHistory` が RPC 形を取っていることで、 同じ socket writer に「broadcast forwarder」 と「query handler」 の 2 経路が同居している。
-
-## 方針
-
-- session log を Pod 状態の単一正本として位置付け、 worker.history は LLM context 投影用の内部 view に格下げる。 ipc 経路には worker.history が現れない。
-- 接続クライアントへの配信を 「session log の prefix (snapshot) + suffix (live)」 という同型ストリームに統合する。 query/reply 型の `Method::GetHistory` を廃止し、 接続自体が暗黙の subscribe-with-replay として動作する。
-- ストリーミング系イベント (`TextDelta` / `ToolCallStart` / `ToolCallArgsDelta` 等) は progressive 描画用の best-effort hint に役割を限定する。 late attach で過去 delta が失われるのは仕様。 確定情報は session log entry の broadcast で別途到達する。
-- entry commit の hook 点は worker 側の確定 callback に置く。 現状 `Pod::run()` 末尾で `persist_turn` が `history_before..` を一括 flush しているが (`crates/pod/src/pod.rs:1491-1502`)、 これを 「worker が assistant block / tool call / tool result を確定した瞬間に append_entry を呼ぶ」 形へ移す。 `wire_event_bridges_on_worker` で worker → event_tx を bridge しているのと同じ箇所に append_entry hook を追加する想定で、 worker 内部構造への介入は確定 callback 受け口の追加に限定する。
-- atomicity の中身は「disk write が成功した entry のみ broadcast に乗る」 順序保証。 alerter は memory-only なので buffer lock + `broadcast::send` で完結するが、 session log は disk I/O が混じるため対称ではない。 `append_entry` は (1) disk write → (2) in-memory mirror 更新 → (3) `Event::Entry` broadcast の順で、 (1) 失敗時は (2)(3) を行わず error を上に返す。 (2)(3) は同一の subscribe lock 下で行い、 `subscribe_with_snapshot` が見る mirror と receiver 側のイベント列に重複・欠落・順序逆転が出ないようにする。
-- `Event::SystemMessage` 廃止に伴い、 system_message を LogEntry に焼く責務は controller 側の `Event::SystemMessage` 送信点 (`crates/pod/src/controller.rs:372`) を `LogEntry::HookInjectedItems` の append_entry 呼び出しに置き換える形で取る。 「context に乗せる前に history に commit する」 という CLAUDE.md の加工原則に揃う。 notify 系の history 焼き込みは `tickets/notify-history-persist.md` が別途扱う領域で、 本チケットは system_message 経路の置換のみを範囲とする。
-
-## 要件
-
-- session log entry の commit は単一経路 (`Pod::append_entry` 相当) を通り、 「永続書き込み + in-memory mirror 更新 + `Event::Entry(LogEntry)` broadcast」 を atomic に行う。 atomicity は alerter と同じパターンの `subscribe_with_snapshot` 用ロックで保証される。
-- entry commit は **per-item / per-block 粒度** で行う。 現在の turn 末尾一括の `persist_turn` / `save_delta` を分解し、 mid-turn 接続で進行中の tool call / 確定済み assistant block / user input すべてが snapshot から見える状態にする。
-- 接続クライアントは接続時に `Event::Snapshot { entries: Vec<LogEntry>, greeting, status }` を受信し、 続けて live `Event::Entry(LogEntry)` を時系列で受信する。 prefix と suffix の境目に重複・欠落が無い。
-- typed user input (`Vec<Segment>`) は flatten/復元の往復なく client に届く。 `PodSharedState.user_segments` と GetHistory の overlay+skip-align ロジックを廃止する。
-- ストリーミング hint は変更なし継続。 ただし「確定情報は entry にあり、 hint は描画進捗のみ」 という分担を protocol 上のドキュメントで明記する。
-- TUI は `Event::Snapshot` / `Event::Entry` 駆動で view を組み立てる。 既存ブロック描画と等価な LogEntry → Block mapping を実装する。
-- inter-pod query (`crates/pod/src/spawn/comm_tools.rs` の GetHistory 経路) は新 snapshot 形式に追従する。
-
-## 完了条件
-
-- session log entry 1 件の commit が、 永続書き込みと `Event::Entry` broadcast を atomic に同期させる経路で行われる。 mid-turn の任意の瞬間で、 session log と Event::Entry の到達順が常に整合する。
-- 接続時に `Event::Snapshot` が必ず流れ、 直後から live `Event::Entry` が同型で連続する。 mid-turn 再アタッチで進行中の user input / 確定済み assistant 出力 / 進行中の tool call / 確定済み tool result が view に再現される。
-- `Method::GetHistory` / `Event::History` / `Event::SystemMessage` が protocol から削除されている。 後者 2 つは `Event::Entry` (`HookInjectedItems` バリアント等) で代替される。
-- `PodSharedState.history` / `PodSharedState.user_segments` が削除されている。 `PodSharedState` は status / greeting / fs_view / workflow / knowledge の lookup ハブとして残る。
-- `crates/pod/src/runtime/dir.rs` の `history.json` write は廃止または用途縮小される (session log が正本)。
-- 既存テスト (`crates/pod/tests/controller_test.rs`、 `crates/session-store/tests/`、 TUI 関連) が通る。 ターン中再アタッチで in-flight turn の user_input が view に含まれることを示すテストが新規追加されている。
-
-## 範囲外
-
-- `LogEntry` スキーマの変更 (バリアントは現状維持)。
-- compaction / fork 動作の変更 (既存の `SessionStart.{compacted_from, forked_from}` がそのまま使われる)。
-- TUI rendering の機能拡張。 LogEntry → 既存 Block の mapping は等価再構成に留め、 装飾追加は別チケット。
-- `PodSharedState` の完全廃止と Pod 借用構造の分解。 controller が `&mut Pod` を握る構造は変えない。
-- broadcast cap (256) の最適化、 ストリーミング hint の replay buffer 化。
-- `Method::ListCompletions` の subscribe 化 (これは真の query なので RPC 形のまま残す)。
-
-## 関連
-
-- `tickets/pod-persistent-state.md` の「session log は引き続き会話状態の唯一の復元ソース」方針と整合する。 Pod identity 永続化は引き続き別チケット領域。
-- `tickets/notify-history-persist.md` の「context に乗せる前に history に commit」 原則と同根。 本チケットは system_message 経路の置換まで、 notify 経路は当該チケットで扱う。