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

scv_tools/delegate/background/
ledger.rs
