feat(cortex): Implement Phase 1 ("The Tick Loop")

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:
 

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<ProcessEvent>` -- a multi-producer, multi-consumer event bus. The channel, all branches, and all workers share the same bus.
+Each agent uses two `broadcast::channel<ProcessEvent>` 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<ProcessE
 | `WorkerComplete` | Worker | Channel | Task done, retrigger |
 | `ToolStarted` | Hook | Channel, UI | Tool call in progress |
 | `ToolCompleted` | Hook | Channel, UI | Tool call finished |
-| `MemorySaved` | Branch, Cortex | UI | New memory created |
+| `MemorySaved` | `memory_save` tool (branch/compactor/cortex contexts) | Cortex | New memory telemetry for signal buffer |
 | `CompactionTriggered` | Compactor | Channel | Context compacted |
 | `StatusUpdate` | Various | UI (SSE) | Typing indicators, lifecycle |
 | `TaskUpdated` | Branch, Worker | UI | Task board change |

docs/design-docs/cortex-implementation.md

@@ -1,58 +1,64 @@
 # Cortex Implementation Plan
 
-The cortex is designed to be the system's self-awareness — supervising processes, maintaining memory coherence, and generating the memory bulletin. Today, only the bulletin works. Everything else is dead code or stubs.
+The cortex is designed to be the system's self-awareness — supervising processes, maintaining memory coherence, and generating the memory bulletin. Phase 1 plumbing is now live; later supervision and maintenance phases remain.
 
 This doc covers the path from "bulletin generator" to "full system supervisor."
 
 ## What Exists Today
 
 **Running:**
-- `spawn_bulletin_loop()` — generates the memory bulletin on startup, refreshes hourly via LLM synthesis of pre-gathered memory data. Fully functional.
+- `spawn_cortex_loop()` — instantiates `Cortex`, subscribes to both control and memory event buses, runs a `tokio::select!` loop for event observation and periodic ticks, and refreshes bulletin/profile on interval.
+- `spawn_bulletin_loop()` — compatibility alias to `spawn_cortex_loop()`.
+- `spawn_warmup_loop()` — asynchronous warmup that keeps bulletin/embedding readiness fresh.
 
-**Defined but never instantiated:**
-- `Cortex` struct — has `observe()` (converts 3 of 12 event types into signals with hardcoded dummy values) and `run_consolidation()` (logs and returns `Ok(())`).
-- `Signal` enum — 6 variants, none ever constructed at runtime.
+**Defined and instantiated:**
+- `Cortex` struct — observes all `ProcessEvent` variants, builds a rolling signal buffer, and runs on configurable tick cadence.
+- `Signal` enum — aligned to current `ProcessEvent` surface (worker/branch/tool/memory/compaction/task/link events).
 - `CortexHook` — all methods return `Continue` with trace logging.
 
 **Implemented but never called:**
 - `memory/maintenance.rs` — `apply_decay()` and `prune_memories()` work. `merge_similar_memories()` is a stub returning `Ok(0)`.
 
-**Wired through config but never read:**
-- `CortexConfig` fields: `tick_interval_secs` (30), `worker_timeout_secs` (300), `branch_timeout_secs` (60), `circuit_breaker_threshold` (3). All resolve through `env > 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<ProcessEvent>` per agent.
-- `MemorySaved` and `CompactionTriggered` variants are defined but never emitted by any code.
+**Event buses:**
+- Two per-agent `broadcast::Sender<ProcessEvent>` 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`

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<u64> },
-    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<String> {
     for message in history.iter().rev() {
         if let rig::message::Message::Assistant { content, .. } = message {

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<ProcessEvent>,
+) -> crate::BroadcastRecvResult<ProcessEvent> {
+    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<std::time::Instant> = 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::<ProcessEvent>(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::<ProcessEvent>(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));
+    }
+}

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()),