Compare commits

...

5 Commits

33 changed files with 1221 additions and 1017 deletions

1
Cargo.lock generated
View File

@ -2160,6 +2160,7 @@ dependencies = [
"manifest",
"memory",
"minijinja",
"parking_lot",
"pod-registry",
"protocol",
"provider",

View File

@ -30,6 +30,7 @@ memory = { workspace = true }
workflow-crate = { package = "workflow", path = "../workflow" }
uuid = { workspace = true, features = ["v7"] }
session-metrics = { workspace = true }
parking_lot = "0.12.5"
[dev-dependencies]
dotenv = "0.15.0"

View File

@ -48,7 +48,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
// 2. Create a persistent store (temp dir for demo)
let tmp = tempfile::tempdir()?;
let store = FsStore::new(tmp.path()).await?;
let store = FsStore::new(tmp.path())?;
// 3. Build the Pod from the single-layer manifest TOML
let mut pod = Pod::from_manifest_toml(&toml, store).await?;

View File

@ -39,7 +39,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let pwd = std::env::current_dir()?;
let toml = manifest_toml(&pwd);
let tmp = tempfile::tempdir()?;
let store = FsStore::new(tmp.path()).await?;
let store = FsStore::new(tmp.path())?;
let pod = pod::Pod::from_manifest_toml(&toml, store).await?;
let runtime_tmp = tempfile::tempdir()?;

View File

@ -6,14 +6,10 @@ use llm_worker::llm_client::client::LlmClient;
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::{LogCommand, LogDrainHandle, Pod, PodError, PodRunResult};
use crate::pod::{Pod, PodError, PodRunResult, SystemItemCommitter};
use crate::runtime::dir::RuntimeDir;
use crate::session_log_sink::SessionLogSink;
use crate::shared_state::PodSharedState;
@ -165,23 +161,21 @@ impl PodController {
}])
.map_err(std::io::Error::other)?;
// === 1.5. Per-item history-commit drain task ===
// === 1.5. Direct writer wiring ===
//
// 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());
// item / tool result that lands in history. With the sync
// writer in place, the callback commits each item directly
// through a `LogWriterHandle` (no mpsc ferry, no drain task).
// The same handle is type-erased into a `SystemItemCommitter`
// and handed to the interceptor for `SystemItem` commits, so
// assistant / tool / system items all share one commit path.
let writer_for_system: Arc<dyn SystemItemCommitter> = Arc::new(pod.log_writer_handle());
pod.attach_log_writer(writer_for_system);
pod.wire_history_persistence();
// === 2. Worker event bridge wiring ===
wire_event_bridges_on_worker(&mut pod, &event_tx, &alerter, log_cmd_tx);
wire_event_bridges_on_worker(&mut pod, &event_tx, &alerter);
// === 3. Tool registration (builtin / memory / spawn-orchestration) ===
let fs_for_view = register_pod_tools(
@ -263,29 +257,20 @@ impl PodController {
/// 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).
/// `Pod::wire_history_persistence` is called separately to wire the
/// per-item history commit callback so every assistant / tool item
/// landing in `worker.history` becomes a singular `LogEntry::AssistantItem`
/// / `ToolResult` commit through the sync writer.
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 });
@ -397,73 +382,6 @@ fn wire_event_bridges_on_worker<C, St>(
// 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,
/// and the Pod-orchestration tools (SpawnPod + comm) on the Pod's
/// Worker. Returns the `ScopedFs` clone used to attach a `PodFsView` to
@ -696,9 +614,11 @@ async fn controller_loop<C, St>(
}
Method::Notify { message } => {
let _ = event_tx.send(Event::Notify {
message: message.clone(),
});
// Client-side live echo is delivered as `Event::SystemItem`
// once the interceptor commits the corresponding
// `LogEntry::SystemItem` entry — drained out of the
// notify buffer + broadcast through the sink. No
// separate echo here.
pod.push_notify(message);
// RUNNING / Paused: the buffer push is the entire
// operation; an in-flight turn (or the next
@ -751,10 +671,12 @@ async fn controller_loop<C, St>(
Method::ListCompletions { .. } => {}
Method::PodEvent(event) => {
// Echo the received event to all subscribers so every
// client sees the input that drove any following
// auto-kicked turn.
let _ = event_tx.send(Event::PodEvent(event.clone()));
// Live echo travels through the SystemItem lane: once
// the interceptor drains the notify buffer, the
// typed `SystemItem::PodEvent` lands as a
// `LogEntry::SystemItem` entry and the sink forwards it
// to clients as `Event::SystemItem`.
//
// (1) system side effects — idempotent and tolerant of
// out-of-order delivery (e.g. `TurnEnded` arriving
// after `ShutDown`).
@ -765,11 +687,10 @@ async fn controller_loop<C, St>(
&self_parent_socket,
)
.await;
// (2) render a one-line summary and push it into the
// notification buffer; the next LLM request will
// inject it as a system message via
// `PodInterceptor::pre_llm_request`.
pod.push_notify(crate::ipc::event::render_event(&event));
// (2) queue the typed event in the notification buffer;
// the next LLM request will inject it as a typed
// `SystemItem::PodEvent` via the interceptor drain.
pod.push_pod_event_notify(event);
// Auto-kick a turn if the Pod is idle so the
// notification is not stranded. Matches the
// `Method::Notify` idle path.
@ -902,23 +823,21 @@ where
});
}
Some(Method::Notify { message }) => {
let _ = event_tx.send(Event::Notify {
message: message.clone(),
});
// Route into the buffer; the in-flight turn will
// drain it at its next pre_llm_request.
notify_buffer.push(message);
// Live echo arrives via `Event::SystemItem` once
// the in-flight turn's next `pre_llm_request`
// drains this entry through the interceptor.
notify_buffer.push_notify(message);
}
Some(Method::ListCompletions { .. }) => {}
Some(Method::PodEvent(event)) => {
let _ = event_tx.send(Event::PodEvent(event.clone()));
// mpsc is consume-once, so we cannot defer this
// to the next main-loop iteration — drop here
// would lose the event entirely (children fire
// and forget). Apply the side effects inline
// and stage the rendered string on the
// notification buffer so the in-flight turn's
// next `pre_llm_request` surfaces it.
// and stage the typed event on the notification
// buffer so the in-flight turn's next
// `pre_llm_request` surfaces it as a typed
// `SystemItem::PodEvent`.
let self_parent_socket = parent_socket.cloned();
crate::ipc::event::apply_event_side_effects(
&event,
@ -927,7 +846,7 @@ where
&self_parent_socket,
)
.await;
notify_buffer.push(crate::ipc::event::render_event(&event));
notify_buffer.push_pod_event(event);
}
None => {
let _ = cancel_tx.try_send(());

View File

@ -22,11 +22,14 @@ use tracing::info;
use tracing::warn;
use crate::compact::state::CompactState;
use session_store::SystemItem;
use crate::hook::{
AbortInfo, HookPromptAction, HookRegistry, PreRequestInfo, PromptSubmitInfo, ToolCallSummary,
ToolResultSummary, TurnEndInfo,
};
use crate::ipc::notify_buffer::{NotifyBuffer, format_notify};
use crate::ipc::notify_buffer::{NotifyBuffer, build_system_item};
use crate::pod::SystemItemCommitter;
use crate::prompt::catalog::PromptCatalog;
use llm_worker::token_counter::total_tokens;
@ -45,13 +48,21 @@ pub(crate) struct PodInterceptor {
/// request. The Worker `extend`s these into its persistent history
/// so the LLM has a visible trigger for any reaction it commits.
pending_notifies: NotifyBuffer,
/// Submit-scoped stash of resolver-produced system messages.
/// Drained inside `on_prompt_submit` and returned via
/// `PromptAction::ContinueWith`. Populated by `Pod::run` immediately
/// before handing off to the worker.
pending_attachments: Arc<Mutex<Vec<Item>>>,
/// Submit-scoped stash of resolver-produced typed system items.
/// Drained inside `on_prompt_submit`, committed as
/// `LogEntry::SystemItem` entries through `log_writer`, and
/// returned to the worker as `Item::system_message` via
/// `PromptAction::ContinueWith`. Populated by `Pod::run`
/// immediately before handing off to the worker.
pending_attachments: Arc<Mutex<Vec<SystemItem>>>,
/// Prompt catalog used to render the injected notification wrapper.
prompts: Arc<PromptCatalog>,
/// Type-erased commit handle. The interceptor uses it to commit
/// `LogEntry::SystemItem` entries directly (sync) before
/// returning the corresponding `Item::system_message`s up to the
/// worker. `None` in tests / `Pod::new` paths where no writer is
/// attached.
log_writer: Option<Arc<dyn SystemItemCommitter>>,
/// Next turn index assigned by `on_prompt_submit`.
next_turn_index: AtomicUsize,
/// Tool calls observed in the current turn (reset on each new prompt).
@ -64,8 +75,9 @@ impl PodInterceptor {
compact_state: Option<Arc<CompactState>>,
usage_history: Option<Arc<Mutex<Vec<UsageRecord>>>>,
pending_notifies: NotifyBuffer,
pending_attachments: Arc<Mutex<Vec<Item>>>,
pending_attachments: Arc<Mutex<Vec<SystemItem>>>,
prompts: Arc<PromptCatalog>,
log_writer: Option<Arc<dyn SystemItemCommitter>>,
) -> Self {
Self {
registry,
@ -74,11 +86,27 @@ impl PodInterceptor {
pending_notifies,
pending_attachments,
prompts,
log_writer,
next_turn_index: AtomicUsize::new(0),
tool_calls_this_turn: AtomicUsize::new(0),
}
}
/// Commit each `SystemItem` as its own `LogEntry::SystemItem`
/// entry through the attached writer (no-op when no writer is
/// wired). Sync — writes complete before the matching
/// `Item::system_message`s reach the worker via
/// `ContinueWith` / `pending_history_appends`, so on-disk order
/// matches worker-history order.
fn commit_system_items(&self, items: &[SystemItem]) {
let Some(writer) = self.log_writer.as_ref() else {
return;
};
for item in items {
writer.commit_system_item(item.clone());
}
}
fn current_turn_index(&self) -> usize {
self.next_turn_index
.load(Ordering::Relaxed)
@ -111,7 +139,7 @@ impl Interceptor for PodInterceptor {
return action.into();
}
}
let extras = std::mem::take(
let extras: Vec<SystemItem> = std::mem::take(
&mut *self
.pending_attachments
.lock()
@ -120,7 +148,14 @@ impl Interceptor for PodInterceptor {
if extras.is_empty() {
PromptAction::Continue
} else {
PromptAction::ContinueWith(extras)
// Commit the typed system items first, then hand the
// matching `Item::system_message`s to the worker. Sync
// commits land BEFORE the worker pushes its
// `Item::system_message`s, so on-disk order matches
// worker-history order.
let items: Vec<Item> = extras.iter().map(SystemItem::to_history_item).collect();
self.commit_system_items(&extras);
PromptAction::ContinueWith(items)
}
}
@ -129,19 +164,31 @@ impl Interceptor for PodInterceptor {
if drained.is_empty() {
return Vec::new();
}
let mut items = Vec::with_capacity(drained.len());
for n in drained {
match format_notify(&n, &self.prompts) {
Ok(item) => items.push(item),
let mut system_items: Vec<SystemItem> = Vec::with_capacity(drained.len());
let mut items: Vec<Item> = Vec::with_capacity(drained.len());
for entry in drained {
match build_system_item(&entry, &self.prompts) {
Ok(system_item) => {
items.push(system_item.to_history_item());
system_items.push(system_item);
}
Err(e) => {
// A render failure here would starve the LLM of
// the notify text. Fall back to the raw message
// so the trigger still lands in history.
// the notify text. Fall back to a raw item so the
// trigger still lands in history; the entry will
// simply be skipped from the SystemItem batch.
warn!(error = %e, "failed to render notify_wrapper; using raw message");
items.push(Item::system_message(n.message.clone()));
let fallback = match &entry {
super::notify_buffer::PendingNotify::Notify { message } => message.clone(),
super::notify_buffer::PendingNotify::PodEvent { event } => {
session_store::render_pod_event(event)
}
};
items.push(Item::system_message(fallback));
}
}
}
self.commit_system_items(&system_items);
items
}
@ -321,6 +368,7 @@ mod tests {
NotifyBuffer::new(),
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
);
let mut ctx = ctx_items;
let action = interceptor.pre_llm_request(&mut ctx).await;
@ -346,6 +394,7 @@ mod tests {
NotifyBuffer::new(),
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
);
let mut ctx = ctx_items;
let action = interceptor.pre_llm_request(&mut ctx).await;
@ -372,6 +421,7 @@ mod tests {
NotifyBuffer::new(),
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
);
let mut ctx = ctx_items;
let action = interceptor.pre_llm_request(&mut ctx).await;
@ -392,6 +442,7 @@ mod tests {
NotifyBuffer::new(),
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
);
let mut ctx: Vec<Item> = Vec::new();
let action = interceptor.pre_llm_request(&mut ctx).await;
@ -414,8 +465,8 @@ mod tests {
async fn pending_history_appends_drains_buffer_into_items() {
let registry = Arc::new(HookRegistryBuilder::new().build());
let buffer = NotifyBuffer::new();
buffer.push("first".into());
buffer.push("second".into());
buffer.push_notify("first".into());
buffer.push_notify("second".into());
let interceptor = PodInterceptor::new(
registry,
@ -424,6 +475,7 @@ mod tests {
buffer.clone(),
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
);
let items = interceptor.pending_history_appends().await;
@ -451,7 +503,7 @@ mod tests {
// anything itself.
let registry = Arc::new(HookRegistryBuilder::new().build());
let buffer = NotifyBuffer::new();
buffer.push("msg".into());
buffer.push_notify("msg".into());
let interceptor = PodInterceptor::new(
registry,
@ -460,6 +512,7 @@ mod tests {
buffer.clone(),
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
);
let mut ctx: Vec<Item> = vec![Item::user_message("hi")];
let action = interceptor.pre_llm_request(&mut ctx).await;
@ -489,6 +542,7 @@ mod tests {
NotifyBuffer::new(),
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
);
let mut ctx: Vec<Item> = Vec::new();
let action = interceptor.pre_llm_request(&mut ctx).await;

View File

@ -3,39 +3,48 @@
//! Entries are queued here by the Controller (on receipt of the
//! corresponding IPC method) and drained by
//! `PodInterceptor::pending_history_appends`, which the Worker calls
//! at the head of each turn loop iteration to `extend` them into the
//! persistent `worker.history`. Each queued entry becomes one
//! `Item::system_message`.
//! at the head of each turn loop iteration. The drain renders each
//! pending entry into a typed `SystemItem` (with the `notify_wrapper`
//! prompt applied), commits a `LogEntry::SystemItem` per entry through
//! the session-log sink, and returns the corresponding
//! `Item::system_message`s for the worker to append to its
//! persistent history.
//!
//! This is the **single lane** for "system messages produced by Pod
//! state that should land in the next LLM request": Notify, PodEvent,
//! and any future `<system-reminder>` injection all ride this queue
//! (or a sibling queue with the same lifecycle). Per
//! `tickets/notify-history-persist.md` and `AGENTS.md` (LLM コンテキスト
//! の加工原則), there is **no** "transient, history-skipping" lane —
//! everything injected into a request is also committed to history so
//! that any LLM reaction has a visible trigger across turns, resume,
//! and compaction, and so the Anthropic prompt cache prefix stays
//! stable across requests.
//! and any future `<system-reminder>` injection all ride this queue.
//! Per `tickets/notify-history-persist.md` and `AGENTS.md` (LLM
//! context の加工原則), there is **no** "transient, history-skipping"
//! lane — everything injected into a request is also committed to
//! history so any LLM reaction has a visible trigger across turns,
//! resume, and compaction, and so the Anthropic prompt cache prefix
//! stays stable across requests.
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use llm_worker::Item;
use protocol::PodEvent;
use session_store::SystemItem;
use tracing::warn;
use crate::prompt::catalog::{CatalogError, PromptCatalog};
/// Maximum queued notify entries. Oldest entries are dropped beyond this.
/// Maximum queued pending entries. Oldest entries are dropped beyond this.
const CAPACITY: usize = 128;
/// One pending notify entry awaiting injection into the next LLM request.
/// One pending entry awaiting drain into the next LLM request.
///
/// The buffer keeps the raw input shape so the drain step can decide
/// the right `SystemItem` kind (and apply `notify_wrapper` to the
/// rendered body) at the moment of commit, when the prompt catalog
/// is available.
#[derive(Debug, Clone)]
pub struct PendingNotify {
pub message: String,
pub enum PendingNotify {
Notify { message: String },
PodEvent { event: PodEvent },
}
/// Shared, mutex-guarded buffer of pending notify entries.
/// Shared, mutex-guarded buffer of pending entries.
///
/// Cloned between the Pod (producer) and PodInterceptor (consumer).
#[derive(Clone, Default)]
@ -51,26 +60,35 @@ impl NotifyBuffer {
/// Push a notify entry onto the queue. If the queue is full, the
/// oldest entry is dropped and a `tracing::warn` is emitted — the
/// caller should never hit this in normal operation.
pub fn push(&self, message: String) {
pub fn push_notify(&self, message: String) {
self.push_entry(PendingNotify::Notify { message });
}
/// Push a typed pod-event entry onto the queue.
pub fn push_pod_event(&self, event: PodEvent) {
self.push_entry(PendingNotify::PodEvent { event });
}
fn push_entry(&self, entry: PendingNotify) {
let mut q = self.inner.lock().expect("notify buffer poisoned");
if q.len() >= CAPACITY {
let dropped = q.pop_front();
warn!(
capacity = CAPACITY,
dropped_message = dropped.as_ref().map(|n| n.message.as_str()),
dropped = ?dropped,
"notify buffer overflow; dropped oldest"
);
}
q.push_back(PendingNotify { message });
q.push_back(entry);
}
/// Remove and return all pending notify entries in FIFO order.
/// Remove and return all pending entries in FIFO order.
pub fn drain(&self) -> Vec<PendingNotify> {
let mut q = self.inner.lock().expect("notify buffer poisoned");
q.drain(..).collect()
}
/// Number of pending notify entries. Primarily for tests.
/// Number of pending entries. Primarily for tests.
pub fn len(&self) -> usize {
self.inner.lock().expect("notify buffer poisoned").len()
}
@ -80,17 +98,30 @@ impl NotifyBuffer {
}
}
/// Format a single pending notify entry into the `Item::system_message`
/// that gets appended to `worker.history` just before the next LLM
/// request. The wrapper body comes from `PodPrompt::NotifyWrapper` so
/// the surrounding phrasing can be customised via a prompt pack
/// (translation, tone, ...).
pub(crate) fn format_notify(
n: &PendingNotify,
/// Render one pending entry into a typed `SystemItem`. The
/// `notify_wrapper` prompt produces the LLM-context body for both
/// `Notify` (raw message) and `PodEvent` (rendered event line).
pub(crate) fn build_system_item(
entry: &PendingNotify,
prompts: &PromptCatalog,
) -> Result<Item, CatalogError> {
let text = prompts.notify_wrapper(&n.message)?;
Ok(Item::system_message(text))
) -> Result<SystemItem, CatalogError> {
match entry {
PendingNotify::Notify { message } => {
let body = prompts.notify_wrapper(message)?;
Ok(SystemItem::Notification {
message: message.clone(),
body,
})
}
PendingNotify::PodEvent { event } => {
let rendered = session_store::render_pod_event(event);
let body = prompts.notify_wrapper(&rendered)?;
Ok(SystemItem::PodEvent {
event: event.clone(),
body,
})
}
}
}
#[cfg(test)]
@ -100,12 +131,14 @@ mod tests {
#[test]
fn push_then_drain_preserves_order() {
let buf = NotifyBuffer::new();
buf.push("one".into());
buf.push("two".into());
buf.push_notify("one".into());
buf.push_notify("two".into());
let drained = buf.drain();
assert_eq!(drained.len(), 2);
assert_eq!(drained[0].message, "one");
assert_eq!(drained[1].message, "two");
match &drained[0] {
PendingNotify::Notify { message } => assert_eq!(message, "one"),
other => panic!("unexpected: {other:?}"),
}
assert!(buf.is_empty());
}
@ -113,28 +146,50 @@ mod tests {
fn capacity_drops_oldest() {
let buf = NotifyBuffer::new();
for i in 0..(CAPACITY + 5) {
buf.push(format!("msg{i}"));
buf.push_notify(format!("msg{i}"));
}
let drained = buf.drain();
assert_eq!(drained.len(), CAPACITY);
// Oldest 5 were dropped; first retained is msg5.
assert_eq!(drained[0].message, "msg5");
assert_eq!(
drained[CAPACITY - 1].message,
format!("msg{}", CAPACITY + 4)
);
match &drained[0] {
PendingNotify::Notify { message } => assert_eq!(message, "msg5"),
other => panic!("unexpected: {other:?}"),
}
}
#[test]
fn format_notify_includes_message_and_nonblocking_hint() {
let n = PendingNotify {
fn build_system_item_for_notify_carries_wrapper_body() {
let entry = PendingNotify::Notify {
message: "hello".into(),
};
let catalog = PromptCatalog::builtins_only().unwrap();
let item = format_notify(&n, &catalog).unwrap();
let text = item.as_text().unwrap_or_default().to_string();
assert!(text.contains("[Notification]"));
assert!(text.contains("hello"));
assert!(text.contains("not a blocking request"));
let item = build_system_item(&entry, &catalog).unwrap();
match item {
SystemItem::Notification { message, body } => {
assert_eq!(message, "hello");
assert!(body.contains("[Notification]"));
assert!(body.contains("hello"));
assert!(body.contains("not a blocking request"));
}
other => panic!("unexpected: {other:?}"),
}
}
#[test]
fn build_system_item_for_pod_event_wraps_rendered_event_text() {
let entry = PendingNotify::PodEvent {
event: PodEvent::TurnEnded {
pod_name: "child".into(),
},
};
let catalog = PromptCatalog::builtins_only().unwrap();
let item = build_system_item(&entry, &catalog).unwrap();
match item {
SystemItem::PodEvent { event, body } => {
assert!(matches!(event, PodEvent::TurnEnded { ref pod_name } if pod_name == "child"));
assert!(body.contains("[Notification]"));
assert!(body.contains("`child`"));
}
other => panic!("unexpected: {other:?}"),
}
}
}

View File

@ -104,25 +104,38 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: PodHandle) {
entry = entry_rx.recv() => {
match entry {
Ok(entry) => {
let outbound = match entry {
session_store::LogEntry::SessionStart { .. } => {
let value = serde_json::to_value(&entry)
.expect("LogEntry is Serialize");
let outbound = match &entry {
session_store::LogEntry::SessionStart { .. } => {
Some(Event::SessionRotated { entry: value })
}
session_store::LogEntry::HookInjectedItems { .. } => {
Some(Event::HookInjectedItems { entry: value })
session_store::LogEntry::SystemItem { item, .. } => {
let value = serde_json::to_value(&item)
.expect("SystemItem is Serialize");
Some(Event::SystemItem { item: value })
}
other => {
// `SessionLogSink::is_live_relevant` keeps
// non-live-relevant variants off the
// broadcast lane; reaching here means the
// two are out of sync and we silently
// dropped a wire event. Log so a future
// regression surfaces instead of vanishing.
tracing::error!(
entry_kind = ?std::mem::discriminant(&other),
"session-log broadcast emitted a non-live-relevant entry; \
sink filter and IPC dispatch are out of sync"
);
None
}
// Defensive: should never reach here per
// `SessionLogSink::is_live_relevant`.
_ => None,
};
if let Some(event) = outbound
&& writer.write(&event).await.is_err()
{
if let Some(event) = outbound {
if writer.write(&event).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

View File

@ -31,5 +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 session_log_sink::SessionLogSink;
pub use shared_state::PodSharedState;

View File

@ -162,7 +162,7 @@ async fn main() -> ExitCode {
}
},
};
let store = match FsStore::new(&store_dir).await {
let store = match FsStore::new(&store_dir) {
Ok(s) => s,
Err(e) => {
eprintln!("error: failed to initialize store at {store_dir:?}: {e}");

View File

@ -1,46 +1,21 @@
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use tokio::sync::Mutex as AsyncMutex;
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::{ToolOutputLimits, UsageRecord, Worker, WorkerError, WorkerResult};
use parking_lot::Mutex as SyncMutex;
use session_store::{
EntryHash, HashedEntry, LogEntry, PodScopeSnapshot, SessionId, Store, StoreError, session_log,
to_logged,
EntryHash, HashedEntry, LogEntry, PodScopeSnapshot, SessionId, Store, StoreError, SystemItem,
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,
@ -73,6 +48,61 @@ pub struct SessionHead {
pub head_hash: Option<EntryHash>,
}
/// Cheap-cloneable bundle of (store + session-head lock + sink) handed
/// to the worker callback and the interceptor so they can commit
/// `LogEntry` values directly without going through an mpsc ferry.
///
/// All three fields are `Clone` (the latter two as `Arc` clones, the
/// store per its `Clone` impl) so the handle itself is a flat triple of
/// cheap copies.
#[derive(Clone)]
pub struct LogWriterHandle<St: Clone> {
pub store: St,
pub session_head: Arc<SyncMutex<SessionHead>>,
pub sink: SessionLogSink,
}
impl<St> LogWriterHandle<St>
where
St: Store + Clone,
{
/// Append `entry` to the log: disk write → in-memory mirror push →
/// broadcast — atomic w.r.t. `subscribe_with_snapshot` callers.
pub fn append_entry(&self, entry: LogEntry) -> Result<EntryHash, StoreError> {
let mut head = self.session_head.lock();
let hash = session_store::append_entry_with_hash(
&self.store,
head.session_id,
&mut head.head_hash,
entry.clone(),
)?;
self.sink.publish(entry);
Ok(hash)
}
}
/// Type-erased commit handle for the interceptor. Lets the
/// interceptor commit `SystemItem`s without being generic over the
/// concrete `Store` type.
pub trait SystemItemCommitter: Send + Sync {
fn commit_system_item(&self, item: SystemItem);
}
impl<St> SystemItemCommitter for LogWriterHandle<St>
where
St: Store + Clone + Send + Sync + 'static,
{
fn commit_system_item(&self, item: SystemItem) {
let entry = LogEntry::SystemItem {
ts: session_log::now_millis(),
item,
};
if let Err(err) = self.append_entry(entry) {
warn!(error = %err, "system item commit failed; dropping");
}
}
}
/// Pre-LLM-request hook that records `history.len()` at send time into a
/// shared `UsageTracker`. The on_usage callback later pairs this with the
/// aggregated UsageEvent to produce one `UsageRecord` per LLM call.
@ -98,7 +128,7 @@ pub struct Pod<C: LlmClient, St: Store> {
worker: Option<Worker<C, Mutable>>,
store: St,
session_id: SessionId,
session_head: Arc<AsyncMutex<SessionHead>>,
session_head: Arc<SyncMutex<SessionHead>>,
/// Absolute working directory of the Pod.
pwd: PathBuf,
/// Shared, atomically-swappable view of the Pod's resolved scope.
@ -158,7 +188,7 @@ pub struct Pod<C: LlmClient, St: Store> {
/// before handing off to the worker; `PodInterceptor::on_prompt_submit`
/// drains it and returns `ContinueWith` so the items land in
/// history right after the user message that referenced them.
pending_attachments: Arc<Mutex<Vec<Item>>>,
pending_attachments: Arc<Mutex<Vec<SystemItem>>>,
/// Scope allocation in the machine-wide lock file. `Some` for
/// Pods built via `from_manifest` / `from_manifest_spawned` /
/// `restore_from_manifest` (production paths); `None` for the
@ -230,12 +260,21 @@ pub struct Pod<C: LlmClient, St: Store> {
/// 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>>,
/// `true` once `wire_history_persistence` has installed the
/// `Worker::on_history_append` callback that commits each appended
/// item as a singular `LogEntry::AssistantItem` / `ToolResult`
/// directly through the writer. Tests that drive `Pod::new` without
/// going through the controller leave this `false`; `persist_turn`
/// then walks the post-`history_before` slice inline so entries
/// still land on disk.
history_persistence_wired: bool,
/// Type-erased commit handle wired by the controller (or by tests
/// via `attach_log_writer`). The interceptor uses it to commit
/// `SystemItem`s directly without being generic over `St`. `None`
/// in low-level test paths that bypass the controller — those
/// paths skip SystemItem disk commits but still see the rendered
/// `Item::system_message` in worker history.
log_writer: Option<Arc<dyn SystemItemCommitter>>,
}
impl<C: LlmClient + 'static, St: Store + 'static> Pod<C, St> {
@ -279,7 +318,7 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Pod<C, St> {
alerter: self.alerter.clone(),
event_tx: self.event_tx.clone(),
pending_notifies: NotifyBuffer::new(),
pending_attachments: Arc::new(Mutex::new(Vec::new())),
pending_attachments: Arc::new(Mutex::new(Vec::<SystemItem>::new())),
scope_allocation: None,
callback_socket: None,
prompts: self.prompts.clone(),
@ -296,21 +335,66 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Pod<C, St> {
// (it only reads `worker.history()`), so a fresh sink is
// fine — nothing observes its broadcast.
sink: SessionLogSink::new(),
log_cmd_tx: None,
history_persistence_wired: false,
log_writer: 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 {
/// Build a `LogWriterHandle` carrying everything the worker
/// callback / interceptor needs to commit `LogEntry` values
/// directly: store handle, the shared session-head lock, and the
/// broadcast sink. All three are cheap clones.
pub fn log_writer_handle(&self) -> LogWriterHandle<St> {
LogWriterHandle {
store: self.store.clone(),
session_head: self.session_head.clone(),
sink: self.sink.clone(),
}
}
/// Attach a type-erased system-item commit handle. The controller
/// calls this once during spawn so the interceptor can commit
/// `SystemItem`s directly without owning a generic store handle.
/// Idempotent: subsequent calls overwrite the previous handle.
pub fn attach_log_writer(&mut self, writer: Arc<dyn SystemItemCommitter>) {
self.log_writer = Some(writer);
}
/// Wire `Worker::on_history_append` to commit each appended item
/// directly as a singular `LogEntry::AssistantItem` / `ToolResult`
/// through the writer. The controller calls this once per spawned
/// Pod after the worker is built; tests that drive `Pod::new` may
/// opt in to the same wiring or leave it off (in which case
/// `persist_turn`'s inline fallback writes entries at turn end).
///
/// `user_message` items are skipped because they are committed
/// up-front via `commit_entry(LogEntry::UserInput { segments })`.
/// `role:system` items are committed by `PodInterceptor` as typed
/// `LogEntry::SystemItem` entries before they reach the worker's
/// history (so this callback would otherwise double-write them).
pub fn wire_history_persistence(&mut self) {
let writer = self.log_writer_handle();
self.worker_mut().on_history_append(move |item| {
if item.is_user_message() {
return;
}
if matches!(
item,
Item::Message {
role: llm_worker::Role::System,
..
}
) {
return;
}
let entry = session_store::classify_history_item(item, session_log::now_millis());
if let Err(err) = writer.append_entry(entry) {
warn!(error = %err, "history append commit failed; dropping");
}
});
self.history_persistence_wired = true;
}
pub fn spawn_post_run_memory_jobs(&mut self) {
// Drop a finished prior handle so we can spawn a fresh task.
// If the prior task is still running, coalesce by skipping —
@ -360,7 +444,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
worker: Some(worker),
store,
session_id,
session_head: Arc::new(AsyncMutex::new(SessionHead {
session_head: Arc::new(SyncMutex::new(SessionHead {
session_id,
head_hash: None,
})),
@ -378,7 +462,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
alerter: None,
event_tx: None,
pending_notifies: NotifyBuffer::new(),
pending_attachments: Arc::new(Mutex::new(Vec::new())),
pending_attachments: Arc::new(Mutex::new(Vec::<SystemItem>::new())),
scope_allocation: None,
callback_socket: None,
prompts,
@ -392,7 +476,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
memory_task: None,
user_segments: Vec::new(),
sink: SessionLogSink::new(),
log_cmd_tx: None,
history_persistence_wired: false,
log_writer: None,
};
pod.apply_permissions_from_manifest();
pod.apply_prune_from_manifest();
@ -486,8 +571,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// process later exits while children keep their allocations, resume
/// can restore the narrowed scope instead of reclaiming delegated
/// writes.
pub async fn persist_scope_snapshot(&mut self) -> Result<(), StoreError> {
if self.session_head.lock().await.head_hash.is_none() {
pub fn persist_scope_snapshot(&mut self) -> Result<(), StoreError> {
if self.session_head.lock().head_hash.is_none() {
return Ok(());
}
let snapshot = {
@ -503,23 +588,21 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
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
/// broadcast sink. Holds the session-head sync 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;
pub(crate) fn commit_entry(&self, entry: LogEntry) -> Result<EntryHash, StoreError> {
let mut head = self.session_head.lock();
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)
}
@ -531,15 +614,6 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
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
/// directly to the async store from a sync tool callback, so it records
/// the latest snapshot and the controller flushes it after the tool
@ -551,7 +625,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
})
}
async fn flush_pending_scope_snapshot(&mut self) -> Result<(), StoreError> {
fn flush_pending_scope_snapshot(&mut self) -> Result<(), StoreError> {
let snapshot = self
.pending_scope_snapshot
.lock()
@ -563,8 +637,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
ts: session_log::now_millis(),
domain: session_store::POD_SCOPE_EXTENSION_DOMAIN.into(),
payload,
})
.await?;
})?;
}
Ok(())
}
@ -726,14 +799,14 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// fail the surrounding turn. On failure the head hash stays put
/// (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) {
fn try_record_metric(&mut self, metric: &session_metrics::Metric) {
let payload = serde_json::to_value(metric).expect("Metric is Serialize");
let entry = LogEntry::Extension {
ts: session_log::now_millis(),
domain: session_metrics::DOMAIN.into(),
payload,
};
if let Err(err) = self.commit_entry(entry).await {
if let Err(err) = self.commit_entry(entry) {
warn!(name = %metric.name, error = %err, "failed to record session metric; dropping");
self.alert(
AlertLevel::Warn,
@ -760,7 +833,17 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// `PodInterceptor::pending_history_appends`. See [`NotifyBuffer`]
/// for overflow behaviour and the lane-of-record rationale.
pub fn push_notify(&self, message: String) {
self.pending_notifies.push(message);
self.pending_notifies.push_notify(message);
}
/// Push a typed `PodEvent` entry onto the pending buffer.
///
/// Same lifecycle as [`push_notify`](Self::push_notify) but
/// preserves the typed `PodEvent` payload so the IPC layer can
/// emit `SystemItem::PodEvent { event, body }` with structured
/// data for clients.
pub fn push_pod_event_notify(&self, event: protocol::PodEvent) {
self.pending_notifies.push_pod_event(event);
}
/// Shared handle to the pending notification buffer.
@ -892,6 +975,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
self.pending_notifies.clone(),
self.pending_attachments.clone(),
self.prompts.clone(),
self.log_writer.clone(),
);
self.worker_mut().set_interceptor(interceptor);
self.interceptor_installed = true;
@ -1032,7 +1116,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
self.ensure_interceptor_installed();
self.ensure_system_prompt_materialized()?;
self.cleanup_finished_memory_task();
self.ensure_session_head().await?;
self.ensure_session_head()?;
if self.should_pre_run_compact() {
self.join_memory_task().await;
}
@ -1057,12 +1141,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;
self.session_id = self.session_head.lock().session_id;
self.commit_entry(LogEntry::UserInput {
ts: session_log::now_millis(),
segments: input.clone(),
})
.await?;
?;
self.user_segments.push(input.clone());
// Resolve `@<path>` refs, `#<slug>` Knowledge refs, and `/<slug>`
@ -1099,7 +1183,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// directory) surface as `AlertLevel::Warn` Alerts and are skipped — the
/// unresolved placeholder stays in the flattened user message so the LLM
/// still sees the intent.
fn resolve_file_refs(&self, segments: &[Segment]) -> Vec<Item> {
fn resolve_file_refs(&self, segments: &[Segment]) -> Vec<SystemItem> {
let view = crate::fs_view::PodFsView::new(tools::ScopedFs::with_shared_scope(
self.scope.clone(),
self.pwd.clone(),
@ -1110,7 +1194,19 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
continue;
};
match view.resolve_file_ref(path, self.manifest.worker.file_upload.max_bytes) {
Ok(item) => out.push(item),
Ok(item) => {
// `resolve_file_ref` returns an `Item::system_message`
// whose text already carries the `[File: <path>]` or
// `[Dir: <path>]` header (plus any truncation hint).
// Persist that body verbatim — it is what the LLM
// actually saw, so resume produces byte-identical
// history.
let body = item.as_text().unwrap_or_default().to_string();
out.push(SystemItem::FileAttachment {
path: path.clone(),
body,
});
}
Err(e) => {
self.alert(
AlertLevel::Warn,
@ -1123,7 +1219,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
out
}
fn resolve_knowledge_refs(&self, segments: &[Segment]) -> Vec<Item> {
fn resolve_knowledge_refs(&self, segments: &[Segment]) -> Vec<SystemItem> {
let Some(layout) = self.memory_layout.as_ref() else {
return Vec::new();
};
@ -1156,7 +1252,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
}
};
let raw = String::from_utf8_lossy(&bytes).into_owned();
let body = match memory::schema::split_frontmatter(&raw) {
let body_text = match memory::schema::split_frontmatter(&raw) {
Ok((_yaml, body)) => body,
Err(e) => {
self.alert(
@ -1173,11 +1269,11 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
&bytes,
);
self.append_memory_use_event(memory::UsageSource::KnowledgeRef, vec![snapshot]);
out.push(Item::system_message(format!(
"[Knowledge #{}]\n{}",
slug,
body.trim_end()
)));
let body = format!("[Knowledge #{}]\n{}", slug, body_text.trim_end());
out.push(SystemItem::Knowledge {
slug: slug.clone(),
body,
});
}
out
}
@ -1247,7 +1343,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
fn resolve_workflow_invocations(
&self,
segments: &[Segment],
) -> Result<Vec<Item>, WorkflowResolveError> {
) -> Result<Vec<SystemItem>, WorkflowResolveError> {
let Some(layout) = self.memory_layout.as_ref() else {
if let Some(slug) = segments.iter().find_map(|seg| match seg {
Segment::WorkflowInvoke { slug } => Some(slug.clone()),
@ -1282,7 +1378,17 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
warn!(workflow = %slug, error = %err, "failed to snapshot workflow usage");
}
}
out.extend(items);
// `resolve_workflow_invocation` returns Item::system_message
// entries (potentially multiple — body + dependency knowledge
// bodies). Persist each as a SystemItem::Workflow keyed on
// the invocation slug.
for item in items {
let body = item.as_text().unwrap_or_default().to_string();
out.push(SystemItem::Workflow {
slug: slug.clone(),
body,
});
}
}
Ok(out)
}
@ -1405,11 +1511,11 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// `ensure_system_prompt_materialized` has just rendered. Subsequent
/// calls fall through to `ensure_head_or_fork`, which auto-forks when
/// another writer has advanced the store head behind our back.
async fn ensure_session_head(&mut self) -> Result<(), PodError> {
fn ensure_session_head(&mut self) -> Result<(), PodError> {
let w = self.worker.as_ref().unwrap();
let prev_session_id;
let initial_state = {
let head = self.session_head.lock().await;
let head = self.session_head.lock();
prev_session_id = head.session_id;
head.head_hash.is_none()
};
@ -1422,17 +1528,16 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
forked_from: None,
compacted_from: None,
};
self.commit_entry(initial).await?;
self.persist_scope_snapshot().await?;
self.commit_entry(initial)?;
self.persist_scope_snapshot()?;
return Ok(());
}
// Check store head + auto-fork if it drifted.
let store_head = self
.store
.read_head_hash(prev_session_id)
.await
.map_err(PodError::from)?;
let mut head = self.session_head.lock().await;
let mut head = self.session_head.lock();
if store_head == head.head_hash {
return Ok(());
}
@ -1456,7 +1561,6 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
};
self.store
.create_session(fork_id, &[hashed])
.await
.map_err(PodError::from)?;
head.session_id = fork_id;
head.head_hash = Some(hash);
@ -1600,83 +1704,52 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
history_before: usize,
result: &Result<WorkerResult, WorkerError>,
) -> Result<(), StoreError> {
// Per-item commits for AssistantItems / ToolResults /
// HookInjectedItems already landed mid-turn through the
// controller-spawned drain task, fed by
// `Worker::on_history_append`. Drain the queue here so every
// in-flight item has actually been committed before the
// trailing `TurnEnd` entry. When no drain is wired (low-level
// tests / direct `Pod::new` usage) we fall back to a synchronous
// 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.
// Per-item commits for AssistantItem / ToolResult / SystemItem
// entries are expected to have landed synchronously: the
// worker `on_history_append` callback (wired by the controller
// via `wire_history_persistence`) commits each appended item
// directly through the writer, and the interceptor commits
// SystemItem entries up-front in `on_prompt_submit` /
// `pending_history_appends` before returning the matching
// `Item::system_message`s.
//
// Low-level test paths that build `Pod::new` without wiring
// the callback fall through this branch: they classify the
// slice from `history_before` inline so the test's
// `restore`-style assertions still see entries on disk.
self.session_id = self.session_head.lock().session_id;
if !self.history_persistence_wired {
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];
for item in &new_items {
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;
continue;
}
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;
if matches!(
item,
Item::Message {
role: llm_worker::Role::System,
..
}
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;
) {
continue;
}
let entry = session_store::classify_history_item(item, ts);
self.commit_entry(entry)?;
}
}
self.flush_pending_scope_snapshot().await?;
self.flush_pending_scope_snapshot()?;
let turn_count = self.worker.as_ref().unwrap().turn_count();
self.commit_entry(LogEntry::TurnEnd {
ts: session_log::now_millis(),
turn_count,
})
.await?;
?;
// Flush any sync-buffered metrics from this run first
// (currently `prune.fire` / `prune.skip` from the prune observer).
@ -1692,7 +1765,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// by this point, and `save_run_completed` still needs to land).
let pending_metrics = self.metrics_tracker.drain();
for metric in pending_metrics {
self.try_record_metric(&metric).await;
self.try_record_metric(&metric);
}
// Persist any LLM Usage measurements collected during this run.
@ -1717,14 +1790,14 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
cache_write_tokens: record.cache_write_tokens,
output_tokens: record.output_tokens,
})
.await?;
?;
if let Some(id) = correlation_id {
let metric = session_metrics::Metric::now("prune.post_request")
.with_correlation_id(&id)
.with_value(record.cache_read_tokens as f64)
.with_dimension("cache_write_tokens", record.cache_write_tokens.to_string())
.with_dimension("history_len", record.history_len.to_string());
self.try_record_metric(&metric).await;
self.try_record_metric(&metric);
}
self.usage_history
.lock()
@ -1740,7 +1813,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
interrupted,
result: r.clone(),
})
.await?;
?;
}
Err(e) => {
self.commit_entry(LogEntry::RunErrored {
@ -1748,7 +1821,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
interrupted,
message: e.to_string(),
})
.await?;
?;
}
}
@ -1982,7 +2055,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// `SessionStart { compacted_from }` and reset their view.
let new_session_id = session_store::new_session_id();
let session_start = {
let mut head = self.session_head.lock().await;
let mut head = self.session_head.lock();
let old_session_id = head.session_id;
let old_head_hash = head
.head_hash
@ -2006,7 +2079,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
prev_hash: None,
entry: entry.clone(),
};
self.store.create_session(new_session_id, &[hashed]).await?;
self.store.create_session(new_session_id, &[hashed])?;
head.session_id = new_session_id;
head.head_hash = Some(hash);
self.session_id = new_session_id;
@ -2054,7 +2127,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
.lock()
.expect("usage_history poisoned")
.clear();
self.persist_scope_snapshot().await?;
self.persist_scope_snapshot()?;
// Reset extract pointer alongside usage_history: the compacted
// session has a fresh log with no `LogEntry::Extension` entries
// yet, so a cold restore here would set extract_pointer to None
@ -2216,7 +2289,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// Read the session log to get the current entry count. This is
// the boundary for the source.range end_entry. Called once per
// extract, on a small local file.
let entries_now = self.store.read_all(self.session_id).await?.len();
let entries_now = self.store.read_all(self.session_id)?.len();
if entries_now == 0 {
return Ok(ExtractDecision::Skipped);
}
@ -2284,7 +2357,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
extract::ExtractedPayload::default()
});
let source_session_id = self.session_head.lock().await.session_id;
let source_session_id = self.session_head.lock().session_id;
let staging_id = if payload.is_empty() {
String::new()
} else {
@ -2309,8 +2382,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
domain: extract::EXTRACT_DOMAIN.into(),
payload: payload_value,
})
.await?;
self.session_id = self.session_head.lock().await.session_id;
?;
self.session_id = self.session_head.lock().session_id;
*self
.extract_pointer
@ -2617,7 +2690,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
worker: Some(worker),
store,
session_id,
session_head: Arc::new(AsyncMutex::new(SessionHead {
session_head: Arc::new(SyncMutex::new(SessionHead {
session_id,
head_hash: None,
})),
@ -2635,7 +2708,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
alerter: None,
event_tx: None,
pending_notifies: NotifyBuffer::new(),
pending_attachments: Arc::new(Mutex::new(Vec::new())),
pending_attachments: Arc::new(Mutex::new(Vec::<SystemItem>::new())),
scope_allocation: Some(scope_allocation),
callback_socket: None,
prompts: common.prompts,
@ -2649,7 +2722,8 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
memory_task: None,
user_segments: Vec::new(),
sink: SessionLogSink::new(),
log_cmd_tx: None,
history_persistence_wired: false,
log_writer: None,
};
pod.apply_permissions_from_manifest();
pod.apply_prune_from_manifest();
@ -2690,7 +2764,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
worker: Some(worker),
store,
session_id,
session_head: Arc::new(AsyncMutex::new(SessionHead {
session_head: Arc::new(SyncMutex::new(SessionHead {
session_id,
head_hash: None,
})),
@ -2708,7 +2782,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
alerter: None,
event_tx: None,
pending_notifies: NotifyBuffer::new(),
pending_attachments: Arc::new(Mutex::new(Vec::new())),
pending_attachments: Arc::new(Mutex::new(Vec::<SystemItem>::new())),
scope_allocation: Some(scope_allocation),
callback_socket: Some(callback_socket),
prompts: common.prompts,
@ -2722,7 +2796,8 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
memory_task: None,
user_segments: Vec::new(),
sink: SessionLogSink::new(),
log_cmd_tx: None,
history_persistence_wired: false,
log_writer: None,
};
pod.apply_permissions_from_manifest();
pod.apply_prune_from_manifest();
@ -2757,7 +2832,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
// 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 raw_entries = store.read_all(session_id)?;
let state = session_store::collect_state(&raw_entries);
if state.head_hash.is_none() {
return Err(PodError::SessionEmpty { session_id });
@ -2832,7 +2907,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
worker: Some(worker),
store,
session_id,
session_head: Arc::new(AsyncMutex::new(SessionHead {
session_head: Arc::new(SyncMutex::new(SessionHead {
session_id,
head_hash: state.head_hash,
})),
@ -2852,7 +2927,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
alerter: None,
event_tx: None,
pending_notifies: NotifyBuffer::new(),
pending_attachments: Arc::new(Mutex::new(Vec::new())),
pending_attachments: Arc::new(Mutex::new(Vec::<SystemItem>::new())),
scope_allocation: Some(scope_allocation),
callback_socket: None,
prompts: common.prompts,
@ -2869,7 +2944,8 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
// late-attaching client sees the full prefix without an
// extra round trip.
sink: SessionLogSink::with_initial(mirror_entries),
log_cmd_tx: None,
history_persistence_wired: false,
log_writer: None,
};
pod.apply_permissions_from_manifest();
pod.apply_prune_from_manifest();

View File

@ -7,7 +7,7 @@
//! 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`):
//! Atomicity contract:
//!
//! 1. Pod writes the entry to disk via the `Store`.
//! 2. Pod calls [`SessionLogSink::publish`] which acquires the mirror
@ -24,10 +24,8 @@
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};
use session_store::LogEntry;
use tokio::sync::broadcast;
/// Broadcast capacity for the live receiver. Slow subscribers that
/// fall behind will see `RecvError::Lagged` and are expected to drop
@ -92,8 +90,8 @@ impl SessionLogSink {
/// Live broadcast fires only for entries that the streaming-event
/// lane does not cover:
/// - `LogEntry::SessionStart` → `Event::SessionRotated` on the wire.
/// - `LogEntry::HookInjectedItems` → `Event::HookInjectedItems`.
/// Everything else (AssistantItems, ToolResults, UserInput, TurnEnd,
/// - `LogEntry::SystemItem` → `Event::SystemItem`.
/// Everything else (AssistantItem, ToolResult, UserInput, TurnEnd,
/// RunCompleted, RunErrored, LlmUsage, Extension, ConfigChanged) is
/// reflected in the mirror so reconnect snapshots stay accurate,
/// but is not sent live — the streaming events (TextDelta /
@ -120,7 +118,7 @@ impl SessionLogSink {
fn is_live_relevant(entry: &LogEntry) -> bool {
matches!(
entry,
LogEntry::SessionStart { .. } | LogEntry::HookInjectedItems { .. }
LogEntry::SessionStart { .. } | LogEntry::SystemItem { .. }
)
}
@ -193,200 +191,6 @@ impl Default for SessionLogSink {
}
}
/// 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::*;
@ -427,12 +231,13 @@ mod tests {
assert!(rx.try_recv().is_err());
}
fn hook_injected(text: &str) -> LogEntry {
LogEntry::HookInjectedItems {
fn notification_entry(text: &str) -> LogEntry {
LogEntry::SystemItem {
ts: now_millis(),
items: vec![session_store::LoggedItem::from(
&llm_worker::Item::system_message(text),
)],
item: session_store::SystemItem::Notification {
message: text.to_owned(),
body: format!("[Notification] {text}"),
},
}
}
@ -448,11 +253,11 @@ mod tests {
sink.publish(turn_end(1));
assert!(rx.try_recv().is_err(), "TurnEnd must not be broadcast live");
// HookInjectedItems is live-relevant.
sink.publish(hook_injected("[Notify] hi"));
// SystemItem is live-relevant.
sink.publish(notification_entry("hi"));
match rx.try_recv() {
Ok(LogEntry::HookInjectedItems { .. }) => {}
other => panic!("expected HookInjectedItems, got {other:?}"),
Ok(LogEntry::SystemItem { .. }) => {}
other => panic!("expected SystemItem, got {other:?}"),
}
// Mirror still grew with both entries (snapshot completeness).
@ -465,11 +270,11 @@ mod tests {
let sink = SessionLogSink::new();
sink.publish(session_start());
let (snapshot, mut rx) = sink.subscribe_with_snapshot();
sink.publish(hook_injected("post-snapshot"));
sink.publish(notification_entry("post-snapshot"));
assert_eq!(snapshot.len(), 1);
match rx.try_recv() {
Ok(LogEntry::HookInjectedItems { .. }) => {}
Ok(LogEntry::SystemItem { .. }) => {}
other => panic!("unexpected: {other:?}"),
}
assert!(rx.try_recv().is_err());

View File

@ -149,7 +149,7 @@ async fn make_pod_with_manifest(
let manifest = pod::PodManifest::from_toml(manifest_toml).unwrap();
let store_tmp = tempfile::tempdir().unwrap();
let store = FsStore::new(store_tmp.path()).await.unwrap();
let store = FsStore::new(store_tmp.path()).unwrap();
std::mem::forget(store_tmp);
let pwd_tmp = tempfile::tempdir().unwrap();

View File

@ -158,7 +158,7 @@ async fn make_pod_with(
let manifest = pod::PodManifest::from_toml(manifest_toml).unwrap();
let store_tmp = tempfile::tempdir().unwrap();
let store = FsStore::new(store_tmp.path()).await.unwrap();
let store = FsStore::new(store_tmp.path()).unwrap();
std::mem::forget(store_tmp);
let scope = pod::Scope::writable(&pwd).unwrap();

View File

@ -29,11 +29,20 @@ fn history_from_sink(handle: &PodHandle) -> Vec<Item> {
let text = protocol::Segment::flatten_to_text(&segments);
items.push(Item::user_message(text));
}
LogEntry::AssistantItem { item, .. } | LogEntry::ToolResult { item, .. } => {
items.push(Item::from(item));
}
LogEntry::SystemItem { item, .. } => {
items.push(item.to_history_item());
}
LogEntry::AssistantItems { items: i, .. }
| LogEntry::ToolResults { items: i, .. }
| LogEntry::HookInjectedItems { items: i, .. } => {
items.extend(i.into_iter().map(Item::from));
}
LogEntry::SystemItems { items: si, .. } => {
items.extend(si.iter().map(|s| s.to_history_item()));
}
_ => {}
}
}
@ -164,7 +173,7 @@ async fn make_pod_with_pwd_and_manifest(
) -> (Pod<MockClient, FsStore>, std::path::PathBuf) {
let manifest = PodManifest::from_toml(manifest_toml).unwrap();
let store_tmp = tempfile::tempdir().unwrap();
let store = FsStore::new(store_tmp.path()).await.unwrap();
let store = FsStore::new(store_tmp.path()).unwrap();
std::mem::forget(store_tmp);
// Separate tempdir to serve as the Pod's pwd/scope — these tests
@ -745,16 +754,12 @@ async fn notify_while_idle_auto_starts_turn_and_injects_system_message() {
.unwrap();
// Wait for the auto-started turn to complete.
let mut saw_notify_echo = false;
let mut saw_turn_end = false;
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
loop {
tokio::select! {
event = rx.recv() => {
match event {
Ok(Event::Notify { ref message }) if message == "turn finished" => {
saw_notify_echo = true;
}
Ok(Event::TurnEnd { .. }) => { saw_turn_end = true; break; }
Err(_) => break,
_ => {}
@ -763,14 +768,26 @@ async fn notify_while_idle_auto_starts_turn_and_injects_system_message() {
_ = tokio::time::sleep_until(deadline) => break,
}
}
assert!(
saw_notify_echo,
"Method::Notify on idle Pod should be echoed as Event::Notify"
);
assert!(saw_turn_end, "auto-triggered turn should complete");
// Status flips back to Idle on the controller thread after RunEnd.
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
assert_eq!(handle.shared_state.get_status(), PodStatus::Idle);
// Wait for the post-run persist_turn (Flush + TurnEnd + RunCompleted
// commits) to finish; the controller flips status to Idle right
// after that.
wait_for_status(&handle, PodStatus::Idle).await;
// The live echo arrives via the sink's `Event::SystemItem` lane,
// not on the `event_tx` broadcast that `handle.subscribe()` taps.
// Verify the notification landed on the sink mirror instead.
let (entries, _) = handle.sink.subscribe_with_snapshot();
let saw_notify_in_mirror = entries.iter().any(|e| matches!(
e,
session_store::LogEntry::SystemItem {
item: session_store::SystemItem::Notification { message, .. },
..
} if message == "turn finished"
));
assert!(
saw_notify_in_mirror,
"Method::Notify should commit a SystemItem::Notification entry; mirror = {entries:?}"
);
// Exactly one request was made; it must contain the formatted
// notification as one of the items (committed to history by
@ -825,18 +842,12 @@ async fn pod_event_turn_ended_while_idle_auto_starts_turn_and_injects_system_mes
.await
.unwrap();
let mut saw_pod_event_echo = false;
let mut saw_turn_end = false;
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
loop {
tokio::select! {
event = rx.recv() => {
match event {
Ok(Event::PodEvent(protocol::PodEvent::TurnEnded { ref pod_name }))
if pod_name == "child" =>
{
saw_pod_event_echo = true;
}
Ok(Event::TurnEnd { .. }) => { saw_turn_end = true; break; }
Err(_) => break,
_ => {}
@ -845,15 +856,29 @@ async fn pod_event_turn_ended_while_idle_auto_starts_turn_and_injects_system_mes
_ = tokio::time::sleep_until(deadline) => break,
}
}
assert!(
saw_pod_event_echo,
"Method::PodEvent on idle Pod should be echoed as Event::PodEvent"
);
assert!(
saw_turn_end,
"PodEvent::TurnEnded on idle Pod should auto-start a turn"
);
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
// Wait for the post-run persist_turn to complete before reading the
// mirror — TurnEnd fires inside the worker loop, persist_turn (and
// its Flush of the drain queue) runs afterwards.
wait_for_status(&handle, PodStatus::Idle).await;
let (entries, _) = handle.sink.subscribe_with_snapshot();
let saw_pod_event_in_mirror = entries.iter().any(|e| matches!(
e,
session_store::LogEntry::SystemItem {
item: session_store::SystemItem::PodEvent {
event: protocol::PodEvent::TurnEnded { pod_name },
..
},
..
} if pod_name == "child"
));
assert!(
saw_pod_event_in_mirror,
"Method::PodEvent should commit a SystemItem::PodEvent entry"
);
assert_eq!(handle.shared_state.get_status(), PodStatus::Idle);
let requests = client_for_assert.captured_requests();
@ -911,8 +936,6 @@ async fn notify_while_running_does_not_emit_already_running_error() {
.unwrap();
// Drain events until the run ends; AlreadyRunning must never appear.
// The in-flight branch must still echo the Notify as a log element.
let mut saw_notify_echo = false;
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
loop {
tokio::select! {
@ -921,9 +944,6 @@ async fn notify_while_running_does_not_emit_already_running_error() {
Ok(Event::Error { code, .. }) if code == pod::ErrorCode::AlreadyRunning => {
panic!("Notify while running must not produce AlreadyRunning");
}
Ok(Event::Notify { ref message }) if message == "ping" => {
saw_notify_echo = true;
}
Ok(Event::TurnEnd { .. }) => break,
Err(_) => break,
_ => {}
@ -932,10 +952,13 @@ async fn notify_while_running_does_not_emit_already_running_error() {
_ = tokio::time::sleep_until(deadline) => break,
}
}
assert!(
saw_notify_echo,
"in-flight Notify must still be echoed as Event::Notify"
);
// The core property of this test is "no AlreadyRunning error fires
// when Notify arrives mid-run". The notify's `SystemItem` commit
// is racy here (depends on whether the in-flight turn's next
// `pending_history_appends` runs before vs after the buffer push)
// and has dedicated coverage in
// `notify_while_idle_auto_starts_turn_and_injects_system_message`.
wait_for_status(&handle, PodStatus::Idle).await;
}
#[tokio::test]
@ -1032,19 +1055,29 @@ async fn socket_pod_event_turn_ended_while_idle_auto_starts_turn() {
let mut saw_turn_end = false;
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
// The SystemItem and TurnEnd events arrive through independent
// broadcast lanes (sink fan-out vs `event_tx`), so their relative
// order on the wire is non-deterministic. Keep reading until both
// are observed (or the deadline trips), rather than breaking on
// the first TurnEnd.
loop {
if saw_pod_event_echo && saw_turn_end {
break;
}
tokio::select! {
event = reader.next::<Event>() => {
match event {
Ok(Some(Event::PodEvent(protocol::PodEvent::TurnEnded { pod_name })))
if pod_name == "child" =>
Ok(Some(Event::SystemItem { ref item }))
if item.get("kind").and_then(|k| k.as_str()) == Some("pod_event")
&& item
.pointer("/event/pod_name")
.and_then(|v| v.as_str()) == Some("child") =>
{
saw_pod_event_echo = true;
}
Ok(Some(Event::TurnStart { .. })) => saw_turn_start = true,
Ok(Some(Event::TurnEnd { .. })) => {
saw_turn_end = true;
break;
}
Ok(None) | Err(_) => break,
_ => {}
@ -1056,7 +1089,7 @@ async fn socket_pod_event_turn_ended_while_idle_auto_starts_turn() {
assert!(
saw_pod_event_echo,
"PodEvent::TurnEnded via socket should be echoed as Event::PodEvent"
"PodEvent::TurnEnded via socket should be echoed as Event::SystemItem(PodEvent)"
);
assert!(
saw_turn_start,

View File

@ -36,7 +36,7 @@ async fn restore_from_manifest_rejects_unknown_session() {
let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let store_tmp = tempfile::tempdir().unwrap();
let store = FsStore::new(store_tmp.path()).await.unwrap();
let store = FsStore::new(store_tmp.path()).unwrap();
let manifest = pod::PodManifest::from_toml(MINIMAL_MANIFEST_TOML).unwrap();
// A freshly-minted id with no jsonl file at all → store returns
@ -59,7 +59,7 @@ async fn restore_from_manifest_rejects_empty_session_log() {
let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let store_tmp = tempfile::tempdir().unwrap();
let store = FsStore::new(store_tmp.path()).await.unwrap();
let store = FsStore::new(store_tmp.path()).unwrap();
let manifest = pod::PodManifest::from_toml(MINIMAL_MANIFEST_TOML).unwrap();
// Pre-create an empty `<id>.jsonl` so `read_all` succeeds with no
@ -86,7 +86,7 @@ async fn restore_from_manifest_rejects_session_without_scope_snapshot() {
let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let store_tmp = tempfile::tempdir().unwrap();
let store = FsStore::new(store_tmp.path()).await.unwrap();
let store = FsStore::new(store_tmp.path()).unwrap();
let manifest = pod::PodManifest::from_toml(MINIMAL_MANIFEST_TOML).unwrap();
let id = session_store::new_session_id();
@ -95,9 +95,7 @@ async fn restore_from_manifest_rejects_session_without_scope_snapshot() {
config: &Default::default(),
history: &[],
};
session_store::create_session_with_id(&store, id, state)
.await
.unwrap();
session_store::create_session_with_id(&store, id, state).unwrap();
let result =
Pod::restore_from_manifest(id, manifest, store, pod::PromptLoader::builtins_only()).await;

View File

@ -174,7 +174,7 @@ async fn make_pod(
) {
let manifest = PodManifest::from_toml(&manifest_toml).unwrap();
let store_tmp = tempfile::tempdir().unwrap();
let store = FsStore::new(store_tmp.path()).await.unwrap();
let store = FsStore::new(store_tmp.path()).unwrap();
let pwd_tmp = tempfile::tempdir().unwrap();
let pwd = pwd_tmp.path().to_path_buf();
let scope = pod::Scope::writable(&pwd).unwrap();
@ -210,7 +210,7 @@ async fn prune_metrics_emit_skip_then_fire_with_post_request_join() {
pod.run_text("first").await.unwrap();
pod.run_text("second").await.unwrap();
let state = session_store::restore(&store, session_id).await.unwrap();
let state = session_store::restore(&store, session_id).unwrap();
let metrics = metrics_from_extensions(&state.extensions);
// Run 1 has 2 LLM iterations (tool loop), each evaluates prune with
@ -296,7 +296,7 @@ async fn prune_metrics_record_below_min_savings_skip() {
pod.run_text("first").await.unwrap();
pod.run_text("second").await.unwrap();
let state = session_store::restore(&store, session_id).await.unwrap();
let state = session_store::restore(&store, session_id).unwrap();
let metrics = metrics_from_extensions(&state.extensions);
let below = metrics
.iter()
@ -329,35 +329,35 @@ struct MetricFailingStore {
}
impl Store for MetricFailingStore {
async fn append(&self, id: SessionId, entry: &HashedEntry) -> Result<(), StoreError> {
fn append(&self, id: SessionId, entry: &HashedEntry) -> Result<(), StoreError> {
if let LogEntry::Extension { domain, .. } = &entry.entry {
if domain == DOMAIN {
return Err(StoreError::Io(std::io::Error::other("synthetic failure")));
}
}
self.inner.append(id, entry).await
self.inner.append(id, entry)
}
async fn read_all(&self, id: SessionId) -> Result<Vec<HashedEntry>, StoreError> {
self.inner.read_all(id).await
fn read_all(&self, id: SessionId) -> Result<Vec<HashedEntry>, StoreError> {
self.inner.read_all(id)
}
async fn list_sessions(&self) -> Result<Vec<SessionId>, StoreError> {
self.inner.list_sessions().await
fn list_sessions(&self) -> Result<Vec<SessionId>, StoreError> {
self.inner.list_sessions()
}
async fn create_session(
fn create_session(
&self,
id: SessionId,
entries: &[HashedEntry],
) -> Result<(), StoreError> {
self.inner.create_session(id, entries).await
self.inner.create_session(id, entries)
}
async fn exists(&self, id: SessionId) -> Result<bool, StoreError> {
self.inner.exists(id).await
fn exists(&self, id: SessionId) -> Result<bool, StoreError> {
self.inner.exists(id)
}
async fn read_head_hash(&self, id: SessionId) -> Result<Option<EntryHash>, StoreError> {
self.inner.read_head_hash(id).await
fn read_head_hash(&self, id: SessionId) -> Result<Option<EntryHash>, StoreError> {
self.inner.read_head_hash(id)
}
async fn append_trace(&self, id: SessionId, entry: &TraceEntry) -> Result<(), StoreError> {
self.inner.append_trace(id, entry).await
fn append_trace(&self, id: SessionId, entry: &TraceEntry) -> Result<(), StoreError> {
self.inner.append_trace(id, entry)
}
}
@ -372,7 +372,7 @@ async fn metric_write_failure_emits_warn_alert_and_does_not_abort_run() {
let manifest_toml = manifest_toml(1, 1);
let manifest = PodManifest::from_toml(&manifest_toml).unwrap();
let store_tmp = tempfile::tempdir().unwrap();
let inner = FsStore::new(store_tmp.path()).await.unwrap();
let inner = FsStore::new(store_tmp.path()).unwrap();
let store = MetricFailingStore { inner };
let pwd_tmp = tempfile::tempdir().unwrap();
let pwd = pwd_tmp.path().to_path_buf();
@ -397,7 +397,7 @@ async fn metric_write_failure_emits_warn_alert_and_does_not_abort_run() {
pod.run_text("hello").await.unwrap();
// No metrics ended up in the log (writes were rejected).
let state = session_store::restore(&store, session_id).await.unwrap();
let state = session_store::restore(&store, session_id).unwrap();
let metrics = metrics_from_extensions(&state.extensions);
assert!(metrics.is_empty(), "metrics must drop on write failure");
@ -444,7 +444,7 @@ permission = "write"
let client = MockClient::new(vec![text_response_with_cache("hi", 0, 0)]);
let manifest = PodManifest::from_toml(manifest_toml).unwrap();
let store_tmp = tempfile::tempdir().unwrap();
let store = FsStore::new(store_tmp.path()).await.unwrap();
let store = FsStore::new(store_tmp.path()).unwrap();
let pwd_tmp = tempfile::tempdir().unwrap();
let pwd = pwd_tmp.path().to_path_buf();
let scope = pod::Scope::writable(&pwd).unwrap();
@ -455,7 +455,7 @@ permission = "write"
let session_id = pod.session_id();
pod.run_text("hello").await.unwrap();
let state = session_store::restore(&store, session_id).await.unwrap();
let state = session_store::restore(&store, session_id).unwrap();
let metrics = metrics_from_extensions(&state.extensions);
assert!(
metrics.is_empty(),

View File

@ -103,7 +103,7 @@ async fn make_pod_with_body(
let manifest = pod::PodManifest::from_toml(MINIMAL_MANIFEST_TOML).unwrap();
let store_tmp = tempfile::tempdir().unwrap();
let store = FsStore::new(store_tmp.path()).await.unwrap();
let store = FsStore::new(store_tmp.path()).unwrap();
std::mem::forget(store_tmp);
let pwd_tmp = tempfile::tempdir().unwrap();
@ -182,7 +182,7 @@ async fn session_start_state_captures_rendered_prompt() {
.unwrap();
pod.run_text("hi").await.unwrap();
let entries = pod.store().read_all(pod.session_id()).await.unwrap();
let entries = pod.store().read_all(pod.session_id()).unwrap();
let first = entries.first().expect("at least one entry");
match &first.entry {
LogEntry::SessionStart { system_prompt, .. } => {

View File

@ -214,20 +214,23 @@ pub enum Event {
UserMessage {
segments: Vec<Segment>,
},
/// Echo of `Method::Notify` received by this Pod. Broadcast on
/// receipt so subscribers can render the external input as a log
/// element. The same `message` is independently pushed into the
/// notification buffer for LLM injection (with prompt-pack
/// wrapping); this echo carries the raw payload and does not
/// imply any turn-boundary semantics.
Notify {
message: String,
/// One agent-injected system item committed to history.
///
/// Carries the JSON form of `session_store::SystemItem`. Covers
/// `Method::Notify` echoes, child-Pod lifecycle events from
/// `Method::PodEvent`, `@<path>` / `#<slug>` / `/<slug>`
/// resolution payloads, and any future agent-side injection kind.
/// Clients dispatch on the `kind` tag for typed rendering instead
/// of parsing free-text prefixes like `[Notification] …` or
/// `[File: …]`.
///
/// One event per `LogEntry::SystemItem` commit. Disk-side and
/// wire-side are 1:1 (singular variant); legacy `SystemItems`
/// entries from older sessions are read-only and never emitted on
/// this lane.
SystemItem {
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
/// context via the notification buffer.
PodEvent(PodEvent),
TurnStart {
turn: usize,
},
@ -335,17 +338,6 @@ pub enum Event {
SessionRotated {
entry: serde_json::Value,
},
/// A non-LLM-driven history append landed in the worker history.
///
/// Carries the JSON form of `session_store::LogEntry::HookInjectedItems`.
/// This is the live counterpart of items that the streaming lane
/// never broadcasts — `Method::Notify` echoes, `@<path>` attachment
/// resolutions, `<system-reminder>` injections — so a connected
/// client can render them in time order without waiting for the
/// next reconnect's `Snapshot`.
HookInjectedItems {
entry: serde_json::Value,
},
/// Current Pod controller status. Broadcast on every controller-level
/// transition and included in `History` snapshots for late attach.
Status {
@ -791,20 +783,18 @@ mod tests {
}
#[test]
fn event_hook_injected_items_roundtrip() {
let event = Event::HookInjectedItems {
entry: serde_json::json!({"kind": "hook_injected_items", "ts": 42, "items": []}),
fn event_system_item_roundtrip() {
let event = Event::SystemItem {
item: serde_json::json!({"kind": "notification", "message": "hello"}),
};
let json = serde_json::to_string(&event).unwrap();
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
assert_eq!(parsed["event"], "hook_injected_items");
assert_eq!(parsed["data"]["entry"]["kind"], "hook_injected_items");
assert_eq!(parsed["event"], "system_item");
assert_eq!(parsed["data"]["item"]["kind"], "notification");
let decoded: Event = serde_json::from_str(&json).unwrap();
match decoded {
Event::HookInjectedItems { entry } => {
assert_eq!(entry["kind"], "hook_injected_items")
}
other => panic!("expected HookInjectedItems, got {other:?}"),
Event::SystemItem { item } => assert_eq!(item["kind"], "notification"),
other => panic!("expected SystemItem, got {other:?}"),
}
}
@ -1066,43 +1056,6 @@ mod tests {
assert_eq!(parsed["data"]["code"], "already_running");
}
#[test]
fn event_notify_roundtrip() {
let event = Event::Notify {
message: "child-pod finished".into(),
};
let json = serde_json::to_string(&event).unwrap();
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
assert_eq!(parsed["event"], "notify");
assert_eq!(parsed["data"]["message"], "child-pod finished");
let decoded: Event = serde_json::from_str(&json).unwrap();
match decoded {
Event::Notify { message } => assert_eq!(message, "child-pod finished"),
other => panic!("expected Notify, got {other:?}"),
}
}
#[test]
fn event_pod_event_roundtrip() {
let event = Event::PodEvent(PodEvent::TurnEnded {
pod_name: "child".into(),
});
let json = serde_json::to_string(&event).unwrap();
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
assert_eq!(parsed["event"], "pod_event");
assert_eq!(parsed["data"]["kind"], "turn_ended");
assert_eq!(parsed["data"]["pod_name"], "child");
let decoded: Event = serde_json::from_str(&json).unwrap();
match decoded {
Event::PodEvent(PodEvent::TurnEnded { pod_name }) => {
assert_eq!(pod_name, "child");
}
other => panic!("expected PodEvent::TurnEnded, got {other:?}"),
}
}
#[test]
fn event_user_message_roundtrip() {
let event = Event::UserMessage {

View File

@ -75,14 +75,14 @@ impl Metric {
///
/// `save_extension` の薄い wrapper。書き込み失敗は呼び出し側に返す
/// (メトリクスのために本体処理を止めるかは呼び出し側の判断)。
pub async fn record_metric(
pub fn record_metric(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
metric: &Metric,
) -> Result<(), StoreError> {
let payload = serde_json::to_value(metric).expect("Metric serialization cannot fail");
save_extension(store, session_id, head_hash, DOMAIN, payload).await
save_extension(store, session_id, head_hash, DOMAIN, payload)
}
/// `RestoredState.extensions` から metrics domain の payload を順に取り出し、

View File

@ -7,10 +7,8 @@ license.workspace = true
[dependencies]
llm-worker = { workspace = true }
async-trait = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
tokio = { workspace = true, features = ["fs", "io-util"] }
uuid = { workspace = true, features = ["v7", "serde"] }
thiserror = { workspace = true }
sha2 = { workspace = true }
@ -19,6 +17,7 @@ protocol = { workspace = true }
tracing.workspace = true
[dev-dependencies]
async-trait = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
tempfile = { workspace = true }
futures = { workspace = true }

View File

@ -8,9 +8,9 @@ use crate::SessionId;
use crate::event_trace::TraceEntry;
use crate::session_log::{EntryHash, HashedEntry};
use crate::store::{Store, StoreError};
use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};
use tokio::fs;
use tokio::io::AsyncWriteExt;
/// Filesystem-backed JSONL store.
///
@ -24,9 +24,9 @@ pub struct FsStore {
impl FsStore {
/// Create a new `FsStore` rooted at the given directory.
/// Creates the directory if it does not exist.
pub async fn new(root: impl Into<PathBuf>) -> Result<Self, StoreError> {
pub fn new(root: impl Into<PathBuf>) -> Result<Self, StoreError> {
let root = root.into();
fs::create_dir_all(&root).await?;
fs::create_dir_all(&root)?;
Ok(Self { root })
}
@ -38,15 +38,13 @@ impl FsStore {
self.root.join(format!("{id}.trace.jsonl"))
}
async fn append_line(&self, path: &Path, line: &str) -> Result<(), StoreError> {
let mut file = fs::OpenOptions::new()
.create(true)
.append(true)
.open(path)
.await?;
file.write_all(line.as_bytes()).await?;
file.write_all(b"\n").await?;
file.flush().await?;
fn append_line(&self, path: &Path, line: &str) -> Result<(), StoreError> {
let mut file = fs::OpenOptions::new().create(true).append(true).open(path)?;
file.write_all(line.as_bytes())?;
file.write_all(b"\n")?;
// Append-mode write is the durability boundary; an explicit
// `sync_all` here would multiply latency by ~10× for no gain
// since the kernel already orders concurrent `O_APPEND` writes.
Ok(())
}
@ -67,24 +65,24 @@ impl FsStore {
}
impl Store for FsStore {
async fn append(&self, id: SessionId, entry: &HashedEntry) -> Result<(), StoreError> {
fn append(&self, id: SessionId, entry: &HashedEntry) -> Result<(), StoreError> {
let line = serde_json::to_string(entry)?;
self.append_line(&self.log_path(id), &line).await
self.append_line(&self.log_path(id), &line)
}
async fn read_all(&self, id: SessionId) -> Result<Vec<HashedEntry>, StoreError> {
fn read_all(&self, id: SessionId) -> Result<Vec<HashedEntry>, StoreError> {
let path = self.log_path(id);
if !path.exists() {
return Err(StoreError::NotFound(id));
}
let content = fs::read_to_string(&path).await?;
let content = fs::read_to_string(&path)?;
Self::parse_jsonl(&content)
}
async fn list_sessions(&self) -> Result<Vec<SessionId>, StoreError> {
fn list_sessions(&self) -> Result<Vec<SessionId>, StoreError> {
let mut sessions = Vec::new();
let mut dir = fs::read_dir(&self.root).await?;
while let Some(entry) = dir.next_entry().await? {
for entry in fs::read_dir(&self.root)? {
let entry = entry?;
let path = entry.path();
// Only match .jsonl files, not .trace.jsonl
let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
@ -100,31 +98,27 @@ impl Store for FsStore {
Ok(sessions)
}
async fn create_session(
&self,
id: SessionId,
entries: &[HashedEntry],
) -> Result<(), StoreError> {
fn create_session(&self, id: SessionId, entries: &[HashedEntry]) -> Result<(), StoreError> {
let path = self.log_path(id);
let mut content = String::new();
for entry in entries {
content.push_str(&serde_json::to_string(entry)?);
content.push('\n');
}
fs::write(&path, content.as_bytes()).await?;
fs::write(&path, content.as_bytes())?;
Ok(())
}
async fn exists(&self, id: SessionId) -> Result<bool, StoreError> {
fn exists(&self, id: SessionId) -> Result<bool, StoreError> {
Ok(self.log_path(id).exists())
}
async fn read_head_hash(&self, id: SessionId) -> Result<Option<EntryHash>, StoreError> {
fn read_head_hash(&self, id: SessionId) -> Result<Option<EntryHash>, StoreError> {
let path = self.log_path(id);
if !path.exists() {
return Err(StoreError::NotFound(id));
}
let content = fs::read_to_string(&path).await?;
let content = fs::read_to_string(&path)?;
let last_line = content.lines().rev().find(|l| !l.trim().is_empty());
match last_line {
Some(line) => {
@ -139,8 +133,8 @@ impl Store for FsStore {
}
}
async fn append_trace(&self, id: SessionId, entry: &TraceEntry) -> Result<(), StoreError> {
fn append_trace(&self, id: SessionId, entry: &TraceEntry) -> Result<(), StoreError> {
let line = serde_json::to_string(entry)?;
self.append_line(&self.trace_path(id), &line).await
self.append_line(&self.trace_path(id), &line)
}
}

View File

@ -32,6 +32,7 @@ pub mod logged_item;
pub mod session;
pub mod session_log;
pub mod store;
pub mod system_item;
pub use event_trace::TraceEntry;
pub use fs_store::FsStore;
@ -39,15 +40,17 @@ 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, append_entry, append_entry_with_hash, create_compacted_session,
create_session, create_session_with_id, ensure_head_or_fork, fork, fork_at, restore,
save_config_changed, save_delta, save_extension, save_pod_scope, save_run_completed,
save_run_errored, save_turn_end, save_usage, save_user_input,
SessionStartState, append_entry, append_entry_with_hash, append_system_item,
classify_history_item, create_compacted_session, create_session, create_session_with_id,
ensure_head_or_fork, fork, fork_at, restore, save_config_changed, save_delta, save_extension,
save_pod_scope, save_run_completed, save_run_errored, save_turn_end, save_usage,
save_user_input,
};
pub use session_log::{
EntryHash, HashedEntry, LogEntry, POD_SCOPE_EXTENSION_DOMAIN, PodScopeSnapshot, RestoredState,
SessionOrigin, build_chain, collect_state, compute_hash,
};
pub use system_item::{SystemItem, render_pod_event};
pub use store::{Store, StoreError};
/// Session identifier. UUID v7 (time-ordered, lexicographically sortable).

View File

@ -8,6 +8,7 @@ use crate::SessionId;
use crate::logged_item::{LoggedItem, to_logged};
use crate::session_log::{self, EntryHash, HashedEntry, LogEntry, PodScopeSnapshot, SessionOrigin};
use crate::store::{Store, StoreError};
use crate::system_item::SystemItem;
use llm_worker::WorkerResult;
use llm_worker::llm_client::RequestConfig;
use llm_worker::llm_client::types::Item;
@ -23,12 +24,12 @@ pub struct SessionStartState<'a> {
/// Create a new session, writing the initial `SessionStart` entry.
///
/// Returns the new session ID and head hash.
pub async fn create_session(
pub fn create_session(
store: &impl Store,
state: SessionStartState<'_>,
) -> Result<(SessionId, EntryHash), StoreError> {
let session_id = crate::new_session_id();
let hash = create_session_with_id(store, session_id, state).await?;
let hash = create_session_with_id(store, session_id, state)?;
Ok((session_id, hash))
}
@ -37,7 +38,7 @@ pub async fn create_session(
/// Used by callers that need to reserve a session ID synchronously but
/// defer the initial log append (e.g. Pod, which resolves a templated
/// system prompt only at first turn). Returns the resulting head hash.
pub async fn create_session_with_id(
pub fn create_session_with_id(
store: &impl Store,
session_id: SessionId,
state: SessionStartState<'_>,
@ -56,7 +57,7 @@ pub async fn create_session_with_id(
prev_hash: None,
entry,
};
store.append(session_id, &hashed_entry).await?;
store.append(session_id, &hashed_entry)?;
Ok(hash)
}
@ -64,7 +65,7 @@ pub async fn create_session_with_id(
///
/// Records `compacted_from` provenance linking back to the source session.
/// Returns the new session ID and head hash.
pub async fn create_compacted_session(
pub fn create_compacted_session(
store: &impl Store,
state: SessionStartState<'_>,
source_session_id: SessionId,
@ -88,7 +89,7 @@ pub async fn create_compacted_session(
prev_hash: None,
entry,
};
store.append(session_id, &hashed_entry).await?;
store.append(session_id, &hashed_entry)?;
Ok((session_id, hash))
}
@ -96,11 +97,11 @@ pub async fn create_compacted_session(
///
/// Returns the reconstructed state. The caller is responsible for
/// applying it to a Worker.
pub async fn restore(
pub fn restore(
store: &impl Store,
session_id: SessionId,
) -> Result<crate::session_log::RestoredState, StoreError> {
let entries = store.read_all(session_id).await?;
let entries = store.read_all(session_id)?;
Ok(session_log::collect_state(&entries))
}
@ -108,13 +109,13 @@ pub async fn restore(
/// If not, auto-fork into a new session.
///
/// Updates `session_id` and `head_hash` in place when a fork occurs.
pub async fn ensure_head_or_fork(
pub fn ensure_head_or_fork(
store: &impl Store,
session_id: &mut SessionId,
head_hash: &mut Option<EntryHash>,
state: SessionStartState<'_>,
) -> Result<(), StoreError> {
let store_head = store.read_head_hash(*session_id).await?;
let store_head = store.read_head_hash(*session_id)?;
if store_head == *head_hash {
return Ok(());
}
@ -133,7 +134,7 @@ pub async fn ensure_head_or_fork(
prev_hash: None,
entry,
};
store.create_session(fork_id, &[hashed_entry]).await?;
store.create_session(fork_id, &[hashed_entry])?;
*session_id = fork_id;
*head_hash = Some(hash);
Ok(())
@ -145,7 +146,7 @@ pub async fn ensure_head_or_fork(
/// the worker pushes its flattened user message into history; replay
/// derives the worker `Item::user_message` from these segments via
/// [`Segment::flatten_to_text`].
pub async fn save_user_input(
pub fn save_user_input(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
@ -160,17 +161,17 @@ pub async fn save_user_input(
segments,
},
)
.await
}
/// Log the history delta — new items added since the previous snapshot.
///
/// Classifies items into AssistantItems, ToolResults, and HookInjectedItems
/// entries automatically. User messages are skipped because they are
/// persisted upfront via [`save_user_input`] at submit time; the worker
/// pushes a flattened copy into its history that arrives here in
/// `new_items` and would otherwise produce a duplicate `UserInput` entry.
pub async fn save_delta(
/// Classifies items into AssistantItem / ToolResult / HookInjectedItems
/// entries automatically (one entry per item). User messages are skipped
/// because they are persisted upfront via [`save_user_input`] at submit
/// time; the worker pushes a flattened copy into its history that
/// arrives here in `new_items` and would otherwise produce a duplicate
/// `UserInput` entry.
pub fn save_delta(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
@ -181,66 +182,63 @@ pub async fn save_delta(
}
let ts = session_log::now_millis();
let mut i = 0;
while i < new_items.len() {
let item = &new_items[i];
for item in new_items {
if item.is_user_message() {
// Already persisted by save_user_input at submit time.
i += 1;
} else if item.is_tool_result() {
let start = i;
while i < new_items.len() && new_items[i].is_tool_result() {
i += 1;
}
append_entry(
store,
session_id,
head_hash,
LogEntry::ToolResults {
ts,
items: to_logged(&new_items[start..i]),
},
)
.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;
}
append_entry(
store,
session_id,
head_hash,
LogEntry::AssistantItems {
ts,
items: to_logged(&new_items[start..i]),
},
)
.await?;
} else {
append_entry(
store,
session_id,
head_hash,
LogEntry::HookInjectedItems {
ts,
items: vec![LoggedItem::from(&new_items[i])],
},
)
.await?;
i += 1;
continue;
}
let entry = classify_history_item(item, ts);
append_entry(store, session_id, head_hash, entry)?;
}
Ok(())
}
/// Map one history item to its singular `LogEntry` form. Used by the
/// fallback `save_delta` path and the controller's worker-callback
/// classifier so write classification lives in one place.
pub fn classify_history_item(item: &Item, ts: u64) -> LogEntry {
if item.is_tool_result() {
LogEntry::ToolResult {
ts,
item: LoggedItem::from(item),
}
} else if item.is_assistant_message() || item.is_tool_call() || item.is_reasoning() {
LogEntry::AssistantItem {
ts,
item: LoggedItem::from(item),
}
} else {
// Defensive: anything else (future Item kinds) routes through
// AssistantItem rather than getting silently dropped.
LogEntry::AssistantItem {
ts,
item: LoggedItem::from(item),
}
}
}
/// Append a single typed system item as `LogEntry::SystemItem`. Helper
/// for the Pod-side interceptor commit path; mirrors the per-item
/// commit shape used for assistant / tool result entries.
pub fn append_system_item(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
item: SystemItem,
) -> Result<EntryHash, StoreError> {
append_entry_with_hash(
store,
session_id,
head_hash,
LogEntry::SystemItem {
ts: session_log::now_millis(),
item,
},
)
}
/// Log a TurnEnd entry.
pub async fn save_turn_end(
pub fn save_turn_end(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
@ -255,11 +253,10 @@ pub async fn save_turn_end(
turn_count,
},
)
.await
}
/// Log a `RunCompleted` entry — `run()` / `resume()` returned `Ok(WorkerResult)`.
pub async fn save_run_completed(
pub fn save_run_completed(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
@ -276,14 +273,13 @@ pub async fn save_run_completed(
result,
},
)
.await
}
/// Log a `RunErrored` entry — `run()` / `resume()` returned `Err(WorkerError)`.
///
/// `WorkerError` is not `Serialize`, so the caller passes a lossy
/// `to_string()` rendering as `message`.
pub async fn save_run_errored(
pub fn save_run_errored(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
@ -300,7 +296,6 @@ pub async fn save_run_errored(
message,
},
)
.await
}
/// Log an `LlmUsage` entry — 1 LLM リクエスト分の Usage スナップショット。
@ -309,7 +304,7 @@ pub async fn save_run_errored(
/// その prefix をプロバイダが実測した占有量(プロンプト全長)で、
/// プロバイダ別の正規化Anthropic では `input + cache_read + cache_creation`)を
/// 済ませた値を渡す。
pub async fn save_usage(
pub fn save_usage(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
@ -332,7 +327,6 @@ pub async fn save_usage(
output_tokens,
},
)
.await
}
/// Log an `Extension` entry — domain-tagged opaque payload.
@ -340,7 +334,7 @@ pub async fn save_usage(
/// session-store treats `payload` as an unstructured `serde_json::Value`.
/// Each domain is responsible for serializing into and folding out of it.
/// Use `RestoredState.extensions` to read entries back at restore time.
pub async fn save_extension(
pub fn save_extension(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
@ -357,11 +351,10 @@ pub async fn save_extension(
payload,
},
)
.await
}
/// Log the Pod's latest runtime scope snapshot.
pub async fn save_pod_scope(
pub fn save_pod_scope(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
@ -375,11 +368,10 @@ pub async fn save_pod_scope(
session_log::POD_SCOPE_EXTENSION_DOMAIN,
payload,
)
.await
}
/// Log a `ConfigChanged` entry.
pub async fn save_config_changed(
pub fn save_config_changed(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
@ -394,14 +386,10 @@ pub async fn save_config_changed(
config: config.clone(),
},
)
.await
}
/// Fork the current state into a new session.
pub async fn fork(
store: &impl Store,
state: SessionStartState<'_>,
) -> Result<SessionId, StoreError> {
pub fn fork(store: &impl Store, state: SessionStartState<'_>) -> Result<SessionId, StoreError> {
let fork_id = crate::new_session_id();
let entry = LogEntry::SessionStart {
ts: session_log::now_millis(),
@ -417,17 +405,17 @@ pub async fn fork(
prev_hash: None,
entry,
};
store.create_session(fork_id, &[hashed_entry]).await?;
store.create_session(fork_id, &[hashed_entry])?;
Ok(fork_id)
}
/// Fork from an arbitrary point in a stored session's log.
pub async fn fork_at(
pub fn fork_at(
store: &impl Store,
source_id: SessionId,
at_hash: &EntryHash,
) -> Result<SessionId, StoreError> {
let entries = store.read_all(source_id).await?;
let entries = store.read_all(source_id)?;
let cut = entries
.iter()
.position(|e| &e.hash == at_hash)
@ -453,7 +441,7 @@ pub async fn fork_at(
prev_hash: None,
entry,
};
store.create_session(fork_id, &[hashed_entry]).await?;
store.create_session(fork_id, &[hashed_entry])?;
Ok(fork_id)
}
@ -462,13 +450,13 @@ pub async fn fork_at(
/// 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(
pub 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?;
append_entry_with_hash(store, session_id, head_hash, entry)?;
Ok(())
}
@ -476,7 +464,7 @@ pub async fn append_entry(
///
/// 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(
pub fn append_entry_with_hash(
store: &impl Store,
session_id: SessionId,
head_hash: &mut Option<EntryHash>,
@ -488,7 +476,7 @@ pub async fn append_entry_with_hash(
prev_hash: head_hash.clone(),
entry,
};
store.append(session_id, &hashed_entry).await?;
store.append(session_id, &hashed_entry)?;
*head_hash = Some(hash.clone());
Ok(hash)
}

View File

@ -15,6 +15,7 @@ use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::logged_item::LoggedItem;
use crate::system_item::SystemItem;
/// SHA-256 hash identifying a specific log entry in the chain.
///
@ -119,13 +120,37 @@ pub enum LogEntry {
/// history; the worker layer never sees segments directly.
UserInput { ts: u64, segments: Vec<Segment> },
/// Assistant response items added to history (worker.rs:1040-1041).
/// One assistant-side item appended to history — assistant message,
/// reasoning, or tool call. Singular: one entry per history item so
/// the wire-side `Event::*` lane and on-disk LogEntry stay 1:1.
AssistantItem { ts: u64, item: LoggedItem },
/// One tool-execution result appended to history.
ToolResult { ts: u64, item: LoggedItem },
/// One typed agent-injected system item: notification, child-Pod
/// lifecycle event, `@<path>` / `#<slug>` / `/<slug>` resolution
/// payload. Each `SystemItem` carries kind metadata that the LLM
/// itself never sees (the LLM gets `Item::system_message` with the
/// item's denormalised `body`), but live clients and replay paths
/// dispatch on `kind` for typed rendering.
SystemItem { ts: u64, item: SystemItem },
/// Legacy plural form: kept **read-only** so old session logs still
/// open. New writes always use the singular `AssistantItem`. Items
/// are flattened on replay.
AssistantItems { ts: u64, items: Vec<LoggedItem> },
/// Tool execution results added to history (worker.rs:897-900, 1072-1076).
/// Legacy plural form: kept **read-only**. New writes use the
/// singular `ToolResult`.
ToolResults { ts: u64, items: Vec<LoggedItem> },
/// Items injected by `on_turn_end` hook via `ContinueWithMessages` (worker.rs:1055).
/// Legacy plural form: kept **read-only**. New writes use the
/// singular `SystemItem`.
SystemItems { ts: u64, items: Vec<SystemItem> },
/// Legacy pre-`SystemItem*` form. Deserialize-only. Items are
/// flattened to `Item::system_message` on replay.
HookInjectedItems { ts: u64, items: Vec<LoggedItem> },
/// Turn boundary. Records the turn count after increment.
@ -270,12 +295,26 @@ pub fn collect_state(entries: &[HashedEntry]) -> RestoredState {
state.history.push(Item::user_message(text));
state.user_segments.push(segments.clone());
}
LogEntry::AssistantItem { item, .. } => {
state.history.push(Item::from(item.clone()));
}
LogEntry::ToolResult { item, .. } => {
state.history.push(Item::from(item.clone()));
}
LogEntry::SystemItem { item, .. } => {
state.history.push(item.to_history_item());
}
LogEntry::AssistantItems { items, .. } => {
state.history.extend(items.iter().cloned().map(Item::from));
}
LogEntry::ToolResults { items, .. } => {
state.history.extend(items.iter().cloned().map(Item::from));
}
LogEntry::SystemItems { items, .. } => {
state
.history
.extend(items.iter().map(|si| si.to_history_item()));
}
LogEntry::HookInjectedItems { items, .. } => {
state.history.extend(items.iter().cloned().map(Item::from));
}

View File

@ -1,12 +1,18 @@
//! Persistence backend abstraction.
//!
//! [`Store`] defines the async interface for reading and writing session logs.
//! [`Store`] defines the sync interface for reading and writing session logs.
//! Implementations handle the physical storage (filesystem, database, etc.).
//!
//! Sync (rather than async) is intentional: a session log append is a single
//! `< 1 KiB` line on local fs and completes well below a millisecond. Going
//! through `tokio::fs` would force every caller — including `Worker`'s sync
//! `on_history_append` callback — to bridge sync → async via a channel +
//! drain task. Keeping the store sync lets the worker callback, Pod commit
//! paths, and `PodInterceptor` all share one direct `append_entry` call.
use crate::SessionId;
use crate::event_trace::TraceEntry;
use crate::session_log::{EntryHash, HashedEntry};
use std::future::Future;
/// Errors from the persistence store.
#[derive(Debug, thiserror::Error)]
@ -24,49 +30,31 @@ pub enum StoreError {
Corrupt { line: usize, message: String },
}
/// Async persistence backend for session logs.
/// Sync persistence backend for session logs.
///
/// All methods take `&self` — implementations should use interior mutability
/// (e.g., append-mode file handles) when needed.
pub trait Store: Send + Sync {
/// Append a single hashed entry to the session log.
fn append(
&self,
id: SessionId,
entry: &HashedEntry,
) -> impl Future<Output = Result<(), StoreError>> + Send;
fn append(&self, id: SessionId, entry: &HashedEntry) -> Result<(), StoreError>;
/// Read all hashed entries for a session, in order.
fn read_all(
&self,
id: SessionId,
) -> impl Future<Output = Result<Vec<HashedEntry>, StoreError>> + Send;
fn read_all(&self, id: SessionId) -> Result<Vec<HashedEntry>, StoreError>;
/// List all session IDs, most recent first.
fn list_sessions(&self) -> impl Future<Output = Result<Vec<SessionId>, StoreError>> + Send;
fn list_sessions(&self) -> Result<Vec<SessionId>, StoreError>;
/// Create a new session with initial entries.
fn create_session(
&self,
id: SessionId,
entries: &[HashedEntry],
) -> impl Future<Output = Result<(), StoreError>> + Send;
fn create_session(&self, id: SessionId, entries: &[HashedEntry]) -> Result<(), StoreError>;
/// Check if a session exists.
fn exists(&self, id: SessionId) -> impl Future<Output = Result<bool, StoreError>> + Send;
fn exists(&self, id: SessionId) -> Result<bool, StoreError>;
/// Read the hash of the last entry in a session (the head).
///
/// Returns `None` if the session is empty.
fn read_head_hash(
&self,
id: SessionId,
) -> impl Future<Output = Result<Option<EntryHash>, StoreError>> + Send;
fn read_head_hash(&self, id: SessionId) -> Result<Option<EntryHash>, StoreError>;
/// Append a trace entry to the debug event trace file.
fn append_trace(
&self,
id: SessionId,
entry: &TraceEntry,
) -> impl Future<Output = Result<(), StoreError>> + Send;
fn append_trace(&self, id: SessionId, entry: &TraceEntry) -> Result<(), StoreError>;
}

View File

@ -0,0 +1,199 @@
//! Typed system-message items injected by the agent system.
//!
//! Items in worker history with `role:system` are never produced by the
//! LLM — they are always inserted by the Pod itself (notifications,
//! file/knowledge/workflow ref resolutions, child-pod lifecycle events,
//! future `<system-reminder>` tags, …). [`SystemItem`] carries the
//! typed shape of each such injection so clients can dispatch on
//! `kind` instead of parsing text prefixes like `[Notification] …` or
//! `[File: …]`.
//!
//! Persisted as the payload of [`crate::LogEntry::SystemItem`] (one
//! entry per item), and broadcast live as the payload of
//! `Event::SystemItem` on the wire.
//!
//! For LLM context replay, each `SystemItem` reduces to an
//! `Item::system_message(...)` whose body matches the legacy free-text
//! shape (see [`SystemItem::history_text`]). The kind metadata is
//! preserved only on the log/wire side; the LLM still sees plain
//! system-message text.
use llm_worker::llm_client::types::Item;
use protocol::PodEvent;
use serde::{Deserialize, Serialize};
/// One agent-injected system item, tagged by origin.
///
/// Each variant carries the kind-specific raw data clients use for
/// typed rendering (`Notification.message`, `PodEvent.event`, file
/// path / knowledge slug / workflow slug / etc.), plus a pre-rendered
/// `body` (where applicable) that is the exact `role:system` text the
/// LLM actually saw at commit time. `body` is denormalised so that
/// session log replay reconstructs worker history byte-identical to
/// what was on the wire — even when prompt overrides (e.g. custom
/// `notify_wrapper` template) re-shape the live rendering on a later
/// resume.
///
/// New variants get added here as fresh injection kinds come online
/// (e.g. `Reminder`). The `kind` JSON tag is the snake_case form of
/// the variant name.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum SystemItem {
/// Free-form notification sent in by an external caller via
/// `Method::Notify`. `message` is the raw caller-supplied text;
/// `body` is the wrapped LLM-context form (Pod renders it via
/// `notify_wrapper` at commit time).
Notification { message: String, body: String },
/// Lifecycle event reported by a child Pod via `Method::PodEvent`.
/// `event` is the typed payload (so the TUI can render per-child
/// banners without re-parsing); `body` is the wrapped LLM-context
/// form (same `notify_wrapper` path as `Notification`).
PodEvent { event: PodEvent, body: String },
/// `@<path>` file reference resolution. `body` is the rendered
/// LLM-context text (`[File: <path>]\n…` for regular files,
/// `[Dir: <path>]\n…` for directory listings, possibly with a
/// truncation hint) so replay reconstructs worker history
/// byte-identical to what was sent.
FileAttachment { path: String, body: String },
/// `#<slug>` Knowledge reference resolution. `body` is the
/// rendered text the LLM saw (Pod composes the `[Knowledge: …]`
/// header + body).
Knowledge { slug: String, body: String },
/// `/<slug>` Workflow invocation. `body` is the workflow's
/// prompt body materialized into the LLM context.
Workflow { slug: String, body: String },
}
impl SystemItem {
/// Free-text body the LLM sees inside its `role:system` message
/// for this item. Returns the variant's stored `body` verbatim.
pub fn history_text(&self) -> String {
match self {
SystemItem::Notification { body, .. } => body.clone(),
SystemItem::PodEvent { body, .. } => body.clone(),
SystemItem::FileAttachment { body, .. } => body.clone(),
SystemItem::Knowledge { body, .. } => body.clone(),
SystemItem::Workflow { body, .. } => body.clone(),
}
}
/// Materialize this `SystemItem` as the `Item::system_message`
/// form that lands in worker history.
pub fn to_history_item(&self) -> Item {
Item::system_message(self.history_text())
}
/// Short human-readable label used for diagnostics. Not on the
/// wire — keep flexible.
pub fn kind_label(&self) -> &'static str {
match self {
SystemItem::Notification { .. } => "notification",
SystemItem::PodEvent { .. } => "pod_event",
SystemItem::FileAttachment { .. } => "file_attachment",
SystemItem::Knowledge { .. } => "knowledge",
SystemItem::Workflow { .. } => "workflow",
}
}
}
/// Render a `PodEvent` as the one-line notification text the agent
/// sees. Centralised here (rather than at the controller's render
/// site) so persistence and broadcast share the same rendering.
pub fn render_pod_event(event: &PodEvent) -> String {
match event {
PodEvent::TurnEnded { pod_name } => format!("pod `{pod_name}` finished a turn"),
PodEvent::Errored { pod_name, message } => {
format!("pod `{pod_name}` errored: {message}")
}
PodEvent::ShutDown { pod_name } => format!("pod `{pod_name}` shut down"),
PodEvent::ScopeSubDelegated {
parent_pod,
sub_pod,
..
} => {
format!("pod `{parent_pod}` sub-delegated scope to `{sub_pod}`")
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn notification_history_text_returns_stored_body() {
let item = SystemItem::Notification {
message: "child done".into(),
body: "[Notification]\nchild done\n\n(non-blocking hint…)".into(),
};
assert_eq!(
item.history_text(),
"[Notification]\nchild done\n\n(non-blocking hint…)"
);
}
#[test]
fn pod_event_history_text_returns_stored_body() {
let item = SystemItem::PodEvent {
event: PodEvent::TurnEnded {
pod_name: "child".into(),
},
body: "[Notification]\npod `child` finished a turn\n\n(non-blocking hint…)".into(),
};
assert!(item.history_text().starts_with("[Notification]\n"));
assert!(item.history_text().contains("`child`"));
}
#[test]
fn file_attachment_history_text_returns_stored_body() {
let item = SystemItem::FileAttachment {
path: "src/main.rs".into(),
body: "[File: src/main.rs]\nfn main() {}".into(),
};
assert_eq!(item.history_text(), "[File: src/main.rs]\nfn main() {}");
}
#[test]
fn round_trip_via_json() {
let item = SystemItem::FileAttachment {
path: "src/main.rs".into(),
body: "[File: src/main.rs]\nfn main() {}".into(),
};
let json = serde_json::to_string(&item).unwrap();
let parsed: SystemItem = serde_json::from_str(&json).unwrap();
match parsed {
SystemItem::FileAttachment { path, body } => {
assert_eq!(path, "src/main.rs");
assert_eq!(body, "[File: src/main.rs]\nfn main() {}");
}
other => panic!("unexpected: {other:?}"),
}
}
#[test]
fn round_trip_pod_event() {
let item = SystemItem::PodEvent {
event: PodEvent::TurnEnded {
pod_name: "child".into(),
},
body: "[Notification] pod `child` finished a turn".into(),
};
let json = serde_json::to_string(&item).unwrap();
let parsed: SystemItem = serde_json::from_str(&json).unwrap();
match parsed {
SystemItem::PodEvent {
event: PodEvent::TurnEnded { pod_name },
body,
} => {
assert_eq!(pod_name, "child");
assert!(body.contains("`child`"));
}
other => panic!("unexpected: {other:?}"),
}
}
}

View File

@ -4,10 +4,10 @@ use session_store::{
FsStore, LogEntry, Store, TraceEntry, build_chain, collect_state, new_session_id,
};
#[tokio::test]
async fn round_trip_write_and_read() {
#[test]
fn round_trip_write_and_read() {
let dir = tempfile::tempdir().unwrap();
let store = FsStore::new(dir.path()).await.unwrap();
let store = FsStore::new(dir.path()).unwrap();
let id = new_session_id();
let raw = vec![
@ -23,9 +23,9 @@ async fn round_trip_write_and_read() {
ts: 2000,
segments: vec![protocol::Segment::text("Hello")],
},
LogEntry::AssistantItems {
LogEntry::AssistantItem {
ts: 3000,
items: vec![Item::assistant_message("Hi there!").into()],
item: Item::assistant_message("Hi there!").into(),
},
LogEntry::TurnEnd {
ts: 3100,
@ -41,11 +41,11 @@ async fn round_trip_write_and_read() {
// Write entries one by one
for entry in &entries {
store.append(id, entry).await.unwrap();
store.append(id, entry).unwrap();
}
// Read back
let read_back = store.read_all(id).await.unwrap();
let read_back = store.read_all(id).unwrap();
assert_eq!(read_back.len(), entries.len());
// Verify hashes survived round-trip
@ -64,10 +64,10 @@ async fn round_trip_write_and_read() {
assert!(state.head_hash.is_some());
}
#[tokio::test]
async fn create_session_writes_all_entries() {
#[test]
fn create_session_writes_all_entries() {
let dir = tempfile::tempdir().unwrap();
let store = FsStore::new(dir.path()).await.unwrap();
let store = FsStore::new(dir.path()).unwrap();
let id = new_session_id();
let entries = build_chain(&[LogEntry::SessionStart {
@ -82,22 +82,22 @@ async fn create_session_writes_all_entries() {
compacted_from: None,
}]);
store.create_session(id, &entries).await.unwrap();
let read_back = store.read_all(id).await.unwrap();
store.create_session(id, &entries).unwrap();
let read_back = store.read_all(id).unwrap();
assert_eq!(read_back.len(), 1);
let state = collect_state(&read_back);
assert_eq!(state.history.len(), 2);
}
#[tokio::test]
async fn list_sessions_returns_newest_first() {
#[test]
fn list_sessions_returns_newest_first() {
let dir = tempfile::tempdir().unwrap();
let store = FsStore::new(dir.path()).await.unwrap();
let store = FsStore::new(dir.path()).unwrap();
let id1 = new_session_id();
// Small delay to ensure different UUID v7 timestamps
tokio::time::sleep(std::time::Duration::from_millis(2)).await;
std::thread::sleep(std::time::Duration::from_millis(2));
let id2 = new_session_id();
let entries1 = build_chain(&[LogEntry::SessionStart {
@ -117,22 +117,22 @@ async fn list_sessions_returns_newest_first() {
compacted_from: None,
}]);
store.append(id1, &entries1[0]).await.unwrap();
store.append(id2, &entries2[0]).await.unwrap();
store.append(id1, &entries1[0]).unwrap();
store.append(id2, &entries2[0]).unwrap();
let sessions = store.list_sessions().await.unwrap();
let sessions = store.list_sessions().unwrap();
assert_eq!(sessions.len(), 2);
assert_eq!(sessions[0], id2); // newest first
assert_eq!(sessions[1], id1);
}
#[tokio::test]
async fn exists_returns_correct_state() {
#[test]
fn exists_returns_correct_state() {
let dir = tempfile::tempdir().unwrap();
let store = FsStore::new(dir.path()).await.unwrap();
let store = FsStore::new(dir.path()).unwrap();
let id = new_session_id();
assert!(!store.exists(id).await.unwrap());
assert!(!store.exists(id).unwrap());
let entries = build_chain(&[LogEntry::SessionStart {
ts: 1000,
@ -142,25 +142,25 @@ async fn exists_returns_correct_state() {
forked_from: None,
compacted_from: None,
}]);
store.append(id, &entries[0]).await.unwrap();
store.append(id, &entries[0]).unwrap();
assert!(store.exists(id).await.unwrap());
assert!(store.exists(id).unwrap());
}
#[tokio::test]
async fn not_found_error_for_missing_session() {
#[test]
fn not_found_error_for_missing_session() {
let dir = tempfile::tempdir().unwrap();
let store = FsStore::new(dir.path()).await.unwrap();
let store = FsStore::new(dir.path()).unwrap();
let id = new_session_id();
let result = store.read_all(id).await;
let result = store.read_all(id);
assert!(result.is_err());
}
#[tokio::test]
async fn trace_entries_in_separate_file() {
#[test]
fn trace_entries_in_separate_file() {
let dir = tempfile::tempdir().unwrap();
let store = FsStore::new(dir.path()).await.unwrap();
let store = FsStore::new(dir.path()).unwrap();
let id = new_session_id();
// Write a log entry
@ -172,7 +172,7 @@ async fn trace_entries_in_separate_file() {
forked_from: None,
compacted_from: None,
}]);
store.append(id, &entries[0]).await.unwrap();
store.append(id, &entries[0]).unwrap();
// Write a trace entry
let trace = TraceEntry {
@ -182,10 +182,10 @@ async fn trace_entries_in_separate_file() {
llm_worker::llm_client::event::PingEvent { timestamp: None },
),
};
store.append_trace(id, &trace).await.unwrap();
store.append_trace(id, &trace).unwrap();
// Log should have 1 entry, unaffected by trace
let log = store.read_all(id).await.unwrap();
let log = store.read_all(id).unwrap();
assert_eq!(log.len(), 1);
// Trace file should exist separately
@ -193,10 +193,10 @@ async fn trace_entries_in_separate_file() {
assert!(trace_path.exists());
}
#[tokio::test]
async fn read_head_hash_returns_last_entry_hash() {
#[test]
fn read_head_hash_returns_last_entry_hash() {
let dir = tempfile::tempdir().unwrap();
let store = FsStore::new(dir.path()).await.unwrap();
let store = FsStore::new(dir.path()).unwrap();
let id = new_session_id();
let entries = build_chain(&[
@ -215,9 +215,9 @@ async fn read_head_hash_returns_last_entry_hash() {
]);
for entry in &entries {
store.append(id, entry).await.unwrap();
store.append(id, entry).unwrap();
}
let head = store.read_head_hash(id).await.unwrap();
let head = store.read_head_hash(id).unwrap();
assert_eq!(head.as_ref(), Some(&entries[1].hash));
}

View File

@ -84,9 +84,9 @@ impl Interceptor for PausePolicy {
}
}
async fn make_store() -> (tempfile::TempDir, FsStore) {
fn make_store() -> (tempfile::TempDir, FsStore) {
let dir = tempfile::tempdir().unwrap();
let store = FsStore::new(dir.path()).await.unwrap();
let store = FsStore::new(dir.path()).unwrap();
(dir, store)
}
@ -108,7 +108,7 @@ async fn run_and_persist(
head_hash,
vec![protocol::Segment::text(input)],
)
.await
.unwrap();
let history_before = worker.history().len();
@ -119,10 +119,10 @@ async fn run_and_persist(
let new_items = &worker.history()[history_before..];
session_store::save_delta(store, session_id, head_hash, new_items)
.await
.unwrap();
session_store::save_turn_end(store, session_id, head_hash, worker.turn_count())
.await
.unwrap();
match &result {
@ -134,7 +134,7 @@ async fn run_and_persist(
r.clone(),
worker.last_run_interrupted(),
)
.await
.unwrap();
}
Err(e) => {
@ -145,7 +145,7 @@ async fn run_and_persist(
e.to_string(),
worker.last_run_interrupted(),
)
.await
.unwrap();
}
}
@ -160,7 +160,7 @@ async fn run_and_persist(
#[tokio::test]
async fn session_run_logs_entries() {
let (_dir, store) = make_store().await;
let (_dir, store) = make_store();
let client = MockLlmClient::new(simple_text_events());
let worker = Worker::new(client);
@ -172,14 +172,14 @@ async fn session_run_logs_entries() {
history: worker.history(),
},
)
.await
.unwrap();
let mut head_hash = Some(head_hash);
let (worker, _) = run_and_persist(worker, &store, sid, &mut head_hash, "Hi").await;
let _ = &worker;
let entries = store.read_all(sid).await.unwrap();
let entries = store.read_all(sid).unwrap();
// SessionStart, UserInput, AssistantItems, TurnEnd, RunCompleted (at minimum)
assert!(
@ -217,7 +217,7 @@ async fn session_run_logs_entries() {
#[tokio::test]
async fn session_restore_round_trip() {
let (_dir, store) = make_store().await;
let (_dir, store) = make_store();
let client = MockLlmClient::new(simple_text_events());
let mut worker = Worker::new(client);
worker.set_system_prompt("You are helpful.");
@ -230,7 +230,7 @@ async fn session_restore_round_trip() {
history: worker.history(),
},
)
.await
.unwrap();
let mut head_hash = Some(head_hash);
@ -240,7 +240,7 @@ async fn session_restore_round_trip() {
let original_turn_count = worker.turn_count();
// Restore
let state = session_store::restore(&store, sid).await.unwrap();
let state = session_store::restore(&store, sid).unwrap();
assert_eq!(state.history.len(), original_history_len);
assert_eq!(state.turn_count, original_turn_count);
@ -250,7 +250,7 @@ async fn session_restore_round_trip() {
#[tokio::test]
async fn session_run_with_tool_call() {
let (_dir, store) = make_store().await;
let (_dir, store) = make_store();
let client = MockLlmClient::with_responses(tool_call_events());
let mut worker = Worker::new(client);
worker.register_tool(weather_tool_definition());
@ -263,29 +263,29 @@ async fn session_run_with_tool_call() {
history: worker.history(),
},
)
.await
.unwrap();
let mut head_hash = Some(head_hash);
let (_worker, _) =
run_and_persist(worker, &store, sid, &mut head_hash, "What's the weather?").await;
let entries = store.read_all(sid).await.unwrap();
let entries = store.read_all(sid).unwrap();
let has_tool_results = entries
.iter()
.any(|e| matches!(&e.entry, LogEntry::ToolResults { .. }));
assert!(has_tool_results, "should have ToolResults entry");
.any(|e| matches!(&e.entry, LogEntry::ToolResult { .. }));
assert!(has_tool_results, "should have ToolResult entry");
let has_assistant = entries
.iter()
.any(|e| matches!(&e.entry, LogEntry::AssistantItems { .. }));
assert!(has_assistant, "should have AssistantItems entry");
.any(|e| matches!(&e.entry, LogEntry::AssistantItem { .. }));
assert!(has_assistant, "should have AssistantItem entry");
}
#[tokio::test]
async fn session_resume_after_pause() {
let (_dir, store) = make_store().await;
let (_dir, store) = make_store();
// First run: tool call with pause policy → Paused
let client = MockLlmClient::with_responses(tool_call_events());
@ -301,7 +301,7 @@ async fn session_resume_after_pause() {
history: worker.history(),
},
)
.await
.unwrap();
let mut head_hash = Some(head_hash);
@ -309,7 +309,7 @@ async fn session_resume_after_pause() {
assert!(matches!(result, llm_worker::WorkerResult::Paused));
// Check RunCompleted is Paused
let entries = store.read_all(sid).await.unwrap();
let entries = store.read_all(sid).unwrap();
let has_paused = entries.iter().any(|e| {
matches!(
&e.entry,
@ -322,13 +322,13 @@ async fn session_resume_after_pause() {
assert!(has_paused, "should have Paused outcome");
// Restore state and verify
let state = session_store::restore(&store, sid).await.unwrap();
let state = session_store::restore(&store, sid).unwrap();
assert!(state.last_run_interrupted);
}
#[tokio::test]
async fn session_fork_preserves_state() {
let (_dir, store) = make_store().await;
let (_dir, store) = make_store();
let client = MockLlmClient::new(simple_text_events());
let mut worker = Worker::new(client);
worker.set_system_prompt("System prompt");
@ -341,7 +341,7 @@ async fn session_fork_preserves_state() {
history: worker.history(),
},
)
.await
.unwrap();
let mut head_hash = Some(head_hash);
@ -356,11 +356,11 @@ async fn session_fork_preserves_state() {
history: worker.history(),
},
)
.await
.unwrap();
// Fork should have a SessionStart with the current history
let fork_entries = store.read_all(fork_id).await.unwrap();
let fork_entries = store.read_all(fork_id).unwrap();
assert_eq!(fork_entries.len(), 1);
assert!(matches!(
&fork_entries[0].entry,
@ -374,7 +374,7 @@ async fn session_fork_preserves_state() {
#[tokio::test]
async fn session_fork_at_truncates() {
let (_dir, store) = make_store().await;
let (_dir, store) = make_store();
let client = MockLlmClient::new(simple_text_events());
let worker = Worker::new(client);
@ -386,20 +386,20 @@ async fn session_fork_at_truncates() {
history: worker.history(),
},
)
.await
.unwrap();
let mut head_hash = Some(head_hash);
let (_worker, _) = run_and_persist(worker, &store, sid, &mut head_hash, "Hello").await;
let all_entries = store.read_all(sid).await.unwrap();
let all_entries = store.read_all(sid).unwrap();
assert!(all_entries.len() > 2);
// Fork at the hash of the 2nd entry (SessionStart + UserInput)
let at_hash = &all_entries[1].hash;
let fork_id = session_store::fork_at(&store, sid, at_hash).await.unwrap();
let fork_id = session_store::fork_at(&store, sid, at_hash).unwrap();
let fork_entries = store.read_all(fork_id).await.unwrap();
let fork_entries = store.read_all(fork_id).unwrap();
assert_eq!(fork_entries.len(), 1); // Just the new SessionStart
let fork_state = collect_state(&fork_entries);
@ -413,7 +413,7 @@ async fn session_fork_at_truncates() {
#[tokio::test]
async fn session_config_changed_logged() {
let (_dir, store) = make_store().await;
let (_dir, store) = make_store();
let client = MockLlmClient::new(vec![]);
let mut worker = Worker::new(client);
@ -425,7 +425,7 @@ async fn session_config_changed_logged() {
history: worker.history(),
},
)
.await
.unwrap();
let mut head_hash = Some(head_hash);
@ -433,10 +433,10 @@ async fn session_config_changed_logged() {
let new_config = RequestConfig::default().with_temperature(0.7);
worker.set_request_config(new_config.clone());
session_store::save_config_changed(&store, sid, &mut head_hash, &new_config)
.await
.unwrap();
let entries = store.read_all(sid).await.unwrap();
let entries = store.read_all(sid).unwrap();
let has_config_changed = entries.iter().any(|e| {
matches!(
&e.entry,
@ -448,7 +448,7 @@ async fn session_config_changed_logged() {
#[tokio::test]
async fn session_auto_forks_on_conflict() {
let (_dir, store) = make_store().await;
let (_dir, store) = make_store();
// Create a session
let client_a = MockLlmClient::new(simple_text_events());
@ -462,7 +462,7 @@ async fn session_auto_forks_on_conflict() {
history: worker_a.history(),
},
)
.await
.unwrap();
let mut session_id = original_sid;
let mut head_hash = Some(head_hash);
@ -472,14 +472,14 @@ async fn session_auto_forks_on_conflict() {
ts: 9999,
segments: vec![protocol::Segment::text("Interloper")],
};
let current_head = store.read_head_hash(original_sid).await.unwrap();
let current_head = store.read_head_hash(original_sid).unwrap();
let hash = session_store::compute_hash(current_head.as_ref(), &extra_entry);
let hashed = session_store::HashedEntry {
hash,
prev_hash: current_head,
entry: extra_entry,
};
store.append(original_sid, &hashed).await.unwrap();
store.append(original_sid, &hashed).unwrap();
// Now head_hash is stale — ensure_head_or_fork should auto-fork
session_store::ensure_head_or_fork(
@ -492,18 +492,18 @@ async fn session_auto_forks_on_conflict() {
history: worker_a.history(),
},
)
.await
.unwrap();
// session_id should now be different
assert_ne!(session_id, original_sid);
// The fork session should exist and have entries
let fork_entries = store.read_all(session_id).await.unwrap();
let fork_entries = store.read_all(session_id).unwrap();
assert!(!fork_entries.is_empty());
// Original session should still have the interloper entry
let original_entries = store.read_all(original_sid).await.unwrap();
let original_entries = store.read_all(original_sid).unwrap();
let has_interloper = original_entries
.iter()
.any(|e| matches!(&e.entry, LogEntry::UserInput { .. }));

View File

@ -483,21 +483,13 @@ impl App {
self.blocks.push(Block::UserMessage { segments });
self.assistant_streaming = false;
}
Event::Notify { message } => {
self.blocks.push(Block::Notify { message });
self.assistant_streaming = false;
}
Event::PodEvent(event) => {
self.blocks.push(Block::PodEvent { event });
self.assistant_streaming = false;
}
Event::SessionRotated { entry } => {
self.reset_for_rotation();
self.apply_log_entry_raw(&entry);
self.assistant_streaming = false;
}
Event::HookInjectedItems { entry } => {
self.apply_log_entry_raw(&entry);
Event::SystemItem { item } => {
self.apply_system_item(&item);
self.assistant_streaming = false;
}
Event::TurnStart { .. } => {
@ -975,6 +967,16 @@ impl App {
self.blocks.push(Block::UserMessage { segments });
}
}
session_store::LogEntry::AssistantItem { item, .. }
| session_store::LogEntry::ToolResult { item, .. } => {
let it: llm_worker::Item = item.into();
let item_value = serde_json::to_value(&it).expect("Item is Serialize");
self.push_history_item(&item_value);
}
session_store::LogEntry::SystemItem { item, .. } => {
let value = serde_json::to_value(&item).expect("SystemItem is Serialize");
self.apply_system_item(&value);
}
session_store::LogEntry::AssistantItems { items, .. }
| session_store::LogEntry::ToolResults { items, .. }
| session_store::LogEntry::HookInjectedItems { items, .. } => {
@ -984,11 +986,51 @@ impl App {
self.push_history_item(&item_value);
}
}
session_store::LogEntry::SystemItems { items, .. } => {
for system_item in items {
let value =
serde_json::to_value(&system_item).expect("SystemItem is Serialize");
self.apply_system_item(&value);
}
}
// Non-history-bearing variants don't affect the block view.
_ => {}
}
}
/// Dispatch one `SystemItem` JSON value into the appropriate block.
///
/// Kind-based routing replaces the old free-text `[Notification]` /
/// `[File: …]` parsing path: each kind maps directly to a typed
/// block (`Block::Notify`, `Block::PodEvent`, …).
fn apply_system_item(&mut self, value: &serde_json::Value) {
let Ok(item) = serde_json::from_value::<session_store::SystemItem>(value.clone()) else {
// Unknown / forward-compat shape: fall back to rendering the
// raw text payload (if any) as a generic system message.
if let Some(text) = value.get("body").and_then(|b| b.as_str()) {
self.task_store.apply_system_message_text(text);
self.blocks.push(Block::SystemMessage {
text: text.to_owned(),
});
}
return;
};
match item {
session_store::SystemItem::Notification { message, .. } => {
self.blocks.push(Block::Notify { message });
}
session_store::SystemItem::PodEvent { event, .. } => {
self.blocks.push(Block::PodEvent { event });
}
session_store::SystemItem::FileAttachment { body, .. }
| session_store::SystemItem::Knowledge { body, .. }
| session_store::SystemItem::Workflow { body, .. } => {
self.task_store.apply_system_message_text(&body);
self.blocks.push(Block::SystemMessage { text: body });
}
}
}
/// 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
@ -1422,18 +1464,14 @@ mod completion_flow_tests {
}
#[test]
fn live_hook_injected_items_event_appends_system_message_block() {
fn live_system_item_workflow_appends_system_message_block() {
let mut app = App::new("test".into());
let entry = serde_json::json!({
"kind": "hook_injected_items",
"ts": 1,
"items": [{
"kind": "message",
"role": "system",
"content": [{ "kind": "text", "text": "[Workflow /build]\nRun the build" }],
}],
let item = serde_json::json!({
"kind": "workflow",
"slug": "build",
"body": "[Workflow /build]\nRun the build",
});
app.handle_pod_event(Event::HookInjectedItems { entry });
app.handle_pod_event(Event::SystemItem { item });
assert!(matches!(
app.blocks.as_slice(),
@ -1441,6 +1479,39 @@ mod completion_flow_tests {
));
}
#[test]
fn live_system_item_notification_appends_notify_block() {
let mut app = App::new("test".into());
let item = serde_json::json!({
"kind": "notification",
"message": "hi",
"body": "[Notification] hi",
});
app.handle_pod_event(Event::SystemItem { item });
assert!(matches!(
app.blocks.as_slice(),
[Block::Notify { message }] if message == "hi"
));
}
#[test]
fn live_system_item_pod_event_appends_pod_event_block() {
let mut app = App::new("test".into());
let item = serde_json::json!({
"kind": "pod_event",
"event": { "kind": "turn_ended", "pod_name": "child" },
"body": "[Notification] pod `child` finished a turn",
});
app.handle_pod_event(Event::SystemItem { item });
assert_eq!(app.blocks.len(), 1);
match &app.blocks[0] {
Block::PodEvent {
event: protocol::PodEvent::TurnEnded { pod_name },
} => assert_eq!(pod_name, "child"),
_ => panic!("expected a PodEvent block"),
}
}
#[test]
fn compact_done_replaces_live_block() {
let mut app = App::new("test".into());
@ -1577,15 +1648,13 @@ 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::HookInjectedItems {
entry: serde_json::json!({
"kind": "hook_injected_items",
"ts": 1,
"items": [{
"kind": "message",
"role": "system",
"content": [{ "kind": "text", "text": snapshot }],
}],
// Snapshot text injected as a workflow body (kind doesn't matter
// for task-store parsing, only the text contents do).
app.handle_pod_event(Event::SystemItem {
item: serde_json::json!({
"kind": "workflow",
"slug": "task-snapshot",
"body": snapshot,
}),
});

View File

@ -80,14 +80,14 @@ struct Row {
}
pub async fn run() -> Result<PickerOutcome, PickerError> {
let store = open_default_store().await?;
let ids = store.list_sessions().await?;
let store = open_default_store()?;
let ids = store.list_sessions()?;
if ids.is_empty() {
return Err(PickerError::NoSessions);
}
let mut rows: Vec<Row> = Vec::with_capacity(MAX_ROWS);
for id in ids.into_iter().take(MAX_ROWS) {
let preview = build_preview(&store, id).await;
let preview = build_preview(&store, id);
// Best-effort live check. A pods.json I/O hiccup downgrades
// the row to "no badge" rather than killing the picker — the
// user still gets to see the listing.
@ -149,7 +149,7 @@ fn close_viewport(terminal: &mut Terminal<CrosstermBackend<io::Stdout>>) -> io::
Ok(())
}
async fn open_default_store() -> Result<FsStore, PickerError> {
fn open_default_store() -> Result<FsStore, PickerError> {
let dir = manifest::paths::sessions_dir().ok_or_else(|| {
PickerError::Io(io::Error::new(
io::ErrorKind::NotFound,
@ -157,11 +157,11 @@ async fn open_default_store() -> Result<FsStore, PickerError> {
(set INSOMNIA_HOME, INSOMNIA_DATA_DIR, or HOME)",
))
})?;
Ok(FsStore::new(&dir).await?)
Ok(FsStore::new(&dir)?)
}
async fn build_preview(store: &FsStore, id: SessionId) -> String {
match store.read_all(id).await {
fn build_preview(store: &FsStore, id: SessionId) -> String {
match store.read_all(id) {
Ok(entries) => last_message_preview(&entries).unwrap_or_else(|| "[empty]".to_string()),
Err(_) => "[corrupt]".to_string(),
}

View File

@ -328,8 +328,8 @@ async fn load_resume_scope(session_id: SessionId) -> Result<ScopeConfig, SpawnEr
"could not resolve sessions directory (set INSOMNIA_HOME, INSOMNIA_DATA_DIR, or HOME)",
)
})?;
let store = session_store::FsStore::new(&store_dir).await?;
let state = session_store::restore(&store, session_id).await?;
let store = session_store::FsStore::new(&store_dir)?;
let state = session_store::restore(&store, session_id)?;
let snapshot = state
.pod_scope
.ok_or(SpawnError::MissingResumeScope { session_id })?;

View File

@ -79,6 +79,23 @@ llm-worker は session 概念を持たない(`Worker` は `history` / `turn_co
- `SessionOrigin.at_hash``at_turn_index` (TurnEnd 由来) に置換。
- `ensure_head_or_fork` の検知ロジックは、Segment 末尾の terminal marker entry または末尾 seq 比較に置換(形式は実装時に決める)。
### 廃止前の足場 (前提)
本セクションを実装に移すタイミングでは、log writer が既に sync 化されていることを前提にする (`tickets/log-entry-singular-and-direct-commit.md`)。具体的には:
- `Store::append` / `read_all` 等が `std::fs` ベースの sync API
- `SessionLogWriter::append_entry()` が sync 関数
- `session_head` mutex は `parking_lot::Mutex` / `std::sync::Mutex`
- `LogCommand` / drain task / Flush バリアは既に撤廃済み
この状態で hash chain を廃止すると追加で取れる単純化:
- **`session_head` mutex そのものを撤去できる**。 hash chain が無いので「`head_hash` を直前 entry から取得して次に渡す」 という serialize 必須の依存が消える。 1 行 < `PIPE_BUF` (Linux 4KB) の `O_APPEND` write は kernel 側で atomic に直列化されるので、 user space で mutex を持つ必要が無い
- `session_head` が消えると Pod / interceptor / worker callback が writer ハンドルだけ持てば良くなる。 `Arc<SessionLogWriter>` は単に `Arc<Store> + sink` を抱えるだけの値で、 hot-path の競合がない
- `compute_hash` 呼び出しが消える分、 append が serialize + open + write + close の 3 syscall まで詰まる
つまり「sync 化」 が先に来て、 「hash 廃止」 で mutex まで消える、 という 2 段階の単純化になる。
## Fork: 2 種類の書き込み方
Session 境界の話ではなく **元 Segment への marker 書き込みの有無**で 2 種類を分ける。Session はどちらの場合も同じで、新 Segment が同 Session 内に生える。