loopd: job files, run ledger, jobs-invalid

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

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<SessionId, JobsError> {
+    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<Vec<Job>, JobsError> {
+    let mut entries: Vec<PathBuf> = 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<Option<SessionId>, JobsError> {
+    let content = read_ledger(path)?;
+    let mut last: Option<Run> = 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<usize, JobsError> {
+    let content = read_ledger(path)?;
+    let mut last_by_name: BTreeMap<String, Run> = 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<String, JobsError> {
+    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)),
+    }
+}

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;

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.