Merge detached_support.rs into detached.rs

The _support suffix was a naming smell — guards, failure persistence,
and progress helpers are all detached-run infrastructure and belong
alongside the detached run entry point.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-27 10:55:39 -04:00
parent 8de4aab7b3
commit f2d90d97ba
6 changed files with 330 additions and 338 deletions

View file

@ -12,7 +12,7 @@ use fabro_util::terminal::Styles;
use fabro_workflows::event::RunNoticeLevel;
use fabro_workflows::run_status::{RunStatus, RunStatusRecord};
use super::detached_support::append_run_notice;
use super::detached::append_run_notice;
use super::run_progress;
#[cfg(test)]

View file

@ -1,6 +1,12 @@
use std::path::PathBuf;
use std::path::{Path, PathBuf};
use anyhow::Result;
use anyhow::{Context, Result};
use chrono::Utc;
use fabro_workflows::event::{RunNoticeLevel, WorkflowRunEvent};
use fabro_workflows::outcome::StageStatus;
use fabro_workflows::records::Conclusion;
use fabro_workflows::run_status::{self, RunStatus, StatusReason};
use serde::Serialize;
use crate::cli_config;
use crate::shared;
@ -19,10 +25,10 @@ pub async fn execute(run_dir: PathBuf, resume: bool) -> Result<()> {
Ok(persisted) => persisted,
Err(err) => {
let anyhow_err: anyhow::Error = anyhow::anyhow!("Failed to load persisted run: {err}");
let _ = super::detached_support::persist_detached_failure(
let _ = persist_detached_failure(
&run_dir,
"bootstrap",
fabro_workflows::run_status::StatusReason::BootstrapFailed,
StatusReason::BootstrapFailed,
&anyhow_err,
);
return Err(anyhow_err);
@ -37,12 +43,8 @@ pub async fn execute(run_dir: PathBuf, resume: bool) -> Result<()> {
)
})
{
let _ = super::detached_support::persist_detached_failure(
&run_dir,
"bootstrap",
fabro_workflows::run_status::StatusReason::BootstrapFailed,
&err,
);
let _ =
persist_detached_failure(&run_dir, "bootstrap", StatusReason::BootstrapFailed, &err);
return Err(err);
}
@ -71,13 +73,326 @@ pub async fn execute(run_dir: PathBuf, resume: bool) -> Result<()> {
match result {
Ok(()) => Ok(()),
Err(err) => {
let _ = super::detached_support::persist_detached_failure(
let _ = persist_detached_failure(
&run_dir,
"bootstrap",
fabro_workflows::run_status::StatusReason::SandboxInitFailed,
StatusReason::SandboxInitFailed,
&err,
);
Err(err)
}
}
}
const POSTRUN_ABORTED_MESSAGE: &str = "Run aborted before post-run finalization completed.";
pub(crate) struct DetachedRunBootstrapGuard {
run_dir: PathBuf,
active: bool,
}
impl DetachedRunBootstrapGuard {
pub(crate) fn arm(run_dir: &Path) -> Result<Self> {
std::fs::write(run_dir.join("run.pid"), std::process::id().to_string())
.with_context(|| format!("Failed to write {}", run_dir.join("run.pid").display()))?;
run_status::write_run_status(
run_dir,
RunStatus::Starting,
Some(StatusReason::SandboxInitializing),
);
Ok(Self {
run_dir: run_dir.to_path_buf(),
active: true,
})
}
pub(crate) fn defuse(&mut self) {
self.active = false;
}
}
impl Drop for DetachedRunBootstrapGuard {
fn drop(&mut self) {
if self.active {
run_status::write_run_status(
&self.run_dir,
RunStatus::Failed,
Some(StatusReason::SandboxInitFailed),
);
}
}
}
pub(crate) struct DetachedRunCompletionGuard {
run_dir: PathBuf,
active: bool,
}
impl DetachedRunCompletionGuard {
pub(crate) fn arm(run_dir: &Path) -> Self {
Self {
run_dir: run_dir.to_path_buf(),
active: true,
}
}
pub(crate) fn defuse(&mut self) {
self.active = false;
}
}
impl Drop for DetachedRunCompletionGuard {
fn drop(&mut self) {
if !self.active {
return;
}
run_status::write_run_status(
&self.run_dir,
RunStatus::Failed,
Some(StatusReason::WorkflowError),
);
if !self.run_dir.join("conclusion.json").exists() {
let _ = write_failure_conclusion(
&self.run_dir,
POSTRUN_ABORTED_MESSAGE,
Some(StatusReason::WorkflowError),
);
}
if let Some(run_id) = load_run_id(&self.run_dir) {
let _ = append_progress_event(
&self.run_dir,
&run_id,
&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Error,
code: "postrun_aborted".to_string(),
message: POSTRUN_ABORTED_MESSAGE.to_string(),
},
);
}
}
}
pub(crate) fn load_run_id(run_dir: &Path) -> Option<String> {
fabro_workflows::records::RunRecord::load(run_dir)
.ok()
.map(|record| record.run_id)
.filter(|run_id| !run_id.trim().is_empty())
.or_else(|| {
std::fs::read_to_string(run_dir.join("id.txt"))
.ok()
.map(|run_id| run_id.trim().to_string())
.filter(|run_id| !run_id.is_empty())
})
}
pub(crate) fn append_progress_event(
run_dir: &Path,
run_id: &str,
event: &WorkflowRunEvent,
) -> Result<()> {
fabro_workflows::event::append_progress_event(run_dir, run_id, event)
}
pub(crate) fn append_run_notice(
run_dir: &Path,
level: RunNoticeLevel,
code: &'static str,
message: impl Into<String>,
) -> Result<()> {
let Some(run_id) = load_run_id(run_dir) else {
return Ok(());
};
append_progress_event(
run_dir,
&run_id,
&WorkflowRunEvent::RunNotice {
level,
code: code.to_string(),
message: message.into(),
},
)
}
pub(crate) fn persist_detached_failure(
run_dir: &Path,
phase: &'static str,
reason: StatusReason,
error: &anyhow::Error,
) -> Result<()> {
#[derive(Serialize)]
struct DetachedFailureRecord<'a> {
timestamp: chrono::DateTime<Utc>,
phase: &'a str,
reason: StatusReason,
error: String,
}
let message = error.to_string();
let record = DetachedFailureRecord {
timestamp: Utc::now(),
phase,
reason,
error: message.clone(),
};
std::fs::write(
run_dir.join("detached_failure.json"),
serde_json::to_string_pretty(&record)?,
)
.with_context(|| {
format!(
"Failed to write {}",
run_dir.join("detached_failure.json").display()
)
})?;
write_failure_conclusion(run_dir, &message, Some(reason))?;
run_status::write_run_status(run_dir, RunStatus::Failed, Some(reason));
if let Some(run_id) = load_run_id(run_dir) {
append_progress_event(
run_dir,
&run_id,
&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Error,
code: format!("{phase}_failed"),
message,
},
)?;
}
Ok(())
}
pub(crate) fn write_failure_conclusion(
run_dir: &Path,
message: &str,
_reason: Option<StatusReason>,
) -> Result<()> {
if run_dir.join("conclusion.json").exists() {
return Ok(());
}
let conclusion = Conclusion {
timestamp: Utc::now(),
status: StageStatus::Fail,
duration_ms: 0,
failure_reason: Some(message.to_string()),
final_git_commit_sha: None,
stages: vec![],
total_cost: None,
total_retries: 0,
total_input_tokens: 0,
total_output_tokens: 0,
total_cache_read_tokens: 0,
total_cache_write_tokens: 0,
total_reasoning_tokens: 0,
has_pricing: false,
};
conclusion.save(&run_dir.join("conclusion.json"))?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use fabro_workflows::run_status::{RunStatusRecord, StatusReason};
#[test]
fn bootstrap_guard_marks_failed_on_drop() {
let dir = tempfile::tempdir().unwrap();
{
let _guard = DetachedRunBootstrapGuard::arm(dir.path()).unwrap();
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Starting);
assert_eq!(record.reason, Some(StatusReason::SandboxInitializing));
}
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Failed);
assert_eq!(record.reason, Some(StatusReason::SandboxInitFailed));
}
#[test]
fn bootstrap_guard_defuse_leaves_starting_intact() {
let dir = tempfile::tempdir().unwrap();
{
let mut guard = DetachedRunBootstrapGuard::arm(dir.path()).unwrap();
guard.defuse();
}
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Starting);
assert_eq!(record.reason, Some(StatusReason::SandboxInitializing));
}
#[test]
fn completion_guard_marks_failed_on_drop() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-123").unwrap();
{
let _guard = DetachedRunCompletionGuard::arm(dir.path());
}
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Failed);
assert_eq!(record.reason, Some(StatusReason::WorkflowError));
assert!(dir.path().join("conclusion.json").exists());
let progress = std::fs::read_to_string(dir.path().join("progress.jsonl")).unwrap();
assert!(progress.contains("postrun_aborted"));
}
#[test]
fn load_run_id_falls_back_to_id_txt() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-xyz").unwrap();
assert_eq!(load_run_id(dir.path()).as_deref(), Some("run-xyz"));
}
#[test]
fn persist_detached_failure_writes_status_conclusion_and_progress() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-err").unwrap();
let err = anyhow::anyhow!("bootstrap exploded");
persist_detached_failure(dir.path(), "bootstrap", StatusReason::BootstrapFailed, &err)
.unwrap();
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Failed);
assert_eq!(record.reason, Some(StatusReason::BootstrapFailed));
let conclusion =
fabro_workflows::records::Conclusion::load(&dir.path().join("conclusion.json"))
.unwrap();
assert_eq!(conclusion.status, StageStatus::Fail);
assert_eq!(
conclusion.failure_reason.as_deref(),
Some("bootstrap exploded")
);
let progress = std::fs::read_to_string(dir.path().join("progress.jsonl")).unwrap();
assert!(progress.contains("bootstrap_failed"));
assert!(dir.path().join("detached_failure.json").exists());
}
#[test]
fn append_run_notice_writes_progress() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-notice").unwrap();
append_run_notice(
dir.path(),
RunNoticeLevel::Warn,
"interview_unanswered",
"The run is still waiting for input.",
)
.unwrap();
let progress = std::fs::read_to_string(dir.path().join("progress.jsonl")).unwrap();
assert!(progress.contains("\"event\":\"RunNotice\""));
assert!(progress.contains("\"code\":\"interview_unanswered\""));
}
}

