moadim 1.7.4

Loop engine for AI agents — routines over REST, MCP, and a built-in web UI
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

//! Durable, append-only run history that survives workbench TTL reaping.
//!
//! A routine's workbenches are reaped once their TTL elapses (the cleanup module's `reap_dir`),
//! deleting the `exit_code` file [`svc_list_runs`](super::svc_list_runs) reads to derive a run's
//! outcome. Right before that deletion, the reaper appends a compact [`PersistedRun`] record to the
//! routine's `runs.log` ([`crate::paths::routine_run_history_path`], keyed by the routine's stable
//! UUID rather than its slug) — one JSON object per line — so the run's outcome survives past that
//! point, even though its `agent.log` body does not.

use std::io::Write as _;
use std::path::Path;

use serde::{Deserialize, Serialize};

use crate::paths::routine_run_history_path;
use crate::routines::model::RunStatus;

/// One durable run record, appended to a routine's `runs.log` right before its workbench is
/// reaped. Unlike [`super::model::RunSummary`], `status` here is never [`RunStatus::Running`] — a
/// run is only ever persisted once its workbench is confirmed finished.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(crate) struct PersistedRun {
    /// Workbench directory name (`{slug}-{unix_secs}`) the run ran under, before it was removed.
    pub workbench: String,
    /// Unix seconds the run was triggered.
    pub started_at: u64,
    /// Unix seconds the run finished.
    pub finished_at: u64,
    /// Success/failure/unknown; never `Running` (see the type-level doc).
    pub status: RunStatus,
    /// Process exit code, when recorded.
    pub exit_code: Option<i32>,
}

/// Parse the `exit_code` file under `workbench_path` (written by the launch command; see
/// `command::build_routine_command`), if present and parseable.
pub(crate) fn read_exit_code(workbench_path: &Path) -> Option<i32> {
    std::fs::read_to_string(workbench_path.join("exit_code"))
        .ok()
        .and_then(|text| text.trim().parse::<i32>().ok())
}

/// Size at which a routine's `runs.log` is rotated into a sibling `.log.1` file on the next
/// append. The reaper appends one record here per finished run for the whole life of the daemon
/// with no other trim point, so a long-lived, frequently-firing routine's history would otherwise
/// grow unbounded — the same shape already fixed for `daemon.log` (see `DAEMON_LOG_MAX_BYTES` in
/// `cli_system`, #316). A routine's `runs.log` is far smaller per entry and per-routine, so it
/// gets a smaller cap.
const RUN_HISTORY_MAX_BYTES: u64 = 1024 * 1024;

/// Rotate `path` into a sibling `.1` file if it has grown past [`RUN_HISTORY_MAX_BYTES`].
///
/// Unlike a bare rename, this merges `path`'s content onto the end of any existing `.1` file
/// instead of overwriting it — a plain rename previously discarded the prior rotation's history
/// on every subsequent rotation (#1277), even though `runs.log` is documented as durable,
/// TTL-surviving history. [`read_persisted_runs`] reads both files back together. Best-effort: a
/// failed rotation (permissions, race) falls through to the caller's own `append(true)` open
/// rather than blocking the reap sweep that triggered it.
fn rotate_run_history_if_oversized(path: &Path) {
    let Ok(metadata) = std::fs::metadata(path) else {
        return;
    };
    if metadata.len() <= RUN_HISTORY_MAX_BYTES {
        return;
    }
    let rotated_path = path.with_extension("log.1");
    // `read` is expected to succeed (metadata() above just confirmed the file exists); on the
    // unlikely race of it disappearing first, falling back to empty just leaves `rotated_path`
    // holding whatever it already had rather than failing the rotation outright.
    let current = std::fs::read(path).unwrap_or_default();
    let mut combined = std::fs::read(&rotated_path).unwrap_or_default();
    combined.extend_from_slice(&current);
    if std::fs::write(&rotated_path, combined).is_ok() {
        let _ = std::fs::remove_file(path);
    }
}

/// Append `run` as one NDJSON line to routine `id`'s `runs.log`.
///
/// Rotates the log first if it's grown past [`RUN_HISTORY_MAX_BYTES`] (see
/// [`rotate_run_history_if_oversized`]), then the append itself is best-effort: a write failure
/// (serializing `run`, creating the routine's directory, opening the log, or the write itself —
/// collapsed into a single chain so there is one failure path to reason about, not four) is
/// logged and swallowed rather than blocking the reap sweep that triggered it — losing one
/// history entry is far cheaper than a stuck cleanup loop.
pub(crate) fn append_persisted_run(id: &str, run: &PersistedRun) {
    let path = routine_run_history_path(id);
    rotate_run_history_if_oversized(&path);
    let result = serde_json::to_string(run)
        .map_err(std::io::Error::other)
        .and_then(|line| {
            crate::utils::fs_perms::parent_or_err(&path, "run history")
                .and_then(crate::utils::fs_perms::create_private_dir_all)
                .and_then(|()| {
                    open_history_append(&path).and_then(|mut file| writeln!(file, "{line}"))
                })
        });
    if let Err(err) = result {
        log::warn!("run history: failed to append for routine {id:?}: {err}");
    }
}

/// Open `path` for append, creating it owner-only (`0600`) on unix if it doesn't already exist —
/// `runs.log` entries can echo `exit_code`s from an agent run and should stay unreadable by other
/// local accounts. Falls back to the standard `OpenOptions` on non-unix.
fn open_history_append(path: &Path) -> std::io::Result<std::fs::File> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .mode(0o600)
            .open(path)
    }
    #[cfg(not(unix))]
    {
        std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(path)
    }
}

/// Read every persisted run for routine `id`, including any rotated-out `runs.log.1` (#1277).
/// Order is not guaranteed — callers merge and sort alongside live workbench-derived runs.
///
/// Malformed lines are skipped rather than failing the whole read: a single corrupted append (e.g.
/// from a crash mid-write) must not hide every run before or after it.
pub(crate) fn read_persisted_runs(id: &str) -> Vec<PersistedRun> {
    let path = routine_run_history_path(id);
    let rotated_path = path.with_extension("log.1");
    [rotated_path, path]
        .iter()
        .filter_map(|file_path| std::fs::read_to_string(file_path).ok())
        .flat_map(|text| {
            text.lines()
                .filter_map(|line| serde_json::from_str::<PersistedRun>(line).ok())
                .collect::<Vec<_>>()
        })
        .collect()
}

/// Whether routine `id` already has a persisted record for `workbench`.
///
/// A reap sweep persists a workbench's outcome *before* removing its directory (see
/// `cleanup::reap_dir`); if that removal then fails (permission hiccup, a still-open file, a crash),
/// the workbench survives and is expired again on the next sweep. Callers use this to skip
/// re-persisting a workbench that already made it into `runs.log`, so a stuck removal doesn't grow
/// an unbounded run of duplicate history entries for the same run.
pub(crate) fn has_persisted_run(id: &str, workbench: &str) -> bool {
    read_persisted_runs(id)
        .iter()
        .any(|run| run.workbench == workbench)
}

#[cfg(test)]
#[path = "run_history_tests.rs"]
mod run_history_tests;