diff --git a/docs/content/docs/(configuration)/config.mdx b/docs/content/docs/(configuration)/config.mdx index d239814a9..7d2f2aafb 100644 --- a/docs/content/docs/(configuration)/config.mdx +++ b/docs/content/docs/(configuration)/config.mdx @@ -503,7 +503,7 @@ Thresholds are fractions of `context_window`. | Key | Type | Default | Description | |-----|------|---------|-------------| -| `tick_interval_secs` | integer | 30 | How often the cortex checks system state | +| `tick_interval_secs` | integer | 30 | How often the cortex runtime loop runs maintenance ticks while continuously observing events | | `worker_timeout_secs` | integer | 300 | Worker timeout before cancellation | | `branch_timeout_secs` | integer | 60 | Branch timeout before cancellation | | `circuit_breaker_threshold` | integer | 3 | Consecutive failures before auto-disable | @@ -517,7 +517,7 @@ Thresholds are fractions of `context_window`. | `refresh_secs` | integer | 900 | Seconds between background warmup passes | | `startup_delay_secs` | integer | 5 | Delay before first warmup pass after boot | -When warmup is enabled, it is the primary bulletin refresh path. The cortex bulletin loop remains as a fallback generator when warmup is disabled or when the cached bulletin is stale (`bulletin_age_secs >= max(1, warmup.refresh_secs)`). +When warmup is enabled, it is the primary bulletin refresh path. The cortex runtime loop still performs fallback bulletin/profile refresh when warmup is disabled or when the cached bulletin is stale (`bulletin_age_secs >= max(1, warmup.refresh_secs)`). Dispatch readiness is derived from warmup runtime state: diff --git a/docs/content/docs/(core)/architecture.mdx b/docs/content/docs/(core)/architecture.mdx index 029bbabaa..5b930fc87 100644 --- a/docs/content/docs/(core)/architecture.mdx +++ b/docs/content/docs/(core)/architecture.mdx @@ -83,7 +83,12 @@ For detailed coverage of each process type, see [Agents](/docs/agents), [Compact ## Inter-Process Communication -All processes within an agent communicate through a `broadcast::channel` -- a multi-producer, multi-consumer event bus. The channel, all branches, and all workers share the same bus. +Each agent uses two `broadcast::channel` buses: + +- `event_tx` -- control/lifecycle events shared by channel, branches, workers, compactor, and UI streams +- `memory_event_tx` -- memory-save telemetry consumed by the cortex (`MemorySaved` events only) + +This split keeps high-volume memory writes off the control bus so channel control events are less likely to lag under load. ### Event Types @@ -96,7 +101,7 @@ All processes within an agent communicate through a `broadcast::channel DB > default` and support hot-reload. +**Wired through config:** +- `tick_interval_secs` and `bulletin_interval_secs` are read by the running cortex loop and hot-reload during runtime. +- `worker_timeout_secs`, `branch_timeout_secs`, and `circuit_breaker_threshold` remain future-facing for Phase 2 supervision work. **Referenced in prompts but don't exist:** - `memory_consolidate` tool - `system_monitor` tool -**Event bus:** -- 12 `ProcessEvent` variants on a `broadcast::Sender` per agent. -- `MemorySaved` and `CompactionTriggered` variants are defined but never emitted by any code. +**Event buses:** +- Two per-agent `broadcast::Sender` streams: + - `event_tx` for control/lifecycle events (channel, branch, worker, compactor, task/link events) + - `memory_event_tx` for `MemorySaved` telemetry emitted by `memory_save` +- `CompactionTriggered` is emitted by the compactor on `event_tx` when thresholds are reached. -## Phase 1: The Tick Loop +## Phase 1: The Tick Loop (Implemented) Get the cortex running as a persistent process that observes the event bus and ticks on an interval. Purely programmatic — no LLM. -### Wire missing event emission +### Wire missing event emission (done) - Emit `MemorySaved` from `memory_save` tool after successful save - Emit `CompactionTriggered` from the compactor when thresholds are hit -### Instantiate the cortex +### Instantiate the cortex (done) -- Call `Cortex::new()` in `main.rs` alongside `spawn_bulletin_loop()` -- Subscribe to the event bus via `deps.event_tx.subscribe()` +- Call `Cortex::new()` from `spawn_cortex_loop()` during agent startup +- Subscribe to both buses via `deps.event_tx.subscribe()` and `deps.memory_event_tx.subscribe()` - Run a `tokio::select!` loop: - - Receive events → feed through `observe()` + - Receive control events → feed through `observe()` + - Receive memory events (`MemorySaved`) → feed through `observe()` - Tick on `cortex_config.tick_interval_secs` - Move bulletin generation into the cortex's tick loop (currently a standalone free function) -### Fix `observe()` to extract real values +### Fix `observe()` to extract real values (done) - Map all 12 `ProcessEvent` variants, not just 3 - Pull real `memory_type`, `importance`, content summaries from events - Enrich `MemorySaved` event variant with `memory_type` and `importance` fields so the cortex gets useful data without querying the store -### Rework `Signal` enum +### Rework `Signal` enum (done) - Align variants with what `ProcessEvent` actually provides - Add `WorkerStarted`, `BranchStarted`, `WorkerStatus` diff --git a/src/agent.rs b/src/agent.rs index 033379ab4..ee8aa0175 100644 --- a/src/agent.rs +++ b/src/agent.rs @@ -15,23 +15,6 @@ mod invariant_harness; pub mod status; pub mod worker; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum EventRecvDisposition { - Continue { lagged_count: Option }, - Stop, -} - -pub(crate) fn classify_event_recv_error( - error: &tokio::sync::broadcast::error::RecvError, -) -> EventRecvDisposition { - match error { - tokio::sync::broadcast::error::RecvError::Lagged(count) => EventRecvDisposition::Continue { - lagged_count: Some(*count), - }, - tokio::sync::broadcast::error::RecvError::Closed => EventRecvDisposition::Stop, - } -} - pub(crate) fn extract_last_assistant_text(history: &[rig::message::Message]) -> Option { for message in history.iter().rev() { if let rig::message::Message::Assistant { content, .. } = message { diff --git a/src/agent/channel.rs b/src/agent/channel.rs index e62dad550..f943563de 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -50,6 +50,29 @@ struct PendingResult { success: bool, } +const EVENT_LAG_WARNING_INTERVAL_SECS: u64 = 30; + +async fn recv_channel_event( + event_rx: &mut broadcast::Receiver, +) -> crate::BroadcastRecvResult { + crate::classify_broadcast_recv_result(event_rx.recv().await) +} + +fn should_process_event_for_channel(event: &ProcessEvent, channel_id: &ChannelId) -> bool { + event_is_for_channel(event, channel_id) +} + +fn should_flush_coalesce_buffer_for_event(event: &ProcessEvent) -> bool { + matches!( + event, + ProcessEvent::BranchStarted { .. } + | ProcessEvent::BranchResult { .. } + | ProcessEvent::WorkerStarted { .. } + | ProcessEvent::WorkerStatus { .. } + | ProcessEvent::WorkerComplete { .. } + ) +} + /// Shared state that channel tools need to act on the channel. /// /// Wrapped in Arc and passed to tools (branch, spawn_worker, route, cancel) @@ -644,6 +667,8 @@ impl Channel { /// Run the channel event loop. pub async fn run(mut self) -> Result<()> { tracing::info!(channel_id = %self.id, "channel started"); + let mut lagged_events_since_warning: u64 = 0; + let mut last_lag_warning: Option = None; loop { // Compute next deadline from coalesce and retrigger timers @@ -680,28 +705,52 @@ impl Channel { } } } - event = self.event_rx.recv() => { + event = recv_channel_event(&mut self.event_rx) => { match event { - Ok(event) => { - // Events bypass coalescing - flush buffer first if needed - if let Err(error) = self.flush_coalesce_buffer().await { - tracing::error!(%error, channel_id = %self.id, "error flushing coalesce buffer"); + crate::BroadcastRecvResult::Event(event) => { + if !should_process_event_for_channel(&event, &self.id) { + continue; + } + // Worker/branch lifecycle events bypass coalescing. + if should_flush_coalesce_buffer_for_event(&event) + && let Err(error) = self.flush_coalesce_buffer().await + { + tracing::error!( + %error, + channel_id = %self.id, + "error flushing coalesce buffer" + ); } if let Err(error) = self.handle_event(event).await { tracing::error!(%error, channel_id = %self.id, "error handling event"); } } - Err(error) => { - match super::classify_event_recv_error(&error) { - super::EventRecvDisposition::Continue { .. } => { - tracing::debug!(channel_id = %self.id, %error, "event receiver lagged, continuing channel loop"); - } - super::EventRecvDisposition::Stop => { - tracing::info!(channel_id = %self.id, %error, "event receiver closed, stopping channel loop"); - break; - } + crate::BroadcastRecvResult::Lagged(skipped) => { + #[cfg(feature = "metrics")] + crate::telemetry::Metrics::global() + .event_receiver_lagged_events_total + .with_label_values(&[&*self.deps.agent_id, "channel_control"]) + .inc_by(skipped); + + if let Some(skipped) = crate::drain_lag_warning_count( + &mut lagged_events_since_warning, + &mut last_lag_warning, + skipped, + std::time::Duration::from_secs( + EVENT_LAG_WARNING_INTERVAL_SECS, + ), + ) { + tracing::warn!( + channel_id = %self.id, + skipped, + "channel event receiver lagged, dropping old events" + ); } } + crate::BroadcastRecvResult::Closed => { + tracing::info!(channel_id = %self.id, "channel event bus closed, stopping channel"); + break; + } } } _ = tokio::time::sleep(sleep_duration), if next_deadline.is_some() => { @@ -1921,7 +1970,6 @@ impl Channel { if !event_is_for_channel(&event, &self.id) { return Ok(()); } - // Update status block { let mut status = self.state.status_block.write().await; @@ -2289,3 +2337,106 @@ impl Channel { } } } + +#[cfg(test)] +mod tests { + use super::{recv_channel_event, should_process_event_for_channel}; + use crate::memory::MemoryType; + use crate::{AgentId, ChannelId, ProcessEvent, ProcessId}; + use std::sync::Arc; + + #[tokio::test] + async fn channel_event_loop_continues_after_lagged_broadcast() { + let (event_tx, mut event_rx) = tokio::sync::broadcast::channel::(2); + let agent_id: AgentId = Arc::from("agent"); + let channel_id: ChannelId = Arc::from("channel"); + let process_id = ProcessId::Channel(channel_id); + + for status in ["one", "two", "three"] { + event_tx + .send(ProcessEvent::StatusUpdate { + agent_id: agent_id.clone(), + process_id: process_id.clone(), + status: status.to_string(), + }) + .ok(); + } + + let first = recv_channel_event(&mut event_rx).await; + assert!( + matches!(first, crate::BroadcastRecvResult::Lagged(skipped) if skipped > 0), + "expected lagged receive, got {first:?}" + ); + + let second = recv_channel_event(&mut event_rx).await; + assert!( + matches!( + second, + crate::BroadcastRecvResult::Event(ProcessEvent::StatusUpdate { .. }) + ), + "expected next event after lagged receive, got {second:?}" + ); + } + + #[tokio::test] + async fn channel_event_loop_stops_when_event_bus_closes() { + let (event_tx, mut event_rx) = tokio::sync::broadcast::channel::(2); + drop(event_tx); + + let event = recv_channel_event(&mut event_rx).await; + assert!(matches!(event, crate::BroadcastRecvResult::Closed)); + } + + #[test] + fn channel_coalesce_ignores_unrelated_memory_saved_events() { + let channel_id: ChannelId = Arc::from("channel-a"); + let event = ProcessEvent::MemorySaved { + agent_id: Arc::from("agent"), + memory_id: "memory-1".to_string(), + channel_id: Some(Arc::from("channel-b")), + memory_type: MemoryType::Fact, + importance: 0.8, + content_summary: "saved memory".to_string(), + }; + + assert!(!should_process_event_for_channel(&event, &channel_id)); + } + + #[test] + fn channel_coalesce_ignores_unrelated_compaction_events() { + let channel_id: ChannelId = Arc::from("channel-a"); + let event = ProcessEvent::CompactionTriggered { + agent_id: Arc::from("agent"), + channel_id: Arc::from("channel-b"), + threshold_reached: 0.85, + }; + + assert!(!should_process_event_for_channel(&event, &channel_id)); + } + + #[test] + fn channel_coalesce_processes_related_worker_events() { + let channel_id: ChannelId = Arc::from("channel-a"); + let event = ProcessEvent::WorkerStatus { + agent_id: Arc::from("agent"), + worker_id: uuid::Uuid::new_v4(), + channel_id: Some(channel_id.clone()), + status: "running".to_string(), + }; + + assert!(should_process_event_for_channel(&event, &channel_id)); + } + + #[test] + fn channel_coalesce_processes_related_branch_events() { + let channel_id: ChannelId = Arc::from("channel-a"); + let event = ProcessEvent::BranchResult { + agent_id: Arc::from("agent"), + branch_id: uuid::Uuid::new_v4(), + channel_id: channel_id.clone(), + conclusion: "done".to_string(), + }; + + assert!(should_process_event_for_channel(&event, &channel_id)); + } +} diff --git a/src/agent/channel_dispatch.rs b/src/agent/channel_dispatch.rs index d406271dd..ab88dc074 100644 --- a/src/agent/channel_dispatch.rs +++ b/src/agent/channel_dispatch.rs @@ -224,6 +224,7 @@ async fn spawn_branch( state.deps.task_store.clone(), state.deps.memory_search.clone(), state.deps.runtime_config.clone(), + state.deps.memory_event_tx.clone(), state.conversation_logger.clone(), state.channel_store.clone(), crate::conversation::ProcessRunLogger::new(state.deps.sqlite_pool.clone()), diff --git a/src/agent/channel_history.rs b/src/agent/channel_history.rs index ceab77e03..820ef6086 100644 --- a/src/agent/channel_history.rs +++ b/src/agent/channel_history.rs @@ -352,35 +352,74 @@ pub(crate) fn extract_message_id(message: &InboundMessage) -> Option { /// channel's workers would leak into sibling channels (e.g. threads). pub(crate) fn event_is_for_channel(event: &ProcessEvent, channel_id: &ChannelId) -> bool { match event { - ProcessEvent::BranchResult { + ProcessEvent::BranchStarted { + channel_id: event_channel, + .. + } + | ProcessEvent::BranchResult { channel_id: event_channel, .. } => event_channel == channel_id, - ProcessEvent::WorkerComplete { + ProcessEvent::WorkerStarted { channel_id: event_channel, .. - } => event_channel.as_ref() == Some(channel_id), - ProcessEvent::WorkerStatus { + } + | ProcessEvent::WorkerComplete { + channel_id: event_channel, + .. + } + | ProcessEvent::WorkerStatus { + channel_id: event_channel, + .. + } + | ProcessEvent::ToolStarted { + channel_id: event_channel, + .. + } + | ProcessEvent::ToolCompleted { + channel_id: event_channel, + .. + } + | ProcessEvent::MemorySaved { + channel_id: event_channel, + .. + } + | ProcessEvent::WorkerPermission { + channel_id: event_channel, + .. + } + | ProcessEvent::WorkerQuestion { channel_id: event_channel, .. } => event_channel.as_ref() == Some(channel_id), + ProcessEvent::CompactionTriggered { + channel_id: event_channel, + .. + } + | ProcessEvent::AgentMessageSent { + channel_id: event_channel, + .. + } + | ProcessEvent::AgentMessageReceived { + channel_id: event_channel, + .. + } => event_channel == channel_id, ProcessEvent::TextDelta { channel_id: event_channel, .. } => event_channel.as_ref() == Some(channel_id), - // Status block updates, tool events, etc. — match on agent_id which - // is already filtered by the event bus subscription. Let them through. - _ => true, + ProcessEvent::StatusUpdate { .. } | ProcessEvent::TaskUpdated { .. } => false, } } #[cfg(test)] mod tests { use super::{apply_history_after_turn, event_is_for_channel}; - use crate::ProcessEvent; + use crate::{ChannelId, ProcessEvent, ProcessId}; use rig::completion::{CompletionError, PromptError}; use rig::message::Message; use rig::tool::ToolSetError; + use std::sync::Arc; fn user_msg(text: &str) -> Message { Message::User { @@ -1028,13 +1067,58 @@ mod tests { ); } + #[test] + fn event_filter_scopes_tool_events_by_channel() { + let channel_id: ChannelId = Arc::from("channel-a"); + let other_channel: ChannelId = Arc::from("channel-b"); + let process_id = ProcessId::Worker(uuid::Uuid::new_v4()); + + let related_event = ProcessEvent::ToolStarted { + agent_id: Arc::from("agent"), + process_id: process_id.clone(), + channel_id: Some(channel_id.clone()), + tool_name: "memory_save".to_string(), + args: "{}".to_string(), + }; + let unrelated_event = ProcessEvent::ToolStarted { + agent_id: Arc::from("agent"), + process_id, + channel_id: Some(other_channel), + tool_name: "memory_save".to_string(), + args: "{}".to_string(), + }; + + assert!(event_is_for_channel(&related_event, &channel_id)); + assert!(!event_is_for_channel(&unrelated_event, &channel_id)); + } + + #[test] + fn event_filter_scopes_agent_message_events_by_channel() { + let channel_id: ChannelId = Arc::from("channel-a"); + let related_event = ProcessEvent::AgentMessageReceived { + from_agent_id: Arc::from("agent-a"), + to_agent_id: Arc::from("agent-b"), + link_id: "link-1".to_string(), + channel_id: channel_id.clone(), + }; + let unrelated_event = ProcessEvent::AgentMessageReceived { + from_agent_id: Arc::from("agent-a"), + to_agent_id: Arc::from("agent-b"), + link_id: "link-1".to_string(), + channel_id: Arc::from("channel-b"), + }; + + assert!(event_is_for_channel(&related_event, &channel_id)); + assert!(!event_is_for_channel(&unrelated_event, &channel_id)); + } + #[test] fn text_delta_events_are_filtered_by_channel_id() { - let target_channel: crate::ChannelId = std::sync::Arc::from("webchat:target"); + let target_channel: ChannelId = Arc::from("webchat:target"); let matching_event = ProcessEvent::TextDelta { - agent_id: std::sync::Arc::from("agent"), - process_id: crate::ProcessId::Channel(target_channel.clone()), + agent_id: Arc::from("agent"), + process_id: ProcessId::Channel(target_channel.clone()), channel_id: Some(target_channel.clone()), text_delta: "hel".to_string(), aggregated_text: "hel".to_string(), @@ -1042,17 +1126,17 @@ mod tests { assert!(event_is_for_channel(&matching_event, &target_channel)); let other_event = ProcessEvent::TextDelta { - agent_id: std::sync::Arc::from("agent"), - process_id: crate::ProcessId::Channel(std::sync::Arc::from("webchat:other")), - channel_id: Some(std::sync::Arc::from("webchat:other")), + agent_id: Arc::from("agent"), + process_id: ProcessId::Channel(Arc::from("webchat:other")), + channel_id: Some(Arc::from("webchat:other")), text_delta: "hel".to_string(), aggregated_text: "hello".to_string(), }; assert!(!event_is_for_channel(&other_event, &target_channel)); let unscoped_event = ProcessEvent::TextDelta { - agent_id: std::sync::Arc::from("agent"), - process_id: crate::ProcessId::Channel(std::sync::Arc::from("webchat:none")), + agent_id: Arc::from("agent"), + process_id: ProcessId::Channel(Arc::from("webchat:none")), channel_id: None, text_delta: "hel".to_string(), aggregated_text: "hello".to_string(), diff --git a/src/agent/compactor.rs b/src/agent/compactor.rs index 120f4d612..04567da89 100644 --- a/src/agent/compactor.rs +++ b/src/agent/compactor.rs @@ -11,7 +11,7 @@ use crate::{AgentDeps, ChannelId, ProcessId, ProcessType}; use rig::agent::AgentBuilder; use rig::completion::CompletionModel; use rig::message::{AssistantContent, Message, UserContent}; -use rig::tool::server::{ToolServer, ToolServerHandle}; +use rig::tool::server::ToolServerHandle; use std::sync::Arc; use tokio::sync::RwLock; use uuid::Uuid; @@ -72,6 +72,21 @@ impl Compactor { ?action, "compaction triggered" ); + if let Err(error) = self + .deps + .event_tx + .send(crate::ProcessEvent::CompactionTriggered { + agent_id: self.deps.agent_id.clone(), + channel_id: self.channel_id.clone(), + threshold_reached: usage, + }) + { + tracing::debug!( + channel_id = %self.channel_id, + %error, + "failed to emit compaction-triggered event" + ); + } match action { CompactionAction::EmergencyTruncate => { @@ -212,11 +227,11 @@ async fn run_compaction( .with_routing((**routing).clone()); // Give the compaction worker memory_save so it can directly persist memories - let tool_server: ToolServerHandle = ToolServer::new() - .tool(crate::tools::MemorySaveTool::new( - deps.memory_search.clone(), - )) - .run(); + let tool_server: ToolServerHandle = crate::tools::create_cortex_tool_server( + deps.agent_id.clone(), + deps.memory_event_tx.clone(), + deps.memory_search.clone(), + ); let agent = AgentBuilder::new(model) .preamble(compactor_prompt) diff --git a/src/agent/cortex.rs b/src/agent/cortex.rs index 4fca18133..e8b6266e3 100644 --- a/src/agent/cortex.rs +++ b/src/agent/cortex.rs @@ -17,16 +17,19 @@ use crate::llm::SpacebotModel; use crate::memory::search::{SearchConfig, SearchMode, SearchSort}; use crate::memory::types::{Association, MemoryType, RelationType}; use crate::tasks::{TaskStatus, UpdateTaskInput}; -use crate::{AgentDeps, ProcessEvent, ProcessType}; +use crate::{ + AgentDeps, AgentId, BranchId, ChannelId, ProcessEvent, ProcessId, ProcessType, WorkerId, +}; use rig::agent::AgentBuilder; use rig::completion::{CompletionModel, Prompt, TypedPrompt}; use serde::Serialize; use sqlx::{Row as _, SqlitePool}; +use std::collections::VecDeque; use std::sync::Arc; use std::time::{Duration, Instant}; -use tokio::sync::RwLock; +use tokio::sync::{RwLock, broadcast}; fn update_warmup_status(deps: &AgentDeps, update: F) where @@ -73,6 +76,63 @@ fn should_generate_bulletin_from_bulletin_loop( age_secs >= warmup_config.refresh_secs.max(1) } +const SIGNAL_BUFFER_CAPACITY: usize = 100; +const BULLETIN_REFRESH_FAILURE_BACKOFF_BASE_SECS: u64 = 30; +const BULLETIN_REFRESH_FAILURE_BACKOFF_MAX_SECS: u64 = 600; +const BULLETIN_REFRESH_CIRCUIT_OPEN_THRESHOLD: u32 = 3; +const BULLETIN_REFRESH_CIRCUIT_OPEN_SECS: u64 = 1800; + +fn bulletin_refresh_failure_backoff(consecutive_failures: u32) -> Duration { + let exponent = consecutive_failures.saturating_sub(1).min(5); + let multiplier = 1_u64 << exponent; + let seconds = BULLETIN_REFRESH_FAILURE_BACKOFF_BASE_SECS + .saturating_mul(multiplier) + .min(BULLETIN_REFRESH_FAILURE_BACKOFF_MAX_SECS); + Duration::from_secs(seconds) +} + +fn record_bulletin_refresh_failure( + bulletin_refresh_failures: &mut u32, + bulletin_refresh_circuit_open: &mut bool, + next_bulletin_refresh_allowed_at: &mut Instant, + now: Instant, +) -> (Duration, bool) { + *bulletin_refresh_failures = bulletin_refresh_failures.saturating_add(1); + let backoff = bulletin_refresh_failure_backoff(*bulletin_refresh_failures); + *next_bulletin_refresh_allowed_at = now + backoff; + + let mut circuit_opened = false; + if *bulletin_refresh_failures >= BULLETIN_REFRESH_CIRCUIT_OPEN_THRESHOLD { + if !*bulletin_refresh_circuit_open { + *bulletin_refresh_circuit_open = true; + circuit_opened = true; + } + let circuit_cooldown = Duration::from_secs(BULLETIN_REFRESH_CIRCUIT_OPEN_SECS); + let circuit_recovery_at = now + circuit_cooldown; + if circuit_recovery_at > *next_bulletin_refresh_allowed_at { + *next_bulletin_refresh_allowed_at = circuit_recovery_at; + } + } + + (backoff, circuit_opened) +} + +fn maybe_close_bulletin_refresh_circuit( + bulletin_refresh_failures: &mut u32, + bulletin_refresh_circuit_open: &mut bool, + next_bulletin_refresh_allowed_at: &mut Instant, + now: Instant, +) -> bool { + if !*bulletin_refresh_circuit_open || now < *next_bulletin_refresh_allowed_at { + return false; + } + + *bulletin_refresh_failures = 0; + *bulletin_refresh_circuit_open = false; + *next_bulletin_refresh_allowed_at = now; + true +} + fn has_completed_initial_warmup(status: &crate::config::WarmupStatus) -> bool { status.last_refresh_unix_ms.is_some() && matches!(status.state, crate::config::WarmupState::Warm) @@ -140,7 +200,7 @@ async fn maybe_generate_bulletin_under_lock( warmup_config: &arc_swap::ArcSwap, warmup_status: &arc_swap::ArcSwap, generate: F, -) -> bool +) -> BulletinRefreshOutcome where F: FnOnce() -> Fut, Fut: std::future::Future, @@ -152,7 +212,11 @@ where let refresh_secs = warmup_config.refresh_secs.max(1); if should_generate_bulletin_from_bulletin_loop(warmup_config, &status) { - generate().await + if generate().await { + BulletinRefreshOutcome::Generated + } else { + BulletinRefreshOutcome::Failed + } } else { tracing::debug!( warmup_enabled = warmup_config.enabled, @@ -160,7 +224,24 @@ where refresh_secs, "skipping bulletin loop generation because warmup bulletin is fresh" ); - true + BulletinRefreshOutcome::SkippedFresh + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum BulletinRefreshOutcome { + Generated, + SkippedFresh, + Failed, +} + +impl BulletinRefreshOutcome { + fn is_success(self) -> bool { + !matches!(self, Self::Failed) + } + + fn generated(self) -> bool { + matches!(self, Self::Generated) } } @@ -169,7 +250,7 @@ pub struct Cortex { pub deps: AgentDeps, pub hook: CortexHook, /// Recent activity signals (rolling window). - pub signal_buffer: Arc>>, + pub signal_buffer: Arc>>, /// System prompt loaded from prompts/CORTEX.md. pub system_prompt: String, } @@ -177,30 +258,106 @@ pub struct Cortex { /// A high-level activity signal (not raw conversation). #[derive(Debug, Clone)] pub enum Signal { - /// Channel started. - ChannelStarted { channel_id: String }, - /// Channel ended. - ChannelEnded { channel_id: String }, + /// Branch started. + BranchStarted { + branch_id: BranchId, + channel_id: ChannelId, + description: String, + }, + /// Branch produced a result. + BranchResult { + branch_id: BranchId, + channel_id: ChannelId, + conclusion: String, + }, + /// Worker started. + WorkerStarted { + worker_id: WorkerId, + channel_id: Option, + task_summary: String, + worker_type: String, + }, + /// Worker status update. + WorkerStatus { + worker_id: WorkerId, + channel_id: Option, + status: String, + }, + /// Worker completed. + WorkerCompleted { + worker_id: WorkerId, + channel_id: Option, + success: bool, + result_summary: String, + }, + /// Tool execution started. + ToolStarted { + process_id: ProcessId, + channel_id: Option, + tool_name: String, + }, + /// Tool execution completed. + ToolCompleted { + process_id: ProcessId, + channel_id: Option, + tool_name: String, + result_summary: String, + }, /// Memory was saved. MemorySaved { - memory_type: String, + memory_id: String, + channel_id: Option, + memory_type: MemoryType, content_summary: String, importance: f32, }, - /// Worker completed. - WorkerCompleted { - task_summary: String, - result_summary: String, + /// Compaction threshold was reached. + CompactionTriggered { + channel_id: ChannelId, + threshold_reached: f32, + }, + /// Generic status update. + StatusUpdate { + process_id: ProcessId, + status: String, + }, + /// Worker requested a permission decision. + WorkerPermission { + worker_id: WorkerId, + channel_id: Option, + permission_id: String, + description: String, + }, + /// Worker asked one or more questions. + WorkerQuestion { + worker_id: WorkerId, + channel_id: Option, + question_id: String, + question_count: usize, }, - /// Compaction occurred. - Compaction { - channel_id: String, - turns_compacted: i64, + /// Agent sent a linked message. + AgentMessageSent { + from_agent_id: AgentId, + to_agent_id: AgentId, + channel_id: ChannelId, }, - /// Error occurred. - Error { - component: String, - error_summary: String, + /// Agent received a linked message. + AgentMessageReceived { + from_agent_id: AgentId, + to_agent_id: AgentId, + channel_id: ChannelId, + }, + /// Task lifecycle update. + TaskUpdated { + task_number: i64, + status: String, + action: String, + }, + /// Streaming text delta emitted by a process. + TextDelta { + process_id: ProcessId, + channel_id: Option, + text_summary: String, }, } @@ -333,68 +490,375 @@ impl Cortex { Self { deps, hook, - signal_buffer: Arc::new(RwLock::new(Vec::with_capacity(100))), + signal_buffer: Arc::new(RwLock::new(VecDeque::with_capacity(SIGNAL_BUFFER_CAPACITY))), system_prompt: system_prompt.into(), } } /// Process a process event and extract signals. pub async fn observe(&self, event: ProcessEvent) { - let signal = match &event { - ProcessEvent::MemorySaved { memory_id, .. } => Some(Signal::MemorySaved { - memory_type: "unknown".into(), - content_summary: format!("memory {}", memory_id), - importance: 0.5, - }), - ProcessEvent::WorkerComplete { result, .. } => Some(Signal::WorkerCompleted { - task_summary: "completed task".into(), - result_summary: result.lines().next().unwrap_or("done").into(), - }), - ProcessEvent::CompactionTriggered { - channel_id, - threshold_reached, - .. - } => Some(Signal::Compaction { - channel_id: channel_id.to_string(), - turns_compacted: (*threshold_reached * 100.0) as i64, - }), - _ => None, - }; - - if let Some(signal) = signal { + let signal = signal_from_event(event); + let buffer_len = { let mut buffer = self.signal_buffer.write().await; - buffer.push(signal); - - if buffer.len() > 100 { - buffer.remove(0); - } + push_signal_into_buffer(&mut buffer, signal); + buffer.len() + }; - tracing::debug!("cortex received signal, buffer size: {}", buffer.len()); - } + tracing::trace!(buffer_len, "cortex received signal"); } /// Run periodic consolidation (future: health monitoring, memory maintenance). pub async fn run_consolidation(&self) -> Result<()> { - tracing::info!("cortex running consolidation"); + tracing::debug!("cortex running consolidation"); Ok(()) } } -/// Spawn the cortex bulletin loop for an agent. -/// -/// Runs bulletin/profile maintenance on a configurable interval. +fn summarize_signal_text(value: &str) -> String { + crate::summarize_first_non_empty_line(value, crate::EVENT_SUMMARY_MAX_CHARS) +} + +fn signal_from_event(event: ProcessEvent) -> Signal { + match event { + ProcessEvent::BranchStarted { + branch_id, + channel_id, + description, + .. + } => Signal::BranchStarted { + branch_id, + channel_id, + description: summarize_signal_text(&description), + }, + ProcessEvent::BranchResult { + branch_id, + channel_id, + conclusion, + .. + } => Signal::BranchResult { + branch_id, + channel_id, + conclusion: summarize_signal_text(&conclusion), + }, + ProcessEvent::WorkerStarted { + worker_id, + channel_id, + task, + worker_type, + .. + } => Signal::WorkerStarted { + worker_id, + channel_id, + task_summary: summarize_signal_text(&task), + worker_type, + }, + ProcessEvent::WorkerStatus { + worker_id, + channel_id, + status, + .. + } => Signal::WorkerStatus { + worker_id, + channel_id, + status: summarize_signal_text(&status), + }, + ProcessEvent::WorkerComplete { + worker_id, + channel_id, + result, + success, + .. + } => Signal::WorkerCompleted { + worker_id, + channel_id, + success, + result_summary: summarize_signal_text(&result), + }, + ProcessEvent::ToolStarted { + process_id, + channel_id, + tool_name, + .. + } => Signal::ToolStarted { + process_id, + channel_id, + tool_name, + }, + ProcessEvent::ToolCompleted { + process_id, + channel_id, + tool_name, + result, + .. + } => Signal::ToolCompleted { + process_id, + channel_id, + tool_name, + result_summary: summarize_signal_text(&result), + }, + ProcessEvent::MemorySaved { + memory_id, + channel_id, + memory_type, + importance, + content_summary, + .. + } => Signal::MemorySaved { + memory_id, + channel_id, + memory_type, + content_summary, + importance, + }, + ProcessEvent::CompactionTriggered { + channel_id, + threshold_reached, + .. + } => Signal::CompactionTriggered { + channel_id, + threshold_reached, + }, + ProcessEvent::StatusUpdate { + process_id, status, .. + } => Signal::StatusUpdate { + process_id, + status: summarize_signal_text(&status), + }, + ProcessEvent::WorkerPermission { + worker_id, + channel_id, + permission_id, + description, + .. + } => Signal::WorkerPermission { + worker_id, + channel_id, + permission_id, + description: summarize_signal_text(&description), + }, + ProcessEvent::WorkerQuestion { + worker_id, + channel_id, + question_id, + questions, + .. + } => Signal::WorkerQuestion { + worker_id, + channel_id, + question_id, + question_count: questions.len(), + }, + ProcessEvent::AgentMessageSent { + from_agent_id, + to_agent_id, + channel_id, + .. + } => Signal::AgentMessageSent { + from_agent_id, + to_agent_id, + channel_id, + }, + ProcessEvent::AgentMessageReceived { + from_agent_id, + to_agent_id, + channel_id, + .. + } => Signal::AgentMessageReceived { + from_agent_id, + to_agent_id, + channel_id, + }, + ProcessEvent::TaskUpdated { + task_number, + status, + action, + .. + } => Signal::TaskUpdated { + task_number, + status: summarize_signal_text(&status), + action, + }, + ProcessEvent::TextDelta { + process_id, + channel_id, + text_delta, + .. + } => Signal::TextDelta { + process_id, + channel_id, + text_summary: summarize_signal_text(&text_delta), + }, + } +} + +fn push_signal_into_buffer(buffer: &mut VecDeque, signal: Signal) { + if let Some(previous) = buffer.back_mut() + && coalesce_signal(previous, &signal) + { + return; + } + + buffer.push_back(signal); + if buffer.len() > SIGNAL_BUFFER_CAPACITY { + buffer.pop_front(); + } +} + +fn coalesce_signal(previous: &mut Signal, next: &Signal) -> bool { + match (previous, next) { + ( + Signal::StatusUpdate { + process_id: previous_process_id, + status: previous_status, + }, + Signal::StatusUpdate { + process_id: next_process_id, + status: next_status, + }, + ) if previous_process_id == next_process_id => { + *previous_status = next_status.clone(); + true + } + ( + Signal::WorkerStatus { + worker_id: previous_worker_id, + channel_id: previous_channel_id, + status: previous_status, + }, + Signal::WorkerStatus { + worker_id: next_worker_id, + channel_id: next_channel_id, + status: next_status, + }, + ) if previous_worker_id == next_worker_id && previous_channel_id == next_channel_id => { + *previous_status = next_status.clone(); + true + } + ( + Signal::TaskUpdated { + task_number: previous_task_number, + status: previous_status, + action: previous_action, + }, + Signal::TaskUpdated { + task_number: next_task_number, + status: next_status, + action: next_action, + }, + ) if previous_task_number == next_task_number => { + *previous_status = next_status.clone(); + *previous_action = next_action.clone(); + true + } + ( + Signal::TextDelta { + process_id: previous_process_id, + channel_id: previous_channel_id, + text_summary: previous_text_summary, + }, + Signal::TextDelta { + process_id: next_process_id, + channel_id: next_channel_id, + text_summary: next_text_summary, + }, + ) if previous_process_id == next_process_id && previous_channel_id == next_channel_id => { + *previous_text_summary = next_text_summary.clone(); + true + } + _ => false, + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ReceiverClosedBehavior { + StopLoop, + DisableStream, +} + +#[derive(Debug, Clone)] +enum CortexReceiverOutcome { + Observe(ProcessEvent), + Lagged { dropped: u64 }, + StopLoop, + DisableStream, +} + +fn handle_cortex_receiver_result( + result: std::result::Result, + receiver_name: &'static str, + close_behavior: ReceiverClosedBehavior, + lagged_since_last_warning: &mut u64, + last_lag_warning: &mut Option, + warning_interval_secs: u64, +) -> CortexReceiverOutcome { + match crate::classify_broadcast_recv_result(result) { + crate::BroadcastRecvResult::Event(event) => CortexReceiverOutcome::Observe(event), + crate::BroadcastRecvResult::Lagged(count) => { + if let Some(dropped) = crate::drain_lag_warning_count( + lagged_since_last_warning, + last_lag_warning, + count, + Duration::from_secs(warning_interval_secs), + ) { + tracing::warn!( + receiver = receiver_name, + dropped, + "cortex event receiver lagged, dropping old events" + ); + } + CortexReceiverOutcome::Lagged { dropped: count } + } + crate::BroadcastRecvResult::Closed => match close_behavior { + ReceiverClosedBehavior::StopLoop => { + tracing::warn!( + receiver = receiver_name, + "cortex event bus closed, stopping cortex loop" + ); + CortexReceiverOutcome::StopLoop + } + ReceiverClosedBehavior::DisableStream => { + tracing::warn!( + receiver = receiver_name, + "cortex memory event bus closed, continuing without memory events" + ); + CortexReceiverOutcome::DisableStream + } + }, + } +} + +/// Spawn the cortex runtime loop for an agent. /// -/// When warmup is enabled, warmup is the primary bulletin refresher and this -/// loop skips duplicate bulletin synthesis while the cached bulletin is fresh. -/// When warmup is disabled (or stale), this loop generates the bulletin. -pub fn spawn_bulletin_loop(deps: AgentDeps, logger: CortexLogger) -> tokio::task::JoinHandle<()> { +/// The loop observes process events and runs periodic cortex maintenance ticks. +/// Bulletin generation and profile refresh happen inside this tick loop. +pub fn spawn_cortex_loop(deps: AgentDeps, logger: CortexLogger) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { - if let Err(error) = run_bulletin_loop(&deps, &logger).await { - tracing::error!(%error, "cortex bulletin loop exited with error"); + let prompt_engine = deps.runtime_config.prompts.load(); + let system_prompt = match prompt_engine.render_static("cortex") { + Ok(prompt) => prompt, + Err(error) => { + tracing::warn!(%error, "failed to render cortex prompt, using empty preamble"); + String::new() + } + }; + drop(prompt_engine); + + let cortex = Cortex::new(deps.clone(), system_prompt); + let mut event_rx = deps.event_tx.subscribe(); + let mut memory_event_rx = deps.memory_event_tx.subscribe(); + if let Err(error) = + run_cortex_loop(&cortex, &logger, &mut event_rx, &mut memory_event_rx).await + { + tracing::error!(%error, "cortex loop exited with error"); } }) } +/// Backwards-compatible alias while callers migrate to `spawn_cortex_loop`. +pub fn spawn_bulletin_loop(deps: AgentDeps, logger: CortexLogger) -> tokio::task::JoinHandle<()> { + spawn_cortex_loop(deps, logger) +} + /// Spawn the warmup loop for an agent. /// /// Warmup runs asynchronously and never blocks channel responsiveness. @@ -554,23 +1018,48 @@ pub fn trigger_forced_warmup(deps: AgentDeps, dispatch_type: &'static str) { }); } -async fn run_bulletin_loop(deps: &AgentDeps, logger: &CortexLogger) -> anyhow::Result<()> { - tracing::info!("cortex bulletin loop started"); +fn spawn_bulletin_refresh_task( + deps: AgentDeps, + logger: CortexLogger, +) -> tokio::task::JoinHandle { + tokio::spawn(async move { + let bulletin_outcome = maybe_generate_bulletin_under_lock( + deps.runtime_config.warmup_lock.as_ref(), + &deps.runtime_config.warmup, + &deps.runtime_config.warmup_status, + || generate_bulletin(&deps, &logger), + ) + .await; + if bulletin_outcome.generated() { + generate_profile(&deps, &logger).await; + } + bulletin_outcome + }) +} + +async fn run_cortex_loop( + cortex: &Cortex, + logger: &CortexLogger, + event_rx: &mut broadcast::Receiver, + memory_event_rx: &mut broadcast::Receiver, +) -> anyhow::Result<()> { + tracing::info!("cortex loop started"); const MAX_RETRIES: u32 = 3; const RETRY_DELAY_SECS: u64 = 15; + const LAG_WARNING_INTERVAL_SECS: u64 = 30; - // Run immediately on startup, with retries + // Run bulletin generation immediately on startup, with retries. for attempt in 0..=MAX_RETRIES { - let bulletin_ok = maybe_generate_bulletin_under_lock( - deps.runtime_config.warmup_lock.as_ref(), - &deps.runtime_config.warmup, - &deps.runtime_config.warmup_status, - || generate_bulletin(deps, logger), + let bulletin_outcome = maybe_generate_bulletin_under_lock( + cortex.deps.runtime_config.warmup_lock.as_ref(), + &cortex.deps.runtime_config.warmup, + &cortex.deps.runtime_config.warmup_status, + || generate_bulletin(&cortex.deps, logger), ) .await; - if bulletin_ok { + if bulletin_outcome.is_success() { break; } if attempt < MAX_RETRIES { @@ -592,23 +1081,197 @@ async fn run_bulletin_loop(deps: &AgentDeps, logger: &CortexLogger) -> anyhow::R } } - // Generate initial profile after bulletin - generate_profile(deps, logger).await; + // Generate an initial profile after startup bulletin synthesis. + generate_profile(&cortex.deps, logger).await; + let mut last_bulletin_refresh = Instant::now(); + let mut tick_interval_secs = cortex + .deps + .runtime_config + .cortex + .load() + .tick_interval_secs + .max(1); + let mut tick_period = Duration::from_secs(tick_interval_secs); + let mut tick_timer = + tokio::time::interval_at(tokio::time::Instant::now() + tick_period, tick_period); + tick_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut lagged_since_last_warning_control: u64 = 0; + let mut last_lag_warning_control: Option = None; + let mut lagged_since_last_warning_memory: u64 = 0; + let mut last_lag_warning_memory: Option = None; + let mut memory_event_stream_open = true; + let mut refresh_task: Option> = None; + let mut bulletin_refresh_failures: u32 = 0; + let mut bulletin_refresh_circuit_open = false; + let mut next_bulletin_refresh_allowed_at = Instant::now(); loop { - let cortex_config = **deps.runtime_config.cortex.load(); - let interval = cortex_config.bulletin_interval_secs; + tokio::select! { + event = event_rx.recv() => { + match handle_cortex_receiver_result( + event, + "control", + ReceiverClosedBehavior::StopLoop, + &mut lagged_since_last_warning_control, + &mut last_lag_warning_control, + LAG_WARNING_INTERVAL_SECS, + ) { + CortexReceiverOutcome::Observe(event) => cortex.observe(event).await, + CortexReceiverOutcome::Lagged { dropped } => { + #[cfg(feature = "metrics")] + crate::telemetry::Metrics::global() + .event_receiver_lagged_events_total + .with_label_values(&[&*cortex.deps.agent_id, "cortex_control"]) + .inc_by(dropped); + #[cfg(not(feature = "metrics"))] + let _ = dropped; + } + CortexReceiverOutcome::StopLoop => { + if let Some(task) = refresh_task.take() { + task.abort(); + } + return Ok(()); + } + CortexReceiverOutcome::DisableStream => unreachable!("control stream cannot disable itself"), + } + }, + event = memory_event_rx.recv(), if memory_event_stream_open => { + match handle_cortex_receiver_result( + event, + "memory", + ReceiverClosedBehavior::DisableStream, + &mut lagged_since_last_warning_memory, + &mut last_lag_warning_memory, + LAG_WARNING_INTERVAL_SECS, + ) { + CortexReceiverOutcome::Observe(event) => cortex.observe(event).await, + CortexReceiverOutcome::Lagged { dropped } => { + #[cfg(feature = "metrics")] + crate::telemetry::Metrics::global() + .event_receiver_lagged_events_total + .with_label_values(&[&*cortex.deps.agent_id, "cortex_memory"]) + .inc_by(dropped); + #[cfg(not(feature = "metrics"))] + let _ = dropped; + } + CortexReceiverOutcome::StopLoop => { + if let Some(task) = refresh_task.take() { + task.abort(); + } + return Ok(()); + } + CortexReceiverOutcome::DisableStream => { + memory_event_stream_open = false; + } + } + }, + _ = tick_timer.tick() => { + if let Err(error) = cortex.run_consolidation().await { + tracing::warn!(%error, "cortex consolidation tick failed"); + } - tokio::time::sleep(Duration::from_secs(interval)).await; + if refresh_task + .as_ref() + .is_some_and(tokio::task::JoinHandle::is_finished) + && let Some(task) = refresh_task.take() + { + match task.await { + Ok(outcome) => { + let now = Instant::now(); + if outcome.is_success() { + last_bulletin_refresh = now; + bulletin_refresh_failures = 0; + bulletin_refresh_circuit_open = false; + next_bulletin_refresh_allowed_at = now; + } else { + let (backoff, circuit_opened) = record_bulletin_refresh_failure( + &mut bulletin_refresh_failures, + &mut bulletin_refresh_circuit_open, + &mut next_bulletin_refresh_allowed_at, + now, + ); + if circuit_opened { + let cooldown_secs = + next_bulletin_refresh_allowed_at.duration_since(now).as_secs(); + tracing::warn!( + failures = bulletin_refresh_failures, + cooldown_secs, + backoff_secs = backoff.as_secs(), + "cortex bulletin refresh circuit opened after consecutive failures" + ); + } else { + tracing::warn!( + failures = bulletin_refresh_failures, + backoff_secs = backoff.as_secs(), + "cortex bulletin refresh failed; applying retry backoff" + ); + } + } + } + Err(error) => { + let now = Instant::now(); + let (backoff, circuit_opened) = record_bulletin_refresh_failure( + &mut bulletin_refresh_failures, + &mut bulletin_refresh_circuit_open, + &mut next_bulletin_refresh_allowed_at, + now, + ); + if circuit_opened { + let cooldown_secs = + next_bulletin_refresh_allowed_at.duration_since(now).as_secs(); + tracing::warn!( + %error, + failures = bulletin_refresh_failures, + cooldown_secs, + backoff_secs = backoff.as_secs(), + "cortex bulletin refresh circuit opened after task failure" + ); + } else { + tracing::warn!( + %error, + failures = bulletin_refresh_failures, + backoff_secs = backoff.as_secs(), + "cortex bulletin refresh task failed" + ); + } + } + } + } - maybe_generate_bulletin_under_lock( - deps.runtime_config.warmup_lock.as_ref(), - &deps.runtime_config.warmup, - &deps.runtime_config.warmup_status, - || generate_bulletin(deps, logger), - ) - .await; - generate_profile(deps, logger).await; + let cortex_config = **cortex.deps.runtime_config.cortex.load(); + let bulletin_interval = Duration::from_secs(cortex_config.bulletin_interval_secs.max(1)); + let now = Instant::now(); + if maybe_close_bulletin_refresh_circuit( + &mut bulletin_refresh_failures, + &mut bulletin_refresh_circuit_open, + &mut next_bulletin_refresh_allowed_at, + now, + ) { + tracing::info!("cortex bulletin refresh circuit closed; retries re-enabled"); + } + if refresh_task.is_none() + && !bulletin_refresh_circuit_open + && last_bulletin_refresh.elapsed() >= bulletin_interval + && now >= next_bulletin_refresh_allowed_at + { + refresh_task = Some(spawn_bulletin_refresh_task( + cortex.deps.clone(), + logger.clone(), + )); + } + + let updated_tick_interval_secs = cortex_config.tick_interval_secs.max(1); + if updated_tick_interval_secs != tick_interval_secs { + tick_interval_secs = updated_tick_interval_secs; + tick_period = Duration::from_secs(tick_interval_secs); + tick_timer = tokio::time::interval_at( + tokio::time::Instant::now() + tick_period, + tick_period, + ); + tick_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + } + } + } } } @@ -1685,12 +2348,19 @@ async fn fetch_memories_for_association( #[cfg(test)] mod tests { use super::{ - apply_cancelled_warmup_status, has_completed_initial_warmup, - maybe_generate_bulletin_under_lock, should_execute_warmup, - should_generate_bulletin_from_bulletin_loop, + BULLETIN_REFRESH_CIRCUIT_OPEN_SECS, BULLETIN_REFRESH_CIRCUIT_OPEN_THRESHOLD, + BulletinRefreshOutcome, CortexReceiverOutcome, ReceiverClosedBehavior, Signal, + apply_cancelled_warmup_status, handle_cortex_receiver_result, has_completed_initial_warmup, + maybe_close_bulletin_refresh_circuit, maybe_generate_bulletin_under_lock, + push_signal_into_buffer, record_bulletin_refresh_failure, should_execute_warmup, + should_generate_bulletin_from_bulletin_loop, signal_from_event, summarize_signal_text, }; + use crate::ProcessEvent; + use crate::memory::MemoryType; + use std::collections::VecDeque; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Instant; #[test] fn run_warmup_once_semantics_skip_when_disabled_without_force() { @@ -1881,7 +2551,400 @@ mod tests { drop(guard); let result = task.await.expect("task should join"); - assert!(result); + assert_eq!(result, BulletinRefreshOutcome::SkippedFresh); assert_eq!(calls.load(Ordering::SeqCst), 0); } + + #[test] + fn summarize_signal_text_uses_first_non_empty_line() { + let text = "\n\nfirst line\nsecond line"; + assert_eq!(summarize_signal_text(text), "first line"); + } + + #[test] + fn summarize_signal_text_truncates_long_text() { + let text = "a".repeat(200); + let summary = summarize_signal_text(&text); + assert_eq!(summary.chars().count(), crate::EVENT_SUMMARY_MAX_CHARS); + } + + #[test] + fn signal_from_event_maps_memory_saved_values() { + let event = ProcessEvent::MemorySaved { + agent_id: Arc::from("agent"), + memory_id: "mem-1".to_string(), + channel_id: Some(Arc::from("channel-1")), + memory_type: MemoryType::Decision, + importance: 0.92, + content_summary: "persisted decision".to_string(), + }; + + let signal = signal_from_event(event); + match signal { + Signal::MemorySaved { + memory_id, + channel_id, + memory_type, + content_summary, + importance, + } => { + assert_eq!(memory_id, "mem-1"); + assert_eq!(channel_id.as_deref(), Some("channel-1")); + assert_eq!(memory_type, MemoryType::Decision); + assert_eq!(content_summary, "persisted decision"); + assert_eq!(importance, 0.92); + } + _ => panic!("expected memory-saved signal"), + } + } + + #[test] + fn signal_from_event_handles_every_process_event_variant() { + let agent_id: crate::AgentId = Arc::from("agent"); + let channel_id: crate::ChannelId = Arc::from("channel"); + let worker_id = uuid::Uuid::new_v4(); + let branch_id = uuid::Uuid::new_v4(); + + let events = vec![ + ProcessEvent::BranchStarted { + agent_id: agent_id.clone(), + branch_id, + channel_id: channel_id.clone(), + description: "branch start".to_string(), + reply_to_message_id: Some("message-1".to_string()), + }, + ProcessEvent::BranchResult { + agent_id: agent_id.clone(), + branch_id, + channel_id: channel_id.clone(), + conclusion: "branch done".to_string(), + }, + ProcessEvent::WorkerStarted { + agent_id: agent_id.clone(), + worker_id, + channel_id: Some(channel_id.clone()), + task: "do work".to_string(), + worker_type: "shell".to_string(), + }, + ProcessEvent::WorkerStatus { + agent_id: agent_id.clone(), + worker_id, + channel_id: Some(channel_id.clone()), + status: "running".to_string(), + }, + ProcessEvent::WorkerComplete { + agent_id: agent_id.clone(), + worker_id, + channel_id: Some(channel_id.clone()), + result: "ok".to_string(), + notify: false, + success: true, + }, + ProcessEvent::ToolStarted { + agent_id: agent_id.clone(), + process_id: crate::ProcessId::Worker(worker_id), + channel_id: Some(channel_id.clone()), + tool_name: "shell".to_string(), + args: "echo hi".to_string(), + }, + ProcessEvent::ToolCompleted { + agent_id: agent_id.clone(), + process_id: crate::ProcessId::Worker(worker_id), + channel_id: Some(channel_id.clone()), + tool_name: "shell".to_string(), + result: "done".to_string(), + }, + ProcessEvent::MemorySaved { + agent_id: agent_id.clone(), + memory_id: "memory-1".to_string(), + channel_id: Some(channel_id.clone()), + memory_type: MemoryType::Fact, + importance: 0.6, + content_summary: "saved memory".to_string(), + }, + ProcessEvent::CompactionTriggered { + agent_id: agent_id.clone(), + channel_id: channel_id.clone(), + threshold_reached: 0.86, + }, + ProcessEvent::StatusUpdate { + agent_id: agent_id.clone(), + process_id: crate::ProcessId::Worker(worker_id), + status: "active".to_string(), + }, + ProcessEvent::WorkerPermission { + agent_id: agent_id.clone(), + worker_id, + channel_id: Some(channel_id.clone()), + permission_id: "perm-1".to_string(), + description: "allow network".to_string(), + patterns: vec!["https://example.com".to_string()], + }, + ProcessEvent::WorkerQuestion { + agent_id: agent_id.clone(), + worker_id, + channel_id: Some(channel_id.clone()), + question_id: "q-1".to_string(), + questions: vec![], + }, + ProcessEvent::AgentMessageSent { + from_agent_id: agent_id.clone(), + to_agent_id: Arc::from("agent-2"), + link_id: "link-1".to_string(), + channel_id: channel_id.clone(), + }, + ProcessEvent::AgentMessageReceived { + from_agent_id: Arc::from("agent-2"), + to_agent_id: agent_id, + link_id: "link-1".to_string(), + channel_id: channel_id.clone(), + }, + ProcessEvent::TaskUpdated { + agent_id: Arc::from("agent"), + task_number: 7, + status: "created".to_string(), + action: "created".to_string(), + }, + ProcessEvent::TextDelta { + agent_id: Arc::from("agent"), + process_id: crate::ProcessId::Worker(worker_id), + channel_id: Some(channel_id.clone()), + text_delta: "he".to_string(), + aggregated_text: "hello".to_string(), + }, + ]; + + for event in events { + let _signal = signal_from_event(event); + } + } + + #[test] + fn push_signal_into_buffer_coalesces_status_updates_for_same_process() { + let mut buffer = VecDeque::new(); + let process_id = crate::ProcessId::Worker(uuid::Uuid::new_v4()); + + push_signal_into_buffer( + &mut buffer, + Signal::StatusUpdate { + process_id: process_id.clone(), + status: "running".to_string(), + }, + ); + push_signal_into_buffer( + &mut buffer, + Signal::StatusUpdate { + process_id, + status: "done".to_string(), + }, + ); + + assert_eq!(buffer.len(), 1); + match buffer.back() { + Some(Signal::StatusUpdate { status, .. }) => assert_eq!(status, "done"), + _ => panic!("expected status-update signal"), + } + } + + #[test] + fn push_signal_into_buffer_keeps_distinct_status_updates() { + let mut buffer = VecDeque::new(); + + push_signal_into_buffer( + &mut buffer, + Signal::StatusUpdate { + process_id: crate::ProcessId::Worker(uuid::Uuid::new_v4()), + status: "running".to_string(), + }, + ); + push_signal_into_buffer( + &mut buffer, + Signal::StatusUpdate { + process_id: crate::ProcessId::Worker(uuid::Uuid::new_v4()), + status: "running".to_string(), + }, + ); + + assert_eq!(buffer.len(), 2); + } + + #[test] + fn memory_receiver_closed_disables_stream_without_stopping_loop() { + let mut lagged_since_last_warning = 0; + let mut last_lag_warning = None; + + let outcome = handle_cortex_receiver_result( + Err(tokio::sync::broadcast::error::RecvError::Closed), + "memory", + ReceiverClosedBehavior::DisableStream, + &mut lagged_since_last_warning, + &mut last_lag_warning, + 30, + ); + + assert!(matches!(outcome, CortexReceiverOutcome::DisableStream)); + } + + #[test] + fn memory_receiver_lagged_continues_loop_and_tracks_drop_count() { + let mut lagged_since_last_warning = 0; + let mut last_lag_warning = Some(Instant::now()); + + let outcome = handle_cortex_receiver_result( + Err(tokio::sync::broadcast::error::RecvError::Lagged(7)), + "memory", + ReceiverClosedBehavior::DisableStream, + &mut lagged_since_last_warning, + &mut last_lag_warning, + 30, + ); + + assert!(matches!( + outcome, + CortexReceiverOutcome::Lagged { dropped: 7 } + )); + assert_eq!(lagged_since_last_warning, 7); + } + + #[test] + fn bulletin_refresh_failure_opens_circuit_at_threshold() { + let mut failures = 0_u32; + let mut circuit_open = false; + let mut next_allowed_at = Instant::now(); + let now = Instant::now(); + + let (_, opened_first) = record_bulletin_refresh_failure( + &mut failures, + &mut circuit_open, + &mut next_allowed_at, + now, + ); + assert!(!opened_first); + assert!(!circuit_open); + + let (_, opened_second) = record_bulletin_refresh_failure( + &mut failures, + &mut circuit_open, + &mut next_allowed_at, + now, + ); + assert!(!opened_second); + assert!(!circuit_open); + + let (_, opened_third) = record_bulletin_refresh_failure( + &mut failures, + &mut circuit_open, + &mut next_allowed_at, + now, + ); + assert!(opened_third); + assert!(circuit_open); + assert_eq!(failures, BULLETIN_REFRESH_CIRCUIT_OPEN_THRESHOLD); + assert!( + next_allowed_at + >= now + std::time::Duration::from_secs(BULLETIN_REFRESH_CIRCUIT_OPEN_SECS), + "circuit-open cooldown should dominate retry window" + ); + } + + #[test] + fn bulletin_refresh_circuit_closes_after_cooldown() { + let mut failures = BULLETIN_REFRESH_CIRCUIT_OPEN_THRESHOLD; + let mut circuit_open = true; + let now = Instant::now(); + let mut next_allowed_at = now + std::time::Duration::from_millis(5); + + let closed_early = maybe_close_bulletin_refresh_circuit( + &mut failures, + &mut circuit_open, + &mut next_allowed_at, + now, + ); + assert!(!closed_early); + assert!(circuit_open); + + let closed = maybe_close_bulletin_refresh_circuit( + &mut failures, + &mut circuit_open, + &mut next_allowed_at, + now + std::time::Duration::from_millis(10), + ); + assert!(closed); + assert!(!circuit_open); + assert_eq!(failures, 0); + } + + #[tokio::test] + async fn run_cortex_loop_tick_not_starved_by_events() { + use std::time::Duration; + + const TEST_DURATION: Duration = Duration::from_millis(750); + const TICK_PERIOD: Duration = Duration::from_millis(25); + const MAX_DROPPED_EVENTS_BUDGET: u64 = 512; + + let (event_tx, mut event_rx) = tokio::sync::broadcast::channel::(1024); + let event_tx_for_sender = event_tx.clone(); + let mut tick_timer = + tokio::time::interval_at(tokio::time::Instant::now() + TICK_PERIOD, TICK_PERIOD); + tick_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + let sender = tokio::spawn(async move { + let agent_id: crate::AgentId = Arc::from("agent"); + let process_id = crate::ProcessId::Worker(uuid::Uuid::new_v4()); + let deadline = tokio::time::Instant::now() + TEST_DURATION; + while tokio::time::Instant::now() < deadline { + for _ in 0..8 { + let _ = event_tx_for_sender.send(ProcessEvent::StatusUpdate { + agent_id: agent_id.clone(), + process_id: process_id.clone(), + status: "busy".to_string(), + }); + } + tokio::task::yield_now().await; + } + }); + + let deadline = tokio::time::Instant::now() + TEST_DURATION + Duration::from_millis(250); + let mut tick_count = 0_u64; + let mut lagged_dropped_events = 0_u64; + let mut receiver_closed = false; + + while tokio::time::Instant::now() < deadline { + tokio::select! { + _ = tick_timer.tick() => { + tick_count = tick_count.saturating_add(1); + } + event = event_rx.recv() => { + match event { + Ok(_) => {} + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + lagged_dropped_events = lagged_dropped_events.saturating_add(skipped); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + receiver_closed = true; + break; + } + } + } + } + } + + sender.await.expect("sender task should complete"); + drop(event_tx); + + assert!( + !receiver_closed, + "receiver should not close while load test sender is active" + ); + assert!( + tick_count >= (TEST_DURATION.as_millis() / TICK_PERIOD.as_millis() / 4) as u64, + "periodic tick should continue firing under sustained event load" + ); + assert!( + lagged_dropped_events <= MAX_DROPPED_EVENTS_BUDGET, + "lagged dropped events exceeded budget: {} > {}", + lagged_dropped_events, + MAX_DROPPED_EVENTS_BUDGET + ); + } } diff --git a/src/agent/ingestion.rs b/src/agent/ingestion.rs index 7b662cbcd..9ffb7eb78 100644 --- a/src/agent/ingestion.rs +++ b/src/agent/ingestion.rs @@ -480,6 +480,7 @@ async fn process_chunk( deps.task_store.clone(), deps.memory_search.clone(), deps.runtime_config.clone(), + deps.memory_event_tx.clone(), conversation_logger, channel_store, crate::conversation::ProcessRunLogger::new(deps.sqlite_pool.clone()), diff --git a/src/agent/invariant_harness.rs b/src/agent/invariant_harness.rs index 309b272f3..7e335c102 100644 --- a/src/agent/invariant_harness.rs +++ b/src/agent/invariant_harness.rs @@ -3,7 +3,6 @@ use super::channel_dispatch::{ WorkerCompletionError, map_worker_completion_result, reserve_worker_slot_local, }; -use super::{EventRecvDisposition, classify_event_recv_error}; use rand::rngs::StdRng; use rand::{Rng, SeedableRng}; @@ -122,14 +121,18 @@ impl HarnessState { } HarnessFault::EventLagged => { self.lagged_event_kept_running &= matches!( - classify_event_recv_error(&tokio::sync::broadcast::error::RecvError::Lagged(1)), - EventRecvDisposition::Continue { .. } + crate::classify_broadcast_recv_result::<()>(Err( + tokio::sync::broadcast::error::RecvError::Lagged(1), + )), + crate::BroadcastRecvResult::Lagged(_) ); } HarnessFault::EventClosed => { self.closed_event_causes_stop &= matches!( - classify_event_recv_error(&tokio::sync::broadcast::error::RecvError::Closed), - EventRecvDisposition::Stop + crate::classify_broadcast_recv_result::<()>(Err( + tokio::sync::broadcast::error::RecvError::Closed, + )), + crate::BroadcastRecvResult::Closed ); } HarnessFault::CortexStart => { diff --git a/src/api/agents.rs b/src/api/agents.rs index 86a6e179e..8469c781a 100644 --- a/src/api/agents.rs +++ b/src/api/agents.rs @@ -403,7 +403,7 @@ pub(super) async fn trigger_warmup( let task_store_registry = state.task_store_registry.clone(); let injection_tx = state.injection_tx.clone(); tokio::spawn(async move { - let (event_tx, _event_rx) = tokio::sync::broadcast::channel(16); + let (event_tx, memory_event_tx) = crate::create_process_event_buses(); let deps = crate::AgentDeps { agent_id: Arc::from(agent_id.as_str()), memory_search, @@ -412,6 +412,7 @@ pub(super) async fn trigger_warmup( cron_tool: None, runtime_config, event_tx, + memory_event_tx, sqlite_pool: sqlite_pool.clone(), messaging_manager: None, sandbox, @@ -630,7 +631,7 @@ pub(super) async fn create_agent( )); let task_store = std::sync::Arc::new(crate::tasks::TaskStore::new(db.sqlite.clone())); - let (event_tx, _) = tokio::sync::broadcast::channel(256); + let (event_tx, memory_event_tx) = crate::create_process_event_buses(); let arc_agent_id: crate::AgentId = std::sync::Arc::from(agent_id.as_str()); crate::identity::scaffold_identity_files(&agent_config.workspace) @@ -713,6 +714,7 @@ pub(super) async fn create_agent( cron_tool: None, runtime_config: runtime_config.clone(), event_tx: event_tx.clone(), + memory_event_tx: memory_event_tx.clone(), sqlite_pool: db.sqlite.clone(), messaging_manager: { let guard = state.messaging_manager.read().await; @@ -778,6 +780,7 @@ pub(super) async fn create_agent( deps.agent_id.clone(), deps.task_store.clone(), memory_search.clone(), + deps.memory_event_tx.clone(), conversation_logger, channel_store, run_logger, @@ -797,8 +800,7 @@ pub(super) async fn create_agent( let cortex_logger = crate::agent::cortex::CortexLogger::new(db.sqlite.clone()); let _warmup_loop = crate::agent::cortex::spawn_warmup_loop(deps.clone(), cortex_logger.clone()); - let _bulletin_loop = - crate::agent::cortex::spawn_bulletin_loop(deps.clone(), cortex_logger.clone()); + let _cortex_loop = crate::agent::cortex::spawn_cortex_loop(deps.clone(), cortex_logger.clone()); let _association_loop = crate::agent::cortex::spawn_association_loop(deps.clone(), cortex_logger); crate::agent::cortex::spawn_ready_task_loop( diff --git a/src/api/state.rs b/src/api/state.rs index 4a361b49a..d5524fbd8 100644 --- a/src/api/state.rs +++ b/src/api/state.rs @@ -505,13 +505,20 @@ impl ApiState { } } Err(error) => { - if let crate::agent::EventRecvDisposition::Continue { lagged_count } = - crate::agent::classify_event_recv_error(&error) - { - let count = lagged_count.unwrap_or(0); - tracing::debug!(agent_id = %agent_id, count, "API event forwarder lagged, skipped events"); - } else { - break; + match crate::classify_broadcast_recv_result::(Err( + error, + )) { + crate::BroadcastRecvResult::Lagged(count) => { + tracing::debug!( + agent_id = %agent_id, + count, + "API event forwarder lagged, skipped events" + ); + } + crate::BroadcastRecvResult::Closed => break, + crate::BroadcastRecvResult::Event(_) => unreachable!( + "classifying an Err recv result should never produce Event" + ), } } } diff --git a/src/api/system.rs b/src/api/system.rs index bd2789596..1156c52cf 100644 --- a/src/api/system.rs +++ b/src/api/system.rs @@ -104,16 +104,17 @@ pub(super) async fn events_sse( } } Err(error) => { - if let crate::agent::EventRecvDisposition::Continue { lagged_count } = - crate::agent::classify_event_recv_error(&error) - { - let count = lagged_count.unwrap_or(0); - tracing::debug!(count, "SSE client lagged"); - yield Ok(axum::response::sse::Event::default() - .event("lagged") - .data(format!("{{\"skipped\":{count}}}"))); - } else { - break; + match crate::classify_broadcast_recv_result::(Err(error)) { + crate::BroadcastRecvResult::Lagged(count) => { + tracing::debug!(count, "SSE client lagged"); + yield Ok(axum::response::sse::Event::default() + .event("lagged") + .data(format!("{{\"skipped\":{count}}}"))); + } + crate::BroadcastRecvResult::Closed => break, + crate::BroadcastRecvResult::Event(_) => unreachable!( + "classifying an Err recv result should never produce Event" + ), } } } diff --git a/src/lib.rs b/src/lib.rs index 53e130036..6a0540b5e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -96,6 +96,51 @@ impl std::fmt::Display for ProcessType { } } +/// Return a short summary from the first non-empty line, truncated to a +/// character limit. +pub const EVENT_SUMMARY_MAX_CHARS: usize = 160; + +pub fn summarize_first_non_empty_line(value: &str, max_chars: usize) -> String { + let first_line = value + .lines() + .map(str::trim) + .find(|line| !line.is_empty()) + .unwrap_or_else(|| value.trim()); + + truncate_to_chars(first_line, max_chars).to_string() +} + +fn truncate_to_chars(value: &str, max_chars: usize) -> &str { + if max_chars == 0 { + return ""; + } + + if let Some((index, _)) = value.char_indices().nth(max_chars) { + &value[..index] + } else { + value + } +} + +#[derive(Debug)] +pub enum BroadcastRecvResult { + Event(T), + Lagged(u64), + Closed, +} + +pub fn classify_broadcast_recv_result( + result: std::result::Result, +) -> BroadcastRecvResult { + match result { + Ok(event) => BroadcastRecvResult::Event(event), + Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { + BroadcastRecvResult::Lagged(count) + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => BroadcastRecvResult::Closed, + } +} + /// Events sent between processes. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] @@ -152,6 +197,9 @@ pub enum ProcessEvent { agent_id: AgentId, memory_id: String, channel_id: Option, + memory_type: crate::memory::MemoryType, + importance: f32, + content_summary: String, }, CompactionTriggered { agent_id: AgentId, @@ -206,6 +254,59 @@ pub enum ProcessEvent { }, } +/// Default broadcast capacity for the per-agent control event bus. +pub const CONTROL_EVENT_BUS_CAPACITY: usize = 256; + +/// Default broadcast capacity for the per-agent memory event bus. +pub const MEMORY_EVENT_BUS_CAPACITY: usize = 1024; + +/// Create the default pair of per-agent process event buses. +/// +/// - `event_tx` carries control/lifecycle events consumed by channels and UI. +/// - `memory_event_tx` carries memory-save telemetry consumed by the cortex. +pub fn create_process_event_buses() -> ( + tokio::sync::broadcast::Sender, + tokio::sync::broadcast::Sender, +) { + create_process_event_buses_with_capacity(CONTROL_EVENT_BUS_CAPACITY, MEMORY_EVENT_BUS_CAPACITY) +} + +/// Create per-agent process event buses with explicit capacities. +pub fn create_process_event_buses_with_capacity( + control_event_capacity: usize, + memory_event_capacity: usize, +) -> ( + tokio::sync::broadcast::Sender, + tokio::sync::broadcast::Sender, +) { + let (event_tx, _event_rx) = tokio::sync::broadcast::channel(control_event_capacity); + let (memory_event_tx, _memory_event_rx) = + tokio::sync::broadcast::channel(memory_event_capacity); + (event_tx, memory_event_tx) +} + +/// Track lagged broadcast events and return the dropped count when a warning +/// should be emitted. Returns `None` when still inside the throttle window. +pub fn drain_lag_warning_count( + lagged_since_last_warning: &mut u64, + last_lag_warning: &mut Option, + newly_lagged_count: u64, + warning_interval: std::time::Duration, +) -> Option { + *lagged_since_last_warning = lagged_since_last_warning.saturating_add(newly_lagged_count); + + let now = std::time::Instant::now(); + let should_warn = + last_lag_warning.is_none_or(|last| now.saturating_duration_since(last) >= warning_interval); + + if !should_warn { + return None; + } + + *last_lag_warning = Some(now); + Some(std::mem::take(lagged_since_last_warning)) +} + /// A message to be injected into a specific channel from outside the normal /// inbound message flow. Used for cross-agent task completion notifications. #[derive(Debug, Clone)] @@ -229,6 +330,7 @@ pub struct AgentDeps { pub cron_tool: Option, pub runtime_config: Arc, pub event_tx: tokio::sync::broadcast::Sender, + pub memory_event_tx: tokio::sync::broadcast::Sender, pub sqlite_pool: sqlx::SqlitePool, pub messaging_manager: Option>, pub sandbox: Arc, diff --git a/src/main.rs b/src/main.rs index 3cd123da3..5c28b4fa9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2128,8 +2128,8 @@ async fn initialize_agents( embedding_model.clone(), )); - // Per-agent event bus (broadcast for fan-out to multiple channels) - let (event_tx, _event_rx) = tokio::sync::broadcast::channel(256); + // Per-agent control and memory event buses (broadcast fan-out). + let (event_tx, memory_event_tx) = spacebot::create_process_event_buses(); let agent_id: spacebot::AgentId = Arc::from(agent_config.id.as_str()); let mcp_manager = Arc::new(spacebot::mcp::McpManager::new(agent_config.mcp.clone())); @@ -2209,6 +2209,7 @@ async fn initialize_agents( cron_tool: None, runtime_config, event_tx, + memory_event_tx, sqlite_pool: db.sqlite.clone(), messaging_manager: None, sandbox, @@ -2772,7 +2773,7 @@ async fn initialize_agents( } } - // Start cortex warmup, bulletin loops, and association loops for each agent + // Start cortex warmup, runtime, and association loops for each agent for (agent_id, agent) in agents.iter() { let cortex_logger = spacebot::agent::cortex::CortexLogger::new(agent.db.sqlite.clone()); let warmup_handle = @@ -2780,10 +2781,10 @@ async fn initialize_agents( cortex_handles.push(warmup_handle); tracing::info!(agent_id = %agent_id, "warmup loop started"); - let bulletin_handle = - spacebot::agent::cortex::spawn_bulletin_loop(agent.deps.clone(), cortex_logger.clone()); - cortex_handles.push(bulletin_handle); - tracing::info!(agent_id = %agent_id, "cortex bulletin loop started"); + let cortex_handle = + spacebot::agent::cortex::spawn_cortex_loop(agent.deps.clone(), cortex_logger.clone()); + cortex_handles.push(cortex_handle); + tracing::info!(agent_id = %agent_id, "cortex loop started"); let association_handle = spacebot::agent::cortex::spawn_association_loop(agent.deps.clone(), cortex_logger); @@ -2812,6 +2813,7 @@ async fn initialize_agents( agent.deps.agent_id.clone(), agent.deps.task_store.clone(), agent.deps.memory_search.clone(), + agent.deps.memory_event_tx.clone(), conversation_logger, channel_store, run_logger, diff --git a/src/telemetry/registry.rs b/src/telemetry/registry.rs index 99cbd8474..55efad796 100644 --- a/src/telemetry/registry.rs +++ b/src/telemetry/registry.rs @@ -79,6 +79,10 @@ pub struct Metrics { /// Labels: agent_id, dispatch_type, reason. pub dispatch_while_cold_count: IntCounterVec, + /// Total broadcast events dropped because a receiver lagged. + /// Labels: agent_id, receiver. + pub event_receiver_lagged_events_total: IntCounterVec, + /// Time-to-recovery for forced warmup passes kicked by dispatch paths, in ms. /// Labels: agent_id, dispatch_type. pub warmup_recovery_latency_ms: HistogramVec, @@ -208,6 +212,15 @@ impl Metrics { ) .expect("hardcoded metric descriptor"); + let event_receiver_lagged_events_total = IntCounterVec::new( + Opts::new( + "spacebot_event_receiver_lagged_events_total", + "Total broadcast events dropped because a receiver lagged", + ), + &["agent_id", "receiver"], + ) + .expect("hardcoded metric descriptor"); + let warmup_recovery_latency_ms = HistogramVec::new( HistogramOpts::new( "spacebot_warmup_recovery_latency_ms", @@ -265,6 +278,9 @@ impl Metrics { registry .register(Box::new(dispatch_while_cold_count.clone())) .expect("hardcoded metric"); + registry + .register(Box::new(event_receiver_lagged_events_total.clone())) + .expect("hardcoded metric"); registry .register(Box::new(warmup_recovery_latency_ms.clone())) .expect("hardcoded metric"); @@ -286,6 +302,7 @@ impl Metrics { process_errors_total, memory_updates_total, dispatch_while_cold_count, + event_receiver_lagged_events_total, warmup_recovery_latency_ms, } } diff --git a/src/tools.rs b/src/tools.rs index ea5bdce24..a87ee2cf3 100644 --- a/src/tools.rs +++ b/src/tools.rs @@ -388,6 +388,14 @@ pub async fn remove_channel_tools( Ok(()) } +fn memory_save_with_events( + memory_search: Arc, + agent_id: AgentId, + memory_event_tx: broadcast::Sender, +) -> MemorySaveTool { + MemorySaveTool::new(memory_search).with_event_bus(agent_id, memory_event_tx) +} + /// Create a per-branch ToolServer with memory tools. /// /// Each branch gets its own isolated ToolServer so `memory_recall` is never @@ -400,12 +408,17 @@ pub fn create_branch_tool_server( task_store: Arc, memory_search: Arc, runtime_config: Arc, + memory_event_tx: broadcast::Sender, conversation_logger: crate::conversation::history::ConversationLogger, channel_store: crate::conversation::ChannelStore, run_logger: crate::conversation::history::ProcessRunLogger, ) -> ToolServerHandle { let mut server = ToolServer::new() - .tool(MemorySaveTool::new(memory_search.clone())) + .tool(memory_save_with_events( + memory_search.clone(), + agent_id.clone(), + memory_event_tx.clone(), + )) .tool(MemoryRecallTool::new(memory_search.clone())) .tool(MemoryDeleteTool::new(memory_search)) .tool(ChannelRecallTool::new(conversation_logger, channel_store)) @@ -500,9 +513,17 @@ pub fn create_worker_tool_server( /// /// The cortex only needs memory_save for consolidation. Additional tools can be /// added later as cortex capabilities expand. -pub fn create_cortex_tool_server(memory_search: Arc) -> ToolServerHandle { +pub fn create_cortex_tool_server( + agent_id: AgentId, + memory_event_tx: broadcast::Sender, + memory_search: Arc, +) -> ToolServerHandle { ToolServer::new() - .tool(MemorySaveTool::new(memory_search)) + .tool(memory_save_with_events( + memory_search, + agent_id, + memory_event_tx, + )) .run() } @@ -518,6 +539,7 @@ pub fn create_cortex_chat_tool_server( agent_id: AgentId, task_store: Arc, memory_search: Arc, + memory_event_tx: broadcast::Sender, conversation_logger: crate::conversation::history::ConversationLogger, channel_store: crate::conversation::ChannelStore, run_logger: crate::conversation::history::ProcessRunLogger, @@ -529,7 +551,11 @@ pub fn create_cortex_chat_tool_server( runtime_config: Arc, ) -> ToolServerHandle { let mut server = ToolServer::new() - .tool(MemorySaveTool::new(memory_search.clone())) + .tool(memory_save_with_events( + memory_search.clone(), + agent_id.clone(), + memory_event_tx, + )) .tool(MemoryRecallTool::new(memory_search.clone())) .tool(MemoryDeleteTool::new(memory_search)) .tool(ChannelRecallTool::new(conversation_logger, channel_store)) diff --git a/src/tools/memory_save.rs b/src/tools/memory_save.rs index 07225bffe..a21d3e687 100644 --- a/src/tools/memory_save.rs +++ b/src/tools/memory_save.rs @@ -3,6 +3,7 @@ use crate::error::Result; use crate::memory::types::Association; use crate::memory::{Memory, MemorySearch, MemoryType}; +use crate::{AgentId, ProcessEvent}; use rig::completion::ToolDefinition; use rig::tool::Tool; use schemars::JsonSchema; @@ -17,12 +18,35 @@ const MAX_MEMORY_CONTENT_BYTES: usize = 50_000; #[derive(Debug, Clone)] pub struct MemorySaveTool { memory_search: Arc, + event_context: Option, +} + +#[derive(Debug, Clone)] +struct MemorySaveEventContext { + agent_id: AgentId, + memory_event_tx: tokio::sync::broadcast::Sender, } impl MemorySaveTool { /// Create a new memory save tool. pub fn new(memory_search: Arc) -> Self { - Self { memory_search } + Self { + memory_search, + event_context: None, + } + } + + /// Enable process event emission for successful memory saves. + pub fn with_event_bus( + mut self, + agent_id: AgentId, + memory_event_tx: tokio::sync::broadcast::Sender, + ) -> Self { + self.event_context = Some(MemorySaveEventContext { + agent_id, + memory_event_tx, + }); + self } } @@ -290,6 +314,26 @@ impl Tool for MemorySaveTool { tracing::warn!(%error, "failed to ensure FTS index after memory save"); } + if let Some(event_context) = &self.event_context + && event_context.memory_event_tx.receiver_count() > 0 + { + let event = ProcessEvent::MemorySaved { + agent_id: event_context.agent_id.clone(), + memory_id: memory.id.clone(), + channel_id: memory.channel_id.clone(), + memory_type: memory.memory_type, + importance: memory.importance, + content_summary: summarize_memory_content(&memory.content), + }; + if let Err(error) = event_context.memory_event_tx.send(event) { + tracing::debug!( + memory_id = %memory.id, + %error, + "failed to emit memory-saved event" + ); + } + } + #[cfg(feature = "metrics")] { let metrics = crate::telemetry::Metrics::global(); @@ -330,3 +374,25 @@ pub async fn save_fact( .map_err(|e| crate::error::AgentError::Other(anyhow::anyhow!(e)))?; Ok(output.memory_id) } + +fn summarize_memory_content(content: &str) -> String { + crate::summarize_first_non_empty_line(content, crate::EVENT_SUMMARY_MAX_CHARS) +} + +#[cfg(test)] +mod tests { + use super::summarize_memory_content; + + #[test] + fn summarize_memory_content_prefers_first_non_empty_line() { + let content = "\n\nFirst line summary\nSecond line details"; + assert_eq!(summarize_memory_content(content), "First line summary"); + } + + #[test] + fn summarize_memory_content_truncates_to_max_chars() { + let content = "a".repeat(200); + let summary = summarize_memory_content(&content); + assert_eq!(summary.chars().count(), crate::EVENT_SUMMARY_MAX_CHARS); + } +} diff --git a/tests/bulletin.rs b/tests/bulletin.rs index c6f43529c..7235e06e0 100644 --- a/tests/bulletin.rs +++ b/tests/bulletin.rs @@ -88,7 +88,7 @@ async fn bootstrap_deps() -> anyhow::Result { skills, )); - let (event_tx, _) = tokio::sync::broadcast::channel(16); + let (event_tx, memory_event_tx) = spacebot::create_process_event_buses_with_capacity(16, 32); let agent_id: spacebot::AgentId = Arc::from(agent_config.id.as_str()); let mcp_manager = Arc::new(spacebot::mcp::McpManager::new(agent_config.mcp.clone())); @@ -115,6 +115,7 @@ async fn bootstrap_deps() -> anyhow::Result { cron_tool: None, runtime_config, event_tx, + memory_event_tx, sqlite_pool: db.sqlite.clone(), messaging_manager: None, sandbox, diff --git a/tests/context_dump.rs b/tests/context_dump.rs index df77d7807..82677953e 100644 --- a/tests/context_dump.rs +++ b/tests/context_dump.rs @@ -87,7 +87,7 @@ async fn bootstrap_deps() -> anyhow::Result<(spacebot::AgentDeps, spacebot::conf skills, )); - let (event_tx, _) = tokio::sync::broadcast::channel(16); + let (event_tx, memory_event_tx) = spacebot::create_process_event_buses_with_capacity(16, 32); let agent_id: spacebot::AgentId = Arc::from(agent_config.id.as_str()); let mcp_manager = Arc::new(spacebot::mcp::McpManager::new(agent_config.mcp.clone())); @@ -114,6 +114,7 @@ async fn bootstrap_deps() -> anyhow::Result<(spacebot::AgentDeps, spacebot::conf cron_tool: None, runtime_config, event_tx, + memory_event_tx, sqlite_pool: db.sqlite.clone(), messaging_manager: None, sandbox, @@ -309,6 +310,7 @@ async fn dump_branch_context() { deps.task_store.clone(), deps.memory_search.clone(), deps.runtime_config.clone(), + deps.memory_event_tx.clone(), conversation_logger, channel_store, run_logger, @@ -506,6 +508,7 @@ async fn dump_all_contexts() { deps.task_store.clone(), deps.memory_search.clone(), deps.runtime_config.clone(), + deps.memory_event_tx.clone(), conversation_logger, channel_store, run_logger,