feat: HeartbeatHandler turn injection — the clockmaker connection

The final piece of the heartbeat-system task. With this wired in,
Aster (or any agent) can schedule a prompt and the runtime fires it
back at the configured time as a fresh user turn.

Plumbing:
- `TurnInjector` trait in core/nervous/handler.rs decouples the
  nervous module from any specific backend; LocalBackend implements it
- HeartbeatHandler now takes Arc<dyn TurnInjector>; on schedule_due
  events it pulls agent_id + prompt from the payload and injects
- CronSensor includes agent_id in event payload so the handler knows
  who owns the schedule
- LocalBackend gains an `active_sessions: Arc<AtomicU32>` counter that
  send() increments and the spawned turn decrements; CronSensors read
  this to skip firing while conversations are active (the existing
  pause-during-presence semantics)
- On startup, LocalBackend::new discovers agents from the inventory
  and spawns one CronSensor per agent + one HeartbeatHandler on the bus
- inject_background_turn reuses the agent's most recent conversation
  (falls back to ensure_conversation), then drains the stream silently

Drive-by fix:
- write.rs append mode now calls file.flush() before drop. tokio's
  File::Drop doesn't sync, which surfaced as a flaky test_append under
  parallel test runs once compilation timing shifted.

102 tests, 0 failures.

src/backend/local.rs

@@ -12,7 +12,7 @@ use anyhow::{Context, Result};
 use async_trait::async_trait;
 use futures::stream::{BoxStream, StreamExt};
 use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
 use std::time::Duration;
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
@@ -288,6 +288,10 @@ pub struct LocalBackend {
     server: Arc<SouveraineServer>,
     event_bus: EventBus,
     seed_id: Arc<SeedId>,
+    /// Live count of in-flight turns (user-initiated or heartbeat-injected).
+    /// CronSensors read this to pause firing while a conversation is active —
+    /// scheduled events shouldn't interrupt presence.
+    active_sessions: Arc<AtomicU32>,
 }
 
 impl LocalBackend {
@@ -320,11 +324,55 @@ impl LocalBackend {
             crate::core::nervous::event_log::EventLog::new(events_dir, event_bus.subscribe());
         tokio::spawn(async move { event_log.run().await });
 
-        Ok(Self {
+        let active_sessions = Arc::new(AtomicU32::new(0));
+        let backend = Self {
             server: Arc::new(server),
-            event_bus,
+            event_bus: event_bus.clone(),
             seed_id,
-        })
+            active_sessions: active_sessions.clone(),
+        };
+
+        // Spawn one CronSensor per agent (each agent owns its own schedules
+        // directory), and one HeartbeatHandler on the bus that injects turns
+        // when a schedule fires. The handler holds an Arc<dyn TurnInjector>
+        // pointing back at us — clean dep direction, no LocalBackend leak
+        // into the nervous module.
+        let agents_dir = base.join("agents");
+        match backend.server.agents.list(None).await {
+            Ok(summaries) => {
+                for summary in summaries {
+                    let schedules_dir = agents_dir.join(&summary.id).join("schedules");
+                    if let Err(e) = std::fs::create_dir_all(&schedules_dir) {
+                        tracing::warn!(
+                            agent = %summary.id,
+                            error = %e,
+                            "could not create schedules dir; skipping cron sensor"
+                        );
+                        continue;
+                    }
+                    let sensor = crate::core::nervous::cron::CronSensor::new(
+                        summary.id.clone(),
+                        schedules_dir,
+                        event_bus.clone(),
+                        active_sessions.clone(),
+                    );
+                    tokio::spawn(async move { sensor.run().await });
+                    tracing::info!(agent = %summary.id, "cron sensor spawned");
+                }
+            }
+            Err(e) => {
+                tracing::warn!(error = %e, "agent listing failed; no cron sensors spawned");
+            }
+        }
+
+        let injector: Arc<dyn crate::core::nervous::handler::TurnInjector> =
+            Arc::new(backend.clone());
+        let mut handler =
+            crate::core::nervous::handler::HeartbeatHandler::new(event_bus.subscribe(), injector);
+        tokio::spawn(async move { handler.run().await });
+        tracing::info!("heartbeat handler spawned");
+
+        Ok(backend)
     }
 
     pub fn from_server(server: Arc<SouveraineServer>) -> Self {
@@ -337,6 +385,7 @@ impl LocalBackend {
             event_bus: EventBus::default(),
             server,
             seed_id,
+            active_sessions: Arc::new(AtomicU32::new(0)),
         }
     }
 
@@ -520,18 +569,48 @@ impl Backend for LocalBackend {
         let server = self.server.clone();
         let conv_id = conversation_id.to_string();
         let event_bus = self.event_bus.clone();
+        let active = self.active_sessions.clone();
 
+        active.fetch_add(1, Ordering::Relaxed);
         tokio::spawn(async move {
             if let Err(e) = run_turn(server, conv_id, &tx, event_bus).await {
                 let _ = tx.send(Err(e)).await;
             }
             let _ = tx.send(Ok(BackendEvent::Done)).await;
+            active.fetch_sub(1, Ordering::Relaxed);
         });
 
         Ok(ReceiverStream::new(rx).boxed())
     }
 }
 
