From 670b5bbabe64a5ae834ae05e3f8204f7d56e0863 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 3 Oct 2026 14:20:19 -0700 Subject: [PATCH] loopd: job files, run ledger, jobs-invalid Co-Authored-By: ornith-1.5-35b-a3b --- crates/loopd/src/jobs.rs | 193 +++++++++++++++++++++++++++++++++++++ crates/loopd/src/lib.rs | 1 + crates/loopd/src/main.rs | 6 ++ crates/loopd/tests/jobs.rs | 151 +++++++++++++++++++++++++++++ docs/implementer-log.md | 1 + docs/runbook.md | 25 +++++ 6 files changed, 377 insertions(+) create mode 100644 crates/loopd/src/jobs.rs create mode 100644 crates/loopd/tests/jobs.rs diff --git a/crates/loopd/src/jobs.rs b/crates/loopd/src/jobs.rs new file mode 100644 index 0000000..f191772 --- /dev/null +++ b/crates/loopd/src/jobs.rs @@ -0,0 +1,193 @@ +//! Owner-written job files and the run ledger (M5d spec, section 4.2). + +use std::collections::BTreeMap; +use std::io::{ErrorKind, Write}; +use std::path::{Path, PathBuf}; + +use bxcore::{SessionId, Timestamp}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Job { + pub name: String, + pub schedule: String, + pub prompt: String, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RunStatus { + Started, + Finished, + Failed, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Run { + pub time: Timestamp, + pub name: String, + pub session: SessionId, + pub status: RunStatus, +} + +#[derive(Debug)] +pub enum JobsError { + Name(String), + Invalid { path: PathBuf, detail: String }, +} + +impl std::fmt::Display for JobsError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + JobsError::Name(name) => write!( + f, + "job name {name:?} is not a valid job; see docs/runbook.md#jobs-invalid" + ), + JobsError::Invalid { path, detail } => write!( + f, + "{}: {detail}; see docs/runbook.md#jobs-invalid", + path.display() + ), + } + } +} + +impl std::error::Error for JobsError {} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct File { + schedule: String, + prompt: String, +} + +fn invalid(path: &Path, detail: impl std::fmt::Display) -> JobsError { + JobsError::Invalid { + path: path.to_path_buf(), + detail: detail.to_string(), + } +} + +pub fn session_id(name: &str, now: Timestamp) -> Result { + let rfc = now.to_rfc3339(); + let ymd = format!("{}{}{}", &rfc[0..4], &rfc[5..7], &rfc[8..10]); + let hm = format!("{}{}", &rfc[11..13], &rfc[14..16]); + let id = format!("job-{name}-{ymd}t{hm}"); + SessionId::new(&id).map_err(|_| JobsError::Name(name.to_string())) +} + +pub fn load(dir: &Path) -> Result, JobsError> { + let mut entries: Vec = match std::fs::read_dir(dir) { + Ok(entries) => { + let mut names = Vec::new(); + for entry in entries { + let entry = entry.map_err(|e| invalid(dir, e))?; + names.push(entry.path()); + } + names + } + Err(e) if e.kind() == ErrorKind::NotFound => return Ok(Vec::new()), + Err(e) => return Err(invalid(dir, e)), + }; + entries.sort(); + + let mut jobs = Vec::new(); + for path in entries { + let file_name = match path.file_name().and_then(|n| n.to_str()) { + Some(name) => name, + None => return Err(invalid(&path, "the job file name is not UTF-8")), + }; + let Some(stem) = file_name.strip_suffix(".toml") else { + continue; + }; + if stem.is_empty() || !stem.bytes().all(is_job_name_byte) { + return Err(JobsError::Name(stem.to_string())); + } + let content = std::fs::read_to_string(&path).map_err(|e| invalid(&path, e))?; + let file: File = toml::from_str(&content).map_err(|e| invalid(&path, e))?; + jobs.push(Job { + name: stem.to_string(), + schedule: file.schedule, + prompt: file.prompt, + }); + } + Ok(jobs) +} + +fn is_job_name_byte(b: u8) -> bool { + matches!(b, b'a'..=b'z' | b'0'..=b'9' | b'-') +} + +pub fn append(path: &Path, run: &Run) -> Result<(), JobsError> { + if let Some(parent) = path.parent() + && !parent.as_os_str().is_empty() + { + std::fs::create_dir_all(parent).map_err(|e| invalid(parent, e))?; + } + let line = serde_json::to_string(run).map_err(|e| invalid(path, e))?; + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(path) + .map_err(|e| invalid(path, e))?; + file.write_all(line.as_bytes()) + .map_err(|e| invalid(path, e))?; + file.write_all(b"\n").map_err(|e| invalid(path, e))?; + file.sync_all().map_err(|e| invalid(path, e))?; + Ok(()) +} + +pub fn live_session(path: &Path, name: &str) -> Result, JobsError> { + let content = read_ledger(path)?; + let mut last: Option = None; + for line in content.lines() { + if line.trim().is_empty() { + continue; + } + let run: Run = serde_json::from_str(line).map_err(|e| invalid(path, e))?; + if run.name == name { + last = Some(run); + } + } + match last { + Some(run) if run.status == RunStatus::Started => Ok(Some(run.session)), + _ => Ok(None), + } +} + +pub fn close_leftovers(path: &Path, now: Timestamp) -> Result { + let content = read_ledger(path)?; + let mut last_by_name: BTreeMap = BTreeMap::new(); + for line in content.lines() { + if line.trim().is_empty() { + continue; + } + let run: Run = serde_json::from_str(line).map_err(|e| invalid(path, e))?; + last_by_name.insert(run.name.clone(), run); + } + let mut closed = 0; + for run in last_by_name.values() { + if run.status == RunStatus::Started { + append( + path, + &Run { + time: now, + name: run.name.clone(), + session: run.session.clone(), + status: RunStatus::Failed, + }, + )?; + closed += 1; + } + } + Ok(closed) +} + +fn read_ledger(path: &Path) -> Result { + match std::fs::read_to_string(path) { + Ok(content) => Ok(content), + Err(e) if e.kind() == ErrorKind::NotFound => Ok(String::new()), + Err(e) => Err(invalid(path, e)), + } +} diff --git a/crates/loopd/src/lib.rs b/crates/loopd/src/lib.rs index 085f094..f129408 100644 --- a/crates/loopd/src/lib.rs +++ b/crates/loopd/src/lib.rs @@ -7,6 +7,7 @@ pub mod compact; pub mod config; pub mod cron; pub mod http; +pub mod jobs; pub mod llama; pub mod memory; pub mod selftest; diff --git a/crates/loopd/src/main.rs b/crates/loopd/src/main.rs index fcdf480..e64d904 100644 --- a/crates/loopd/src/main.rs +++ b/crates/loopd/src/main.rs @@ -12,6 +12,7 @@ use bxcore::instance; use loopd::broker_port::{BrokerPort, NoBroker, not_configured_line}; use loopd::channel::{self, Context}; use loopd::config::Config; +use loopd::jobs; use loopd::llama::Client; use loopd::memory::{MEMORY_WAIT, MemoryPort}; use loopd::selftest::{SelfTestError, run}; @@ -84,6 +85,11 @@ fn run_serve(path: &Path) -> ExitCode { return ExitCode::from(1); } + if let Err(e) = jobs::load(&cfg.paths.home.join("jobs")) { + eprintln!("loopd: {e}"); + return ExitCode::from(1); + } + // One loopd per home (M5b spec, section 9a). The lock comes before the socket is touched and // before the self-test, so a second copy neither takes the socket of the one that runs nor // sends its slot a prompt. Held until the process ends. diff --git a/crates/loopd/tests/jobs.rs b/crates/loopd/tests/jobs.rs new file mode 100644 index 0000000..6e28f1a --- /dev/null +++ b/crates/loopd/tests/jobs.rs @@ -0,0 +1,151 @@ +//! Tests for job files and the run ledger. Do not edit. + +use bxcore::{SessionId, Timestamp}; +use loopd::jobs::{self, JobsError, RunStatus}; + +fn ts() -> Timestamp { + Timestamp::parse("2026-10-03T07:00:00.000Z").unwrap() +} + +struct Scratch { + dir: std::path::PathBuf, +} + +impl Scratch { + fn new() -> Scratch { + let n = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let dir = std::env::temp_dir().join(format!("loopd-jobs-{}-{n}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + Scratch { dir } + } +} + +impl Drop for Scratch { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.dir); + } +} + +#[test] +fn a_valid_job_file_loads() { + let home = Scratch::new(); + let jobs = home.dir.join("jobs"); + std::fs::create_dir(&jobs).unwrap(); + std::fs::write( + jobs.join("mail.toml"), + "schedule = \"0 7 * * *\"\nprompt = \"Check mail.\"\n", + ) + .unwrap(); + let loaded = jobs::load(&jobs).unwrap(); + assert_eq!(loaded.len(), 1); + assert_eq!(loaded[0].name, "mail"); + assert_eq!(loaded[0].schedule, "0 7 * * *"); + assert_eq!(loaded[0].prompt, "Check mail."); +} + +#[test] +fn unknown_keys_are_errors_with_the_runbook() { + let home = Scratch::new(); + let jobs = home.dir.join("jobs"); + std::fs::create_dir(&jobs).unwrap(); + std::fs::write( + jobs.join("mail.toml"), + "schedule = \"* * * * *\"\nprompt = \"x\"\nwhen = \"now\"\n", + ) + .unwrap(); + let err = jobs::load(&jobs).unwrap_err(); + let s = err.to_string(); + assert!(s.contains("mail.toml"), "{s}"); + assert!(s.ends_with("see docs/runbook.md#jobs-invalid"), "{s}"); +} + +#[test] +fn a_bad_name_is_an_error() { + let home = Scratch::new(); + let jobs = home.dir.join("jobs"); + std::fs::create_dir(&jobs).unwrap(); + std::fs::write( + jobs.join("Mail.toml"), + "schedule = \"* * * * *\"\nprompt = \"x\"\n", + ) + .unwrap(); + let err = jobs::load(&jobs).unwrap_err(); + assert!(matches!(err, JobsError::Name(_)), "{err:?}"); +} + +#[test] +fn session_id_fits_and_is_lowercase() { + let id = jobs::session_id("mail", ts()).unwrap(); + assert_eq!(id.as_str(), "job-mail-20261003t0700"); + SessionId::new(id.as_str()).unwrap(); +} + +#[test] +fn a_name_too_long_for_a_session_id_is_an_error() { + let long = "n".repeat(50); + let err = jobs::session_id(&long, ts()).unwrap_err(); + assert!(matches!(err, JobsError::Name(_)), "{err:?}"); +} + +#[test] +fn a_missing_jobs_directory_is_empty() { + let home = Scratch::new(); + let loaded = jobs::load(&home.dir.join("jobs")).unwrap(); + assert!(loaded.is_empty()); +} + +#[test] +fn leftover_started_is_closed_as_failed() { + let home = Scratch::new(); + let path = home.dir.join("runs.jsonl"); + let session = jobs::session_id("mail", ts()).unwrap(); + jobs::append( + &path, + &jobs::Run { + time: ts(), + name: "mail".to_string(), + session: session.clone(), + status: RunStatus::Started, + }, + ) + .unwrap(); + let n = jobs::close_leftovers(&path, ts()).unwrap(); + assert_eq!(n, 1); + assert!(jobs::live_session(&path, "mail").unwrap().is_none()); +} + +#[test] +fn a_started_run_without_a_close_is_live() { + let home = Scratch::new(); + let path = home.dir.join("runs.jsonl"); + let session = jobs::session_id("mail", ts()).unwrap(); + jobs::append( + &path, + &jobs::Run { + time: ts(), + name: "mail".to_string(), + session: session.clone(), + status: RunStatus::Started, + }, + ) + .unwrap(); + assert_eq!( + jobs::live_session(&path, "mail").unwrap().as_ref(), + Some(&session) + ); + jobs::append( + &path, + &jobs::Run { + time: ts(), + name: "mail".to_string(), + session, + status: RunStatus::Finished, + }, + ) + .unwrap(); + assert!(jobs::live_session(&path, "mail").unwrap().is_none()); +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index e2f8e19..e3b7a06 100644 --- a/docs/implementer-log.md +++ b/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/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--t` 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 | | M5d2/03-loopd-heartbeat-config | 2026-10-03 | done | 1 | pass | none | Copied the given test `crates/loopd/tests/config_heartbeat.rs` and the starter `config/heartbeat.md` from `docs/plans/M5d2/files/`. Step 2 (`cargo build --tests --keep-going -p loopd`) showed exactly the named failure: `config_heartbeat` with E0609 (`no field heartbeat`) and E0599 (`no method heartbeat_prompt` / `check_summarise`); every other target compiled. Made H1 (`Config` gains `#[serde(default)] pub heartbeat: Heartbeat` after `compact`), H2 (new `Heartbeat { every_s: u64, prompt: PathBuf }` with `#[serde(deny_unknown_fields, default)]`, `Default` = `every_s` 0 and `prompt` `heartbeat.md`, placed after `Compact`'s Default impl), H3 (`load` joins `heartbeat.prompt` to the config file's parent beside `baseline.system`), H4 (`heartbeat_prompt` returns `self.heartbeat.prompt.clone()`, `check_summarise` walks `summarise` and returns Err naming both `summarise` and the unknown name, both after `compact_prompt`). Made M1 (`run_selftest`) and M2 (`run_serve`): after a successful `load` each calls `cfg.check_summarise(&loopd::tools::Registry::m5c())`, printing `loopd: {e}` and returning exit 1 on failure, before `Client::new` / the instance lock. Made A1 (`ask_memory_tool.rs` `loopd_serve_uses_the_m5c_registry` counts `Registry::m5c()` as 3). `cargo check -p loopd` clean, no error or warning. `cargo test -p loopd --test config_heartbeat --test config --test config_compact --test ask_memory_tool` passes (8 + 11 + 8 + 4). `git diff --stat` lists only `config.rs`, `main.rs` and `ask_memory_tool.rs`; `git status --short` shows the new test and `heartbeat.md` as untracked. `cargo fmt --all` changed nothing after the edits. `make gate` printed `gate: ok` with 1526 tests (1518 before, plus the 8 of `config_heartbeat.rs`), matching the task's expected count. No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b | | M5d2/03-loopd-heartbeat-config | 2026-10-03 | done | 1 | pass | none | Copied the given test `crates/loopd/tests/config_heartbeat.rs` and the starter `config/heartbeat.md` from `docs/plans/M5d2/files/`. Step 2 (`cargo build --tests --keep-going -p loopd`) showed exactly the named failure: `config_heartbeat` with E0609 (`no field heartbeat`) and E0599 (`no method heartbeat_prompt` / `check_summarise`); every other target compiled. Made H1 (`Config` gains `#[serde(default)] pub heartbeat: Heartbeat` after `compact`), H2 (new `Heartbeat { every_s: u64, prompt: PathBuf }` with `#[serde(deny_unknown_fields, default)]`, `Default` = `every_s` 0 and `prompt` `heartbeat.md`, placed after `Compact`'s Default impl), H3 (`load` joins `heartbeat.prompt` to the config file's parent beside `baseline.system`), H4 (`heartbeat_prompt` returns `self.heartbeat.prompt.clone()`, `check_summarise` walks `summarise` and returns Err naming both `summarise` and the unknown name, both after `compact_prompt`). Made M1 (`run_selftest`) and M2 (`run_serve`): after a successful `load` each calls `cfg.check_summarise(&loopd::tools::Registry::m5c())`, printing `loopd: {e}` and returning exit 1 on failure, before `Client::new` / the instance lock. Made A1 (`ask_memory_tool.rs` `loopd_serve_uses_the_m5c_registry` counts `Registry::m5c()` as 3). `cargo check -p loopd` clean, no error or warning. `cargo test -p loopd --test config_heartbeat --test config --test config_compact --test ask_memory_tool` passes (8 + 11 + 8 + 4). `git diff --stat` lists only `config.rs`, `main.rs` and `ask_memory_tool.rs`; `git status --short` shows the new test and `heartbeat.md` as untracked. `cargo fmt --all` changed nothing after the edits. `make gate` printed `gate: ok` with 1526 tests (1518 before, plus the 8 of `config_heartbeat.rs`), matching the task's expected count. No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b | diff --git a/docs/runbook.md b/docs/runbook.md index 78bbc85..0a61e5b 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -1192,3 +1192,28 @@ delete `EpochEnd`; the next compact continues from it. **Check.** The next turn on that session either runs on the current epoch, or compact succeeds and the next turn's log is `.jsonl` with `epoch` n. + +## jobs-invalid + +**What you see.** `loopd` does not start. It prints a line naming `{home}/jobs/.toml` and +this entry: + +```text +loopd: jobs/mail.toml: unknown field `when`, expected `schedule` or `prompt`; see docs/runbook.md#jobs-invalid +``` + +**Why.** Job files are owner-written. `loopd` reads them and cannot write them. Unknown keys, a +name that is not `[a-z0-9-]+`, a name so long that `job--t` is not a valid +session id (64 bytes, `[a-z0-9-]`), a schedule that is not five UTC fields, or a missing +`prompt`, all refuse start so a misspelt job cannot sit silently. + +**Confirm.** + +```sh +ls -l "$BOXMAKER_HOME/jobs/" +``` + +**Fix.** Rename or edit the named file so it matches the example in `config/jobs/tier.toml.example`. +Then start `loopd` again. + +**Check.** `loopd serve` prints `serving on` and `{home}/jobs/runs.jsonl` exists (empty is fine).