scv_tools/delegate/background/
ledger.rs

1//! The background job ledger: a durable record of every job, so jobs stay
2//! visible after their session, a restart, or a crash, and an interrupted
3//! one can be resumed.
4//!
5//! Each job is one file, `<jobs>/<session>/<job>.json` (directories 0700,
6//! files 0600, replaced atomically), written when the job starts, when a
7//! queued job's turn begins, when it ends, and when its result is reported.
8//! A job its session dropped unfinished is recorded `interrupted` then; one
9//! whose SCV process died without a word is found by [`settle_orphans`],
10//! which the daemon runs at startup and periodically. Bookkeeping never
11//! fails a job: a write that fails is logged and skipped.
12
13use std::{
14    collections::HashMap,
15    io::Read as _,
16    path::{Path, PathBuf},
17    sync::{Arc, Mutex, MutexGuard},
18    time::{Duration, SystemTime, UNIX_EPOCH},
19};
20
21use scv_protocol::{AgentState, JobOrigin, WorktreeRef};
22use serde::{Deserialize, Serialize};
23use serde_json::Value;
24
25use crate::delegate::records::{ProcessIdentity, write_private_json};
26
27/// Longest prompt kept for a resume, in bytes.
28const MAX_PROMPT_BYTES: usize = 16 * 1024;
29/// Longest reply or error kept, in bytes.
30const MAX_REPLY_BYTES: usize = 8 * 1024;
31/// Largest record file read. A prompt, reply, and error at their bounds
32/// expand when JSON escapes control characters (`\u00XX` is six bytes), so
33/// the cap is above that expansion. [`fit_record`] still shrinks a record
34/// that would not fit, or a later read would skip it and the job would
35/// vanish.
36const MAX_RECORD_BYTES: u64 = 512 * 1024;
37
38/// Held around every write of a record, and around every change that reads
39/// one first (dismiss, update, settle, prune), so those changes are not
40/// reordered with each other. A runtime snapshot is taken earlier, outside
41/// this lock; [`write`] is what keeps that snapshot from undoing a
42/// dismissal or a resume link. Nothing that holds the lock takes it again.
43static WRITES: Mutex<()> = Mutex::new(());
44
45fn writing() -> MutexGuard<'static, ()> {
46    crate::sync::lock(&WRITES)
47}
48
49/// One background job, as the ledger keeps it.
50#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
51pub struct JobRecord {
52    pub v: u32,
53    pub session: String,
54    pub job: String,
55    pub agent: String,
56    pub task: String,
57    /// The delegated prompt, bounded.
58    pub prompt: String,
59    /// The call's other arguments (`cwd`, `model`, `effort`, `session`,
60    /// `worktree`, ...), without the prompt.
61    #[serde(default)]
62    pub arguments: Value,
63    pub workspace: PathBuf,
64    /// Who the session serves, such as `TUI` or `workbench`.
65    pub label: String,
66    #[serde(default)]
67    pub started_by: JobOrigin,
68    #[serde(default)]
69    pub reviewed: bool,
70    /// The SCV process that runs it.
71    pub owner: ProcessIdentity,
72    pub state: AgentState,
73    pub created_unix: u64,
74    pub updated_unix: u64,
75    #[serde(default, skip_serializing_if = "Option::is_none")]
76    pub finished_unix: Option<u64>,
77    /// The conversation it ran a turn of, once known.
78    #[serde(default, skip_serializing_if = "Option::is_none")]
79    pub conversation: Option<String>,
80    #[serde(default, skip_serializing_if = "Option::is_none")]
81    pub worktree: Option<WorktreeRef>,
82    #[serde(default, skip_serializing_if = "Option::is_none")]
83    pub reply: Option<String>,
84    #[serde(default, skip_serializing_if = "Option::is_none")]
85    pub error: Option<String>,
86    /// What SCV knew of the permission mode the agent ran in: `verified
87    /// <mode>`, or `unverified` for an agent that reports none SCV reads.
88    #[serde(default, skip_serializing_if = "Option::is_none")]
89    pub mode: Option<String>,
90    /// The result reached the model or the client.
91    #[serde(default)]
92    pub reported: bool,
93    /// Why it stopped before finishing, for an interrupted job.
94    #[serde(default, skip_serializing_if = "Option::is_none")]
95    pub interrupted: Option<String>,
96    /// The job it was resumed as, `<session>/<job>`.
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    pub resumed_as: Option<String>,
99    /// The interrupted job this one resumes, `<session>/<job>`.
100    #[serde(default, skip_serializing_if = "Option::is_none")]
101    pub resumed_from: Option<String>,
102    /// The owner dismissed it: what is left only keeps its number taken
103    /// until it is pruned, and nothing lists or loads it.
104    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
105    pub dismissed: bool,
106}
107
108impl JobRecord {
109    /// Whether the owner can resume it: interrupted, not a reviewed job
110    /// (its journal is the record; a new reviewed call restates it), and
111    /// not resumed already.
112    pub fn resumable(&self) -> bool {
113        self.state == AgentState::Interrupted && !self.reviewed && self.resumed_as.is_none()
114    }
115}
116
117/// A job about to be recorded.
118pub(crate) struct NewJob<'a> {
119    pub(crate) job: &'a str,
120    pub(crate) agent: &'a str,
121    pub(crate) task: &'a str,
122    /// The call's arguments, prompt included.
123    pub(crate) arguments: &'a Value,
124    pub(crate) started_by: JobOrigin,
125    pub(crate) reviewed: bool,
126    /// It waits behind its conversation's running turn.
127    pub(crate) queued: bool,
128}
129
130/// Where one session's jobs are recorded.
131#[derive(Debug, Clone)]
132pub struct JobLedger {
133    root: PathBuf,
134    session: String,
135    label: String,
136    workspace: PathBuf,
137    owner: Option<ProcessIdentity>,
138    /// The newest revision of each job's record written, so writes made
139    /// outside the job table's lock never put an older one back.
140    written: Arc<Mutex<HashMap<String, u64>>>,
141}
142
143impl JobLedger {
144    /// The ledger of session `session` under `root` (the instance's
145    /// [`Layout::jobs`](scv_client::Layout::jobs)); `label` says who the
146    /// session serves. `None` for a session ID that cannot name a directory.
147    pub fn new(root: &Path, session: &str, label: &str, workspace: &Path) -> Option<Self> {
148        valid_name(session).then(|| Self {
149            root: root.to_owned(),
150            session: session.to_owned(),
151            label: label.to_owned(),
152            workspace: workspace.to_owned(),
153            owner: ProcessIdentity::current(),
154            written: Arc::default(),
155        })
156    }
157
158    /// A new record of the job `new` describes, not yet written.
159    pub(crate) fn record(&self, new: &NewJob<'_>) -> Option<JobRecord> {
160        let owner = self.owner?;
161        let now = unix_now();
162        let mut rest = new.arguments.clone();
163        let prompt = rest
164            .as_object_mut()
165            .and_then(|object| object.remove("prompt"))
166            .and_then(|prompt| prompt.as_str().map(str::to_owned))
167            .unwrap_or_default();
168        Some(JobRecord {
169            v: 1,
170            session: self.session.clone(),
171            job: new.job.to_owned(),
172            agent: new.agent.to_owned(),
173            task: new.task.to_owned(),
174            prompt: bounded(&prompt, MAX_PROMPT_BYTES),
175            arguments: rest,
176            workspace: self.workspace.clone(),
177            label: self.label.clone(),
178            started_by: new.started_by,
179            reviewed: new.reviewed,
180            owner,
181            state: if new.queued {
182                AgentState::Queued
183            } else {
184                AgentState::Working
185            },
186            created_unix: now,
187            updated_unix: now,
188            finished_unix: None,
189            conversation: None,
190            worktree: None,
191            reply: None,
192            error: None,
193            mode: None,
194            reported: false,
195            interrupted: None,
196            resumed_as: None,
197            resumed_from: None,
198            dismissed: false,
199        })
200    }
201
202    /// The highest `job-<n>` this session already has a record of, so a
203    /// session that comes back under its ID (a daemon's workbench host)
204    /// numbers its jobs after them instead of over them.
205    pub(crate) fn last_number(&self) -> u64 {
206        let Ok(entries) = std::fs::read_dir(self.root.join(&self.session)) else {
207            return 0;
208        };
209        entries
210            .filter_map(Result::ok)
211            .filter_map(|entry| {
212                entry
213                    .file_name()
214                    .to_str()?
215                    .strip_suffix(".json")?
216                    .strip_prefix("job-")?
217                    .parse::<u64>()
218                    .ok()
219            })
220            .max()
221            .unwrap_or(0)
222    }
223
224    /// The conversation handles this session's records name, so a session
225    /// that comes back under its ID numbers new conversations after them;
226    /// `None` when the session has no records, so it was not here before.
227    pub(crate) fn conversations(&self) -> Option<Vec<String>> {
228        let entries = std::fs::read_dir(self.root.join(&self.session)).ok()?;
229        let handles = entries
230            .filter_map(Result::ok)
231            .filter_map(|entry| read_record(&entry.path()))
232            .flat_map(|record| {
233                let continued = record
234                    .arguments
235                    .get("session")
236                    .and_then(Value::as_str)
237                    .map(str::to_owned);
238                record.conversation.into_iter().chain(continued)
239            })
240            .collect();
241        Some(handles)
242    }
243
244    /// Save `record`.
245    pub(crate) fn save(&self, record: &JobRecord) {
246        if let Err(error) = save(&self.root, record) {
247            tracing::debug!(job = %record.job, %error, "could not record a background job");
248        }
249    }
250
251    /// Save revision `revision` of `record`, unless a newer one of the job
252    /// was written already. Callers write after letting go of the job
253    /// table, so writes may arrive out of order; this keeps the newest.
254    pub(crate) fn save_revision(&self, record: &JobRecord, revision: u64) {
255        let mut written = crate::sync::lock(&self.written);
256        if written
257            .get(&record.job)
258            .is_some_and(|newest| *newest >= revision)
259        {
260            return;
261        }
262        self.save(record);
263        written.insert(record.job.clone(), revision);
264    }
265}
266
267/// Write `record` in its session's directory under `root`.
268pub fn save(root: &Path, record: &JobRecord) -> std::io::Result<()> {
269    let _writing = writing();
270    write(root, record)
271}
272
273/// [`save`], for a caller that holds [`WRITES`].
274///
275/// The job runtime writes a snapshot it took before this lock. A dismissal
276/// already on disk stays, prompt and reply included. A resume link already
277/// recorded stays when the snapshot was copied without it. When the job was
278/// not dismissed, the snapshot's other fields still land.
279fn write(root: &Path, record: &JobRecord) -> std::io::Result<()> {
280    if !valid_name(&record.session) || !valid_name(&record.job) {
281        return Err(std::io::Error::new(
282            std::io::ErrorKind::InvalidInput,
283            "not a session or job name",
284        ));
285    }
286    let dir = root.join(&record.session);
287    let name = format!("{}.json", record.job);
288    let mut record = record.clone();
289    if let Some(existing) = read_record(&dir.join(&name)) {
290        // The owner's dismissal wins over a completion that was copied first
291        // and written second: the tombstone keeps the prompt and the reply
292        // off disk.
293        if existing.dismissed && !record.dismissed {
294            return Ok(());
295        }
296        // Resume records the link before the runtime's next snapshot, which
297        // was copied without it.
298        if record.resumed_as.is_none() {
299            record.resumed_as = existing.resumed_as;
300        }
301        if record.resumed_from.is_none() {
302            record.resumed_from = existing.resumed_from;
303        }
304    }
305    fit_record(&mut record);
306    write_private_json(&dir, &name, &record)
307}
308
309/// Shrink `record` until its JSON fits in [`MAX_RECORD_BYTES`]. A file over
310/// that cap is not read back, so the job would disappear from the ledger.
311fn fit_record(record: &mut JobRecord) {
312    let Some(limit) = usize::try_from(MAX_RECORD_BYTES).ok() else {
313        return;
314    };
315    // A full prompt, reply, and error can each need many halvings before
316    // the JSON fits. Stop only after more halvings than that worst case.
317    for _ in 0..64 {
318        let Ok(bytes) = serde_json::to_vec(record) else {
319            return;
320        };
321        if bytes.len() <= limit {
322            return;
323        }
324        let prompt = record.prompt.len();
325        let reply = record.reply.as_ref().map_or(0, String::len);
326        let error = record.error.as_ref().map_or(0, String::len);
327        let longest = prompt.max(reply).max(error);
328        if longest == 0 {
329            record.arguments = Value::Null;
330            record.task.clear();
331            continue;
332        }
333        let keep = longest / 2;
334        if prompt == longest {
335            let next = scv_client::text::utf8_prefix(&record.prompt, keep).to_owned();
336            record.prompt = next;
337        } else if reply == longest {
338            let next = record
339                .reply
340                .as_deref()
341                .map(|text| scv_client::text::utf8_prefix(text, keep).to_owned());
342            record.reply = next;
343        } else {
344            let next = record
345                .error
346                .as_deref()
347                .map(|text| scv_client::text::utf8_prefix(text, keep).to_owned());
348            record.error = next;
349        }
350    }
351}
352
353/// The record of `job` in `session`, unless the owner dismissed it.
354pub fn load(root: &Path, session: &str, job: &str) -> Option<JobRecord> {
355    load_including_dismissed(root, session, job).filter(|record| !record.dismissed)
356}
357
358/// The record of `job` in `session`, a dismissed one included. Dismissing
359/// keeps the worktree the job ran in, which a queued follow-up still needs
360/// after that job is forgotten.
361pub fn load_including_dismissed(root: &Path, session: &str, job: &str) -> Option<JobRecord> {
362    if !valid_name(session) || !valid_name(job) {
363        return None;
364    }
365    read_record(&root.join(session).join(format!("{job}.json")))
366}
367
368/// Forget `job` of `session` for the owner: its record keeps only its
369/// name, so the job's number is not given to another job before the
370/// record is pruned.
371pub fn dismiss(root: &Path, session: &str, job: &str) -> std::io::Result<()> {
372    let _writing = writing();
373    let mut record = load(root, session, job)
374        .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::NotFound, "no such job record"))?;
375    record.dismissed = true;
376    // Settled, so it is pruned like any other: nothing settles a dismissed
377    // record later.
378    if !record.state.is_final() {
379        interrupt(&mut record, "dismissed by the owner");
380    }
381    record.prompt.clear();
382    record.task.clear();
383    record.arguments = Value::Null;
384    record.reply = None;
385    record.error = None;
386    record.interrupted = None;
387    record.updated_unix = unix_now();
388    write(root, &record)
389}
390
391/// Remove `job`'s record of `session` from disk.
392pub fn remove(root: &Path, session: &str, job: &str) -> std::io::Result<()> {
393    if !valid_name(session) || !valid_name(job) {
394        return Err(std::io::Error::new(
395            std::io::ErrorKind::InvalidInput,
396            "not a session or job name",
397        ));
398    }
399    std::fs::remove_file(root.join(session).join(format!("{job}.json")))?;
400    // The session's directory goes with its last job.
401    let _ = std::fs::remove_dir(root.join(session));
402    Ok(())
403}
404
405/// Change `job`'s record with `change` and save it, when it exists.
406pub fn update(
407    root: &Path,
408    session: &str,
409    job: &str,
410    change: impl FnOnce(&mut JobRecord),
411) -> Option<JobRecord> {
412    update_if(root, session, job, |record| {
413        change(record);
414        true
415    })
416}
417
418/// Change `job`'s record with `change` and save it, when it exists and
419/// `change` says it changed it.
420fn update_if(
421    root: &Path,
422    session: &str,
423    job: &str,
424    change: impl FnOnce(&mut JobRecord) -> bool,
425) -> Option<JobRecord> {
426    let _writing = writing();
427    let mut record = load(root, session, job)?;
428    if !change(&mut record) {
429        return None;
430    }
431    record.updated_unix = unix_now();
432    write(root, &record).ok()?;
433    Some(record)
434}
435
436/// Every job recorded under `root` that the owner has not dismissed.
437pub fn list(root: &Path) -> Vec<JobRecord> {
438    let mut records = list_all(root);
439    records.retain(|record| !record.dismissed);
440    records
441}
442
443/// Every record under `root`, dismissed ones too.
444pub fn list_all(root: &Path) -> Vec<JobRecord> {
445    let Ok(sessions) = std::fs::read_dir(root) else {
446        return Vec::new();
447    };
448    let mut records = Vec::new();
449    for session in sessions.filter_map(Result::ok) {
450        let Some(name) = session.file_name().to_str().map(str::to_owned) else {
451            continue;
452        };
453        if !valid_name(&name)
454            || !std::fs::symlink_metadata(session.path()).is_ok_and(|meta| meta.is_dir())
455        {
456            continue;
457        }
458        let Ok(jobs) = std::fs::read_dir(session.path()) else {
459            continue;
460        };
461        for job in jobs.filter_map(Result::ok) {
462            if let Some(record) = read_record(&job.path())
463                && record.session == name
464            {
465                records.push(record);
466            }
467        }
468    }
469    records
470}
471
472fn read_record(path: &Path) -> Option<JobRecord> {
473    use std::os::unix::fs::OpenOptionsExt as _;
474    let file = std::fs::OpenOptions::new()
475        .read(true)
476        .custom_flags(libc::O_NOFOLLOW)
477        .open(path)
478        .ok()?;
479    let mut bytes = Vec::new();
480    file.take(MAX_RECORD_BYTES).read_to_end(&mut bytes).ok()?;
481    let record: JobRecord = serde_json::from_slice(&bytes).ok()?;
482    // Only a record named after its own job is trusted.
483    let stem = path.file_stem()?.to_str()?;
484    (stem == record.job).then_some(record)
485}
486
487/// Mark interrupted every unfinished job whose SCV process no longer runs
488/// (a crash, a kill, or a restart that could not wait). Returns them.
489pub fn settle_orphans(root: &Path) -> Vec<JobRecord> {
490    let mut settled = Vec::new();
491    for record in list(root) {
492        if record.state.is_final() || record.owner.is_alive() {
493            continue;
494        }
495        // Checked again on the record as it is now: it may have finished,
496        // been dismissed, or been resumed since it was listed.
497        if let Some(record) = update_if(root, &record.session, &record.job, |record| {
498            let orphaned = !record.state.is_final() && !record.owner.is_alive();
499            if orphaned {
500                interrupt(record, "the SCV process that ran it stopped");
501            }
502            orphaned
503        }) {
504            settled.push(record);
505        }
506    }
507    settled
508}
509
510/// Record `record` as interrupted for `reason`.
511pub(crate) fn interrupt(record: &mut JobRecord, reason: &str) {
512    let now = unix_now();
513    record.state = AgentState::Interrupted;
514    record.interrupted = Some(reason.to_owned());
515    record.finished_unix = Some(now);
516    record.updated_unix = now;
517}
518
519/// Remove finished jobs whose record has not changed for `keep`, and
520/// interrupted and dismissed ones too once they are that old.
521pub fn prune(root: &Path, keep: Duration) -> usize {
522    let now = unix_now();
523    let mut removed = 0;
524    let old = |record: &JobRecord| {
525        record.state.is_final() && now.saturating_sub(record.updated_unix) > keep.as_secs()
526    };
527    for record in list_all(root) {
528        if !old(&record) {
529            continue;
530        }
531        // Checked again on the record as it is now, which a change since
532        // it was listed keeps.
533        let _writing = writing();
534        let path = root
535            .join(&record.session)
536            .join(format!("{}.json", record.job));
537        if read_record(&path).is_some_and(|record| old(&record))
538            && remove(root, &record.session, &record.job).is_ok()
539        {
540            removed += 1;
541        }
542    }
543    removed
544}
545
546/// The record's reply and error from a finished job's result.
547pub(crate) fn finish(record: &mut JobRecord, state: AgentState, result: &Value, reported: bool) {
548    let now = unix_now();
549    record.state = state;
550    record.finished_unix = Some(now);
551    record.updated_unix = now;
552    record.reported |= reported;
553    let text = |key: &str| result.get(key).and_then(Value::as_str).map(str::to_owned);
554    record.reply = text("reply").map(|reply| bounded(&reply, MAX_REPLY_BYTES));
555    record.error = text("error").map(|error| bounded(&error, MAX_REPLY_BYTES));
556    record.mode = text("mode").map(|mode| bounded(&mode, 200));
557    if let Some(conversation) = text("session") {
558        record.conversation = Some(conversation);
559    }
560    if let Some(worktree) = result
561        .get("worktree")
562        .and_then(|value| serde_json::from_value::<WorktreeRef>(value.clone()).ok())
563    {
564        record.worktree = Some(worktree);
565    }
566}
567
568/// Whether `name` can name a ledger file or directory: 1-80 ASCII letters,
569/// digits, `-`, or `_`, as session IDs and job handles are.
570pub fn valid_name(name: &str) -> bool {
571    !name.is_empty()
572        && name.len() <= 80
573        && name
574            .bytes()
575            .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_')
576}
577
578fn bounded(text: &str, max: usize) -> String {
579    scv_client::text::utf8_prefix(text, max).to_owned()
580}
581
582/// The current time, as records hold it.
583pub(crate) fn now() -> u64 {
584    unix_now()
585}
586
587fn unix_now() -> u64 {
588    SystemTime::now()
589        .duration_since(UNIX_EPOCH)
590        .map_or(0, |elapsed| elapsed.as_secs())
591}
592
593#[cfg(test)]
594mod tests;
scv_tools/delegate/background/ledger.rs

scv_tools/delegate/background/
ledger.rs
