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

orch/cluster/
progress.rs
