loopd: cron tick, new session per run

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>

crates/loopd/src/scheduler.rs

@@ -3,6 +3,7 @@
 use crate::baseline::Baseline;
 use crate::channel::{Busy, Context};
 use crate::compact;
+use crate::cron;
 use crate::jobs::{self, Run, RunStatus};
 use crate::session::{Session, SessionError};
 use crate::turn::{Runtime, run_turn};
@@ -42,6 +43,133 @@ impl Scheduler {
             self.leftovers_closed = true;
         }
         self.heartbeat(ctx, now, &mut work);
+
+        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: {e}");
+                return work;
+            }
+        };
+        for job in &jobs {
+            match cron::due(&job.schedule, now) {
+                Ok(true) => {}
+                Ok(false) => continue,
+                Err(e) => {
+                    eprintln!("loopd: {e}");
+                    continue;
+                }
+            }
+            match jobs::live_session(&runs_path, &job.name) {
+                Ok(Some(_)) => {
+                    work.push(TickWork::SkippedJob {
+                        name: job.name.clone(),
+                    });
+                    continue;
+                }
+                Ok(None) => {}
+                Err(e) => {
+                    eprintln!("loopd: cannot read the live session for {}: {e}", job.name);
+                    continue;
+                }
+            }
+            let id = match jobs::session_id(&job.name, now) {
+                Ok(id) => id,
+                Err(e) => {
+                    eprintln!("loopd: {e}");
+                    continue;
+                }
+            };
+            let held = match ctx.hold(id.clone()) {
+                Ok(held) => held,
+                Err(Busy) => {
+                    work.push(TickWork::SkippedJob {
+                        name: job.name.clone(),
+                    });
+                    continue;
+                }
+            };
+            let baseline = match Baseline::assemble(&ctx.cfg, &ctx.registry) {
+                Ok(baseline) => baseline,
+                Err(e) => {
+                    eprintln!("loopd: job {} baseline: {e}", job.name);
+                    drop(held);
+                    continue;
+                }
+            };
+            let mut session = match Session::create(
+                &ctx.cfg.paths.home,
+                id.clone(),
+                baseline,
+                ctx.cfg.slots.background,
+            ) {
+                Ok(session) => session,
+                Err(SessionError::Exists(_)) => {
+                    match Session::open(&ctx.cfg.paths.home, id.clone()) {
+                        Ok(session) => session,
+                        Err(e) => {
+                            eprintln!("loopd: job {} session {}: {e}", job.name, id.as_str());
+                            drop(held);
+                            continue;
+                        }
+                    }
+                }
+                Err(e) => {
+                    eprintln!("loopd: job {} session {}: {e}", job.name, id.as_str());
+                    drop(held);
+                    continue;
+                }
+            };
+            let _ = jobs::append(
+                &runs_path,
+                &Run {
+                    time: now,
+                    name: job.name.clone(),
+                    session: id.clone(),
+                    status: RunStatus::Started,
+                },
+            );
+            let runtime = Runtime {
+                cfg: &ctx.cfg,
+                client: &ctx.client,
+                port: &*ctx.port,
+                registry: &ctx.registry,
+            };
+            let text = format!("It is {}.\n\n{}", now.to_rfc3339(), job.prompt);
+            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,
+            );
+            let status = if outcome.is_ok() {
+                RunStatus::Finished
+            } else {
+                RunStatus::Failed
+            };
+            let _ = jobs::append(
+                &runs_path,
+                &Run {
+                    time: now,
+                    name: job.name.clone(),
+                    session: id.clone(),
+                    status,
+                },
+            );
+            drop(held);
+            work.push(TickWork::Job {
+                name: job.name.clone(),
+                session: id,
+            });
+        }
+
         work
     }
 

crates/loopd/tests/scheduler_cron.rs

