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`.

**First action.** Run this before any Read of crate source. Do not `ls` `crates/loopd/tests/`.
Do not Read `tests/support/`, `session.rs`, `turn.rs`, `compact.rs`, `config.rs`, `jobs.rs`,
`baseline.rs`, `bxcore`, or `docs/implementer-log.md`. The signatures below are complete.

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

## 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 `tick`s 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.

At step 7, read only the first 12 lines of `docs/implementer-log.md`. The rest of that file
fills the session (~360 kB).

## Facts you need

- **`Context::hold`.** Same insert as `handle`: if the id is already in `busy`, `Err(Busy)`;
  else insert and return `Held`. `Held` must be `pub` so the given test can keep it. `Drop`
  already removes the id. Do not return `Result<_, ()>`: clippy `result_unit_err` is denied.
- **Due.** `every_s == 0` → no heartbeat work (do not `return` from `tick`; task 07 runs jobs after this). 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 `hold`ing 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`:

  ```rust
  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.
- **Uses.** The skeleton compiles with only `Context` and `Timestamp`. When filling `heartbeat`,
  add `use std::fs`, `Baseline`, `compact`, `Session`/`SessionError`, `Runtime`/`run_turn`,
  `TurnEvent`. When filling leftovers, add `jobs::{self, Run, RunStatus}`.
- The items the edits use. This table is complete.

  | Item | Signature |
  |---|---|
  | `Session::open` | `open(home: &Path, id: SessionId) -> Result<Session, SessionError>` |
  | `Session::create` | `create(home, id, baseline, slot: u32) -> Result<Session, SessionError>` |
  | `SessionError::NotFound` | `_` |
  | `Baseline::assemble` | `assemble(&ctx.cfg, &ctx.registry) -> Result<Baseline, _>` |
  | `Runtime` | `{ cfg: &ctx.cfg, client: &ctx.client, port: &*ctx.port, registry: &ctx.registry }` |
  | `run_turn` | `run_turn(&mut session, &runtime, &text, None, &mut emit)` |
  | `compact::finish_turn` | `finish_turn(&mut session, &runtime, &*ctx.memory, &text, None, &mut emit, outcome)` |
  | `jobs::load` | `load(dir: &Path) -> Result<Vec<Job>, JobsError>` |
  | `jobs::live_session` | `live_session(ledger, name) -> Result<Option<SessionId>, _>` |
  | `jobs::append` | `append(ledger, &Run { time, name, session, status })` |
  | `RunStatus::Failed` | leftover close |
  | `Timestamp` | `unix_millis() -> u64`, `to_rfc3339() -> String`, `now()` |

## The edits, exactly

### `crates/loopd/src/channel.rs`

**C1.** `struct Held` becomes `pub struct Held`. Immediately after that struct (before
`impl Drop for Held`), add:

```rust
/// The session is already in the busy set.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Busy;
```

**C2.** Inside `impl Context`, after `new`, add:

```rust
    /// 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 })
    }
```

### `crates/loopd/src/scheduler.rs`

Write this module. `tick` is already glue. Fill `heartbeat` then `close_idle_leftovers`,
`cargo check -p loopd` after each (T25). Do not iterate job files (task 07). Do not add
`run_jobs`.

```rust
//! 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 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>) {
        let _ = (ctx, now, work);
        todo!()
    }
}

fn close_idle_leftovers(ctx: &Context, now: Timestamp) {
    let _ = (ctx, now);
    todo!()
}
```

## Steps

- [ ] **1. Copy the given test.**

  ```sh
  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` (fill `heartbeat` then `close_idle_leftovers`,
  `cargo check -p loopd` after each), `pub mod scheduler;`, main thread. `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.**

  ```sh
  { 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). After `gate: ok`, do not add up
  `test result:` lines.
- [ ] **7. Log and commit.** Task cell `M5d2/06-loopd-scheduler-heartbeat`. Read only the first
  12 lines of `docs/implementer-log.md`; insert the row under the header. 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's last line is not `gate: ok`; you sleep in a test; you post to
  Mattermost; you send `slots.main`; you set a speaker; you `chmod 000`; `Held` stays private;
  `hold` returns `Result<_, ()>`.

## 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`.