Files
codecraft/server/src/jobs.rs
T
Richard ZinkandClaude Opus 5.5 01c1484033 Add backups with S3, daily cost push, automation templates and a browser tool
- Daily backups (database + config) into <data_dir>/backups, keep N,
  optional upload to any S3-compatible bucket (SigV4, bucket pruning);
  settings panel with run-now, list and download
- Daily cost summary push at 20:00 (opt-in per user, preview button)
- Automations from templates (dependency check, security review, test
  health, daily summary, TODO report, docs freshness) via POST /jobs/create
- browser tool: headless Chromium over CDP – click, type, keys, waits,
  JavaScript, navigation, screenshots, console errors; requests to private
  network addresses are blocked (sub-resources and redirects included)
- Secrets masked in the config view: S3 secret key, metrics token
- Replace the remaining text arrows/refresh glyphs with SVG icons

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-09 06:07:03 +01:00

515 lines
22 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! Scheduled agents ("automations"): a Markdown file with a cron schedule in
//! its frontmatter starts an agent session at the given times.
//!
//! ```markdown
//! ---
//! name: Nightly dependency check
//! schedule: "0 3 * * *" # cron (minute hour day month weekday), server time zone
//! agent: build # mode (default build)
//! model: provider/model # optional, default: server default model
//! permission_mode: auto # default auto; "bypass" runs without asking
//! timeout_minutes: 60 # optional, default 60
//! ---
//! Check for outdated dependencies and open a summary …
//! ```
//!
//! The file is re-read on every run, so edits apply to the next run. Each run
//! is a normal chat session owned by the job's owner.
use crate::app::App;
use crate::tools::{Tool, ToolCtx, ToolOutput, arg_str};
use crate::types::{Part, Session, new_id, now_ms};
use anyhow::{Context, Result, anyhow, bail};
use async_trait::async_trait;
use serde::Serialize;
use serde_json::{Value, json};
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
#[derive(Debug, Clone, Serialize)]
pub struct Job {
pub id: String,
pub owner_id: String,
pub path: String,
pub enabled: bool,
pub last_run: Option<i64>,
pub last_session_id: Option<String>,
pub last_status: Option<String>,
pub created_at: i64,
/// Webhook enabled (the secret – also the HMAC key – is only shown once when created).
pub webhook: bool,
}
/// Parsed job file.
#[derive(Debug, Clone, Serialize)]
pub struct JobSpec {
pub name: String,
pub schedule: String,
pub agent: String,
pub model: Option<String>,
pub permission_mode: String,
pub timeout_minutes: u64,
pub cwd: Option<String>,
pub prompt: String,
}
pub fn parse(path: &Path) -> Result<JobSpec> {
let text = std::fs::read_to_string(path).with_context(|| format!("cannot read {}", path.display()))?;
let (meta, body) = crate::prompts::frontmatter(&text);
let unquote = |s: &String| s.trim().trim_matches('"').trim_matches('\'').to_string();
let schedule = meta.get("schedule").map(unquote).ok_or_else(|| anyhow!("{}: frontmatter needs `schedule: \"<cron>\"`", path.display()))?;
croner::Cron::from_str(&schedule).map_err(|e| anyhow!("invalid schedule '{schedule}': {e}"))?;
let permission_mode = meta.get("permission_mode").map(unquote).unwrap_or_else(|| "auto".into());
if !crate::permission::MODES.contains(&permission_mode.as_str()) {
bail!("invalid permission_mode '{permission_mode}'");
}
if body.trim().is_empty() {
bail!("{}: the job has no prompt (text after the frontmatter)", path.display());
}
Ok(JobSpec {
name: meta.get("name").map(unquote).unwrap_or_else(|| path.file_stem().and_then(|s| s.to_str()).unwrap_or("job").to_string()),
schedule,
agent: meta.get("agent").map(unquote).unwrap_or_else(|| "build".into()),
model: meta.get("model").map(unquote).filter(|m| !m.is_empty()),
permission_mode,
timeout_minutes: meta.get("timeout_minutes").and_then(|v| unquote(v).parse().ok()).unwrap_or(60).clamp(1, 24 * 60),
cwd: meta.get("cwd").map(unquote).filter(|c| !c.is_empty()),
prompt: body.trim().to_string(),
})
}
/// Next run after `after` (ms) in the server's local time zone.
pub fn next_run(schedule: &str, after_ms: i64) -> Option<i64> {
let cron = croner::Cron::from_str(schedule).ok()?;
let start = chrono::DateTime::from_timestamp_millis(after_ms)?.with_timezone(&chrono::Local);
cron.find_next_occurrence(&start, false).ok().map(|t| t.timestamp_millis())
}
/// Working directory of a job: `cwd` from the file, else the project containing `.codecraft/jobs/`.
fn job_cwd(path: &Path, spec: &JobSpec) -> PathBuf {
if let Some(c) = &spec.cwd {
let p = PathBuf::from(c);
return if p.is_absolute() { p } else { path.parent().unwrap_or(Path::new("/")).join(p) };
}
let mut dir = path.parent();
while let Some(d) = dir {
if d.file_name().and_then(|n| n.to_str()) == Some(".codecraft") {
return d.parent().unwrap_or(d).to_path_buf();
}
dir = d.parent();
}
path.parent().unwrap_or(Path::new("/")).to_path_buf()
}
/// Job with its parsed file and next run, for the UI and the tool.
pub fn describe(job: &Job) -> Value {
let mut v = json!(job);
match parse(Path::new(&job.path)) {
Ok(spec) => {
v["spec"] = json!(spec);
let base = job.last_run.unwrap_or(job.created_at).max(now_ms() - 60_000);
v["next_run"] = json!(if job.enabled { next_run(&spec.schedule, base) } else { None });
}
Err(e) => v["error"] = json!(format!("{e:#}")),
}
v
}
/// Start a run now. Returns the new session id.
pub async fn run(app: &Arc<App>, job: &Job, trigger: &str) -> Result<String> {
run_with(app, job, trigger, None).await
}
/// Start a run; `payload` (webhook body) is appended to the instructions.
pub async fn run_with(app: &Arc<App>, job: &Job, trigger: &str, payload: Option<String>) -> Result<String> {
let path = PathBuf::from(&job.path);
let spec = match parse(&path) {
Ok(s) => s,
Err(e) => {
app.store.job_finished(&job.id, None, &format!("error: {e:#}"))?;
return Err(e);
}
};
let owner = app.store.user(&job.owner_id)?.ok_or_else(|| anyhow!("the job's owner no longer exists"))?;
if owner.disabled {
bail!("the job's owner is disabled");
}
let cwd = job_cwd(&path, &spec);
let roots = app.user_roots(Some(&owner.id));
if !roots.iter().any(|r| cwd.starts_with(r)) {
bail!("{} is outside the owner's folders", cwd.display());
}
let cfg = app.config();
let model = spec.model.clone().or_else(|| Some(cfg.server.default_model.clone()).filter(|m| !m.is_empty())).or_else(|| app.registry().first_model()).ok_or_else(|| anyhow!("no model configured"))?;
let now = now_ms();
let stamp = chrono::Local::now().format("%Y-%m-%d %H:%M");
let session = Session {
id: new_id(),
title: format!("{} – {stamp}", spec.name),
cwd: cwd.display().to_string(),
model,
agent: spec.agent.clone(),
permission_mode: spec.permission_mode.clone(),
parent_id: None,
owner_id: Some(owner.id.clone()),
created_at: now,
updated_at: now,
todos: Value::Null,
last_usage: None,
clean_room: None,
phase: None,
worktree: None,
pinned: false,
archived: false,
folder: None,
};
let session = app.save_session(session)?;
app.store.job_started(&job.id, &session.id, now)?;
let text = format!(
"<automation name=\"{}\" trigger=\"{trigger}\" file=\"{}\">\nThis is an unattended, scheduled run – nobody is watching live. Work autonomously, avoid questions unless you are truly blocked, and end with a short report of what you did and found.\n</automation>\n\n{}",
spec.name, job.path, spec.prompt
);
let text = match payload {
Some(p) => format!("{text}\n\n<webhook-payload>\n{p}\n</webhook-payload>\nThe automation was triggered by this webhook call – use the payload as input."),
None => text,
};
crate::runner::submit(app.clone(), session.id.clone(), vec![Part::Text { text }], "user")?;
// Watch for the end (or the timeout).
let app2 = app.clone();
let job_id = job.id.clone();
let sid = session.id.clone();
let timeout = Duration::from_secs(spec.timeout_minutes * 60);
tokio::spawn(async move {
let started = std::time::Instant::now();
tokio::time::sleep(Duration::from_secs(2)).await;
while app2.is_busy(&sid) {
if started.elapsed() > timeout {
crate::runner::abort(&app2, &sid);
let _ = app2.store.job_finished(&job_id, Some(&sid), "timeout");
return;
}
tokio::time::sleep(Duration::from_secs(5)).await;
}
let _ = app2.store.job_finished(&job_id, Some(&sid), "done");
});
Ok(session.id)
}
/// Background scheduler: checks every 30 s which jobs are due.
pub fn spawn_scheduler(app: Arc<App>) {
tokio::spawn(async move {
loop {
tokio::time::sleep(Duration::from_secs(30)).await;
let Ok(jobs) = app.store.jobs(None) else { continue };
let now = now_ms();
for job in jobs.into_iter().filter(|j| j.enabled) {
let Ok(spec) = parse(Path::new(&job.path)) else { continue };
// Due if a scheduled time passed since the last run (or since creation).
let base = job.last_run.unwrap_or(job.created_at);
let Some(next) = next_run(&spec.schedule, base) else { continue };
if next > now {
continue;
}
// Missed runs while the server was down: run once, not for every slot.
if job.last_status.as_deref() == Some("running") && job.last_session_id.as_deref().is_some_and(|s| app.is_busy(s)) {
continue;
}
if let Err(e) = run(&app, &job, "schedule").await {
tracing::warn!("job {} ({}): {e:#}", job.id, job.path);
let _ = app.store.job_finished(&job.id, None, &format!("error: {e:#}"));
// Avoid retrying every 30 s.
let _ = app.store.job_touch(&job.id, now);
}
}
}
});
}
// --- tool -------------------------------------------------------------------------
/// Tool: create, list, run and delete automations.
pub struct Schedule;
/// Fields of a job file written by the agent tool or the "from template" dialog.
#[derive(Debug, serde::Deserialize)]
pub struct NewJob {
pub name: String,
pub schedule: String,
#[serde(default = "default_agent")]
pub agent: String,
#[serde(default)]
pub model: Option<String>,
#[serde(default = "default_mode")]
pub permission_mode: String,
#[serde(default = "default_timeout")]
pub timeout_minutes: u64,
pub prompt: String,
}
fn default_agent() -> String {
"build".into()
}
fn default_mode() -> String {
"auto".into()
}
fn default_timeout() -> u64 {
60
}
/// Write (and validate) a job file.
pub fn write_file(path: &Path, j: &NewJob) -> Result<JobSpec> {
croner::Cron::from_str(&j.schedule).map_err(|e| anyhow!("invalid schedule '{}': {e}", j.schedule))?;
let q = |s: &str| format!("\"{}\"", s.replace('"', "'").replace('\n', " "));
let plain = |s: &str| s.chars().filter(|c| c.is_alphanumeric() || "-_/.:".contains(*c)).collect::<String>();
let mut fm = format!("---\nname: {}\nschedule: {}\nagent: {}\n", q(&j.name), q(&j.schedule), plain(&j.agent));
if let Some(m) = j.model.as_deref().filter(|m| !m.is_empty()) {
fm.push_str(&format!("model: {}\n", plain(m)));
}
fm.push_str(&format!("permission_mode: {}\n", plain(&j.permission_mode)));
fm.push_str(&format!("timeout_minutes: {}\n---\n", j.timeout_minutes.clamp(1, 24 * 60)));
if let Some(dir) = path.parent() {
std::fs::create_dir_all(dir)?;
}
std::fs::write(path, format!("{fm}{}\n", j.prompt.trim()))?;
match parse(path) {
Ok(spec) => Ok(spec),
Err(e) => {
let _ = std::fs::remove_file(path);
Err(e)
}
}
}
/// Free file name for a new job in `<project>/.codecraft/jobs/`.
pub fn new_job_path(project: &Path, name: &str) -> PathBuf {
let dir = project.join(".codecraft/jobs");
let base = slug(name);
(1..100).map(|i| dir.join(if i == 1 { format!("{base}.md") } else { format!("{base}-{i}.md") })).find(|p| !p.exists()).unwrap_or_else(|| dir.join(format!("{base}-{}.md", new_id())))
}
fn slug(s: &str) -> String {
let lower = s.to_lowercase().replace('ä', "ae").replace('ö', "oe").replace('ü', "ue").replace('ß', "ss");
let out: String = lower.chars().map(|c| if c.is_ascii_alphanumeric() { c } else { '-' }).collect();
out.split('-').filter(|p| !p.is_empty()).collect::<Vec<_>>().join("-").chars().take(50).collect()
}
#[async_trait]
impl Tool for Schedule {
fn name(&self) -> &str {
"schedule"
}
fn description(&self) -> String {
"Manage automations: agents that run on a cron schedule from a Markdown job file \
(.codecraft/jobs/<name>.md in the working directory). Actions: \
`create` (name, schedule, prompt, optional agent/model/permission_mode/timeout_minutes) writes the job file and registers it; \
`register` (path) registers an existing job file; `list`; `run` (id) starts a run now; \
`enable`/`disable` (id); `delete` (id, delete_file). Schedules are 5-field cron expressions \
in the server's time zone, e.g. \"0 7 * * 1-5\" = weekdays 07:00, \"*/30 * * * *\" = every 30 minutes. \
The prompt must be self-contained: the run starts a fresh session without this conversation."
.into()
}
fn schema(&self) -> Value {
json!({
"type": "object",
"properties": {
"action": {"type": "string", "enum": ["create", "register", "list", "run", "enable", "disable", "delete"]},
"name": {"type": "string"},
"schedule": {"type": "string", "description": "cron, e.g. \"0 3 * * *\""},
"prompt": {"type": "string", "description": "Complete instructions for the scheduled agent"},
"agent": {"type": "string", "description": "Mode, default build"},
"model": {"type": "string"},
"permission_mode": {"type": "string", "enum": crate::permission::MODES},
"timeout_minutes": {"type": "integer"},
"path": {"type": "string"},
"id": {"type": "string"},
"delete_file": {"type": "boolean"}
},
"required": ["action"]
})
}
fn read_only(&self) -> bool {
false
}
fn title(&self, input: &Value) -> String {
format!("Automation: {} {}", input["action"].as_str().unwrap_or(""), input["name"].as_str().or(input["id"].as_str()).unwrap_or(""))
}
async fn run(&self, input: Value, ctx: &ToolCtx) -> Result<ToolOutput> {
let owner = ctx.app.session(&ctx.session_id).ok().and_then(|s| s.owner_id).ok_or_else(|| anyhow!("session has no owner"))?;
let own_job = |id: &str| -> Result<Job> {
ctx.app.store.job(id)?.filter(|j| j.owner_id == owner).ok_or_else(|| anyhow!("no automation with id {id}"))
};
match arg_str(&input, "action")? {
"create" => {
let name = arg_str(&input, "name")?;
let schedule = arg_str(&input, "schedule")?;
let prompt = arg_str(&input, "prompt")?;
let rel = format!(".codecraft/jobs/{}.md", slug(name));
let path = ctx.resolve(&rel)?;
write_file(
&path,
&NewJob {
name: name.into(),
schedule: schedule.into(),
agent: input["agent"].as_str().unwrap_or("build").into(),
model: input["model"].as_str().map(String::from),
permission_mode: input["permission_mode"].as_str().unwrap_or("auto").into(),
timeout_minutes: input["timeout_minutes"].as_u64().unwrap_or(60),
prompt: prompt.into(),
},
)?;
let job = ctx.app.store.job_register(&owner, &path.display().to_string())?;
let d = describe(&job);
Ok(ToolOutput::text(format!(
"Automation '{name}' created: {} (id {}). Schedule {schedule}, next run {}. The user can manage it on the Automations page.",
rel,
job.id,
fmt_time(d["next_run"].as_i64())
))
.with_meta(d))
}
"register" => {
let path = ctx.resolve(arg_str(&input, "path")?)?;
parse(&path)?;
let job = ctx.app.store.job_register(&owner, &path.display().to_string())?;
Ok(ToolOutput::text(format!("Registered {} as automation {}.", path.display(), job.id)).with_meta(describe(&job)))
}
"list" => {
let jobs = ctx.app.store.jobs(Some(&owner))?;
if jobs.is_empty() {
return Ok(ToolOutput::text("No automations."));
}
let mut out = String::new();
for j in &jobs {
let d = describe(j);
out.push_str(&format!(
"- {} [{}] {} – schedule {} – next {} – last {} ({})\n",
j.id,
if j.enabled { "on" } else { "off" },
d["spec"]["name"].as_str().unwrap_or("?"),
d["spec"]["schedule"].as_str().unwrap_or("?"),
fmt_time(d["next_run"].as_i64()),
fmt_time(j.last_run),
j.last_status.as_deref().unwrap_or("-"),
));
}
Ok(ToolOutput::text(out))
}
"run" => {
let job = own_job(arg_str(&input, "id")?)?;
let sid = run(&ctx.app, &job, "manual").await?;
Ok(ToolOutput::text(format!("Started run in session {sid}.")))
}
a @ ("enable" | "disable") => {
let job = own_job(arg_str(&input, "id")?)?;
ctx.app.store.job_set_enabled(&job.id, a == "enable")?;
Ok(ToolOutput::text(format!("Automation {} {a}d.", job.id)))
}
"delete" => {
let job = own_job(arg_str(&input, "id")?)?;
ctx.app.store.job_delete(&job.id)?;
if input["delete_file"].as_bool().unwrap_or(false) {
let _ = std::fs::remove_file(&job.path);
}
Ok(ToolOutput::text(format!("Automation {} deleted.", job.id)))
}
other => bail!("unknown action '{other}'"),
}
}
}
fn fmt_time(ms: Option<i64>) -> String {
ms.and_then(chrono::DateTime::from_timestamp_millis).map(|t| t.with_timezone(&chrono::Local).format("%Y-%m-%d %H:%M %Z").to_string()).unwrap_or_else(|| "-".into())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn next_run_works() {
let base = chrono::DateTime::parse_from_rfc3339("2026-11-08T10:15:00+00:00").unwrap().timestamp_millis();
let n = next_run("*/30 * * * *", base).unwrap();
assert_eq!(n - base, 15 * 60 * 1000);
assert!(next_run("not a cron", base).is_none());
}
#[test]
fn parses_job_file() {
let dir = std::env::temp_dir().join(format!("cc-job-{}", std::process::id()));
std::fs::create_dir_all(dir.join(".codecraft/jobs")).unwrap();
let p = dir.join(".codecraft/jobs/x.md");
std::fs::write(&p, "---\nname: \"Nightly\"\nschedule: \"0 3 * * *\"\n---\nDo things.\n").unwrap();
let s = parse(&p).unwrap();
assert_eq!((s.name.as_str(), s.schedule.as_str(), s.agent.as_str(), s.permission_mode.as_str()), ("Nightly", "0 3 * * *", "build", "auto"));
assert_eq!(job_cwd(&p, &s), dir);
std::fs::remove_dir_all(&dir).unwrap();
}
}
// --- incoming webhooks -------------------------------------------------------------
fn constant_eq(a: &[u8], b: &[u8]) -> bool {
a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
}
fn hmac_hex(secret: &str, body: &[u8]) -> String {
use hmac::{KeyInit, Mac};
let mut mac = <hmac::Hmac<sha2::Sha256> as KeyInit>::new_from_slice(secret.as_bytes()).expect("hmac key");
mac.update(body);
mac.finalize().into_bytes().iter().map(|b| format!("{b:02x}")).collect()
}
/// Check a webhook call: shared secret (header/query) or Gitea/GitHub HMAC signature.
pub fn verify_webhook(secret: &str, headers: &axum::http::HeaderMap, query_secret: Option<&str>, body: &[u8]) -> bool {
let h = |k: &str| headers.get(k).and_then(|v| v.to_str().ok()).map(|v| v.trim().to_string());
if let Some(given) = h("x-codecraft-secret").or(query_secret.map(String::from)) {
return constant_eq(given.as_bytes(), secret.as_bytes());
}
let expected = hmac_hex(secret, body);
if let Some(sig) = h("x-gitea-signature").or(h("x-gogs-signature")) {
return constant_eq(sig.to_lowercase().as_bytes(), expected.as_bytes());
}
if let Some(sig) = h("x-hub-signature-256") {
return constant_eq(sig.trim_start_matches("sha256=").to_lowercase().as_bytes(), expected.as_bytes());
}
false
}
/// The payload as text for the prompt (pretty JSON, event name, size-limited).
pub fn payload_text(headers: &axum::http::HeaderMap, body: &[u8]) -> String {
let event = ["x-gitea-event", "x-github-event", "x-gitlab-event", "x-event-key"]
.iter()
.find_map(|k| headers.get(*k).and_then(|v| v.to_str().ok()))
.map(|e| format!("event: {e}\n"))
.unwrap_or_default();
let text = match serde_json::from_slice::<Value>(body) {
Ok(v) => serde_json::to_string_pretty(&v).unwrap_or_default(),
Err(_) => String::from_utf8_lossy(body).into_owned(),
};
let mut text: String = text.chars().take(20_000).collect();
if body.len() > 20_000 {
text.push_str("\n… (payload truncated)");
}
format!("{event}{text}")
}
#[cfg(test)]
mod webhook_tests {
#[test]
fn signatures() {
let mut h = axum::http::HeaderMap::new();
let sig = super::hmac_hex("s3cret", b"{\"a\":1}");
h.insert("x-hub-signature-256", format!("sha256={sig}").parse().unwrap());
assert!(super::verify_webhook("s3cret", &h, None, b"{\"a\":1}"));
assert!(!super::verify_webhook("other", &h, None, b"{\"a\":1}"));
let mut h2 = axum::http::HeaderMap::new();
h2.insert("x-gitea-signature", sig.parse().unwrap());
assert!(super::verify_webhook("s3cret", &h2, None, b"{\"a\":1}"));
assert!(super::verify_webhook("abc", &axum::http::HeaderMap::new(), Some("abc"), b""));
assert!(!super::verify_webhook("abc", &axum::http::HeaderMap::new(), None, b""));
}
}