Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

update: SystemItem１本化

Browse Source
This commit is contained in:
Hare
2026-05-14 14:36:29 +09:00
parent e4b66345aa
commit f9def2d5bb
14 changed files with 752 additions and 379 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/pod/src/controller.rs
+78 -45
View File
@@ -411,24 +411,17 @@ where
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");
}
commit_via_drain(&ctx, entry).await;
}
LogCommand::SystemItems(items) => {
if items.is_empty() {
continue;
}
let entry = LogEntry::SystemItems {
ts: session_log::now_millis(),
items,
};
commit_via_drain(&ctx, entry).await;
}
LogCommand::Flush(ack) => {
let _ = ack.send(());
@@ -437,15 +430,52 @@ where
}
}
/// 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.
async fn commit_via_drain<St>(ctx: &LogDrainHandle<St>, entry: LogEntry)
where
St: session_store::Store + Clone + Send + 'static,
{
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");
}
}
}
/// Map one LLM-driven worker-history append to its `LogEntry` form.
///
/// `None` is the skip signal for items that the drain must not commit:
/// - `user_message` items are committed by `Pod::run` up-front as
/// `LogEntry::UserInput { segments }`.
/// - `system_message` items are committed by `PodInterceptor` as part
/// of a `LogEntry::SystemItems` batch (with typed kind metadata)
/// before they reach the worker's history.
fn classify_history_item(item: Item) -> Option<LogEntry> {
let ts = session_log::now_millis();
if item.is_user_message() {
return None;
}
if matches!(
item,
Item::Message {
role: llm_worker::Role::System,
..
}
) {
return None;
}
if item.is_tool_result() {
return Some(LogEntry::ToolResults {
ts,
@@ -458,7 +488,9 @@ fn classify_history_item(item: Item) -> Option<LogEntry> {
items: vec![session_store::LoggedItem::from(&item)],
});
}
Some(LogEntry::HookInjectedItems {
// Defensive: anything else (future Item kinds) routes through
// AssistantItems rather than getting silently dropped.
Some(LogEntry::AssistantItems {
ts,
items: vec![session_store::LoggedItem::from(&item)],
})
@@ -696,9 +728,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::SystemItems` 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 +785,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::SystemItems` entry and the sink fans it
// out 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 +801,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 +937,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 +960,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(());
crates/pod/src/ipc/interceptor.rs
+72 -19
View File
@@ -22,11 +22,15 @@ use tracing::info;
use tracing::warn;
use crate::compact::state::CompactState;
use session_store::SystemItem;
use tokio::sync::mpsc;
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::LogCommand;
use crate::prompt::catalog::PromptCatalog;
use llm_worker::token_counter::total_tokens;
@@ -45,13 +49,20 @@ 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 a
/// `LogEntry::SystemItems` through `log_cmd_tx`, 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>,
/// Sender into the Pod's history-drain task. The interceptor uses
/// it to commit `LogCommand::SystemItems` batches before returning
/// the corresponding `Item::system_message`s up to the worker.
/// `None` in tests / `Pod::new` paths where no drain is wired.
log_cmd_tx: Option<mpsc::UnboundedSender<LogCommand>>,
/// 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_cmd_tx: Option<mpsc::UnboundedSender<LogCommand>>,
) -> Self {
Self {
registry,
@@ -74,11 +86,26 @@ impl PodInterceptor {
pending_notifies,
pending_attachments,
prompts,
log_cmd_tx,
next_turn_index: AtomicUsize::new(0),
tool_calls_this_turn: AtomicUsize::new(0),
}
}
/// Send a `LogCommand::SystemItems` batch down the drain channel
/// (no-op if no drain is wired). The drain task commits the entry
/// before the corresponding `Item::system_message`s reach the
/// worker via `ContinueWith` / `pending_history_appends`, so the
/// drain barrier in `persist_turn` covers system commits too.
fn send_system_items(&self, items: Vec<SystemItem>) {
if items.is_empty() {
return;
}
if let Some(tx) = self.log_cmd_tx.as_ref() {
let _ = tx.send(LogCommand::SystemItems(items));
}
}
fn current_turn_index(&self) -> usize {
self.next_turn_index
.load(Ordering::Relaxed)
@@ -111,7 +138,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 +147,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. The
// drain task processes the `SystemItems` command BEFORE
// any subsequent `Item` commands from `on_history_append`,
// so on-disk order matches worker-history order.
let items: Vec<Item> = extras.iter().map(SystemItem::to_history_item).collect();
self.send_system_items(extras);
PromptAction::ContinueWith(items)
}
}
@@ -129,19 +163,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 SystemItems 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.send_system_items(system_items);
items
}
@@ -321,6 +367,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 +393,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 +420,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 +441,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 +464,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 +474,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 +502,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 +511,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 +541,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;
crates/pod/src/ipc/notify_buffer.rs
+105 -50
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::SystemItems` 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:?}"),
}
}
}
crates/pod/src/ipc/server.rs
+27 -10
View File
@@ -104,22 +104,39 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: PodHandle) {
entry = entry_rx.recv() => {
match entry {
Ok(entry) => {
let value = serde_json::to_value(&entry)
.expect("LogEntry is Serialize");
let outbound = match &entry {
let outbound = match entry {
session_store::LogEntry::SessionStart { .. } => {
Some(Event::SessionRotated { entry: value })
let value = serde_json::to_value(&entry)
.expect("LogEntry is Serialize");
vec![Event::SessionRotated { entry: value }]
}
session_store::LogEntry::HookInjectedItems { .. } => {
Some(Event::HookInjectedItems { entry: value })
session_store::LogEntry::SystemItems { items, .. } => {
// Fan out per-item so each `SystemItem`
// arrives as its own `Event::SystemItem`
// on the wire. Batching on disk is an
// implementation detail of the drain
// task; clients see them one at a time.
items
.into_iter()
.map(|si| {
let value = serde_json::to_value(&si)
.expect("SystemItem is Serialize");
Event::SystemItem { item: value }
})
.collect()
}
// Defensive: should never reach here per
// `SessionLogSink::is_live_relevant`.
_ => None,
_ => Vec::new(),
};
if let Some(event) = outbound
&& writer.write(&event).await.is_err()
{
let mut hit_error = false;
for event in outbound {
if writer.write(&event).await.is_err() {
hit_error = true;
break;
}
}
if hit_error {
break;
}
}
crates/pod/src/pod.rs
+65 -27
View File
@@ -9,8 +9,8 @@ use llm_worker::llm_client::client::LlmClient;
use llm_worker::state::Mutable;
use llm_worker::{ToolOutputLimits, UsageRecord, Worker, WorkerError, WorkerResult};
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};
@@ -18,16 +18,21 @@ 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.
/// - `Item`: one worker-history append observed via
/// `Worker::on_history_append`; the drain classifies it into
/// `LogEntry::AssistantItems` / `LogEntry::ToolResults` and commits
/// through the sink. `role:system` items are explicitly skipped
/// because they are committed up-front through `SystemItems`.
/// - `SystemItems`: typed agent-injected items committed as a single
/// `LogEntry::SystemItems` entry. Used by the interceptor when it
/// drains the notify buffer or pending attachments.
/// - `Flush(ack)`: barrier used by `persist_turn` to ensure every
/// queued command has been processed before the trailing `TurnEnd`
/// entry lands.
#[derive(Debug)]
pub enum LogCommand {
Item(Item),
SystemItems(Vec<SystemItem>),
Flush(tokio::sync::oneshot::Sender<()>),
}
@@ -158,7 +163,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
@@ -279,7 +284,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(),
@@ -378,7 +383,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,
@@ -760,7 +765,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 +907,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
self.pending_notifies.clone(),
self.pending_attachments.clone(),
self.prompts.clone(),
self.log_cmd_tx.clone(),
);
self.worker_mut().set_interceptor(interceptor);
self.interceptor_installed = true;
@@ -1099,7 +1115,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 +1126,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 +1151,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 +1184,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 +1201,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 +1275,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 +1310,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)
}
@@ -2635,7 +2673,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,
@@ -2708,7 +2746,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,
@@ -2852,7 +2890,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,
crates/pod/src/session_log_sink.rs
+13 -12
View File
@@ -120,7 +120,7 @@ impl SessionLogSink {
fn is_live_relevant(entry: &LogEntry) -> bool {
matches!(
entry,
LogEntry::SessionStart { .. } | LogEntry::HookInjectedItems { .. }
LogEntry::SessionStart { .. } | LogEntry::SystemItems { .. }
)
}
@@ -427,12 +427,13 @@ mod tests {
assert!(rx.try_recv().is_err());
}
fn hook_injected(text: &str) -> LogEntry {
LogEntry::HookInjectedItems {
fn notification_entry(text: &str) -> LogEntry {
LogEntry::SystemItems {
ts: now_millis(),
items: vec![session_store::LoggedItem::from(
&llm_worker::Item::system_message(text),
)],
items: vec![session_store::SystemItem::Notification {
message: text.to_owned(),
body: format!("[Notification] {text}"),
}],
}
}
@@ -448,11 +449,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"));
// SystemItems is live-relevant.
sink.publish(notification_entry("hi"));
match rx.try_recv() {
Ok(LogEntry::HookInjectedItems { .. }) => {}
other => panic!("expected HookInjectedItems, got {other:?}"),
Ok(LogEntry::SystemItems { .. }) => {}
other => panic!("expected SystemItems, got {other:?}"),
}
// Mirror still grew with both entries (snapshot completeness).
@@ -465,11 +466,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::SystemItems { .. }) => {}
other => panic!("unexpected: {other:?}"),
}
assert!(rx.try_recv().is_err());
crates/pod/tests/controller_test.rs
+63 -35
View File
@@ -34,6 +34,9 @@ fn history_from_sink(handle: &PodHandle) -> Vec<Item> {
| 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()));
}
_ => {}
}
}
@@ -745,16 +748,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 +762,28 @@ 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::SystemItems { items, .. }
if items.iter().any(|si| matches!(
si,
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 +838,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 +852,28 @@ 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::SystemItems { items, .. }
if items.iter().any(|si| matches!(
si,
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 +931,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 +939,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 +947,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 +1050,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 +1084,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,