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

Method::NotifyとEvent::Notificationが紛らわしい問題

Browse Source
This commit is contained in:
Hare
2026-04-26 23:25:50 +09:00
parent 82f08b966b
commit 2ee536ed71
18 changed files with 461 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/pod/src/ipc/alerter.rs
+168
View File
@@ -0,0 +1,168 @@
//! User-facing alert channel for Pod → client.
//!
//! Separate from `tracing` (which is for developer logs). Alerts
//! are short human-readable messages the Pod layer wants a client to
//! see — for example "compaction failed", "tool output truncated".
//!
//! Each alert is broadcast on the shared `Event` channel and
//! also appended to an in-memory buffer so that clients connecting
//! after the fact still see everything emitted during the session.
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::sync::broadcast;
use protocol::{Alert, AlertLevel, AlertSource, Event};
/// Upper bound on buffered alerts. When exceeded, the oldest
/// entries are discarded so a long-running session cannot leak
/// memory through a pathological loop of recurring alerts
/// (e.g. compaction failing every turn).
const MAX_BUFFERED_ALERTS: usize = 512;
#[derive(Clone)]
pub struct Alerter {
inner: Arc<Inner>,
}
struct Inner {
event_tx: broadcast::Sender<Event>,
buffer: Mutex<VecDeque<Alert>>,
}
impl Alerter {
pub fn new(event_tx: broadcast::Sender<Event>) -> Self {
Self {
inner: Arc::new(Inner {
event_tx,
buffer: Mutex::new(VecDeque::with_capacity(MAX_BUFFERED_ALERTS)),
}),
}
}
/// Record and broadcast an alert.
///
/// The broadcast may have no subscribers (e.g. during Pod
/// construction before any client has connected); the buffer
/// guarantees the message is still delivered once a client
/// attaches.
///
/// The buffer mutex is held across `broadcast::send` to make
/// `subscribe_with_snapshot` race-free — a client that snapshots
/// the buffer while holding the same lock sees every alert
/// exactly once: older ones from the snapshot, newer ones from
/// the freshly-subscribed receiver.
pub fn alert(&self, level: AlertLevel, source: AlertSource, message: String) {
let alert = Alert {
level,
source,
message,
timestamp_ms: now_ms(),
};
if let Ok(mut buf) = self.inner.buffer.lock() {
if buf.len() >= MAX_BUFFERED_ALERTS {
buf.pop_front();
}
buf.push_back(alert.clone());
let _ = self.inner.event_tx.send(Event::Alert(alert));
}
}
/// Subscribe and atomically snapshot the current buffer.
///
/// The returned snapshot contains alerts emitted before
/// this call; the receiver will deliver alerts emitted
/// after. An alert cannot appear in both.
pub fn subscribe_with_snapshot(&self) -> (Vec<Alert>, broadcast::Receiver<Event>) {
let buf = self
.inner
.buffer
.lock()
.expect("alerter buffer mutex poisoned");
let rx = self.inner.event_tx.subscribe();
let snapshot: Vec<Alert> = buf.iter().cloned().collect();
(snapshot, rx)
}
}
fn now_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn alert_broadcasts_to_existing_subscriber() {
let (tx, _keep) = broadcast::channel::<Event>(8);
let alerter = Alerter::new(tx);
let (_snapshot, mut rx) = alerter.subscribe_with_snapshot();
alerter.alert(
AlertLevel::Warn,
AlertSource::Compactor,
"test message".into(),
);
match rx.try_recv() {
Ok(Event::Alert(a)) => assert_eq!(a.message, "test message"),
other => panic!("unexpected event: {other:?}"),
}
}
#[test]
fn late_subscriber_sees_earlier_alerts_via_snapshot() {
let (tx, _keep) = broadcast::channel::<Event>(8);
let alerter = Alerter::new(tx);
alerter.alert(AlertLevel::Error, AlertSource::Pod, "first".into());
alerter.alert(AlertLevel::Warn, AlertSource::AgentsMd, "second".into());
let (snapshot, mut rx) = alerter.subscribe_with_snapshot();
assert_eq!(snapshot.len(), 2);
assert_eq!(snapshot[0].message, "first");
assert_eq!(snapshot[1].message, "second");
assert!(rx.try_recv().is_err()); // nothing pending on the receiver
}
#[test]
fn buffer_discards_oldest_past_cap() {
let (tx, _keep) = broadcast::channel::<Event>(1024);
let alerter = Alerter::new(tx);
for i in 0..(MAX_BUFFERED_ALERTS + 50) {
alerter.alert(AlertLevel::Warn, AlertSource::Worker, format!("msg-{i}"));
}
let (snapshot, _rx) = alerter.subscribe_with_snapshot();
assert_eq!(snapshot.len(), MAX_BUFFERED_ALERTS);
// First 50 were evicted; the oldest remaining is msg-50.
assert_eq!(snapshot.first().unwrap().message, "msg-50");
let last = format!("msg-{}", MAX_BUFFERED_ALERTS + 49);
assert_eq!(snapshot.last().unwrap().message, last);
}
#[test]
fn subscribe_snapshot_and_live_do_not_overlap() {
let (tx, _keep) = broadcast::channel::<Event>(8);
let alerter = Alerter::new(tx);
alerter.alert(AlertLevel::Warn, AlertSource::Worker, "historic".into());
let (snapshot, mut rx) = alerter.subscribe_with_snapshot();
alerter.alert(AlertLevel::Error, AlertSource::Worker, "live".into());
assert_eq!(snapshot.len(), 1);
assert_eq!(snapshot[0].message, "historic");
match rx.try_recv() {
Ok(Event::Alert(a)) => assert_eq!(a.message, "live"),
other => panic!("unexpected: {other:?}"),
}
assert!(rx.try_recv().is_err());
}
}
crates/pod/src/ipc/interceptor.rs
+16 -16
View File
@@ -25,7 +25,7 @@ use crate::hook::{
AbortInfo, HookRegistry, PreRequestInfo, PromptSubmitInfo, ToolCallSummary, ToolResultSummary,
TurnEndInfo,
};
use crate::ipc::notification_buffer::{NotificationBuffer, format_notification};
use crate::ipc::notify_buffer::{NotifyBuffer, format_notify};
use crate::prompt::catalog::PromptCatalog;
use crate::compact::token_counter::total_tokens_impl;
use tracing::warn;
@@ -42,7 +42,7 @@ pub(crate) struct PodInterceptor {
usage_history: Option<Arc<Mutex<Vec<UsageRecord>>>>,
/// Pending-notification buffer drained into the per-request
/// context at the head of `pre_llm_request`.
pending_notifications: NotificationBuffer,
pending_notifies: NotifyBuffer,
/// Prompt catalog used to render the injected notification wrapper.
prompts: Arc<PromptCatalog>,
/// Next turn index assigned by `on_prompt_submit`.
@@ -56,14 +56,14 @@ impl PodInterceptor {
registry: Arc<HookRegistry>,
compact_state: Option<Arc<CompactState>>,
usage_history: Option<Arc<Mutex<Vec<UsageRecord>>>>,
pending_notifications: NotificationBuffer,
pending_notifies: NotifyBuffer,
prompts: Arc<PromptCatalog>,
) -> Self {
Self {
registry,
compact_state,
usage_history,
pending_notifications,
pending_notifies,
prompts,
next_turn_index: AtomicUsize::new(0),
tool_calls_this_turn: AtomicUsize::new(0),
@@ -127,16 +127,16 @@ impl Interceptor for PodInterceptor {
// into the per-request context as transient system messages.
// These are not persisted to the Worker history; they exist only
// for this single LLM request.
for notification in self.pending_notifications.drain() {
match format_notification(&notification, &self.prompts) {
for n in self.pending_notifies.drain() {
match format_notify(&n, &self.prompts) {
Ok(item) => context.push(item),
Err(e) => {
// A render failure here would starve the LLM of the
// notification text. Fall back to the raw message —
// notify text. Fall back to the raw message —
// it still carries the intent, just without the
// wrapper phrasing.
warn!(error = %e, "failed to render notify_wrapper; using raw message");
context.push(Item::system_message(notification.message.clone()));
context.push(Item::system_message(n.message.clone()));
}
}
}
@@ -296,7 +296,7 @@ mod tests {
registry,
Some(state),
Some(history),
NotificationBuffer::new(),
NotifyBuffer::new(),
PromptCatalog::builtins_only().unwrap(),
);
let mut ctx = ctx_items;
@@ -320,7 +320,7 @@ mod tests {
registry,
Some(state),
Some(history),
NotificationBuffer::new(),
NotifyBuffer::new(),
PromptCatalog::builtins_only().unwrap(),
);
let mut ctx = ctx_items;
@@ -345,7 +345,7 @@ mod tests {
registry,
Some(state),
Some(history),
NotificationBuffer::new(),
NotifyBuffer::new(),
PromptCatalog::builtins_only().unwrap(),
);
let mut ctx = ctx_items;
@@ -364,7 +364,7 @@ mod tests {
registry,
None,
None,
NotificationBuffer::new(),
NotifyBuffer::new(),
PromptCatalog::builtins_only().unwrap(),
);
let mut ctx: Vec<Item> = Vec::new();
@@ -385,9 +385,9 @@ mod tests {
}
#[tokio::test]
async fn pre_llm_request_drains_pending_notifications_into_context() {
async fn pre_llm_request_drains_pending_notifies_into_context() {
let registry = Arc::new(HookRegistryBuilder::new().build());
let buffer = NotificationBuffer::new();
let buffer = NotifyBuffer::new();
buffer.push("first".into());
buffer.push("second".into());
@@ -419,7 +419,7 @@ mod tests {
// When compaction yields, notifications remain in the buffer for
// the next pre_llm_request (after compaction + resume).
let registry = Arc::new(HookRegistryBuilder::new().build());
let buffer = NotificationBuffer::new();
let buffer = NotifyBuffer::new();
buffer.push("msg".into());
let state = Arc::new(CompactState::new(None, Some(100), 2));
@@ -455,7 +455,7 @@ mod tests {
registry,
None,
None,
NotificationBuffer::new(),
NotifyBuffer::new(),
PromptCatalog::builtins_only().unwrap(),
);
let mut ctx: Vec<Item> = Vec::new();
crates/pod/src/ipc/mod.rs
+2 -2
View File
@@ -1,6 +1,6 @@
pub mod alerter;
pub mod event;
pub mod notifier;
pub mod server;
pub(crate) mod interceptor;
pub(crate) mod notification_buffer;
pub(crate) mod notify_buffer;
crates/pod/src/ipc/notifier.rs
-191
View File
@@ -1,191 +0,0 @@
//! User-facing notification channel for Pod → client.
//!
//! Separate from `tracing` (which is for developer logs). Notifications
//! are short human-readable messages the Pod layer wants a client to
//! see — for example "compaction failed", "tool output truncated".
//!
//! Each notification is broadcast on the shared `Event` channel and
//! also appended to an in-memory buffer so that clients connecting
//! after the fact still see everything emitted during the session.
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::sync::broadcast;
use protocol::{Event, Notification, NotificationLevel, NotificationSource};
/// Upper bound on buffered notifications. When exceeded, the oldest
/// entries are discarded so a long-running session cannot leak
/// memory through a pathological loop of recurring notifications
/// (e.g. compaction failing every turn).
const MAX_BUFFERED_NOTIFICATIONS: usize = 512;
#[derive(Clone)]
pub struct Notifier {
inner: Arc<Inner>,
}
struct Inner {
event_tx: broadcast::Sender<Event>,
buffer: Mutex<VecDeque<Notification>>,
}
impl Notifier {
pub fn new(event_tx: broadcast::Sender<Event>) -> Self {
Self {
inner: Arc::new(Inner {
event_tx,
buffer: Mutex::new(VecDeque::with_capacity(MAX_BUFFERED_NOTIFICATIONS)),
}),
}
}
/// Record and broadcast a notification.
///
/// The broadcast may have no subscribers (e.g. during Pod
/// construction before any client has connected); the buffer
/// guarantees the message is still delivered once a client
/// attaches.
///
/// The buffer mutex is held across `broadcast::send` to make
/// `subscribe_with_snapshot` race-free — a client that snapshots
/// the buffer while holding the same lock sees every notification
/// exactly once: older ones from the snapshot, newer ones from
/// the freshly-subscribed receiver.
pub fn notify(&self, level: NotificationLevel, source: NotificationSource, message: String) {
let notification = Notification {
level,
source,
message,
timestamp_ms: now_ms(),
};
if let Ok(mut buf) = self.inner.buffer.lock() {
if buf.len() >= MAX_BUFFERED_NOTIFICATIONS {
buf.pop_front();
}
buf.push_back(notification.clone());
let _ = self
.inner
.event_tx
.send(Event::Notification(notification));
}
}
/// Subscribe and atomically snapshot the current buffer.
///
/// The returned snapshot contains notifications emitted before
/// this call; the receiver will deliver notifications emitted
/// after. A notification cannot appear in both.
pub fn subscribe_with_snapshot(&self) -> (Vec<Notification>, broadcast::Receiver<Event>) {
let buf = self
.inner
.buffer
.lock()
.expect("notifier buffer mutex poisoned");
let rx = self.inner.event_tx.subscribe();
let snapshot: Vec<Notification> = buf.iter().cloned().collect();
(snapshot, rx)
}
}
fn now_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn notify_broadcasts_to_existing_subscriber() {
let (tx, _keep) = broadcast::channel::<Event>(8);
let notifier = Notifier::new(tx);
let (_snapshot, mut rx) = notifier.subscribe_with_snapshot();
notifier.notify(
NotificationLevel::Warn,
NotificationSource::Compactor,
"test message".into(),
);
match rx.try_recv() {
Ok(Event::Notification(n)) => assert_eq!(n.message, "test message"),
other => panic!("unexpected event: {other:?}"),
}
}
#[test]
fn late_subscriber_sees_earlier_notifications_via_snapshot() {
let (tx, _keep) = broadcast::channel::<Event>(8);
let notifier = Notifier::new(tx);
notifier.notify(
NotificationLevel::Error,
NotificationSource::Pod,
"first".into(),
);
notifier.notify(
NotificationLevel::Warn,
NotificationSource::AgentsMd,
"second".into(),
);
let (snapshot, mut rx) = notifier.subscribe_with_snapshot();
assert_eq!(snapshot.len(), 2);
assert_eq!(snapshot[0].message, "first");
assert_eq!(snapshot[1].message, "second");
assert!(rx.try_recv().is_err()); // nothing pending on the receiver
}
#[test]
fn buffer_discards_oldest_past_cap() {
let (tx, _keep) = broadcast::channel::<Event>(1024);
let notifier = Notifier::new(tx);
for i in 0..(MAX_BUFFERED_NOTIFICATIONS + 50) {
notifier.notify(
NotificationLevel::Warn,
NotificationSource::Worker,
format!("msg-{i}"),
);
}
let (snapshot, _rx) = notifier.subscribe_with_snapshot();
assert_eq!(snapshot.len(), MAX_BUFFERED_NOTIFICATIONS);
// First 50 were evicted; the oldest remaining is msg-50.
assert_eq!(snapshot.first().unwrap().message, "msg-50");
let last = format!("msg-{}", MAX_BUFFERED_NOTIFICATIONS + 49);
assert_eq!(snapshot.last().unwrap().message, last);
}
#[test]
fn subscribe_snapshot_and_live_do_not_overlap() {
let (tx, _keep) = broadcast::channel::<Event>(8);
let notifier = Notifier::new(tx);
notifier.notify(
NotificationLevel::Warn,
NotificationSource::Worker,
"historic".into(),
);
let (snapshot, mut rx) = notifier.subscribe_with_snapshot();
notifier.notify(
NotificationLevel::Error,
NotificationSource::Worker,
"live".into(),
);
assert_eq!(snapshot.len(), 1);
assert_eq!(snapshot[0].message, "historic");
match rx.try_recv() {
Ok(Event::Notification(n)) => assert_eq!(n.message, "live"),
other => panic!("unexpected: {other:?}"),
}
assert!(rx.try_recv().is_err());
}
}
crates/pod/src/ipc/notification_buffer.rs → crates/pod/src/ipc/notify_buffer.rs
+26 -29
View File
@@ -1,6 +1,6 @@
//! Pending-notification buffer for `Method::Notify`.
//! Pending-notify buffer for `Method::Notify`.
//!
//! Notifications are queued here by the Controller and drained by
//! Notify entries are queued here by the Controller and drained by
//! `PodInterceptor::pre_llm_request` into the per-request context
//! (never into the Worker's persistent history). Each queued entry
//! becomes one `Item::system_message` in the outgoing request.
@@ -13,56 +13,53 @@ use tracing::warn;
use crate::prompt::catalog::{CatalogError, PromptCatalog};
/// Maximum queued notifications. Oldest entries are dropped beyond this.
/// Maximum queued notify entries. Oldest entries are dropped beyond this.
const CAPACITY: usize = 128;
/// One pending notification awaiting injection into the next LLM request.
/// One pending notify entry awaiting injection into the next LLM request.
#[derive(Debug, Clone)]
pub struct PendingNotification {
pub struct PendingNotify {
pub message: String,
}
/// Shared, mutex-guarded buffer of pending notifications.
/// Shared, mutex-guarded buffer of pending notify entries.
///
/// Cloned between the Pod (producer) and PodInterceptor (consumer).
#[derive(Clone, Default)]
pub struct NotificationBuffer {
inner: Arc<Mutex<VecDeque<PendingNotification>>>,
pub struct NotifyBuffer {
inner: Arc<Mutex<VecDeque<PendingNotify>>>,
}
impl NotificationBuffer {
impl NotifyBuffer {
pub fn new() -> Self {
Self::default()
}
/// Push a notification onto the queue. If the queue is full, the
/// 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) {
let mut q = self.inner.lock().expect("notification buffer poisoned");
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()),
"notification buffer overflow; dropped oldest"
"notify buffer overflow; dropped oldest"
);
}
q.push_back(PendingNotification { message });
q.push_back(PendingNotify { message });
}
/// Remove and return all pending notifications in FIFO order.
pub fn drain(&self) -> Vec<PendingNotification> {
let mut q = self.inner.lock().expect("notification buffer poisoned");
/// Remove and return all pending notify 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 notifications. Primarily for tests.
/// Number of pending notify entries. Primarily for tests.
pub fn len(&self) -> usize {
self.inner
.lock()
.expect("notification buffer poisoned")
.len()
self.inner.lock().expect("notify buffer poisoned").len()
}
pub fn is_empty(&self) -> bool {
@@ -70,12 +67,12 @@ impl NotificationBuffer {
}
}
/// Format a single pending notification into the `Item::system_message`
/// Format a single pending notify entry into the `Item::system_message`
/// that gets injected into the per-request context. The wrapper body
/// comes from `PodPrompt::NotifyWrapper` so the surrounding phrasing
/// can be customised via a prompt pack (translation, tone, ...).
pub(crate) fn format_notification(
n: &PendingNotification,
pub(crate) fn format_notify(
n: &PendingNotify,
prompts: &PromptCatalog,
) -> Result<Item, CatalogError> {
let text = prompts.notify_wrapper(&n.message)?;
@@ -88,7 +85,7 @@ mod tests {
#[test]
fn push_then_drain_preserves_order() {
let buf = NotificationBuffer::new();
let buf = NotifyBuffer::new();
buf.push("one".into());
buf.push("two".into());
let drained = buf.drain();
@@ -100,7 +97,7 @@ mod tests {
#[test]
fn capacity_drops_oldest() {
let buf = NotificationBuffer::new();
let buf = NotifyBuffer::new();
for i in 0..(CAPACITY + 5) {
buf.push(format!("msg{i}"));
}
@@ -112,12 +109,12 @@ mod tests {
}
#[test]
fn format_notification_includes_message_and_nonblocking_hint() {
let n = PendingNotification {
fn format_notify_includes_message_and_nonblocking_hint() {
let n = PendingNotify {
message: "hello".into(),
};
let catalog = PromptCatalog::builtins_only().unwrap();
let item = format_notification(&n, &catalog).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"));
crates/pod/src/ipc/server.rs
+5 -9
View File
@@ -62,17 +62,13 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: PodHandle) {
let mut reader = JsonLineReader::new(reader);
let mut writer = JsonLineWriter::new(writer);
// Atomically subscribe and snapshot buffered notifications so that
// Atomically subscribe and snapshot buffered alerts so that
// warnings emitted before this client connected are replayed
// exactly once — they appear in the snapshot, and any notification
// exactly once — they appear in the snapshot, and any alert
// arriving afterwards reaches us through `rx`.
let (notification_snapshot, mut rx) = handle.notifier.subscribe_with_snapshot();
for notification in notification_snapshot {
if writer
.write(&Event::Notification(notification))
.await
.is_err()
{
let (alert_snapshot, mut rx) = handle.alerter.subscribe_with_snapshot();
for alert in alert_snapshot {
if writer.write(&Event::Alert(alert)).await.is_err() {
return;
}
}