loopd: heartbeat tick on the background slot

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
2026-10-03 16:48:23 -07:00
parent 4dd3b87db3
commit ea0708e305
7 changed files with 377 additions and 1 deletions
@@ -47,16 +47,32 @@ impl Context {
            busy: Mutex::new(HashSet::new()),
        }
    }
+
+    /// Mark `id` busy. `Err(Busy)` if it already is. Tests and the scheduler use this so a skip
+    /// is not a sleep (T27).
+    pub fn hold(&self, id: SessionId) -> Result<Held<'_>, Busy> {
+        let mut busy = self.busy.lock().unwrap_or_else(|p| p.into_inner());
+        if busy.contains(&id) {
+            return Err(Busy);
+        }
+        busy.insert(id.clone());
+        drop(busy);
+        Ok(Held { ctx: self, id })
+    }
 }

 /// While a turn runs, the session is in the busy set. The guard holds only an immutable borrow of
 /// the context and a clone of the id, never the lock, so another connection can still check
 /// membership while this turn is slow. The id is removed when the guard drops.
-struct Held<'a> {
+pub struct Held<'a> {
    ctx: &'a Context,
    id: SessionId,
 }

+/// The session is already in the busy set.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Busy;
+
 impl Drop for Held<'_> {
    fn drop(&mut self) {
        let mut busy = self.ctx.busy.lock().unwrap_or_else(|p| p.into_inner());
@@ -10,6 +10,7 @@ pub mod http;
 pub mod jobs;
 pub mod llama;
 pub mod memory;
+pub mod scheduler;
 pub mod selftest;
 pub mod session;
 pub mod socket;
@@ -151,6 +151,17 @@ fn run_serve(path: &Path) -> ExitCode {

    let memory = Box::new(MemoryPort::new(cfg.memory_socket(), MEMORY_WAIT));
    let ctx = Arc::new(Context::new(cfg, client, port, Registry::m5c(), memory));
+
+    // A heartbeat on the background slot, independent of the channel (M5b spec, section 4).
+    let ctx_sched = Arc::clone(&ctx);
+    std::thread::spawn(move || {
+        let mut sched = loopd::scheduler::Scheduler::new(bxcore::Timestamp::now());
+        loop {
+            sched.tick(&ctx_sched, bxcore::Timestamp::now());
+            std::thread::sleep(std::time::Duration::from_secs(1));
+        }
+    });
+
    if let Err(e) = channel::serve(listener, ctx) {
        eprintln!("loopd: the channel server stopped: {e}");
        return ExitCode::from(1);
@@ -0,0 +1,176 @@
+//! Isolated heartbeat (and, from task 07, cron) turns (M5b spec, section 4).
+
+use crate::baseline::Baseline;
+use crate::channel::{Busy, Context};
+use crate::compact;
+use crate::jobs::{self, Run, RunStatus};
+use crate::session::{Session, SessionError};
+use crate::turn::{Runtime, run_turn};
+use bxcore::{SessionId, Timestamp, TurnEvent};
+use std::fs;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum TickWork {
+    Heartbeat,
+    SkippedBusy { session: SessionId },
+    HeartbeatUnreadable,
+    Job { name: String, session: SessionId },
+    SkippedJob { name: String },
+}
+
+pub struct Scheduler {
+    started: Timestamp,
+    last_heartbeat: Option<Timestamp>,
+    leftovers_closed: bool,
+    heartbeat_unreadable_logged: bool,
+}
+
+impl Scheduler {
+    pub fn new(started: Timestamp) -> Scheduler {
+        Scheduler {
+            started,
+            last_heartbeat: None,
+            leftovers_closed: false,
+            heartbeat_unreadable_logged: false,
+        }
+    }
+
+    pub fn tick(&mut self, ctx: &Context, now: Timestamp) -> Vec<TickWork> {
+        let mut work = Vec::new();
+        if !self.leftovers_closed {
+            close_idle_leftovers(ctx, now);
+            self.leftovers_closed = true;
+        }
+        self.heartbeat(ctx, now, &mut work);
+        work
+    }
+
+    fn heartbeat(&mut self, ctx: &Context, now: Timestamp, work: &mut Vec<TickWork>) {
+        if ctx.cfg.heartbeat.every_s == 0 {
+            return;
+        }
+        let last = self.last_heartbeat.unwrap_or(self.started);
+        let due = now.unix_millis().saturating_sub(last.unix_millis())
+            >= ctx.cfg.heartbeat.every_s.saturating_mul(1000);
+        if !due {
+            return;
+        }
+
+        let id = match SessionId::new("heartbeat") {
+            Ok(id) => id,
+            Err(_) => return,
+        };
+
+        let prompt = ctx.cfg.heartbeat_prompt();
+        let file = match fs::read_to_string(&prompt) {
+            Ok(file) => file,
+            Err(e) => {
+                if !self.heartbeat_unreadable_logged {
+                    eprintln!(
+                        "loopd: {}: {e}; see docs/runbook.md#heartbeat-prompt-unreadable",
+                        prompt.display()
+                    );
+                    self.heartbeat_unreadable_logged = true;
+                }
+                work.push(TickWork::HeartbeatUnreadable);
+                return;
+            }
+        };
+
+        let held = match ctx.hold(id.clone()) {
+            Ok(held) => held,
+            Err(Busy) => {
+                work.push(TickWork::SkippedBusy { session: id });
+                return;
+            }
+        };
+
+        let mut session = match Session::open(&ctx.cfg.paths.home, id.clone()) {
+            Ok(session) => session,
+            Err(SessionError::NotFound(_)) => {
+                let baseline = match Baseline::assemble(&ctx.cfg, &ctx.registry) {
+                    Ok(baseline) => baseline,
+                    Err(e) => {
+                        eprintln!("loopd: heartbeat baseline: {e}");
+                        return;
+                    }
+                };
+                match Session::create(
+                    &ctx.cfg.paths.home,
+                    id.clone(),
+                    baseline,
+                    ctx.cfg.slots.background,
+                ) {
+                    Ok(session) => session,
+                    Err(e) => {
+                        eprintln!("loopd: heartbeat session {}: {e}", id.as_str());
+                        return;
+                    }
+                }
+            }
+            Err(e) => {
+                eprintln!("loopd: heartbeat session {}: {e}", id.as_str());
+                return;
+            }
+        };
+
+        let runtime = Runtime {
+            cfg: &ctx.cfg,
+            client: &ctx.client,
+            port: &*ctx.port,
+            registry: &ctx.registry,
+        };
+
+        let text = format!("It is {}.\n\n{file}", now.to_rfc3339());
+        let mut emit = |_event: &TurnEvent| {};
+        let outcome = run_turn(&mut session, &runtime, &text, None, &mut emit);
+        let outcome = compact::finish_turn(
+            &mut session,
+            &runtime,
+            &*ctx.memory,
+            &text,
+            None,
+            &mut emit,
+            outcome,
+        );
+        if outcome.is_ok() {
+            self.last_heartbeat = Some(now);
+            work.push(TickWork::Heartbeat);
+        }
+        drop(held);
+    }
+}
+
+fn close_idle_leftovers(ctx: &Context, now: Timestamp) {
+    let jobs_dir = ctx.cfg.paths.home.join("jobs");
+    let runs_path = jobs_dir.join("runs.jsonl");
+    let jobs = match jobs::load(&jobs_dir) {
+        Ok(jobs) => jobs,
+        Err(e) => {
+            eprintln!("loopd: cannot read the jobs: {e}");
+            return;
+        }
+    };
+    for job in &jobs {
+        let session = match jobs::live_session(&runs_path, &job.name) {
+            Ok(Some(session)) => session,
+            Ok(None) => continue,
+            Err(e) => {
+                eprintln!("loopd: cannot read the live session for {}: {e}", job.name);
+                continue;
+            }
+        };
+        if let Ok(held) = ctx.hold(session.clone()) {
+            drop(held);
+            let _ = jobs::append(
+                &runs_path,
+                &Run {
+                    time: now,
+                    name: job.name.clone(),
+                    session,
+                    status: RunStatus::Failed,
+                },
+            );
+        }
+    }
+}
@@ -0,0 +1,145 @@
+//! Tests for the heartbeat tick. Do not edit.
+
+mod support;
+
+use bxcore::{SessionId, Timestamp};
+use loopd::channel::Context;
+use loopd::llama::Client;
+use loopd::memory::NoMemory;
+use loopd::scheduler::{Scheduler, TickWork};
+use loopd::tools::Registry;
+use std::sync::Arc;
+use support::{FakeServer, Home, Reply, ScriptedPort};
+
+const CHAT: &str = "/v1/chat/completions";
+const PROMPT: &str = "Look around.\n";
+
+fn sse() -> Reply {
+    let data1 = r#"data: {"choices":[{"finish_reason":null,"index":0,"delta":{"content":"noted"}}],"id":"h","timings":{"cache_n":0,"prompt_n":10,"predicted_n":1}}"#;
+    let data2 = r#"data: {"choices":[{"finish_reason":"stop","index":0,"delta":{}}],"id":"h","timings":{"cache_n":0,"prompt_n":10,"predicted_n":1}}"#;
+    let body = format!("{data1}\n\n{data2}\n\ndata: [DONE]\n\n");
+    Reply::raw(format!(
+        "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
+        body.len()
+    ))
+}
+
+fn ts(s: &str) -> Timestamp {
+    Timestamp::parse(s).unwrap()
+}
+
+fn world(every_s: u64) -> (Home, FakeServer, Arc<Context>) {
+    let home = Home::new();
+    home.write("heartbeat.md", PROMPT);
+    home.write("compact.md", "Summarise.\n");
+    let server = FakeServer::start();
+    let mut cfg = home.config(&server.socket);
+    cfg.heartbeat.every_s = every_s;
+    cfg.heartbeat.prompt = home.dir.join("heartbeat.md");
+    server.route(CHAT, vec![sse()]);
+    let ctx = Arc::new(Context::new(
+        cfg.clone(),
+        Client::new(cfg),
+        Box::new(ScriptedPort::new(vec![])),
+        Registry::m2b(),
+        Box::new(NoMemory),
+    ));
+    (home, server, ctx)
+}
+
+#[test]
+fn every_s_zero_does_nothing() {
+    let (_home, server, ctx) = world(0);
+    let mut sched = Scheduler::new(ts("2026-10-03T07:00:00.000Z"));
+    let work = sched.tick(&ctx, ts("2026-10-03T08:00:00.000Z"));
+    assert!(work.is_empty());
+    assert!(server.requests_to(CHAT).is_empty());
+}
+
+#[test]
+fn a_due_heartbeat_runs_on_the_background_slot_with_a_time_line() {
+    let (home, server, ctx) = world(60);
+    let mut sched = Scheduler::new(ts("2026-10-03T07:00:00.000Z"));
+    let work = sched.tick(&ctx, ts("2026-10-03T07:01:00.000Z"));
+    assert_eq!(work, vec![TickWork::Heartbeat]);
+    let sent = server.requests_to(CHAT);
+    assert_eq!(sent.len(), 1);
+    assert_eq!(sent[0].json()["id_slot"], 1);
+    let body = sent[0].json();
+    let last = body["messages"].as_array().unwrap().last().unwrap();
+    assert_eq!(last["role"], "user");
+    let content = last["content"].as_str().unwrap();
+    assert!(
+        content.starts_with("It is 2026-10-03T07:01:00.000Z."),
+        "{content}"
+    );
+    assert!(content.contains("Look around."), "{content}");
+    let s = loopd::session::Session::open(&home.dir, SessionId::new("heartbeat").unwrap()).unwrap();
+    assert_eq!(s.slot(), 1);
+}
+
+#[test]
+fn a_second_tick_before_every_s_does_not_run() {
+    let (_home, server, ctx) = world(60);
+    let mut sched = Scheduler::new(ts("2026-10-03T07:00:00.000Z"));
+    sched.tick(&ctx, ts("2026-10-03T07:01:00.000Z"));
+    let work = sched.tick(&ctx, ts("2026-10-03T07:01:30.000Z"));
+    assert!(work.is_empty());
+    assert_eq!(server.requests_to(CHAT).len(), 1);
+}
+
+#[test]
+fn a_busy_heartbeat_session_is_skipped() {
+    let (_home, server, ctx) = world(60);
+    let held = ctx
+        .hold(SessionId::new("heartbeat").unwrap())
+        .expect("hold");
+    let mut sched = Scheduler::new(ts("2026-10-03T07:00:00.000Z"));
+    let work = sched.tick(&ctx, ts("2026-10-03T07:01:00.000Z"));
+    assert_eq!(
+        work,
+        vec![TickWork::SkippedBusy {
+            session: SessionId::new("heartbeat").unwrap()
+        }]
+    );
+    assert!(server.requests_to(CHAT).is_empty());
+    drop(held);
+}
+
+#[test]
+fn an_unreadable_prompt_is_skipped_and_names_the_runbook() {
+    let home = Home::new();
+    let path = home.dir.join("heartbeat.md");
+    std::fs::create_dir(&path).unwrap();
+    let server = FakeServer::start();
+    let mut cfg = home.config(&server.socket);
+    cfg.heartbeat.every_s = 60;
+    cfg.heartbeat.prompt = path;
+    let ctx = Arc::new(Context::new(
+        cfg.clone(),
+        Client::new(cfg),
+        Box::new(ScriptedPort::new(vec![])),
+        Registry::m2b(),
+        Box::new(NoMemory),
+    ));
+    let mut sched = Scheduler::new(ts("2026-10-03T07:00:00.000Z"));
+    let work = sched.tick(&ctx, ts("2026-10-03T07:01:00.000Z"));
+    assert_eq!(work, vec![TickWork::HeartbeatUnreadable]);
+    assert!(server.requests_to(CHAT).is_empty());
+}
+
+#[test]
+fn heartbeat_has_no_speaker() {
+    let (home, _server, ctx) = world(60);
+    let mut sched = Scheduler::new(ts("2026-10-03T07:00:00.000Z"));
+    sched.tick(&ctx, ts("2026-10-03T07:01:00.000Z"));
+    let recs = home.records("heartbeat");
+    let user = recs
+        .iter()
+        .find_map(|r| match r {
+            bxcore::LogRecord::User { speaker, .. } => Some(speaker),
+            _ => None,
+        })
+        .unwrap();
+    assert_eq!(user, &None);
+}