@@ -0,0 +1,189 @@
+//! Tests for the cron tick. Do not edit.
+
+mod support;
+
+use bxcore::Timestamp;
+use loopd::channel::Context;
+use loopd::jobs::{self, RunStatus};
+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";
+
+fn sse() -> Reply {
+    let data1 = r#"data: {"choices":[{"finish_reason":null,"index":0,"delta":{"content":"done"}}],"id":"j","timings":{"cache_n":0,"prompt_n":10,"predicted_n":1}}"#;
+    let data2 = r#"data: {"choices":[{"finish_reason":"stop","index":0,"delta":{}}],"id":"j","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(schedule: &str) -> (Home, FakeServer, Arc<Context>) {
+    let home = Home::new();
+    home.write("compact.md", "Summarise.\n");
+    home.write(
+        "jobs/mail.toml",
+        &format!("schedule = \"{schedule}\"\nprompt = \"Check mail.\"\n"),
+    );
+    let server = FakeServer::start();
+    let cfg = home.config(&server.socket);
+    server.route(CHAT, vec![sse(), 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 a_due_job_starts_a_new_session_on_the_background_slot() {
+    let (home, server, ctx) = world("0 7 * * *");
+    let mut sched = Scheduler::new(ts("2026-10-03T06:00:00.000Z"));
+    let work = sched.tick(&ctx, ts("2026-10-03T07:00:00.000Z"));
+    let want = jobs::session_id("mail", ts("2026-10-03T07:00:00.000Z")).unwrap();
+    assert_eq!(
+        work,
+        vec![TickWork::Job {
+            name: "mail".to_string(),
+            session: want.clone()
+        }]
+    );
+    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();
+    let content = last["content"].as_str().unwrap();
+    assert!(
+        content.starts_with("It is 2026-10-03T07:00:00.000Z."),
+        "{content}"
+    );
+    assert!(content.contains("Check mail."), "{content}");
+    let s = loopd::session::Session::open(&home.dir, want).unwrap();
+    assert_eq!(s.slot(), 1);
+}
+
+#[test]
+fn a_job_not_due_does_not_run() {
+    let (_home, server, ctx) = world("0 7 * * *");
+    let mut sched = Scheduler::new(ts("2026-10-03T06:00:00.000Z"));
+    let work = sched.tick(&ctx, ts("2026-10-03T06:59:00.000Z"));
+    assert!(work.is_empty());
+    assert!(server.requests_to(CHAT).is_empty());
+}
+
+#[test]
+fn a_live_previous_run_skips_the_tick() {
+    let (home, server, ctx) = world("* * * * *");
+    let ledger = home.dir.join("jobs/runs.jsonl");
+    let previous = jobs::session_id("mail", ts("2026-10-03T06:59:00.000Z")).unwrap();
+    jobs::append(
+        &ledger,
+        &jobs::Run {
+            time: ts("2026-10-03T06:59:00.000Z"),
+            name: "mail".to_string(),
+            session: previous.clone(),
+            status: RunStatus::Started,
+        },
+    )
+    .unwrap();
+    let held = ctx.hold(previous).expect("hold");
+    let mut sched = Scheduler::new(ts("2026-10-03T06:00:00.000Z"));
+    let work = sched.tick(&ctx, ts("2026-10-03T07:00:00.000Z"));
+    assert_eq!(
+        work,
+        vec![TickWork::SkippedJob {
+            name: "mail".to_string()
+        }]
+    );
+    assert!(server.requests_to(CHAT).is_empty());
+    let text = home.read("jobs/runs.jsonl");
+    assert!(text.contains("started"), "{text}");
+    assert!(
+        !text.contains("failed"),
+        "a leftover still held is not closed as failed: {text}"
+    );
+    drop(held);
+}
+
+#[test]
+fn leftover_started_is_failed_on_the_first_tick() {
+    let (home, _server, ctx) = world("* * * * *");
+    let ledger = home.dir.join("jobs/runs.jsonl");
+    let previous = jobs::session_id("mail", ts("2026-10-03T06:59:00.000Z")).unwrap();
+    jobs::append(
+        &ledger,
+        &jobs::Run {
+            time: ts("2026-10-03T06:59:00.000Z"),
+            name: "mail".to_string(),
+            session: previous.clone(),
+            status: RunStatus::Started,
+        },
+    )
+    .unwrap();
+    let mut sched = Scheduler::new(ts("2026-10-03T07:00:00.000Z"));
+    sched.tick(&ctx, ts("2026-10-03T07:00:00.000Z"));
+    let text = home.read("jobs/runs.jsonl");
+    assert!(
+        text.contains("failed"),
+        "leftover started was closed: {text}"
+    );
+    assert!(text.contains(previous.as_str()), "{text}");
+}
+
+#[test]
+fn a_cron_turn_has_no_speaker() {
+    let (home, _server, ctx) = world("0 7 * * *");
+    let mut sched = Scheduler::new(ts("2026-10-03T06:00:00.000Z"));
+    sched.tick(&ctx, ts("2026-10-03T07:00:00.000Z"));
+    let id = jobs::session_id("mail", ts("2026-10-03T07:00:00.000Z")).unwrap();
+    let recs = {
+        let text = home.read(&format!("sessions/{}/0.jsonl", id.as_str()));
+        text.lines()
+            .map(|l| serde_json::from_str::<bxcore::LogRecord>(l).unwrap())
+            .collect::<Vec<_>>()
+    };
+    let user = recs
+        .iter()
+        .find_map(|r| match r {
+            bxcore::LogRecord::User { speaker, .. } => Some(speaker),
+            _ => None,
+        })
+        .unwrap();
+    assert_eq!(user, &None);
+}
+
+#[test]
+fn two_due_minutes_make_two_sessions() {
+    let (home, server, ctx) = world("* * * * *");
+    let mut sched = Scheduler::new(ts("2026-10-03T06:58:00.000Z"));
+    sched.tick(&ctx, ts("2026-10-03T06:59:00.000Z"));
+    sched.tick(&ctx, ts("2026-10-03T07:00:00.000Z"));
+    assert_eq!(server.requests_to(CHAT).len(), 2);
+    let a = jobs::session_id("mail", ts("2026-10-03T06:59:00.000Z")).unwrap();
+    let b = jobs::session_id("mail", ts("2026-10-03T07:00:00.000Z")).unwrap();
+    assert_ne!(a, b);
+    assert!(
+        home.dir
+            .join(format!("sessions/{}/0.jsonl", a.as_str()))
+            .is_file()
+    );
+    assert!(
+        home.dir
+            .join(format!("sessions/{}/0.jsonl", b.as_str()))
+            .is_file()
+    );
+}

docs/implementer-log.md

@@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone.
 
 | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model |
 |---|---|---|---|---|---|---|---|
+| M5d2/07-loopd-scheduler-cron | 2026-10-03 | done | 1 | pass | none | Copied the given test `crates/loopd/tests/scheduler_cron.rs` from `docs/plans/M5d2/files/`. Added the cron loop into `tick` in `crates/loopd/src/scheduler.rs`, right after `self.heartbeat(...)` and before `work` is returned (heartbeat untouched): load `jobs` via `jobs::load(&ctx.cfg.paths.home.join("jobs"))`, on `Err` log `loopd: {e}` and return; for each job in order check `cron::due` (`Err` log+skip, `Ok(false)` skip), then `jobs::live_session` on `{home}/jobs/runs.jsonl` (`Some` push `SkippedJob` and continue), then `session_id` (`Err` log+skip), then `ctx.hold` (`Busy` push `SkippedJob`). On a hold: `Baseline::assemble`, `Session::create` on `slots.background` (on `Exists`, `open` instead), append `Started`, run `run_turn` then `compact::finish_turn` with user text `format!("It is {}.\n\n{}", now.to_rfc3339(), job.prompt)` and speaker `None`, append `Finished` on an ok turn else `Failed`, drop the hold, push `TickWork::Job { name, session }`. Added `use crate::cron;` to the imports. `cargo check -p loopd` clean, no error or warning; `wc -l` prints 304, under 500. All four suites pass: `cron` 11, `jobs` 8, `scheduler_cron` 6, `scheduler_heartbeat` 6. `cargo fmt --all` wrapped the `Session::create` call and its `Exists` arm as a nested match; `make gate` printed `gate: ok` with 1557 tests (1551 before, plus the 6 of `scheduler_cron.rs`). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b |
 | M5d2/06-loopd-scheduler-heartbeat | 2026-10-03 | done | 2 | fail | none | Copied the given test `crates/loopd/tests/scheduler_heartbeat.rs` from `docs/plans/M5d2/files/`. Made `crates/loopd/src/scheduler.rs` from the skeleton: `tick` closes idle leftovers once on the first tick, then runs `heartbeat`; `heartbeat` returns early when `every_s == 0` or when not due (`now - last_heartbeat.unwrap_or(started) >= every_s * 1000`), reads the prompt first (pushing `HeartbeatUnreadable` and logging once via `heartbeat_unreadable_logged` on a failed read, without holding), then holds `heartbeat` (pushing `SkippedBusy` on `Busy`), opens it or creates it on `NotFound` with `Baseline::assemble` and the background slot, and runs the turn with `speaker None` and a no-op emit, updating `last_heartbeat` and pushing `Heartbeat` only on an ok outcome. `close_idle_leftovers` loads the jobs and, for each with a live `Started` session, holds it -- dropping the hold and appending a `Failed` run when the hold succeeds and skipping when it is `Busy`, rather than calling `jobs::close_leftovers` blindly, so a busy session is never closed. Added `pub mod scheduler;` after `pub mod llama;` in `lib.rs`. Wired `run_serve` to spawn a thread that `tick`s once a second before `channel::serve`. Made `Held` pub, added `pub struct Busy` and `Context::hold` in `channel.rs`. Appended the given `heartbeat-prompt-unreadable` entry to `docs/runbook.md` (heading count 1). First `make gate` failed on clippy `single_match` in `close_idle_leftovers`, fixed by converting the match to `if let Ok(held)` (the `Busy` case is the skip); the second run printed `gate: ok` with 1551 tests (1545 before, plus the 6 of `scheduler_heartbeat.rs`). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b |
 | M5d2/05-loopd-jobs | 2026-10-03 | done | 2 | fail | none | Copied the given test `crates/loopd/tests/jobs.rs` from `docs/plans/M5d2/files/`. Made `crates/loopd/src/jobs.rs` from the skeleton (192 lines, under 500): `session_id` builds `job-<name>-<yyyymmdd>t<hhmm>` from `to_rfc3339()` and maps a `SessionId::new` failure to `JobsError::Name`; `load` reads the dir (NotFound -> empty, any other IO error or an unreadable file -> Invalid naming the path), skips names not ending in `.toml`, validates the stem against `[a-z0-9-]+` on the bytes (empty or bad -> Name), parses each with a private `File { schedule, prompt }` and `deny_unknown_fields`; `append` creates the parent dir, opens in append mode, writes one JSON object plus a newline and `sync_all`; `live_session` walks every line of the ledger (missing file -> None) and returns the last matching run's session only when that run is `Started`; `close_leftovers` closes each name whose last run is `Started` with a `Failed` run carrying the same session and `now`, and returns the count (missing file -> 0). Added `pub mod jobs;` after `pub mod llama;` in `lib.rs` (rustfmt order). Wired `run_serve` to call `jobs::load(&cfg.paths.home.join("jobs"))` after `check_summarise`, printing `loopd: {e}` and exiting 1 on error, without creating the directory. Appended the given `jobs-invalid` entry to `docs/runbook.md` (heading count 1). `cargo check -p loopd` clean, no error or warning; all 8 `jobs` tests pass. First `make gate` failed on clippy `collapsible-if` in `append` (the parent-dir guard), which I fixed by combining the two guards into one `if let Some(parent) = path.parent() && !parent.as_os_str().is_empty()`; the second run printed `gate: ok` with 1545 tests (1537 before, plus the 8 of `jobs.rs`). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b |
 | M5d2/04-loopd-cron | 2026-10-03 | done | 2 | pass | none | Copied the given test `crates/loopd/tests/cron.rs` from `docs/plans/M5d2/files/`. Step 2 (`cargo build --tests --keep-going -p loopd`) showed exactly the named failure: `cron` with E0432 (`unresolved import loopd::cron`); every other target compiled. Made the module `cron.rs` from the skeleton: `due` splits on whitespace, requires five pieces, reads the civil fields from `to_rfc3339`, computes DOW as `(unix_millis/1000/86400 + 4) % 7`, ANDs the five `match_field` checks; `match_field` handles `*`, `*/n` (with `n == 0` rejected via `is_multiple_of`), `n-m` ranges, and bare numbers, each range-checked against the field bounds. Added `pub mod cron;` after `pub mod config;` in `lib.rs` (rustfmt order). `cargo check -p loopd` clean, no error or warning. `cargo test -p loopd --test cron` passes all 11. `wc -l crates/loopd/src/cron.rs` prints 150, well under 500. `git diff --stat` lists only `cron.rs` and `lib.rs`; `git status --short` shows `cron.rs` and the test as untracked. `cargo fmt --all` changed nothing after the edits. `make gate` printed `gate: ok` with 1537 tests (1526 before, plus the 11 of `cron.rs`), matching the task's expected count. No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b |