boxmaker/docs/plans/M5d2/06-loopd-scheduler-heartbeat.md

M5d-2 task 06: heartbeat tick

Branch: m5d2. Check it: git branch --show-current prints m5d2 and git status --short is empty; otherwise stop. Never switch branch. Commit subject: loopd: heartbeat tick on the background slot Log row: the Task cell is M5d2/06-loopd-scheduler-heartbeat.

Goal

A Scheduler whose tick(&Context, now) runs the standing heartbeat session when every_s has elapsed since the last heartbeat turn, or since Scheduler::new(started) if none (spec 4.1). The user text is It is <RFC 3339 UTC>.\n\n plus the prompt file. Slot is background. Speaker is None. Skip if the session is busy. Skip and log once if the prompt cannot be read (heartbeat-prompt-unreadable).

Tests call tick(ctx, now) with injected time. No sleeps (T27).

Every place that changes

Place	What happens there
crates/loopd/src/channel.rs	Held is pub; Context::hold
crates/loopd/src/scheduler.rs	new: Scheduler, TickWork, tick
crates/loopd/src/lib.rs	pub mod scheduler;
crates/loopd/src/main.rs	a thread that ticks once a second, before channel::serve
docs/runbook.md	append the given heartbeat-prompt-unreadable entry
crates/loopd/tests/scheduler_heartbeat.rs	new, given
everything else	no change: not cron job runs (task 07)

Files

• Copy, from docs/plans/M5d2/files/ (the same path under it): the new test crates/loopd/tests/scheduler_heartbeat.rs.
• Append, from docs/plans/M5d2/files/runbook-heartbeat-prompt-unreadable.md, to docs/runbook.md.
• Create: crates/loopd/src/scheduler.rs (skeleton below).
• Modify: crates/loopd/src/channel.rs, crates/loopd/src/lib.rs, crates/loopd/src/main.rs, docs/implementer-log.md.
• Do not touch: any other file.

Facts you need

• Context::hold. Same insert as handle: if the id is already in busy, Err(()); else insert and return Held. Held must be pub so the given test can keep it. Drop already removes the id.

• Due. every_s == 0 → no heartbeat work. Else now.unix_millis().saturating_sub(last.unix_millis()) >= every_s.saturating_mul(1000) where last is last_heartbeat.unwrap_or(started). 07:00 to 07:01 with every_s = 60 is due. 07:01 then 07:01:30 is not.

• Prompt first, then hold. If read_to_string(cfg.heartbeat_prompt()) fails, do not hold, push HeartbeatUnreadable, log loopd: {path}: {e}; see docs/runbook.md#heartbeat-prompt-unreadable only while heartbeat_unreadable_logged is false, then set that flag. Do not update last_heartbeat. A directory at the prompt path is the given-test stand-in for an unreadable file (T27: no chmod 000).

• Session id heartbeat. Session::open; NotFound → Session::create with Baseline::assemble(&ctx.cfg, &ctx.registry) and ctx.cfg.slots.background. Other open errors: log and skip (do not panic).

• The turn. User text format!("It is {}.\n\n{file}", now.to_rfc3339()) where file is the prompt contents as read (no extra trim beyond what the file has). run_turn then compact::finish_turn with &*ctx.memory, speaker None, on_event a no-op. Then last_heartbeat = Some(now), push Heartbeat.

• TickWork. Include Job / SkippedJob now so task 07 does not change the type. Task 06 never pushes them.

• close_leftovers. On the first tick only, call jobs::close_leftovers(&ctx.cfg.paths.home.join("jobs").join("runs.jsonl"), now) and ignore a missing-file Ok(0). A leftover that is currently holding its session is not closed: skip names whose live_session session is in busy (or: close only when hold of that session would succeed, then immediately drop that hold). Task 07's live-skip test depends on this. If close_leftovers has no busy-set filter, add a loop here: for each live name, if hold(session) is Ok, drop it and treat as leftover (append Failed yourself or call a helper). Simplest: do not call close_leftovers blindly. Walk: if live_session is Some(id) and hold(id) is Ok, drop the hold and append a Failed for that name (or call close_leftovers only when no job session is held). The given leftover test does not hold; the live-skip test does. Closing only leftovers that are not busy is required.