+#[async_trait]
+impl crate::core::nervous::handler::TurnInjector for LocalBackend {
+    /// Heartbeat-driven turn injection. The cron loop pauses while
+    /// `active_sessions > 0`, so by the time we get here the agent is
+    /// idle. We grab the most recent conversation (or create a fresh one
+    /// if the agent has none), append the scheduled prompt as a user
+    /// message, and drain the resulting stream — the turn runs silently
+    /// in the background. Anything Aster surfaces lands in the inbox.
+    async fn inject_background_turn(
+        &self,
+        agent_id: &str,
+        text: &str,
+    ) -> anyhow::Result<()> {
+        let conv_id = match self.server.sessions.list_for_agent(agent_id).last().cloned() {
+            Some(id) => id,
+            None => self.ensure_conversation(agent_id).await?,
+        };
+        let stream = self.send(&conv_id, text).await?;
+        // Drain the stream in the background — no UI is listening.
+        tokio::spawn(async move {
+            let mut s = stream;
+            while s.next().await.is_some() {}
+        });
+        Ok(())
+    }
+}
+
 // ── Turn Loop ────────────────────────────────────────────────────
 
 async fn run_turn(

src/core/nervous/cron.rs

@@ -251,6 +251,7 @@ impl CronSensor {
                     target: Some(entry.name.clone()),
                     urgency: entry.urgency,
                     payload: Some(serde_json::json!({
+                        "agent_id": self.agent_id,
                         "kind": entry.kind,
                         "prompt": entry.prompt,
                         "source": entry.source,

src/core/nervous/handler.rs

@@ -1,15 +1,35 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
 use tokio::sync::broadcast;
-use tracing::{debug, warn};
+use tracing::{debug, info, warn};
 
 use super::SensorEvent;
 
+/// The seam between the nervous system (which knows *when* to fire) and
+/// the backend (which knows *how* to start a turn). The handler doesn't
+/// know about LocalBackend; LocalBackend implements this trait. Keeps the
+/// dependency direction sane and lets remote/federated agents inject too.
+#[async_trait]
+pub trait TurnInjector: Send + Sync {
+    /// Start a background turn for `agent_id` with `text` as the user
+    /// message. The stream of events is drained inside; callers don't
+    /// see them — heartbeats are silent unless the agent surfaces.
+    async fn inject_background_turn(
+        &self,
+        agent_id: &str,
+        text: &str,
+    ) -> anyhow::Result<()>;
+}
+
 pub struct HeartbeatHandler {
     rx: broadcast::Receiver<SensorEvent>,
+    injector: Arc<dyn TurnInjector>,
 }
 
 impl HeartbeatHandler {
-    pub fn new(rx: broadcast::Receiver<SensorEvent>) -> Self {
-        Self { rx }
+    pub fn new(rx: broadcast::Receiver<SensorEvent>, injector: Arc<dyn TurnInjector>) -> Self {
+        Self { rx, injector }
     }
 
     pub async fn run(&mut self) {
@@ -33,6 +53,11 @@ impl HeartbeatHandler {
 
     async fn handle_schedule_event(&self, event: &SensorEvent) {
         let name = event.target.as_deref().unwrap_or("unknown");
+        let agent_id = event
+            .payload
+            .as_ref()
+            .and_then(|p| p.get("agent_id"))
+            .and_then(|v| v.as_str());
         let prompt = event
             .payload
             .as_ref()
@@ -40,16 +65,30 @@ impl HeartbeatHandler {
             .and_then(|v| v.as_str())
             .unwrap_or("");
 
-        debug!(
+        let Some(agent_id) = agent_id else {
+            warn!(schedule = name, "schedule_due missing agent_id in payload");
+            return;
+        };
+
+        if prompt.is_empty() {
+            warn!(schedule = name, agent = agent_id, "schedule_due missing prompt");
+            return;
+        }
+
+        info!(
             schedule = name,
+            agent = agent_id,
             urgency = event.urgency,
-            "heartbeat: schedule due"
+            "heartbeat firing — injecting background turn"
         );
 
-        // TODO(phase 3): inject turn through run_turn() when idle,
-        // route to subconscious inbox when active session exists.
-        // For now, log the event. The wiring into LocalBackend's
-        // turn injection path comes in Phase 4 integration.
-        let _ = prompt;
+        if let Err(e) = self.injector.inject_background_turn(agent_id, prompt).await {
+            warn!(
+                schedule = name,
+                agent = agent_id,
+                error = %e,
+                "background turn injection failed"
+            );
+        }
     }
 }

src/core/tools/write.rs

@@ -108,6 +108,8 @@ Paths inside my memory directories contain frontmatter, are git-tracked, and hav
                 }
                 file.write_all(content.as_bytes()).await
                     .map_err(|e| ToolError::io_error(resolved.clone(), e))?;
+                file.flush().await
+                    .map_err(|e| ToolError::io_error(resolved.clone(), e))?;
 
                 Ok(ToolOutput {
                     content: format!("Appended {} chars to {}", content.len(), resolved.display()),