View file

@ -1,322 +0,0 @@
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use chrono::Utc;
use fabro_workflows::event::{RunNoticeLevel, WorkflowRunEvent};
use fabro_workflows::outcome::StageStatus;
use fabro_workflows::records::Conclusion;
use fabro_workflows::run_status::{self, RunStatus, StatusReason};
use serde::Serialize;
const POSTRUN_ABORTED_MESSAGE: &str = "Run aborted before post-run finalization completed.";
pub(crate) struct DetachedRunBootstrapGuard {
run_dir: PathBuf,
active: bool,
}
impl DetachedRunBootstrapGuard {
pub(crate) fn arm(run_dir: &Path) -> Result<Self> {
std::fs::write(run_dir.join("run.pid"), std::process::id().to_string())
.with_context(|| format!("Failed to write {}", run_dir.join("run.pid").display()))?;
run_status::write_run_status(
run_dir,
RunStatus::Starting,
Some(StatusReason::SandboxInitializing),
);
Ok(Self {
run_dir: run_dir.to_path_buf(),
active: true,
})
}
pub(crate) fn defuse(&mut self) {
self.active = false;
}
}
impl Drop for DetachedRunBootstrapGuard {
fn drop(&mut self) {
if self.active {
run_status::write_run_status(
&self.run_dir,
RunStatus::Failed,
Some(StatusReason::SandboxInitFailed),
);
}
}
}
pub(crate) struct DetachedRunCompletionGuard {
run_dir: PathBuf,
active: bool,
}
impl DetachedRunCompletionGuard {
pub(crate) fn arm(run_dir: &Path) -> Self {
Self {
run_dir: run_dir.to_path_buf(),
active: true,
}
}
pub(crate) fn defuse(&mut self) {
self.active = false;
}
}
impl Drop for DetachedRunCompletionGuard {
fn drop(&mut self) {
if !self.active {
return;
}
run_status::write_run_status(
&self.run_dir,
RunStatus::Failed,
Some(StatusReason::WorkflowError),
);
if !self.run_dir.join("conclusion.json").exists() {
let _ = write_failure_conclusion(
&self.run_dir,
POSTRUN_ABORTED_MESSAGE,
Some(StatusReason::WorkflowError),
);
}
if let Some(run_id) = load_run_id(&self.run_dir) {
let _ = append_progress_event(
&self.run_dir,
&run_id,
&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Error,
code: "postrun_aborted".to_string(),
message: POSTRUN_ABORTED_MESSAGE.to_string(),
},
);
}
}
}
pub(crate) fn load_run_id(run_dir: &Path) -> Option<String> {
fabro_workflows::records::RunRecord::load(run_dir)
.ok()
.map(|record| record.run_id)
.filter(|run_id| !run_id.trim().is_empty())
.or_else(|| {
std::fs::read_to_string(run_dir.join("id.txt"))
.ok()
.map(|run_id| run_id.trim().to_string())
.filter(|run_id| !run_id.is_empty())
})
}
pub(crate) fn append_progress_event(
run_dir: &Path,
run_id: &str,
event: &WorkflowRunEvent,
) -> Result<()> {
fabro_workflows::event::append_progress_event(run_dir, run_id, event)
}
pub(crate) fn append_run_notice(
run_dir: &Path,
level: RunNoticeLevel,
code: &'static str,
message: impl Into<String>,
) -> Result<()> {
let Some(run_id) = load_run_id(run_dir) else {
return Ok(());
};
append_progress_event(
run_dir,
&run_id,
&WorkflowRunEvent::RunNotice {
level,
code: code.to_string(),
message: message.into(),
},
)
}
pub(crate) fn persist_detached_failure(
run_dir: &Path,
phase: &'static str,
reason: StatusReason,
error: &anyhow::Error,
) -> Result<()> {
#[derive(Serialize)]
struct DetachedFailureRecord<'a> {
timestamp: chrono::DateTime<Utc>,
phase: &'a str,
reason: StatusReason,
error: String,
}
let message = error.to_string();
let record = DetachedFailureRecord {
timestamp: Utc::now(),
phase,
reason,
error: message.clone(),
};
std::fs::write(
run_dir.join("detached_failure.json"),
serde_json::to_string_pretty(&record)?,
)
.with_context(|| {
format!(
"Failed to write {}",
run_dir.join("detached_failure.json").display()
)
})?;
write_failure_conclusion(run_dir, &message, Some(reason))?;
run_status::write_run_status(run_dir, RunStatus::Failed, Some(reason));
if let Some(run_id) = load_run_id(run_dir) {
append_progress_event(
run_dir,
&run_id,
&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Error,
code: format!("{phase}_failed"),
message,
},
)?;
}
Ok(())
}
pub(crate) fn write_failure_conclusion(
run_dir: &Path,
message: &str,
_reason: Option<StatusReason>,
) -> Result<()> {
if run_dir.join("conclusion.json").exists() {
return Ok(());
}
let conclusion = Conclusion {
timestamp: Utc::now(),
status: StageStatus::Fail,
duration_ms: 0,
failure_reason: Some(message.to_string()),
final_git_commit_sha: None,
stages: vec![],
total_cost: None,
total_retries: 0,
total_input_tokens: 0,
total_output_tokens: 0,
total_cache_read_tokens: 0,
total_cache_write_tokens: 0,
total_reasoning_tokens: 0,
has_pricing: false,
};
conclusion.save(&run_dir.join("conclusion.json"))?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use fabro_workflows::run_status::{RunStatusRecord, StatusReason};
#[test]
fn bootstrap_guard_marks_failed_on_drop() {
let dir = tempfile::tempdir().unwrap();
{
let _guard = DetachedRunBootstrapGuard::arm(dir.path()).unwrap();
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Starting);
assert_eq!(record.reason, Some(StatusReason::SandboxInitializing));
}
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Failed);
assert_eq!(record.reason, Some(StatusReason::SandboxInitFailed));
}
#[test]
fn bootstrap_guard_defuse_leaves_starting_intact() {
let dir = tempfile::tempdir().unwrap();
{
let mut guard = DetachedRunBootstrapGuard::arm(dir.path()).unwrap();
guard.defuse();
}
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Starting);
assert_eq!(record.reason, Some(StatusReason::SandboxInitializing));
}
#[test]
fn completion_guard_marks_failed_on_drop() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-123").unwrap();
{
let _guard = DetachedRunCompletionGuard::arm(dir.path());
}
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Failed);
assert_eq!(record.reason, Some(StatusReason::WorkflowError));
assert!(dir.path().join("conclusion.json").exists());
let progress = std::fs::read_to_string(dir.path().join("progress.jsonl")).unwrap();
assert!(progress.contains("postrun_aborted"));
}
#[test]
fn load_run_id_falls_back_to_id_txt() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-xyz").unwrap();
assert_eq!(load_run_id(dir.path()).as_deref(), Some("run-xyz"));
}
#[test]
fn persist_detached_failure_writes_status_conclusion_and_progress() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-err").unwrap();
let err = anyhow::anyhow!("bootstrap exploded");
persist_detached_failure(dir.path(), "bootstrap", StatusReason::BootstrapFailed, &err)
.unwrap();
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Failed);
assert_eq!(record.reason, Some(StatusReason::BootstrapFailed));
let conclusion =
fabro_workflows::records::Conclusion::load(&dir.path().join("conclusion.json"))
.unwrap();
assert_eq!(conclusion.status, StageStatus::Fail);
assert_eq!(
conclusion.failure_reason.as_deref(),
Some("bootstrap exploded")
);
let progress = std::fs::read_to_string(dir.path().join("progress.jsonl")).unwrap();
assert!(progress.contains("bootstrap_failed"));
assert!(dir.path().join("detached_failure.json").exists());
}
#[test]
fn append_run_notice_writes_progress() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-notice").unwrap();
append_run_notice(
dir.path(),
RunNoticeLevel::Warn,
"interview_unanswered",
"The run is still waiting for input.",
)
.unwrap();
let progress = std::fs::read_to_string(dir.path().join("progress.jsonl")).unwrap();
assert!(progress.contains("\"event\":\"RunNotice\""));
assert!(progress.contains("\"code\":\"interview_unanswered\""));
}
}

View file

@ -30,7 +30,7 @@ use indicatif::HumanDuration;
use std::time::Duration;
use tracing::debug;
use super::detached_support::{DetachedRunBootstrapGuard, DetachedRunCompletionGuard};
use super::detached::{DetachedRunBootstrapGuard, DetachedRunCompletionGuard};
use super::run_progress;
use crate::args::{CliSandboxProvider, GlobalArgs, RunArgs};
use crate::cli_config;

View file

@ -6,7 +6,6 @@ pub(crate) mod attach;
pub(crate) mod cp;
pub(crate) mod create;
pub(crate) mod detached;
pub(crate) mod detached_support;
pub(crate) mod diff;
pub(crate) mod execute;
pub(crate) mod fork;

View file

@ -3,7 +3,7 @@ use std::path::Path;
use anyhow::{bail, Result};
use fabro_workflows::run_status::{RunStatus, StatusReason};
use super::detached_support::persist_detached_failure;
use super::detached::persist_detached_failure;
/// Spawn a detached engine process for the given run directory.
///