scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Create RuntimeState for run runtime and cache paths

Browse source
This commit is contained in:
Bryan Helmkamp 2026-03-28 01:14:14 -04:00
parent ffd0e1de11
commit bf9fd2cea2
25 changed files with 515 additions and 182 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
Cargo.lock generated
View file

@ -1555,6 +1555,7 @@ dependencies = [
"fabro-openai-oauth",
"fabro-retro",
"fabro-sandbox",
"fabro-store",
"fabro-telemetry",
"fabro-types",
"fabro-util",
@ -1881,6 +1882,7 @@ dependencies = [
"serde",
"serde_json",
"slatedb",
"tempfile",
"thiserror 2.0.18",
"tokio",
"tokio-stream",
@ -2000,6 +2002,7 @@ dependencies = [
"fabro-model",
"fabro-retro",
"fabro-sandbox",
"fabro-store",
"fabro-types",
"fabro-util",
"fabro-validate",

39
docs/agents/outputs.mdx
View file

@ -139,8 +139,8 @@ When a stage produces a large context value -- an LLM response, command output,
After each node completes, Fabro checks every context update. If the serialized JSON of a value exceeds **100KB**, it is written to the artifact store on disk and replaced in the context with a `file://` pointer:
```
response.plan --> file:///path/to/logs/artifacts/values/response.plan.json
command.output --> file:///path/to/logs/artifacts/values/command.output.json
response.plan --> file:///path/to/logs/cache/artifacts/values/response.plan.json
command.output --> file:///path/to/logs/cache/artifacts/values/command.output.json
```
Values under 100KB remain in the context as-is.
@ -151,11 +151,12 @@ Offloaded artifacts are written to the run's directory:
```
~/.fabro/runs/{run_id}/
artifacts/
values/
response.plan.json
response.implement.json
command.output.json
cache/
artifacts/
values/
response.plan.json
response.implement.json
command.output.json
```
Each file contains the full serialized JSON value. The `ArtifactStore` manages reads and writes, and cleans up files when artifacts are removed.
@ -169,10 +170,10 @@ When Fabro builds a [preamble](/execution/context#preamble-construction) for a d
- **plan**: success
- Model: claude-sonnet-4-5, 12.4k tokens in / 3.2k out
- Files: src/main.rs, tests/api_test.rs
- Response: See: /path/to/logs/artifacts/values/response.plan.json
- Response: See: /path/to/logs/cache/artifacts/values/response.plan.json
- **test**: success
- Script: `cargo test 2>&1 || true`
- Stdout: See: /path/to/logs/artifacts/values/command.output.json
- Stdout: See: /path/to/logs/cache/artifacts/values/command.output.json
```
This keeps preambles concise while still giving agents a path to read the full output if needed.
@ -206,7 +207,7 @@ For each pointer in the context updates:
```
# Before sync (host path)
file:///home/user/.fabro/runs/01JK.../artifacts/values/response.plan.json
file:///home/user/.fabro/runs/01JK.../cache/artifacts/values/response.plan.json
# After sync (sandbox path)
file:///workspace/.fabro/artifacts/response.plan.json
@ -256,13 +257,15 @@ Collected assets are written to the run's directory, organized by node and retry
```
~/.fabro/runs/{run_id}/
assets/
{node_slug}/
retry_1/
test-results/
screenshot.png
video.webm
manifest.json
cache/
artifacts/
assets/
{node_slug}/
retry_1/
test-results/
screenshot.png
video.webm
manifest.json
```
Each collection writes a `manifest.json` summarizing what was captured:
@ -291,4 +294,4 @@ Outputs and artifacts appear in several observability surfaces:
| `WorkflowRunCompleted` event | `artifact_count` -- total number of offloaded artifacts across the run |
| [Retros](/execution/retros) | Per-stage `files_touched` and aggregate `files_touched` across all stages |
| [Preambles](/execution/context#preamble-construction) | File list and artifact pointer references for completed stages |
| Stage logs | `status.json` in each stage's run directory contains the full outcome including `files_touched` |
| Stage logs | `status.json` in each stage's run directory contains the full outcome including `files_touched` |

2
docs/execution/context.mdx
View file

@ -199,7 +199,7 @@ Internal keys (prefixed with `internal.`, `current`, `graph.`, `thread.`, `respo
When a stage produces a large output (over 100KB of serialized JSON), Fabro automatically offloads it to the **artifact store** on disk rather than keeping it in the in-memory context. The context value is replaced with a `file://` pointer:
```
response.plan → file:///tmp/logs/artifacts/values/response.plan.json
response.plan → file:///tmp/logs/cache/artifacts/values/response.plan.json
```
The preamble renderer resolves these pointers and displays a reference to the file path. For remote sandboxes (Docker, Daytona), Fabro syncs artifact files to the sandbox at `{working_directory}/.fabro/artifacts/` so agents can read them.

26
docs/reference/run-directory.mdx
View file

@ -69,7 +69,12 @@ Manager nodes that run sub-workflows write a nested `child/` directory containin
**`worktree/`** — When running in git checkpoint mode, Fabro creates a Git worktree here as the working directory for agents and commands.
**`assets/`** — Test artifacts collected from the execution environment (Playwright reports, JUnit XML, Cypress screenshots/videos). Located at `worktree/assets/` for local runs or within node directories for remote sandbox runs.
**`runtime/`** — Local-only runtime files, including interview IPC files used by detached runs and `fabro attach`.
**`cache/`** — Local filesystem cache for file-backed artifacts and captured test assets:
- `cache/artifacts/values/` — large context values offloaded from checkpoints
- `cache/artifacts/assets/` — captured test artifacts organized by node and retry
## Browsing runs
@ -98,6 +103,21 @@ fabro ps --filter workflow=my-workflow
│ ├── final.patch
│ ├── retro.json
│ ├── cli.log
│ ├── runtime/
│ │ ├── interview_request.json
│ │ ├── interview_response.json
│ │ └── interview_request.claim
│ ├── cache/
│ │ └── artifacts/
│ │ ├── values/
│ │ │ ├── response.plan.json
│ │ │ └── command.output.json
│ │ └── assets/
│ │ └── test/
│ │ └── retry_1/
│ │ ├── test-results/
│ │ │ └── screenshot.png
│ │ └── manifest.json
│ ├── nodes/
│ │ ├── plan/
│ │ │ ├── prompt.md
@ -126,7 +146,5 @@ fabro ps --filter workflow=my-workflow
│ │ ├── start.json
│ │ ├── checkpoint.json
│ │ └── nodes/
│ ├── worktree/ # Git worktree (git checkpoint mode)
│ │ └── assets/ # Test artifacts
│ └── assets/ # Or here for remote sandboxes
│ └── worktree/ # Git worktree (git checkpoint mode)
```

1
lib/crates/fabro-cli/Cargo.toml
View file

@ -34,6 +34,7 @@ fabro-validate = { path = "../fabro-validate" }
fabro-workflows = { path = "../fabro-workflows" }
fabro-api = { path = "../fabro-api", optional = true }
fabro-telemetry = { path = "../fabro-telemetry" }
fabro-store = { path = "../fabro-store" }
fabro-types = { path = "../fabro-types" }
fabro-util = { path = "../fabro-util" }
clap.workspace = true

5
lib/crates/fabro-cli/src/commands/asset/cp.rs
View file

@ -2,6 +2,7 @@ use std::path::{Path, PathBuf};
use anyhow::{bail, Context, Result};
use fabro_config::FabroSettingsExt;
use fabro_store::RuntimeState;
use crate::args::AssetCpArgs;
use crate::shared::split_run_path;
@ -11,7 +12,9 @@ pub fn cp_command(args: &AssetCpArgs) -> Result<()> {
let base = fabro_workflows::run_lookup::runs_base(&cli_config.storage_dir());
let (run_id, asset_path) = parse_source(&args.source);
let run = fabro_workflows::run_lookup::resolve_run(&base, run_id)?;
let entries = fabro_workflows::assets::scan_assets(&run.path, args.node.as_deref())?;
let runtime_state = RuntimeState::new(&run.path);
let entries =
fabro_workflows::assets::scan_assets(&runtime_state.assets_dir(), args.node.as_deref())?;
if entries.is_empty() {
bail!("No assets found for this run");

5
lib/crates/fabro-cli/src/commands/asset/list.rs
View file

@ -1,5 +1,6 @@
use anyhow::Result;
use fabro_config::FabroSettingsExt;
use fabro_store::RuntimeState;
use crate::args::AssetListArgs;
use crate::shared::format_size;
@ -8,7 +9,9 @@ pub fn list_command(args: &AssetListArgs) -> Result<()> {
let cli_config = crate::cli_config::load_cli_settings(None)?;
let base = fabro_workflows::run_lookup::runs_base(&cli_config.storage_dir());
let run = fabro_workflows::run_lookup::resolve_run(&base, &args.run_id)?;
let entries = fabro_workflows::assets::scan_assets(&run.path, args.node.as_deref())?;
let runtime_state = RuntimeState::new(&run.path);
let entries =
fabro_workflows::assets::scan_assets(&runtime_state.assets_dir(), args.node.as_deref())?;
if args.json {
println!("{}", serde_json::to_string_pretty(&entries)?);

147
lib/crates/fabro-cli/src/commands/run/attach.rs
View file

@ -8,6 +8,7 @@ use std::time::{Duration, Instant};
use anyhow::{bail, Result};
use fabro_interview::{AnswerValue, ConsoleInterviewer};
use fabro_store::RuntimeState;
use fabro_util::terminal::Styles;
use fabro_workflows::records::{ConclusionExt, RunRecordExt};
use fabro_workflows::run_status::{RunStatus, RunStatusRecord, RunStatusRecordExt};
@ -33,8 +34,9 @@ pub async fn attach_run(
let progress_path = run_dir.join("progress.jsonl");
let conclusion_path = run_dir.join("conclusion.json");
let status_path = run_dir.join("status.json");
let interview_request_path = run_dir.join("interview_request.json");
let interview_response_path = run_dir.join("interview_response.json");
let runtime_state = RuntimeState::new(run_dir);
let runtime_interview_paths = InterviewPaths::from_runtime_state(&runtime_state);
let legacy_interview_paths = InterviewPaths::from_base_dir(run_dir);
let mut engine_guard = engine_child.map(EngineChildGuard::new);
@ -163,32 +165,41 @@ pub async fn attach_run(
}
// Check for interview request
if interview_request_path.exists() && !interview_response_path.exists() {
if let Some(_claim_guard) = InterviewClaimGuard::acquire(run_dir) {
if let Ok(request_data) = std::fs::read_to_string(&interview_request_path) {
if let Ok(question) =
serde_json::from_str::<fabro_interview::Question>(&request_data)
if let Some(interview_paths) =
active_interview_paths(&runtime_interview_paths, &legacy_interview_paths)
{
if !interview_paths.response_path.exists() {
if let Some(_claim_guard) = InterviewClaimGuard::acquire(&interview_paths.base_dir)
{
if let Ok(request_data) = std::fs::read_to_string(&interview_paths.request_path)
{
// Hide progress bars during interview
progress_ui.hide_bars();
if let Ok(question) =
serde_json::from_str::<fabro_interview::Question>(&request_data)
{
// Hide progress bars during interview
progress_ui.hide_bars();
// Prompt user via ConsoleInterviewer
let interviewer = ConsoleInterviewer::new(styles);
let answer =
fabro_interview::Interviewer::ask(&interviewer, question).await;
// Prompt user via ConsoleInterviewer
let interviewer = ConsoleInterviewer::new(styles);
let answer =
fabro_interview::Interviewer::ask(&interviewer, question).await;
// Show progress bars again before any return path.
progress_ui.show_bars();
// Show progress bars again before any return path.
progress_ui.show_bars();
if answer_requires_reattach(&answer) {
if let Some(guard) = engine_guard.as_mut() {
guard.defuse();
if answer_requires_reattach(&answer) {
if let Some(guard) = engine_guard.as_mut() {
guard.defuse();
}
eprintln!("{INTERVIEW_UNANSWERED_MESSAGE}");
return Ok(ExitCode::from(1));
}
eprintln!("{INTERVIEW_UNANSWERED_MESSAGE}");
return Ok(ExitCode::from(1));
}
write_interview_response_atomically(&interview_response_path, &answer)?;
write_interview_response_atomically(
&interview_paths.response_path,
&answer,
)?;
}
}
}
}
@ -287,8 +298,47 @@ fn progress_file_is_empty(path: &Path) -> bool {
.unwrap_or(true)
}
fn interview_claim_path(run_dir: &Path) -> std::path::PathBuf {
run_dir.join("interview_request.claim")
#[derive(Debug, Clone, PartialEq, Eq)]
struct InterviewPaths {
base_dir: PathBuf,
request_path: PathBuf,
response_path: PathBuf,
}
impl InterviewPaths {
fn from_base_dir(base_dir: &Path) -> Self {
let base_dir = base_dir.to_path_buf();
Self {
request_path: base_dir.join("interview_request.json"),
response_path: base_dir.join("interview_response.json"),
base_dir,
}
}
fn from_runtime_state(runtime_state: &RuntimeState) -> Self {
Self {
base_dir: runtime_state.runtime_dir(),
request_path: runtime_state.interview_request_path(),
response_path: runtime_state.interview_response_path(),
}
}
}
fn active_interview_paths(
runtime_paths: &InterviewPaths,
legacy_paths: &InterviewPaths,
) -> Option<InterviewPaths> {
if runtime_paths.request_path.exists() {
Some(runtime_paths.clone())
} else if legacy_paths.request_path.exists() {
Some(legacy_paths.clone())
} else {
None
}
}
fn interview_claim_path(base_dir: &Path) -> PathBuf {
base_dir.join("interview_request.claim")
}
struct InterviewClaimGuard {
@ -296,10 +346,10 @@ struct InterviewClaimGuard {
}
impl InterviewClaimGuard {
fn acquire(run_dir: &Path) -> Option<Self> {
if try_claim_interview_request(run_dir) {
fn acquire(base_dir: &Path) -> Option<Self> {
if try_claim_interview_request(base_dir) {
Some(Self {
claim_path: interview_claim_path(run_dir),
claim_path: interview_claim_path(base_dir),
})
} else {
None
@ -340,8 +390,12 @@ impl Drop for EngineChildGuard {
}
}
fn try_claim_interview_request(run_dir: &Path) -> bool {
let claim_path = interview_claim_path(run_dir);
fn try_claim_interview_request(base_dir: &Path) -> bool {
if std::fs::create_dir_all(base_dir).is_err() {
return false;
}
let claim_path = interview_claim_path(base_dir);
if let Ok(existing) = std::fs::read_to_string(&claim_path) {
if let Ok(pid) = existing.trim().parse::<u32>() {
if process_alive(pid) {
@ -373,6 +427,9 @@ fn write_interview_response_atomically(
answer: &fabro_interview::Answer,
) -> Result<()> {
let response_json = serde_json::to_string_pretty(answer)?;
if let Some(parent) = response_path.parent() {
std::fs::create_dir_all(parent)?;
}
let temp_path = response_path.with_extension("json.tmp");
std::fs::write(&temp_path, response_json)?;
std::fs::rename(temp_path, response_path)?;
@ -522,6 +579,38 @@ mod tests {
assert!(!interview_claim_path(dir.path()).exists());
}
#[test]
fn active_interview_paths_prefers_runtime_paths() {
let dir = tempfile::tempdir().unwrap();
let runtime_state = RuntimeState::new(dir.path());
let runtime_paths = InterviewPaths::from_runtime_state(&runtime_state);
let legacy_paths = InterviewPaths::from_base_dir(dir.path());
std::fs::create_dir_all(runtime_state.runtime_dir()).unwrap();
std::fs::write(&runtime_paths.request_path, "{}").unwrap();
std::fs::write(&legacy_paths.request_path, "{}").unwrap();
assert_eq!(
active_interview_paths(&runtime_paths, &legacy_paths),
Some(runtime_paths)
);
}
#[test]
fn active_interview_paths_falls_back_to_legacy_paths() {
let dir = tempfile::tempdir().unwrap();
let runtime_state = RuntimeState::new(dir.path());
let runtime_paths = InterviewPaths::from_runtime_state(&runtime_state);
let legacy_paths = InterviewPaths::from_base_dir(dir.path());
std::fs::write(&legacy_paths.request_path, "{}").unwrap();
assert_eq!(
active_interview_paths(&runtime_paths, &legacy_paths),
Some(legacy_paths)
);
}
#[test]
fn answer_requires_reattach_for_aborted_and_skipped_answers() {
let aborted = Answer {

8
lib/crates/fabro-cli/src/commands/run/detached.rs
View file

@ -3,6 +3,7 @@ use std::sync::Arc;
use anyhow::Result;
use fabro_interview::FileInterviewer;
use fabro_store::RuntimeState;
use fabro_workflows::event::EventEmitter;
use crate::cli_config;
@ -19,11 +20,16 @@ pub async fn execute(run_dir: PathBuf, launcher_path: PathBuf, resume: bool) ->
let _launcher_guard = scopeguard::guard(launcher_path.clone(), |path| {
super::launcher::remove_launcher_record(&path);
});
let runtime_state = RuntimeState::new(&run_dir);
let services = fabro_workflows::operations::StartServices {
cancel_token: None,
emitter: Arc::new(EventEmitter::new()),
interviewer: Arc::new(FileInterviewer::new(run_dir.clone())),
interviewer: Arc::new(FileInterviewer::new(
runtime_state.interview_request_path(),
runtime_state.interview_response_path(),
runtime_state.interview_claim_path(),
)),
git_author,
github_app,
registry_override: None,

4
lib/crates/fabro-cli/src/commands/run/output.rs
View file

@ -1,6 +1,7 @@
use std::path::Path;
use std::time::Duration;
use fabro_store::RuntimeState;
use fabro_util::terminal::Styles;
use fabro_workflows::outcome::{format_cost, StageStatus};
use fabro_workflows::pipeline::{Persisted, Validated};
@ -199,7 +200,8 @@ pub(crate) fn print_final_output(run_dir: &Path, styles: &Styles) {
}
pub(crate) fn print_assets(run_dir: &Path, styles: &Styles) {
let paths = fabro_workflows::asset_snapshot::collect_asset_paths(run_dir);
let runtime_state = RuntimeState::new(run_dir);
let paths = fabro_workflows::asset_snapshot::collect_asset_paths(&runtime_state.assets_dir());
if paths.is_empty() {
return;
}

72
lib/crates/fabro-cli/tests/cli.rs
View file

@ -1,6 +1,7 @@
use assert_cmd::Command;
use fabro_config::mcp::McpTransport;
use fabro_config::FabroSettings;
use fabro_store::RuntimeState;
use predicates::prelude::*;
#[allow(deprecated)]
@ -1303,8 +1304,10 @@ fn bug3_attach_leaves_interview_request_until_engine_consumes_response() {
"stage": "gate",
"metadata": {}
});
let runtime_state = RuntimeState::new(&run_dir);
std::fs::create_dir_all(runtime_state.runtime_dir()).unwrap();
std::fs::write(
run_dir.join("interview_request.json"),
runtime_state.interview_request_path(),
serde_json::to_string(&question).unwrap(),
)
.unwrap();
@ -1324,14 +1327,14 @@ fn bug3_attach_leaves_interview_request_until_engine_consumes_response() {
// The attach loop should leave the request durable until the engine consumes
// the response, so a crashed attach can be retried safely.
assert!(
run_dir.join("interview_request.json").exists(),
runtime_state.interview_request_path().exists(),
"bug3: interview_request.json should stay present until the engine consumes the answer"
);
assert!(
run_dir.join("interview_response.json").exists(),
runtime_state.interview_response_path().exists(),
"bug3: attach should write interview_response.json after handling the prompt"
);
let response = std::fs::read_to_string(run_dir.join("interview_response.json")).unwrap();
let response = std::fs::read_to_string(runtime_state.interview_response_path()).unwrap();
assert!(response.contains("\"value\": \"Yes\""));
}
@ -1364,8 +1367,10 @@ fn attach_closed_stdin_keeps_interview_pending() {
"stage": "gate",
"metadata": {}
});
let runtime_state = RuntimeState::new(&run_dir);
std::fs::create_dir_all(runtime_state.runtime_dir()).unwrap();
std::fs::write(
run_dir.join("interview_request.json"),
runtime_state.interview_request_path(),
serde_json::to_string(&question).unwrap(),
)
.unwrap();
@ -1386,19 +1391,70 @@ fn attach_closed_stdin_keeps_interview_pending() {
"attach should explain that the run is still waiting for a human answer.\nstderr: {stderr}"
);
assert!(
run_dir.join("interview_request.json").exists(),
runtime_state.interview_request_path().exists(),
"attach with closed stdin must leave the request pending"
);
assert!(
!run_dir.join("interview_response.json").exists(),
!runtime_state.interview_response_path().exists(),
"attach with closed stdin must not fabricate a response"
);
assert!(
!run_dir.join("interview_request.claim").exists(),
!runtime_state.interview_claim_path().exists(),
"attach with closed stdin must release the claim so a later attach can answer"
);
}
#[test]
fn attach_supports_legacy_root_interview_paths() {
let home = tempfile::tempdir().unwrap();
let run_dir = setup_run_dir(
home.path(),
"attach-legacy-interview-paths",
serde_json::json!({}),
&[
r#"{"ts":"2026-01-01T00:00:01Z","run_id":"attach-legacy-interview-paths","event":"StageStarted","node_id":"gate","name":"Gate","index":0,"attempt":1,"max_attempts":1}"#,
],
);
std::fs::write(
run_dir.join("status.json"),
serde_json::json!({"status": "running", "updated_at": "2026-01-01T00:00:00Z"}).to_string(),
)
.unwrap();
let question = serde_json::json!({
"text": "Approve?",
"question_type": "YesNo",
"options": [],
"allow_freeform": false,
"default": {"value": "Yes", "selected_option": null, "text": null},
"timeout_seconds": 1.0,
"stage": "gate",
"metadata": {}
});
std::fs::write(
run_dir.join("interview_request.json"),
serde_json::to_string(&question).unwrap(),
)
.unwrap();
std::fs::write(run_dir.join("run.pid"), "99999999").unwrap();
let _ = arc()
.env("HOME", home.path())
.env("NO_COLOR", "1")
.args(["attach", "attach-legacy-interview-paths"])
.write_stdin("y\n")
.timeout(std::time::Duration::from_secs(5))
.output();
assert!(run_dir.join("interview_request.json").exists());
assert!(run_dir.join("interview_response.json").exists());
assert!(!RuntimeState::new(&run_dir)
.interview_response_path()
.exists());
}
// Bug 4: attach should respect the verbose flag from run.json.
// Currently ProgressUI is created with verbose=false regardless of config.
#[test]

3
lib/crates/fabro-cli/tests/scenario.rs
View file

@ -2,6 +2,7 @@ use std::path::{Path, PathBuf};
use std::time::Duration;
use assert_cmd::Command;
use fabro_store::RuntimeState;
use predicates;
use serde_json::Value;
@ -782,7 +783,7 @@ fn local_run_lifecycle() {
);
// 6. Seed a synthetic asset so asset list/cp have something to work with.
let asset_dir = run_dir.join("artifacts/assets/step1/retry_0");
let asset_dir = RuntimeState::new(&run_dir).asset_stage_dir("step1", 0);
std::fs::create_dir_all(&asset_dir).unwrap();
std::fs::write(asset_dir.join("output.txt"), "asset-content-42").unwrap();
std::fs::write(

73
lib/crates/fabro-interview/src/file.rs
View file

@ -10,35 +10,47 @@ const REATTACH_WINDOW: Duration = Duration::from_millis(300);
#[cfg(not(test))]
const REATTACH_WINDOW: Duration = Duration::from_secs(30);
#[cfg(test)]
use std::path::Path;
/// An interviewer that communicates via JSON files in the run directory.
///
/// The engine process writes `interview_request.json` and polls for
/// `interview_response.json`. The attach process watches for the request
/// file, prompts the user, and writes the response file.
pub struct FileInterviewer {
run_dir: PathBuf,
request_path: PathBuf,
response_path: PathBuf,
claim_path: PathBuf,
}
impl FileInterviewer {
pub fn new(run_dir: PathBuf) -> Self {
Self { run_dir }
pub fn new(request_path: PathBuf, response_path: PathBuf, claim_path: PathBuf) -> Self {
Self {
request_path,
response_path,
claim_path,
}
}
fn request_path(&self) -> PathBuf {
self.run_dir.join("interview_request.json")
self.request_path.clone()
}
fn response_path(&self) -> PathBuf {
self.run_dir.join("interview_response.json")
self.response_path.clone()
}
fn claim_path(&self) -> PathBuf {
self.run_dir.join("interview_request.claim")
self.claim_path.clone()
}
async fn write_request_atomically(&self, question: &Question) -> std::io::Result<()> {
let json = serde_json::to_string_pretty(question).expect("Question serialization failed");
let request_path = self.request_path();
if let Some(parent) = request_path.parent() {
tokio::fs::create_dir_all(parent).await?;
}
let temp_path = request_path.with_extension("json.tmp");
tokio::fs::write(&temp_path, json).await?;
tokio::fs::rename(temp_path, request_path).await
@ -133,11 +145,24 @@ mod tests {
use super::*;
use crate::{AnswerValue, QuestionType};
fn interviewer_paths(run_dir: &Path) -> (PathBuf, PathBuf, PathBuf) {
(
run_dir.join("interview_request.json"),
run_dir.join("interview_response.json"),
run_dir.join("interview_request.claim"),
)
}
#[tokio::test]
async fn write_request_poll_response() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir.path().to_path_buf();
let interviewer = FileInterviewer::new(run_dir.clone());
let (request_path, response_path, claim_path) = interviewer_paths(&run_dir);
let interviewer = FileInterviewer::new(
request_path.clone(),
response_path.clone(),
claim_path.clone(),
);
let question = Question::new("approve?", QuestionType::YesNo);
@ -145,7 +170,6 @@ mod tests {
let ask_handle = tokio::spawn(async move { interviewer.ask(question).await });
// Wait for the request file to appear
let request_path = run_dir.join("interview_request.json");
for _ in 0..50 {
if request_path.exists() {
break;
@ -162,7 +186,6 @@ mod tests {
// Write a response
let answer = Answer::yes();
let response_json = serde_json::to_string_pretty(&answer).unwrap();
let response_path = run_dir.join("interview_response.json");
tokio::fs::write(&response_path, response_json)
.await
.unwrap();
@ -174,13 +197,14 @@ mod tests {
// Both files should be cleaned up
assert!(!request_path.exists());
assert!(!response_path.exists());
assert!(!run_dir.join("interview_request.claim").exists());
assert!(!claim_path.exists());
}
#[tokio::test]
async fn timeout_returns_default() {
let dir = tempfile::tempdir().unwrap();
let interviewer = FileInterviewer::new(dir.path().to_path_buf());
let (request_path, response_path, claim_path) = interviewer_paths(dir.path());
let interviewer = FileInterviewer::new(request_path, response_path, claim_path);
let mut question = Question::new("approve?", QuestionType::YesNo);
question.timeout_seconds = Some(0.1);
@ -194,14 +218,15 @@ mod tests {
async fn claim_released_without_response_returns_timeout() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir.path().to_path_buf();
let interviewer = FileInterviewer::new(run_dir.clone());
let (request_path, response_path, claim_path) = interviewer_paths(&run_dir);
let interviewer =
FileInterviewer::new(request_path.clone(), response_path, claim_path.clone());
let question = Question::new("approve?", QuestionType::YesNo);
let ask_handle = tokio::spawn(async move { interviewer.ask(question).await });
// Wait for request file to appear
let request_path = run_dir.join("interview_request.json");
for _ in 0..50 {
if request_path.exists() {
break;
@ -211,7 +236,6 @@ mod tests {
assert!(request_path.exists());
// Simulate attacher creating claim file
let claim_path = run_dir.join("interview_request.claim");
std::fs::write(&claim_path, "12345\n").unwrap();
// Let the poll loop see the claim
@ -238,7 +262,9 @@ mod tests {
async fn claim_released_without_response_returns_default() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir.path().to_path_buf();
let interviewer = FileInterviewer::new(run_dir.clone());
let (request_path, response_path, claim_path) = interviewer_paths(&run_dir);
let interviewer =
FileInterviewer::new(request_path.clone(), response_path, claim_path.clone());
let mut question = Question::new("approve?", QuestionType::YesNo);
question.default = Some(Answer::no());
@ -246,7 +272,6 @@ mod tests {
let ask_handle = tokio::spawn(async move { interviewer.ask(question).await });
// Wait for request file
let request_path = run_dir.join("interview_request.json");
for _ in 0..50 {
if request_path.exists() {
break;
@ -256,7 +281,6 @@ mod tests {
assert!(request_path.exists());
// Simulate attacher creating then deleting claim
let claim_path = run_dir.join("interview_request.claim");
std::fs::write(&claim_path, "12345\n").unwrap();
tokio::time::sleep(Duration::from_millis(150)).await;
std::fs::remove_file(&claim_path).unwrap();
@ -273,15 +297,18 @@ mod tests {
async fn claim_released_then_new_attacher_answers() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir.path().to_path_buf();
let interviewer = FileInterviewer::new(run_dir.clone());
let (request_path, response_path, claim_path) = interviewer_paths(&run_dir);
let interviewer = FileInterviewer::new(
request_path.clone(),
response_path.clone(),
claim_path.clone(),
);
let question = Question::new("approve?", QuestionType::YesNo);
let run_dir2 = run_dir.clone();
let ask_handle = tokio::spawn(async move { interviewer.ask(question).await });
// Wait for request file
let request_path = run_dir.join("interview_request.json");
for _ in 0..50 {
if request_path.exists() {
break;
@ -291,7 +318,6 @@ mod tests {
assert!(request_path.exists());
// First attacher creates then releases claim
let claim_path = run_dir.join("interview_request.claim");
std::fs::write(&claim_path, "12345\n").unwrap();
tokio::time::sleep(Duration::from_millis(150)).await;
std::fs::remove_file(&claim_path).unwrap();
@ -302,7 +328,7 @@ mod tests {
let answer = Answer::yes();
let response_json = serde_json::to_string_pretty(&answer).unwrap();
tokio::fs::write(run_dir2.join("interview_response.json"), response_json)
tokio::fs::write(response_path, response_json)
.await
.unwrap();
@ -317,7 +343,8 @@ mod tests {
#[tokio::test]
async fn timeout_without_default_returns_timeout() {
let dir = tempfile::tempdir().unwrap();
let interviewer = FileInterviewer::new(dir.path().to_path_buf());
let (request_path, response_path, claim_path) = interviewer_paths(dir.path());
let interviewer = FileInterviewer::new(request_path, response_path, claim_path);
let mut question = Question::new("approve?", QuestionType::YesNo);
question.timeout_seconds = Some(0.1);

1
lib/crates/fabro-store/Cargo.toml
View file

@ -24,3 +24,4 @@ futures.workspace = true
[dev-dependencies]
tokio = { workspace = true, features = ["test-util", "macros"] }
tempfile = "3"

2
lib/crates/fabro-store/src/lib.rs
View file

@ -8,11 +8,13 @@ use futures::Stream;
mod error;
mod keys;
mod memory;
mod runtime;
mod slate;
mod types;
pub use error::{Result, StoreError};
pub use memory::InMemoryStore;
pub use runtime::RuntimeState;
pub use slate::SlateStore;
pub use types::{
CatalogRecord, EventEnvelope, EventPayload, NodeSnapshot, NodeVisitRef, RunSnapshot, RunSummary,

134
lib/crates/fabro-store/src/runtime.rs Normal file
View file

@ -0,0 +1,134 @@
use std::path::{Path, PathBuf};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RuntimeState {
root: PathBuf,
}
impl RuntimeState {
#[must_use]
pub fn new(run_dir: impl AsRef<Path>) -> Self {
Self {
root: run_dir.as_ref().to_path_buf(),
}
}
#[must_use]
pub fn root(&self) -> &Path {
&self.root
}
#[must_use]
pub fn runtime_dir(&self) -> PathBuf {
self.root.join("runtime")
}
#[must_use]
pub fn interview_request_path(&self) -> PathBuf {
self.runtime_dir().join("interview_request.json")
}
#[must_use]
pub fn interview_response_path(&self) -> PathBuf {
self.runtime_dir().join("interview_response.json")
}
#[must_use]
pub fn interview_claim_path(&self) -> PathBuf {
self.runtime_dir().join("interview_request.claim")
}
#[must_use]
pub fn artifact_values_dir(&self) -> PathBuf {
self.root.join("cache").join("artifacts").join("values")
}
#[must_use]
pub fn artifact_value_path(&self, artifact_id: &str) -> PathBuf {
self.artifact_values_dir()
.join(format!("{artifact_id}.json"))
}
#[must_use]
pub fn assets_dir(&self) -> PathBuf {
self.root.join("cache").join("artifacts").join("assets")
}
#[must_use]
pub fn asset_stage_dir(&self, node_slug: &str, attempt: u32) -> PathBuf {
self.assets_dir()
.join(node_slug)
.join(format!("retry_{attempt}"))
}
pub fn ensure_runtime_dir(&self) -> std::io::Result<()> {
std::fs::create_dir_all(self.runtime_dir())
}
pub fn ensure_artifact_values_dir(&self) -> std::io::Result<()> {
std::fs::create_dir_all(self.artifact_values_dir())
}
}
#[cfg(test)]
mod tests {
use super::RuntimeState;
#[test]
fn computes_runtime_and_cache_paths() {
let dir = tempfile::tempdir().unwrap();
let state = RuntimeState::new(dir.path());
assert_eq!(state.root(), dir.path());
assert_eq!(state.runtime_dir(), dir.path().join("runtime"));
assert_eq!(
state.interview_request_path(),
dir.path().join("runtime").join("interview_request.json")
);
assert_eq!(
state.interview_response_path(),
dir.path().join("runtime").join("interview_response.json")
);
assert_eq!(
state.interview_claim_path(),
dir.path().join("runtime").join("interview_request.claim")
);
assert_eq!(
state.artifact_values_dir(),
dir.path().join("cache").join("artifacts").join("values")
);
assert_eq!(
state.artifact_value_path("response.plan"),
dir.path()
.join("cache")
.join("artifacts")
.join("values")
.join("response.plan.json")
);
assert_eq!(
state.assets_dir(),
dir.path().join("cache").join("artifacts").join("assets")
);
assert_eq!(
state.asset_stage_dir("plan", 2),
dir.path()
.join("cache")
.join("artifacts")
.join("assets")
.join("plan")
.join("retry_2")
);
}
#[test]
fn ensure_methods_create_directories() {
let dir = tempfile::tempdir().unwrap();
let state = RuntimeState::new(dir.path());
state.ensure_runtime_dir().unwrap();
state.ensure_artifact_values_dir().unwrap();
assert!(state.runtime_dir().is_dir());
assert!(state.artifact_values_dir().is_dir());
}
}

1
lib/crates/fabro-workflows/Cargo.toml
View file

@ -35,6 +35,7 @@ fabro-llm = { path = "../fabro-llm" }
fabro-model = { path = "../fabro-model" }
fabro-retro = { path = "../fabro-retro" }
fabro-core = { path = "../fabro-core" }
fabro-store = { path = "../fabro-store" }
fabro-types = { path = "../fabro-types" }
thiserror.workspace = true
serde.workspace = true

67
lib/crates/fabro-workflows/src/artifact.rs
View file

@ -32,28 +32,28 @@ enum StoredData {
/// Named, typed storage for large stage outputs.
pub struct ArtifactStore {
base_dir: Option<PathBuf>,
values_dir: Option<PathBuf>,
artifacts: RwLock<HashMap<String, (ArtifactInfo, StoredData)>>,
}
impl std::fmt::Debug for ArtifactStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ArtifactStore")
.field("base_dir", &self.base_dir)
.field("values_dir", &self.values_dir)
.finish_non_exhaustive()
}
}
impl ArtifactStore {
#[must_use]
pub fn new(base_dir: Option<PathBuf>) -> Self {
pub fn new(values_dir: Option<PathBuf>) -> Self {
Self {
base_dir,
values_dir,
artifacts: RwLock::new(HashMap::new()),
}
}
/// Store an artifact. Large artifacts with a configured `base_dir` are written to disk.
/// Store an artifact. Large artifacts with a configured `values_dir` are written to disk.
///
/// # Errors
///
@ -74,13 +74,12 @@ impl ArtifactStore {
.map_err(|e| FabroError::engine(format!("artifact serialize failed: {e}")))?;
let size_bytes = serialized.len();
let is_file_backed = size_bytes > FILE_BACKING_THRESHOLD && self.base_dir.is_some();
let is_file_backed = size_bytes > FILE_BACKING_THRESHOLD && self.values_dir.is_some();
let (stored, file_path) = if is_file_backed {
let base = self.base_dir.as_ref().expect("base_dir checked above");
let artifacts_dir = base.join("artifacts").join("values");
std::fs::create_dir_all(&artifacts_dir)?;
let path = artifacts_dir.join(format!("{id}.json"));
let values_dir = self.values_dir.as_ref().expect("values_dir checked above");
std::fs::create_dir_all(values_dir)?;
let path = values_dir.join(format!("{id}.json"));
std::fs::write(&path, &serialized)?;
(StoredData::FileBacked(path.clone()), Some(path))
} else {
@ -175,13 +174,11 @@ impl ArtifactStore {
}
}
/// Returns the absolute path to the artifacts directory under this store's base_dir.
/// Returns `None` if no `base_dir` is configured.
/// Returns the configured values directory.
/// Returns `None` if no file-backed storage is configured.
#[must_use]
pub fn artifacts_dir(&self) -> Option<PathBuf> {
self.base_dir
.as_ref()
.map(|b| b.join("artifacts").join("values"))
pub fn values_dir(&self) -> Option<PathBuf> {
self.values_dir.clone()
}
/// Remove all artifacts. Also deletes file-backed data from disk.
@ -247,7 +244,8 @@ pub fn is_artifact_pointer(value: &Value) -> bool {
/// Resolve an artifact pointer to the base name displayed in preamble rendering.
///
/// Given `"file:///tmp/logs/artifacts/response.plan.json"`, returns `"See: /tmp/logs/artifacts/response.plan.json"`.
/// Given `"file:///tmp/logs/cache/artifacts/values/response.plan.json"`, returns
/// `"See: /tmp/logs/cache/artifacts/values/response.plan.json"`.
#[must_use]
pub fn format_artifact_reference(path: &str) -> String {
format!("See: {path}")
@ -377,10 +375,7 @@ mod tests {
let info = store.store("big", "large artifact", data.clone()).unwrap();
assert!(info.is_file_backed);
assert!(info.size_bytes > FILE_BACKING_THRESHOLD);
assert_eq!(
info.file_path,
Some(dir.path().join("artifacts").join("values").join("big.json"))
);
assert_eq!(info.file_path, Some(dir.path().join("big.json")));
let retrieved = store.retrieve("big").unwrap();
assert_eq!(retrieved, data);
@ -395,7 +390,7 @@ mod tests {
let data = serde_json::json!(large_string);
store.store("big", "large", data).unwrap();
let file_path = dir.path().join("artifacts").join("values").join("big.json");
let file_path = dir.path().join("big.json");
assert!(file_path.exists());
store.remove("big");
@ -403,7 +398,7 @@ mod tests {
}
#[test]
fn small_artifact_stays_in_memory_even_with_base_dir() {
fn small_artifact_stays_in_memory_even_with_values_dir() {
let dir = tempfile::tempdir().unwrap();
let store = ArtifactStore::new(Some(dir.path().to_path_buf()));
@ -438,12 +433,7 @@ mod tests {
let path = artifact_path(pointer).expect("should be an artifact pointer");
assert_eq!(
path,
dir.path()
.join("artifacts")
.join("values")
.join("response.plan.json")
.to_str()
.unwrap()
dir.path().join("response.plan.json").to_str().unwrap()
);
// The artifact store should contain the original value
@ -451,12 +441,7 @@ mod tests {
assert_eq!(retrieved, serde_json::json!(large_string));
// File should exist on disk
assert!(dir
.path()
.join("artifacts")
.join("values")
.join("response.plan.json")
.exists());
assert!(dir.path().join("response.plan.json").exists());
}
#[test]
@ -476,13 +461,21 @@ mod tests {
#[test]
fn artifact_path_extracts_path_from_pointer() {
let value = serde_json::json!("file:///tmp/logs/artifacts/response.plan.json");
let value = serde_json::json!("file:///tmp/logs/cache/artifacts/values/response.plan.json");
assert_eq!(
artifact_path(&value),
Some("/tmp/logs/artifacts/response.plan.json")
Some("/tmp/logs/cache/artifacts/values/response.plan.json")
);
}
#[test]
fn values_dir_returns_configured_directory() {
let dir = tempfile::tempdir().unwrap();
let store = ArtifactStore::new(Some(dir.path().to_path_buf()));
assert_eq!(store.values_dir(), Some(dir.path().to_path_buf()));
}
#[test]
fn artifact_path_returns_none_for_plain_string() {
let value = serde_json::json!("just a normal string");

22
lib/crates/fabro-workflows/src/asset_snapshot.rs
View file

@ -317,12 +317,11 @@ pub async fn collect_assets(
Ok(summary)
}
/// Collect all asset paths from manifest files under `{run_dir}/artifacts/assets/*/retry_*/manifest.json`.
/// Collect all asset paths from manifest files under `{assets_dir}/*/retry_*/manifest.json`.
///
/// Returns the full on-disk paths to the downloaded asset files.
pub fn collect_asset_paths(run_dir: &Path) -> Vec<String> {
let assets_dir = run_dir.join("artifacts/assets");
let Ok(nodes) = std::fs::read_dir(&assets_dir) else {
pub fn collect_asset_paths(assets_dir: &Path) -> Vec<String> {
let Ok(nodes) = std::fs::read_dir(assets_dir) else {
return Vec::new();
};
@ -700,9 +699,10 @@ mod tests {
fn collect_asset_paths_from_manifests() {
let tmp = tempfile::tempdir().unwrap();
let base = tmp.path();
let assets_dir = base.join("cache/artifacts/assets");
// Create two node directories with manifests
let node_a = base.join("artifacts/assets/node_a/retry_1");
let node_a = assets_dir.join("node_a/retry_1");
std::fs::create_dir_all(&node_a).unwrap();
std::fs::write(
node_a.join("manifest.json"),
@ -720,7 +720,7 @@ mod tests {
)
.unwrap();
let node_b = base.join("artifacts/assets/node_b/retry_1");
let node_b = assets_dir.join("node_b/retry_1");
std::fs::create_dir_all(&node_b).unwrap();
std::fs::write(
node_b.join("manifest.json"),
@ -735,24 +735,24 @@ mod tests {
)
.unwrap();
let paths = collect_asset_paths(base);
let paths = collect_asset_paths(&assets_dir);
assert_eq!(paths.len(), 3);
let base_str = base.to_string_lossy();
assert!(paths.contains(&format!(
"{base_str}/artifacts/assets/node_a/retry_1/test-results/report.xml"
"{base_str}/cache/artifacts/assets/node_a/retry_1/test-results/report.xml"
)));
assert!(paths.contains(&format!(
"{base_str}/artifacts/assets/node_a/retry_1/test-results/screenshot.png"
"{base_str}/cache/artifacts/assets/node_a/retry_1/test-results/screenshot.png"
)));
assert!(paths.contains(&format!(
"{base_str}/artifacts/assets/node_b/retry_1/coverage/lcov.info"
"{base_str}/cache/artifacts/assets/node_b/retry_1/coverage/lcov.info"
)));
}
#[test]
fn collect_asset_paths_empty_when_no_assets() {
let tmp = tempfile::tempdir().unwrap();
let paths = collect_asset_paths(tmp.path());
let paths = collect_asset_paths(&tmp.path().join("cache/artifacts/assets"));
assert!(paths.is_empty());
}

7
lib/crates/fabro-workflows/src/assets.rs
View file

@ -19,10 +19,9 @@ fn serialize_path<S: serde::Serializer>(path: &Path, serializer: S) -> Result<S:
serializer.serialize_str(&path.display().to_string())
}
/// Walk `{run_dir}/artifacts/assets/*/retry_*/manifest.json`, stat each file, and return entries.
pub fn scan_assets(run_dir: &Path, node_filter: Option<&str>) -> Result<Vec<AssetEntry>> {
let assets_dir = run_dir.join("artifacts/assets");
let nodes = match std::fs::read_dir(&assets_dir) {
/// Walk `{assets_dir}/*/retry_*/manifest.json`, stat each file, and return entries.
pub fn scan_assets(assets_dir: &Path, node_filter: Option<&str>) -> Result<Vec<AssetEntry>> {
let nodes = match std::fs::read_dir(assets_dir) {
Ok(read_dir) => read_dir,
Err(_) => return Ok(Vec::new()),
};

18
lib/crates/fabro-workflows/src/lifecycle/artifact.rs
View file

@ -23,9 +23,9 @@ type WfNodeDecision = NodeDecision<Option<StageUsage>>;
pub struct ArtifactLifecycle {
pub sandbox: Arc<dyn fabro_sandbox::Sandbox>,
pub artifact_store: Arc<Mutex<ArtifactStore>>,
pub artifact_base_dir: Option<PathBuf>,
pub artifact_values_dir: Option<PathBuf>,
pub emitter: Arc<EventEmitter>,
pub run_dir: PathBuf,
pub assets_dir: PathBuf,
pub asset_globs: Vec<String>,
/// Per-attempt state: epoch seconds when the attempt started.
attempt_start_epoch: Mutex<Option<f64>>,
@ -36,17 +36,17 @@ impl ArtifactLifecycle {
pub fn new(
sandbox: Arc<dyn fabro_sandbox::Sandbox>,
artifact_store: Arc<Mutex<ArtifactStore>>,
artifact_base_dir: Option<PathBuf>,
artifact_values_dir: Option<PathBuf>,
emitter: Arc<EventEmitter>,
run_dir: PathBuf,
assets_dir: PathBuf,
asset_globs: Vec<String>,
) -> Self {
Self {
sandbox,
artifact_store,
artifact_base_dir,
artifact_values_dir,
emitter,
run_dir,
assets_dir,
asset_globs,
attempt_start_epoch: Mutex::new(None),
}
@ -62,7 +62,7 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
) -> fabro_core::error::Result<()> {
// Swap in a fresh artifact store on restart (don't call clear() — preserves files on disk)
let mut store = self.artifact_store.lock().unwrap();
*store = ArtifactStore::new(self.artifact_base_dir.clone());
*store = ArtifactStore::new(self.artifact_values_dir.clone());
*self.attempt_start_epoch.lock().unwrap() = None;
Ok(())
}
@ -98,9 +98,7 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
format!("{node_id}-visit_{visit}")
};
let stage_dir = self
.run_dir
.join("artifacts")
.join("assets")
.assets_dir
.join(node_slug)
.join(format!("retry_{}", ctx.attempt));
let _ = std::fs::create_dir_all(&stage_dir);

10
lib/crates/fabro-workflows/src/lifecycle/mod.rs
View file

@ -14,6 +14,7 @@ use std::sync::{Arc, Mutex};
use std::time::Instant;
use async_trait::async_trait;
use fabro_store::RuntimeState;
use fabro_core::error::Result as CoreResult;
use fabro_core::graph::NodeSpec;
@ -83,12 +84,15 @@ impl WorkflowLifecycle {
run_options: Arc<RunOptions>,
is_resume: bool,
) -> Self {
let runtime_state = RuntimeState::new(&run_dir);
let restarted_from: Arc<Mutex<Option<(String, String)>>> = Arc::new(Mutex::new(None));
let loop_restart_signature_limit = graph.loop_restart_signature_limit();
let checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>> =
Arc::new(Mutex::new(None));
let last_git_sha: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
let artifact_store = Arc::new(Mutex::new(ArtifactStore::new(Some(run_dir.clone()))));
let artifact_store = Arc::new(Mutex::new(ArtifactStore::new(Some(
runtime_state.artifact_values_dir(),
))));
let circuit_breaker = Arc::new(CircuitBreakerLifecycle::new(loop_restart_signature_limit));
@ -157,9 +161,9 @@ impl WorkflowLifecycle {
let artifact = ArtifactLifecycle::new(
Arc::clone(&sandbox),
Arc::clone(&artifact_store),
Some(run_dir.clone()),
Some(runtime_state.artifact_values_dir()),
Arc::clone(&emitter),
run_dir,
runtime_state.assets_dir(),
run_options.asset_globs().to_vec(),
);

11
lib/crates/fabro-workflows/src/operations/resume.rs
View file

@ -1,5 +1,7 @@
use std::path::Path;
use fabro_store::RuntimeState;
use crate::error::FabroError;
use crate::outcome::StageStatus;
use crate::records::{Checkpoint, CheckpointExt, Conclusion, ConclusionExt};
@ -38,6 +40,15 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
}
fn cleanup_resume_artifacts(run_dir: &Path) {
let runtime_state = RuntimeState::new(run_dir);
for path in [
runtime_state.interview_request_path(),
runtime_state.interview_response_path(),
runtime_state.interview_claim_path(),
] {
let _ = std::fs::remove_file(path);
}
for name in [
"conclusion.json",
"pull_request.json",

8
lib/crates/fabro-workflows/tests/daytona_integration.rs
View file

@ -13,6 +13,7 @@ use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
use fabro_llm::provider::Provider;
use fabro_sandbox::daytona::{DaytonaConfig, DaytonaSandbox, DaytonaSnapshotConfig};
use fabro_sandbox::SandboxRecordExt;
use fabro_store::RuntimeState;
use fabro_workflows::artifact::sync_artifacts_to_env;
use fabro_workflows::context::Context;
use fabro_workflows::error::FabroError;
@ -1307,12 +1308,7 @@ async fn daytona_asset_collection() {
.expect("pipeline should succeed");
assert_eq!(outcome.status, StageStatus::Success);
let assets_dir = dir
.path()
.join("artifacts")
.join("assets")
.join("create_assets")
.join("retry_1");
let assets_dir = RuntimeState::new(dir.path()).asset_stage_dir("create_assets", 1);
let report_path = assets_dir.join("test-results/report.xml");
assert!(

28
lib/crates/fabro-workflows/tests/integration.rs
View file

@ -11,6 +11,7 @@ use fabro_interview::{
QueueInterviewer, RecordingInterviewer,
};
use fabro_llm::provider::Provider;
use fabro_store::RuntimeState;
use fabro_validate::{validate, validate_or_raise, Severity};
use fabro_workflows::context::Context;
use fabro_workflows::error::{FabroError, FailureSignatureExt};
@ -8729,11 +8730,7 @@ async fn large_context_values_are_offloaded_to_artifact_store() {
);
// The artifact file should exist on disk
let artifact_file = dir
.path()
.join("artifacts")
.join("values")
.join("response.big_output.json");
let artifact_file = RuntimeState::new(dir.path()).artifact_value_path("response.big_output");
assert!(
artifact_file.exists(),
"artifact file should exist at {artifact_file:?}"
@ -12840,12 +12837,7 @@ async fn asset_collection_local_sandbox_success() {
assert_eq!(outcome.status, StageStatus::Success);
// Check that asset files were collected into the stage directory
let assets_dir = run_dir
.path()
.join("artifacts")
.join("assets")
.join("create_assets")
.join("retry_1");
let assets_dir = RuntimeState::new(run_dir.path()).asset_stage_dir("create_assets", 1);
let report_path = assets_dir.join("test-results/report.xml");
assert!(
@ -12955,12 +12947,7 @@ async fn asset_collection_local_sandbox_on_failure() {
// Assets should still be collected regardless of intermediate node failures.
assert_eq!(outcome.status, StageStatus::Success);
let assets_dir = run_dir
.path()
.join("artifacts")
.join("assets")
.join("create_assets")
.join("retry_1");
let assets_dir = RuntimeState::new(run_dir.path()).asset_stage_dir("create_assets", 1);
let report_path = assets_dir.join("test-results/report.xml");
assert!(
@ -13050,12 +13037,7 @@ async fn asset_collection_docker_sandbox() {
.expect("pipeline should succeed");
assert_eq!(outcome.status, StageStatus::Success);
let assets_dir = run_dir
.path()
.join("artifacts")
.join("assets")
.join("create_assets")
.join("retry_1");
let assets_dir = RuntimeState::new(run_dir.path()).asset_stage_dir("create_assets", 1);
let report_path = assets_dir.join("test-results/report.xml");
assert!(