orch/cluster/
progress.rs

1//! `orch progress`: what an agent finished, what is left, and where the work is. The record goes
2//! through the log, the session and the worktree stay on the node that runs the task
3
4use std::path::Path;
5
6use serde::{Deserialize, Serialize};
7use tokio::process::Command;
8
9use crate::engine::runner::commit::When;
10use crate::engine::runner::{git_ops, worktree};
11use crate::store::TaskStore;
12
13/// Each text field is cut to this many characters, a record is a pointer and not a log
14const MAX_TEXT: usize = 500;
15/// Records kept per task, oldest dropped first
16const KEEP: usize = 5;
17
18/// What the CLI sends after it tried to commit in the worktree itself
19#[derive(Debug, Clone, Serialize, Deserialize)]
20pub struct Report {
21    pub worktree: String,
22    pub done: String,
23    pub left: String,
24    pub state: String,
25    pub cli_committed: bool,
26    #[serde(default)]
27    pub cli_error: Option<String>,
28}
29
30#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
31pub struct Record {
32    pub node: String,
33    pub ts: String,
34    pub branch: String,
35    pub commit: String,
36    /// `agent`, `cli`, `orch` or `none`
37    pub committed_by: String,
38    /// `agent`, `orch` or `none`
39    pub pushed_by: String,
40    #[serde(default)]
41    pub error: Option<String>,
42    pub done: String,
43    pub left: String,
44    pub state: String,
45}
46
47fn cut(s: &str) -> String {
48    s.chars().take(MAX_TEXT).collect()
49}
50
51fn key(task_id: i64) -> String {
52    format!("progress:{task_id}")
53}
54
55pub async fn list(store: &TaskStore, task_id: i64) -> anyhow::Result<Vec<Record>> {
56    match store.kv_get(&key(task_id)).await? {
57        Some(json) => Ok(serde_json::from_str(&json)?),
58        None => Ok(Vec::new()),
59    }
60}
61
62async fn git_out(dir: &Path, args: &[&str]) -> Option<String> {
63    let out = Command::new("git")
64        .args(args)
65        .current_dir(dir)
66        .output()
67        .await
68        .ok()?;
69    out.status
70        .success()
71        .then(|| String::from_utf8_lossy(&out.stdout).trim().to_string())
72}
73
74/// Commit what is still uncommitted, push the branch, and record it. Only a known task's worktree
75/// is touched, so a caller cannot point the daemon at another directory
76pub async fn handle(store: &TaskStore, node: &str, report: Report) -> anyhow::Result<Record> {
77    let task_id = store
78        .resolve_id_by_worktree(&report.worktree)
79        .await?
80        .ok_or_else(|| anyhow::anyhow!("{} is not the worktree of a task", report.worktree))?;
81    let task = store.get(task_id).await?;
82    let dir = Path::new(&report.worktree);
83    let previous = list(store, task_id).await?;
84    let mut errors: Vec<String> = report.cli_error.iter().cloned().collect();
85
86    let head_before = git_out(dir, &["rev-parse", "HEAD"])
87        .await
88        .unwrap_or_default();
89    let mut committed_by = if report.cli_committed { "cli" } else { "none" };
90    let ext = task.external_id.clone().unwrap_or_default();
91    match git_ops::auto_commit(
92        dir,
93        &ext,
94        &report.done,
95        "orch",
96        u32::try_from(task.attempts).unwrap_or(0),
97        When::Tested,
98    )
99    .await
100    {
101        Ok(true) => committed_by = "orch",
102        Ok(false) => {}
103        Err(e) => errors.push(format!("commit: {e:#}")),
104    }
105
106    if committed_by == "none"
107        && previous
108            .last()
109            .is_some_and(|last| !head_before.is_empty() && last.commit != head_before)
110    {
111        committed_by = "agent";
112    }
113
114    let branch = git_ops::get_current_branch(dir).await;
115    let commit = git_out(dir, &["rev-parse", "HEAD"])
116        .await
117        .unwrap_or_default();
118    let remote = git_out(
119        dir,
120        &["rev-parse", &format!("refs/remotes/origin/{branch}")],
121    )
122    .await;
123    let pushed_by = if remote.as_deref() == Some(commit.as_str()) {
124        "agent"
125    } else {
126        let default_branch = worktree::detect_default_branch(dir).await;
127        match git_ops::push_branch(dir, &branch, &default_branch).await {
128            Ok(true) => "orch",
129            Ok(false) => "none",
130            Err(e) => {
131                errors.push(format!("push: {e:#}"));
132                "none"
133            }
134        }
135    };
136
137    let record = Record {
138        node: node.to_string(),
139        ts: super::now_stamp(),
140        branch,
141        commit,
142        committed_by: committed_by.to_string(),
143        pushed_by: pushed_by.to_string(),
144        error: (!errors.is_empty()).then(|| cut(&errors.join("; "))),
145        done: cut(&report.done),
146        left: cut(&report.left),
147        state: cut(&report.state),
148    };
149    let mut records = previous;
150    records.push(record.clone());
151    let skip = records.len().saturating_sub(KEEP);
152    let records: Vec<Record> = records.into_iter().skip(skip).collect();
153    store
154        .kv_set(&key(task_id), &serde_json::to_string(&records)?)
155        .await?;
156    Ok(record)
157}
orch/cluster/progress.rs

orch/cluster/
progress.rs