• Main thread. After Context::new and before channel::serve:

  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));
      }
  });
  

  That sleep is production only. Tests never sleep.

• channel.rs stays ≤ 500 lines. hold is about 15 lines.

• The items the edits use. This table is complete.

  Item	What it does
  Session::slot / create / open	standing session
  run_turn / finish_turn	the pulse
  Baseline::assemble	new session
  jobs::close_leftovers / live_session / append	leftovers
  Timestamp::unix_millis / to_rfc3339	due and the time line

The edits, exactly

crates/loopd/src/channel.rs

C1. struct Held becomes pub struct Held.

C2. Inside impl Context, after new, add:

    /// Mark `id` busy. `Err(())` 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<'_>, ()> {
        let mut busy = self.busy.lock().unwrap_or_else(|p| p.into_inner());
        if busy.contains(&id) {
            return Err(());
        }
        busy.insert(id.clone());
        drop(busy);
        Ok(Held { ctx: self, id })
    }

crates/loopd/src/scheduler.rs

Write this module. Fill tick. Cron jobs stay a comment until task 07: after heartbeat, do not iterate job files yet.

//! Isolated heartbeat (and, from task 07, cron) turns (M5d spec, section 4).

use crate::channel::Context;
use bxcore::{SessionId, Timestamp};

#[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 _ = (ctx, now);
        Vec::new()
    }
}

Replace the tick body in this task. leftovers_closed is set true after the first leftover pass so it runs once.

Steps

• 1. Copy the given test.

  cp docs/plans/M5d2/files/crates/loopd/tests/scheduler_heartbeat.rs crates/loopd/tests/
  

• 2. See it fail. cargo build --tests --keep-going -p loopd, timeout 600000 ms: test target scheduler_heartbeat does not compile (E0432 unresolved loopd::scheduler, E0599 no method hold). Every other target compiles.

• 3. The edits. C1, C2, scheduler.rs, pub mod scheduler;, main thread. Fill tick. cargo check -p loopd prints no error and no warning. wc -l crates/loopd/src/scheduler.rs and wc -l crates/loopd/src/channel.rs each ≤ 500.

• 4. The tests. cargo test -p loopd --test scheduler_heartbeat --test channel --test session_slot: every test passes, 6 in scheduler_heartbeat.

• 5. The runbook.

  { echo; cat docs/plans/M5d2/files/runbook-heartbeat-prompt-unreadable.md; } >> docs/runbook.md
  

  grep -c '^## heartbeat-prompt-unreadable$' docs/runbook.md prints 1.

• 6. Run the gate. cargo fmt --all, then make gate, timeout 600000 ms. Last line gate: ok. Test count 1551 (1545 before, plus 6).

• 7. Log and commit. Task cell M5d2/06-loopd-scheduler-heartbeat. Stage with git add crates/loopd docs/runbook.md docs/implementer-log.md Cargo.lock, then commit as AGENTS.md says.

Done when

• make gate prints gate: ok with 1551 tests.
• git diff HEAD~1 --stat lists only crates/loopd/src/channel.rs, crates/loopd/src/scheduler.rs, crates/loopd/src/lib.rs, crates/loopd/src/main.rs, crates/loopd/tests/scheduler_heartbeat.rs, docs/runbook.md and the log.

Stop and report if

• The branch check fails; a "before" block is not found exactly once; step 2 differs; cargo check warns; a test fails; the gate count is not 1551; you sleep in a test; you post to Mattermost; you send slots.main; you set a speaker; you chmod 000; Held stays private.

What the given tests pin

every_s 0 does nothing; a due heartbeat sends id_slot 1 with a time line and the file text, session heartbeat has slot() == 1; a second tick before every_s does not run; a held session is SkippedBusy with no request; a directory at the prompt path is HeartbeatUnreadable with no request; the user record has speaker: None.