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

refactor(workflow): write metadata snapshots with git2

Browse source 
Replace the sandbox-side fast-import metadata writer with an in-process git2 writer that builds metadata commits locally and pushes them with worker-side GitHub credentials. Keep sandbox git probing separate from metadata runtime state so checkpoint commits and metadata snapshots have independent lifecycles.
This commit is contained in:
Bryan Helmkamp 2026-05-01 00:22:01 -04:00
parent 7cb120b96c
commit 165b38c3ed
No known key found for this signature in database
19 changed files with 1362 additions and 901 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
Cargo.lock generated
View file

@ -2930,6 +2930,8 @@ dependencies = [
"libc",
"libgit2-sys",
"log",
"openssl-probe 0.1.6",
"openssl-sys",
"url",
]
@ -3806,6 +3808,7 @@ dependencies = [
"cc",
"libc",
"libz-sys",
"openssl-sys",
"pkg-config",
]
@ -4190,7 +4193,7 @@ dependencies = [
"libc",
"log",
"openssl",
"openssl-probe",
"openssl-probe 0.2.1",
"openssl-sys",
"schannel",
"security-framework",
@ -4649,6 +4652,12 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "openssl-probe"
version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e"
[[package]]
name = "openssl-probe"
version = "0.2.1"
@ -5742,7 +5751,7 @@ version = "0.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63"
dependencies = [
"openssl-probe",
"openssl-probe 0.2.1",
"rustls-pki-types",
"schannel",
"security-framework",

2
Cargo.toml
View file

@ -39,7 +39,7 @@ tar = "0.4"
cli-table = { version = "0.5", default-features = false }
console = "0.15"
dialoguer = "0.12"
git2 = { version = "0.20", default-features = false, features = ["vendored-libgit2"] }
git2 = { version = "0.20", default-features = false, features = ["vendored-libgit2", "vendored-openssl", "https"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
tracing-appender = "0.2"

2
lib/crates/fabro-workflow/src/handler/parallel.rs
View file

@ -186,7 +186,7 @@ impl Handler for ParallelHandler {
// --- Git isolation: checkpoint "parallel base" before fan-out ---
let base_sha: Option<String> = if let Some(ref gs) = git_state {
let result = checked_git_checkpoint(
&services.run.metadata_runtime,
&services.run.sandbox_git,
&*services.run.sandbox,
&gs.run_id.to_string(),
&node.id,

3
lib/crates/fabro-workflow/src/lib.rs
View file

@ -146,11 +146,12 @@ pub mod run_lookup;
pub use error::{Error, FailureCategory, FailureSignature, FailureSignatureExt, Result};
pub use manifest_path::ManifestPath;
pub mod run_materialization;
pub(crate) mod run_metadata;
pub mod run_options;
pub mod run_status;
pub mod runtime_store;
pub mod sandbox_git;
pub(crate) mod sandbox_metadata;
pub(crate) mod sandbox_git_runtime;
pub mod services;
#[doc(hidden)]
pub mod test_support;

112
lib/crates/fabro-workflow/src/lifecycle/git.rs
View file

@ -18,10 +18,11 @@ use crate::graph::{WorkflowGraph, WorkflowNode};
use crate::lifecycle::event::stage_scope_for;
use crate::outcome::BilledModelUsage;
use crate::run_dump::RunDump;
use crate::run_metadata::{MetadataSnapshot, RunMetadataRuntime, RunMetadataWriterHandle};
use crate::run_options::RunOptions;
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git::{checked_git_checkpoint, git_diff};
use crate::sandbox_metadata::{MetadataSnapshot, SandboxGitRuntime, SandboxMetadataWriter};
use crate::sandbox_git_runtime::SandboxGitRuntime;
type WfRunState = ExecutionState<Option<BilledModelUsage>>;
type WfNodeResult = NodeResult<Option<BilledModelUsage>>;
@ -70,7 +71,9 @@ pub(crate) struct GitLifecycle {
pub run_id: RunId,
pub run_store: RunStoreHandle,
pub run_options: Arc<RunOptions>,
pub metadata_runtime: Arc<SandboxGitRuntime>,
pub sandbox_git: Arc<SandboxGitRuntime>,
pub metadata_runtime: Arc<RunMetadataRuntime>,
pub metadata_writer: Option<RunMetadataWriterHandle>,
pub start_node_id: Option<String>,
// Cross-lifecycle data (shared with EventLifecycle)
pub checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>>,
@ -84,7 +87,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
*self.last_git_sha.lock().unwrap() = None;
*self.checkpoint_git_result.lock().unwrap() = None;
if let Some(meta_branch) = self.metadata_branch().map(str::to_string) {
if self.metadata_runtime.metadata_degraded() {
if self.metadata_writer.is_none() || self.metadata_runtime.metadata_degraded() {
return Ok(());
}
let phase = MetadataSnapshotPhase::Init;
@ -151,7 +154,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
None,
);
let shadow_sha = if let Some(meta_branch) = self.metadata_branch().map(str::to_string) {
if self.metadata_runtime.metadata_degraded() {
if self.metadata_writer.is_none() || self.metadata_runtime.metadata_degraded() {
None
} else {
let phase = MetadataSnapshotPhase::Checkpoint;
@ -200,7 +203,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
let completed_count = state.completed_nodes.len();
let git_author = self.run_options.git_author();
let commit_result = checked_git_checkpoint(
&self.metadata_runtime,
&self.sandbox_git,
&*self.sandbox,
&self.run_id.to_string(),
node_id,
@ -311,15 +314,8 @@ impl GitLifecycle {
if self.metadata_runtime.metadata_degraded() {
return None;
}
let writer = self.metadata_writer.as_ref()?;
let run_id = self.run_id.to_string();
let writer = SandboxMetadataWriter::new(
&*self.sandbox,
&self.metadata_runtime,
&run_id,
meta_branch,
self.run_options.git_author(),
);
match writer.write_snapshot(dump, message).await {
Ok(snapshot) => {
if let Some(detail) = snapshot.push_error.as_deref() {
@ -596,12 +592,61 @@ mod tests {
events
}
#[expect(
clippy::disallowed_methods,
reason = "metadata event tests use synchronous git commands to set up temporary bare remotes"
)]
fn metadata_writer_for_repo(repo: &Path, branch: &str) -> RunMetadataWriterHandle {
let remote = repo.with_extension("metadata-remote.git");
let init = std::process::Command::new("git")
.args(["init", "--bare"])
.current_dir(repo)
.arg(&remote)
.output()
.unwrap();
assert!(
init.status.success(),
"git init --bare failed: {}",
String::from_utf8_lossy(&init.stderr)
);
RunMetadataWriterHandle::new_for_test(
format!("file://{}", remote.display()),
branch.to_string(),
crate::git::GitAuthor::default(),
None,
)
.unwrap()
}
fn git_lifecycle(
repo: &Path,
emitter: Arc<Emitter>,
run_store: RunStoreHandle,
run_options: Arc<RunOptions>,
metadata_runtime: Arc<SandboxGitRuntime>,
metadata_runtime: Arc<RunMetadataRuntime>,
) -> GitLifecycle {
let metadata_writer = run_options
.git
.as_ref()
.and_then(|git| git.meta_branch.as_deref())
.map(|branch| metadata_writer_for_repo(repo, branch));
git_lifecycle_with_writer(
repo,
emitter,
run_store,
run_options,
metadata_runtime,
metadata_writer,
)
}
fn git_lifecycle_with_writer(
repo: &Path,
emitter: Arc<Emitter>,
run_store: RunStoreHandle,
run_options: Arc<RunOptions>,
metadata_runtime: Arc<RunMetadataRuntime>,
metadata_writer: Option<RunMetadataWriterHandle>,
) -> GitLifecycle {
GitLifecycle {
sandbox: Arc::new(fabro_agent::LocalSandbox::new(repo.to_path_buf())),
@ -609,7 +654,9 @@ mod tests {
run_id: fixtures::RUN_1,
run_store,
run_options,
sandbox_git: Arc::new(SandboxGitRuntime::new()),
metadata_runtime,
metadata_writer,
start_node_id: Some("start".to_string()),
checkpoint_git_result: Arc::new(Mutex::new(None)),
last_git_sha: Arc::new(Mutex::new(None)),
@ -637,7 +684,7 @@ mod tests {
emitter,
handle,
run_options(repo_dir.path(), branch),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
);
let graph = workflow_graph();
let state = ExecutionState::new(&graph).unwrap();
@ -674,7 +721,7 @@ mod tests {
emitter,
RunStoreHandle::new(Arc::new(FailingStateStore)),
run_options(repo_dir.path(), branch),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
);
let graph = workflow_graph();
let state = ExecutionState::new(&graph).unwrap();
@ -701,20 +748,9 @@ mod tests {
}
#[tokio::test]
#[expect(
clippy::disallowed_methods,
reason = "metadata push-failure test uses a synchronous git command to configure a temporary remote"
)]
async fn init_metadata_push_failure_emits_failed_with_snapshot_accounting() {
let repo_dir = tempfile::tempdir().unwrap();
init_git_repo(repo_dir.path());
let missing_origin = repo_dir.path().join("missing-origin.git");
let remote = std::process::Command::new("git")
.args(["remote", "add", "origin", missing_origin.to_str().unwrap()])
.current_dir(repo_dir.path())
.output()
.unwrap();
assert!(remote.status.success());
let branch = "fabro/metadata/run";
let run_store = run_store(fixtures::RUN_1).await;
let handle = RunStoreHandle::local(run_store.clone());
@ -727,13 +763,21 @@ mod tests {
.sum::<u64>();
let emitter = Arc::new(Emitter::new(fixtures::RUN_1));
let events = record_events(&emitter);
let runtime = Arc::new(SandboxGitRuntime::new());
let lifecycle = git_lifecycle(
let runtime = Arc::new(RunMetadataRuntime::new());
let metadata_writer = RunMetadataWriterHandle::new_for_test(
format!("file://{}", repo_dir.path().display()),
branch.to_string(),
crate::git::GitAuthor::default(),
None,
)
.unwrap();
let lifecycle = git_lifecycle_with_writer(
repo_dir.path(),
emitter,
handle,
run_options(repo_dir.path(), branch),
Arc::clone(&runtime),
Some(metadata_writer),
);
let graph = workflow_graph();
let state = ExecutionState::new(&graph).unwrap();
@ -772,7 +816,7 @@ mod tests {
emitter,
RunStoreHandle::new(Arc::new(FailingStateStore)),
run_options(repo_dir.path(), branch),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
);
let graph = workflow_graph();
let node = graph.get_node("build").unwrap();
@ -815,7 +859,7 @@ mod tests {
emitter,
RunStoreHandle::local(run_store),
run_options(repo_dir.path(), branch),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
);
let graph = workflow_graph();
let node = graph.get_node("build").unwrap();
@ -855,7 +899,7 @@ mod tests {
async fn degraded_metadata_runtime_skips_snapshot_events() {
let repo_dir = tempfile::tempdir().unwrap();
init_git_repo(repo_dir.path());
let runtime = Arc::new(SandboxGitRuntime::new());
let runtime = Arc::new(RunMetadataRuntime::new());
runtime.mark_metadata_degraded();
let emitter = Arc::new(Emitter::new(fixtures::RUN_1));
let events = record_events(&emitter);
@ -880,7 +924,7 @@ mod tests {
init_git_repo(repo_dir.path());
let emitter = Arc::new(Emitter::new(fixtures::RUN_1));
let events = record_events(&emitter);
let runtime = Arc::new(SandboxGitRuntime::new());
let runtime = Arc::new(RunMetadataRuntime::new());
let lifecycle = git_lifecycle(
repo_dir.path(),
emitter,
@ -911,7 +955,9 @@ mod tests {
None,
fabro_model::Provider::Anthropic,
Arc::new(fabro_auth::EnvCredentialSource::new()),
Arc::new(SandboxGitRuntime::new()),
Arc::clone(&lifecycle.metadata_runtime),
lifecycle.metadata_writer.clone(),
);
let conclusion = Conclusion {
timestamp: chrono::Utc::now(),

9
lib/crates/fabro-workflow/src/lifecycle/mod.rs
View file

@ -39,9 +39,10 @@ use crate::event::Emitter;
use crate::graph::{WorkflowGraph, WorkflowNode};
use crate::outcome::{BilledModelUsage, Outcome, OutcomeExt};
use crate::run_control::RunControlState;
use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle};
use crate::run_options::RunOptions;
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_metadata::SandboxGitRuntime;
use crate::sandbox_git_runtime::SandboxGitRuntime;
type WfRunState = ExecutionState<Option<BilledModelUsage>>;
type WfNodeResult = NodeResult<Option<BilledModelUsage>>;
@ -88,7 +89,9 @@ impl WorkflowLifecycle {
run_store: &RunStoreHandle,
artifact_sink: Option<ArtifactSink>,
run_options: &Arc<RunOptions>,
metadata_runtime: Arc<SandboxGitRuntime>,
sandbox_git: Arc<SandboxGitRuntime>,
metadata_runtime: Arc<RunMetadataRuntime>,
metadata_writer: Option<RunMetadataWriterHandle>,
is_resume: bool,
on_node: crate::OnNodeCallback,
run_control: Option<Arc<RunControlState>>,
@ -150,7 +153,9 @@ impl WorkflowLifecycle {
run_id: run_options.run_id,
run_store: run_store.clone(),
run_options: Arc::clone(run_options),
sandbox_git,
metadata_runtime,
metadata_writer,
start_node_id,
checkpoint_git_result: Arc::clone(&checkpoint_git_result),
last_git_sha,

2
lib/crates/fabro-workflow/src/operations/start.rs
View file

@ -47,10 +47,10 @@ use crate::pipeline::{
};
use crate::records::Checkpoint;
use crate::run_control::RunControlState;
use crate::run_metadata::metadata_branch_name;
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use crate::run_status::{FailureReason, RunStatus};
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_metadata::metadata_branch_name;
use crate::workflow_bundle::{RunDefinition, WorkflowBundle};
struct RunSession {

2
lib/crates/fabro-workflow/src/pipeline/execute.rs
View file

@ -84,7 +84,9 @@ pub async fn execute(init: Initialized) -> Executed {
&engine.run.run_store,
artifact_sink,
&settings_arc,
Arc::clone(&engine.run.sandbox_git),
Arc::clone(&engine.run.metadata_runtime),
engine.run.metadata_writer.clone(),
checkpoint.is_some(),
on_node,
run_control,

72
lib/crates/fabro-workflow/src/pipeline/finalize.rs
View file

@ -12,11 +12,11 @@ use crate::event::{Event, RunNoticeLevel};
use crate::outcome::{Outcome, OutcomeExt, StageOutcome};
use crate::records::{Checkpoint, Conclusion, StageSummary};
use crate::run_dump::RunDump;
use crate::run_metadata::MetadataSnapshot;
use crate::run_options::RunOptions;
use crate::run_status::{FailureReason, RunStatus, SuccessReason};
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git::git_diff_with_timeout;
use crate::sandbox_metadata::{MetadataSnapshot, SandboxMetadataWriter};
use crate::services::RunServices;
pub fn classify_engine_result(
@ -154,6 +154,9 @@ pub async fn write_finalize_commit(
if services.metadata_runtime.metadata_degraded() {
return;
}
let Some(writer) = services.metadata_writer.as_ref() else {
return;
};
let Some(meta_branch) = run_options
.git
.as_ref()
@ -188,14 +191,6 @@ pub async fn write_finalize_commit(
};
projection.conclusion = Some(conclusion.clone());
let dump = RunDump::from_projection(&projection);
let run_id = run_options.run_id.to_string();
let writer = SandboxMetadataWriter::new(
&*services.sandbox,
&services.metadata_runtime,
&run_id,
meta_branch,
run_options.git_author(),
);
match writer.write_snapshot(&dump, "finalize run").await {
Ok(snapshot) => {
if let Some(detail) = snapshot.push_error.as_deref() {
@ -540,9 +535,10 @@ mod tests {
use super::*;
use crate::event::{Emitter, StoreProgressLogger, append_event};
use crate::pipeline::types::Retroed;
use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle};
use crate::run_options::{GitCheckpointOptions, RunOptions};
use crate::runtime_store::{RunStoreBackend, RunStoreHandle};
use crate::sandbox_metadata::SandboxGitRuntime;
use crate::sandbox_git_runtime::SandboxGitRuntime;
fn test_run_id() -> RunId {
fixtures::RUN_1
@ -644,11 +640,38 @@ mod tests {
events
}
#[expect(
clippy::disallowed_methods,
reason = "metadata event tests use synchronous git commands to set up temporary bare remotes"
)]
fn metadata_writer_for_repo(repo: &Path, branch: &str) -> RunMetadataWriterHandle {
let remote = repo.with_extension("metadata-remote.git");
let init = std::process::Command::new("git")
.args(["init", "--bare"])
.current_dir(repo)
.arg(&remote)
.output()
.unwrap();
assert!(
init.status.success(),
"git init --bare failed: {}",
String::from_utf8_lossy(&init.stderr)
);
RunMetadataWriterHandle::new_for_test(
format!("file://{}", remote.display()),
branch.to_string(),
crate::git::GitAuthor::default(),
None,
)
.unwrap()
}
fn test_services(
run_store: RunStoreHandle,
emitter: Arc<Emitter>,
sandbox: Arc<dyn fabro_agent::Sandbox>,
metadata_runtime: Arc<SandboxGitRuntime>,
metadata_runtime: Arc<RunMetadataRuntime>,
metadata_writer: Option<RunMetadataWriterHandle>,
) -> Arc<RunServices> {
RunServices::new(
run_store,
@ -658,7 +681,9 @@ mod tests {
None,
fabro_model::Provider::Anthropic,
Arc::new(fabro_auth::EnvCredentialSource::new()),
Arc::new(SandboxGitRuntime::new()),
metadata_runtime,
metadata_writer,
)
}
@ -682,7 +707,9 @@ mod tests {
None,
fabro_model::Provider::Anthropic,
Arc::new(fabro_auth::EnvCredentialSource::new()),
Arc::new(crate::sandbox_metadata::SandboxGitRuntime::new()),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
None,
);
let retroed = Retroed {
graph: Graph::new("test"),
@ -732,7 +759,8 @@ mod tests {
Arc::new(fabro_agent::LocalSandbox::new(
repo_dir.path().to_path_buf(),
)),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
Some(metadata_writer_for_repo(repo_dir.path(), branch)),
);
let run_options = test_git_run_options(repo_dir.path(), branch);
@ -765,7 +793,11 @@ mod tests {
Arc::new(fabro_agent::LocalSandbox::new(
repo_dir.path().to_path_buf(),
)),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
Some(metadata_writer_for_repo(
repo_dir.path(),
"fabro/metadata/run",
)),
);
let run_options = test_git_run_options(repo_dir.path(), "fabro/metadata/run");
let conclusion = Conclusion {
@ -804,7 +836,7 @@ mod tests {
let run_store = seeded_run_store().await;
let emitter = Arc::new(Emitter::new(test_run_id()));
let events = record_events(&emitter);
let runtime = Arc::new(SandboxGitRuntime::new());
let runtime = Arc::new(RunMetadataRuntime::new());
runtime.mark_metadata_degraded();
let services = test_services(
RunStoreHandle::local(run_store),
@ -813,6 +845,10 @@ mod tests {
repo_dir.path().to_path_buf(),
)),
runtime,
Some(metadata_writer_for_repo(
repo_dir.path(),
"fabro/metadata/run",
)),
);
let run_options = test_git_run_options(repo_dir.path(), "fabro/metadata/run");
let conclusion = Conclusion {
@ -844,7 +880,11 @@ mod tests {
Arc::new(fabro_agent::LocalSandbox::new(
repo_dir.path().to_path_buf(),
)),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
Some(metadata_writer_for_repo(
repo_dir.path(),
"fabro/metadata/run",
)),
);
let retroed = Retroed {
graph: Graph::new("test"),

58
lib/crates/fabro-workflow/src/pipeline/initialize.rs
View file

@ -31,9 +31,12 @@ use crate::event::{Emitter, Event, RunNoticeLevel};
use crate::git::RUN_BRANCH_PREFIX;
use crate::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter};
use crate::handler::{HandlerRegistry, default_registry, sandbox_cancel_token};
use crate::run_metadata::{
RunMetadataRuntime, build_metadata_writer, metadata_branch_name, mint_token,
};
use crate::run_options::{GitCheckpointOptions, RunOptions};
use crate::sandbox_git::GIT_REMOTE;
use crate::sandbox_metadata::{SandboxGitRuntime, metadata_branch_name};
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::services::{EngineServices, RunServices};
struct WorktreePlan {
@ -216,32 +219,9 @@ async fn mint_github_token(
origin_url: &str,
permissions: &HashMap<String, String>,
) -> Result<String, Error> {
if let fabro_github::GitHubCredentials::Token(token) = creds {
return Ok(token.clone());
}
let https_url = fabro_github::ssh_url_to_https(origin_url);
let (owner, repo) =
fabro_github::parse_github_owner_repo(&https_url).map_err(|e| Error::engine(e.clone()))?;
let fabro_github::GitHubCredentials::App(creds) = creds else {
unreachable!("token credentials return early");
};
let jwt = fabro_github::sign_app_jwt(&creds.app_id, &creds.private_key_pem)
.map_err(|e| Error::engine(e.clone()))?;
let client = fabro_http::http_client().map_err(|e| Error::engine(e.to_string()))?;
let perms_json = serde_json::to_value(permissions).map_err(|e| Error::engine(e.to_string()))?;
let install_url = creds.installation_url(&owner);
fabro_github::create_installation_access_token_with_permissions_and_install_url(
&client,
&jwt,
&owner,
&repo,
&fabro_github::github_api_base_url(),
perms_json,
install_url.as_deref(),
)
.await
.map_err(|e| Error::engine(e.clone()))
mint_token(creds, origin_url, permissions)
.await
.map_err(Error::engine)
}
async fn build_sandbox_env(
@ -465,7 +445,8 @@ pub async fn initialize(
let llm_source = build_llm_source(options.vault.clone());
let cli_resolver = options.vault.clone().map(CredentialResolver::new);
let metadata_runtime = Arc::new(SandboxGitRuntime::new());
let sandbox_git = Arc::new(SandboxGitRuntime::new());
let metadata_runtime = Arc::new(RunMetadataRuntime::new());
let hook_runner = if options.hooks.hooks.is_empty() {
None
@ -494,7 +475,7 @@ pub async fn initialize(
.await
.map_err(|e| Error::engine(e.to_string()))?;
if let Some(base_sha) = resolve_worktree_base_sha(&*inner, plan).await? {
metadata_runtime
sandbox_git
.ensure_git_available(&*inner)
.await
.map_err(|err| Error::engine(format!("sandbox git unavailable: {err}")))?;
@ -612,7 +593,7 @@ pub async fn initialize(
if !has_run_branch {
let intent = git_setup_intent(&options.run_options);
if sandbox.origin_url().is_some() {
metadata_runtime
sandbox_git
.ensure_git_available(&*sandbox)
.await
.map_err(|err| Error::engine(format!("sandbox git unavailable: {err}")))?;
@ -711,6 +692,21 @@ pub async fn initialize(
.await?;
}
let metadata_writer = match build_metadata_writer(&options.run_options) {
Ok(writer) => writer,
Err(err) => {
let message = format!("failed to initialize checkpoint metadata writer: {err}");
if metadata_runtime.mark_metadata_degraded() {
options.emitter.notice(
RunNoticeLevel::Warn,
"checkpoint_metadata_write_failed",
message,
);
}
None
}
};
let run_services = RunServices::new(
options.run_store.clone(),
Arc::clone(&options.emitter),
@ -719,7 +715,9 @@ pub async fn initialize(
options.run_options.cancel_token.clone(),
options.llm.provider,
Arc::clone(&llm_source),
sandbox_git,
metadata_runtime,
metadata_writer,
);
let engine = Arc::new(EngineServices {
run: Arc::clone(&run_services),

8
lib/crates/fabro-workflow/src/pipeline/retro.rs
View file

@ -327,7 +327,9 @@ mod tests {
None,
fabro_llm::Provider::Anthropic,
test_llm_source(),
Arc::new(crate::sandbox_metadata::SandboxGitRuntime::new()),
Arc::new(crate::sandbox_git_runtime::SandboxGitRuntime::new()),
Arc::new(crate::run_metadata::RunMetadataRuntime::new()),
None,
);
let mut engine = EngineServices::test_default();
engine.run = Arc::clone(&services);
@ -381,7 +383,9 @@ mod tests {
None,
fabro_llm::Provider::Anthropic,
test_llm_source(),
Arc::new(crate::sandbox_metadata::SandboxGitRuntime::new()),
Arc::new(crate::sandbox_git_runtime::SandboxGitRuntime::new()),
Arc::new(crate::run_metadata::RunMetadataRuntime::new()),
None,
);
let retro = run_retro(

10
lib/crates/fabro-workflow/src/run_dump.rs
View file

@ -170,6 +170,16 @@ impl RunDump {
self.entries.push(RunDumpEntry::bytes(path, contents));
}
#[cfg(test)]
pub(crate) fn from_raw_entries(entries: Vec<(String, Vec<u8>)>) -> Self {
Self {
entries: entries
.into_iter()
.map(|(path, contents)| RunDumpEntry::bytes(path, contents))
.collect(),
}
}
pub async fn hydrate_referenced_blobs_with_reader<'a, F>(
&mut self,
mut read_blob: F,

1038
lib/crates/fabro-workflow/src/run_metadata.rs Normal file
View file

File diff suppressed because it is too large Load diff

373
lib/crates/fabro-workflow/src/sandbox_git.rs
View file

@ -8,7 +8,7 @@ use fabro_types::RunId;
use crate::artifact_snapshot;
use crate::git::GitAuthor;
use crate::sandbox_metadata::SandboxGitRuntime;
use crate::sandbox_git_runtime::SandboxGitRuntime;
/// Captured git state for a workflow run, shared with handlers.
#[derive(Debug, Clone)]
@ -804,8 +804,8 @@ mod tests {
reason = "These unit tests use the real git CLI to construct sandbox-git fixture repositories and sync-write fixtures to disk."
)]
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};
use std::collections::VecDeque;
use std::sync::Mutex;
use async_trait::async_trait;
use fabro_agent::{DirEntry, ExecResult, GrepOptions};
@ -814,144 +814,6 @@ mod tests {
use super::*;
struct RecordingSandbox {
inner: Arc<dyn fabro_sandbox::Sandbox>,
commands: Arc<Mutex<Vec<String>>>,
pushes: Arc<Mutex<Vec<String>>>,
}
impl RecordingSandbox {
fn new(inner: Arc<dyn fabro_sandbox::Sandbox>) -> Self {
Self {
inner,
commands: Arc::new(Mutex::new(Vec::new())),
pushes: Arc::new(Mutex::new(Vec::new())),
}
}
fn commands_after_probe(&self) -> Vec<String> {
self.commands
.lock()
.unwrap()
.iter()
.filter(|command| !command.contains("probe.txt"))
.cloned()
.collect()
}
fn pushes(&self) -> Vec<String> {
self.pushes.lock().unwrap().clone()
}
}
#[async_trait]
impl Sandbox for RecordingSandbox {
async fn read_file(
&self,
path: &str,
offset: Option<usize>,
limit: Option<usize>,
) -> fabro_sandbox::Result<String> {
self.inner.read_file(path, offset, limit).await
}
async fn write_file(&self, path: &str, content: &str) -> fabro_sandbox::Result<()> {
self.inner.write_file(path, content).await
}
async fn delete_file(&self, path: &str) -> fabro_sandbox::Result<()> {
self.inner.delete_file(path).await
}
async fn file_exists(&self, path: &str) -> fabro_sandbox::Result<bool> {
self.inner.file_exists(path).await
}
async fn list_directory(
&self,
path: &str,
depth: Option<usize>,
) -> fabro_sandbox::Result<Vec<DirEntry>> {
self.inner.list_directory(path, depth).await
}
async fn exec_command(
&self,
command: &str,
timeout_ms: u64,
working_dir: Option<&str>,
env_vars: Option<&std::collections::HashMap<String, String>>,
cancel_token: Option<CancellationToken>,
) -> fabro_sandbox::Result<ExecResult> {
self.commands.lock().unwrap().push(command.to_string());
self.inner
.exec_command(command, timeout_ms, working_dir, env_vars, cancel_token)
.await
}
async fn grep(
&self,
pattern: &str,
path: &str,
options: &GrepOptions,
) -> fabro_sandbox::Result<Vec<String>> {
self.inner.grep(pattern, path, options).await
}
async fn glob(
&self,
pattern: &str,
path: Option<&str>,
) -> fabro_sandbox::Result<Vec<String>> {
self.inner.glob(pattern, path).await
}
async fn download_file_to_local(
&self,
remote_path: &str,
local_path: &std::path::Path,
) -> fabro_sandbox::Result<()> {
self.inner
.download_file_to_local(remote_path, local_path)
.await
}
async fn upload_file_from_local(
&self,
local_path: &std::path::Path,
remote_path: &str,
) -> fabro_sandbox::Result<()> {
self.inner
.upload_file_from_local(local_path, remote_path)
.await
}
async fn initialize(&self) -> fabro_sandbox::Result<()> {
self.inner.initialize().await
}
async fn cleanup(&self) -> fabro_sandbox::Result<()> {
self.inner.cleanup().await
}
fn working_directory(&self) -> &str {
self.inner.working_directory()
}
fn platform(&self) -> &str {
self.inner.platform()
}
fn os_version(&self) -> String {
self.inner.os_version()
}
async fn git_push_ref(&self, refspec: &str) -> fabro_sandbox::Result<()> {
self.pushes.lock().unwrap().push(refspec.to_string());
self.inner.git_push_ref(refspec).await
}
}
struct ScriptedSandbox {
exec_results: Mutex<VecDeque<ExecResult>>,
}
@ -1122,7 +984,7 @@ mod tests {
#[tokio::test]
async fn checked_git_checkpoint_fails_before_checkpoint_when_probe_fails() {
let sandbox = ScriptedSandbox::new(vec![exec_failed(127, "", "git missing\n")]);
let runtime = crate::sandbox_metadata::SandboxGitRuntime::new();
let runtime = crate::sandbox_git_runtime::SandboxGitRuntime::new();
let err = checked_git_checkpoint(
&runtime,
@ -1307,233 +1169,6 @@ mod tests {
String::from_utf8(out.stdout).unwrap().trim().to_string()
}
#[tokio::test]
async fn sandbox_metadata_writer_preserves_worktree_and_writes_binary_run_dump() {
let repo_dir = tempfile::tempdir().unwrap();
let repo = repo_dir.path();
init_git_repo(repo);
std::fs::write(repo.join("tracked.txt"), "seed\n").unwrap();
let head = git_commit_all(repo, "initial");
let sandbox =
RecordingSandbox::new(Arc::new(fabro_agent::LocalSandbox::new(repo.to_path_buf())));
let run_id = fabro_types::fixtures::RUN_1;
let mut projection = fabro_store::RunProjection::default();
projection.spec = Some(fabro_types::RunSpec {
run_id,
settings: fabro_types::WorkflowSettings::default(),
graph: fabro_types::Graph::new("metadata"),
workflow_slug: Some("metadata".to_string()),
source_directory: Some("/Users/client/project".to_string()),
git: Some(fabro_types::GitContext {
origin_url: "https://github.com/fabro-sh/fabro.git".to_string(),
branch: "main".to_string(),
sha: None,
dirty: fabro_types::DirtyStatus::Clean,
push_outcome: fabro_types::PreRunPushOutcome::NotAttempted,
}),
labels: HashMap::new(),
provenance: None,
manifest_blob: None,
definition_blob: None,
fork_source_ref: None,
in_place: false,
});
let mut dump = crate::run_dump::RunDump::from_projection(&projection);
dump.add_file_bytes("binary/payload.bin", vec![0, 159, 146, 150]);
dump.add_file_bytes("path with spaces.txt", b"quoted path\n".to_vec());
let runtime = crate::sandbox_metadata::SandboxGitRuntime::new();
let run_id_string = run_id.to_string();
let branch = crate::sandbox_metadata::metadata_branch_name(&run_id_string);
let writer = crate::sandbox_metadata::SandboxMetadataWriter::new(
&sandbox,
&runtime,
&run_id_string,
&branch,
crate::git::GitAuthor::default(),
);
let expected_entries = dump.git_entries().unwrap();
let expected_entry_count = expected_entries.len();
let expected_bytes = expected_entries
.iter()
.map(|(_, bytes)| u64::try_from(bytes.len()).unwrap_or(u64::MAX))
.sum::<u64>();
let snapshot = writer.write_snapshot(&dump, "checkpoint").await.unwrap();
assert_eq!(snapshot.push_error, None);
assert_eq!(snapshot.entry_count, expected_entry_count);
assert_eq!(snapshot.bytes, expected_bytes);
let commit_sha = snapshot.commit_sha;
let current = std::process::Command::new("git")
.args(["rev-parse", "--abbrev-ref", "HEAD"])
.current_dir(repo)
.output()
.unwrap();
assert_eq!(String::from_utf8(current.stdout).unwrap().trim(), "main");
let head_after = std::process::Command::new("git")
.args(["rev-parse", "HEAD"])
.current_dir(repo)
.output()
.unwrap();
assert_eq!(String::from_utf8(head_after.stdout).unwrap().trim(), head);
let run_json = std::process::Command::new("git")
.args(["show", &format!("{commit_sha}:run.json")])
.current_dir(repo)
.output()
.unwrap();
assert!(run_json.status.success());
let stored_projection: serde_json::Value =
serde_json::from_slice(&run_json.stdout).unwrap();
assert!(stored_projection.get("spec").is_some());
let binary = std::process::Command::new("git")
.args(["show", &format!("{commit_sha}:binary/payload.bin")])
.current_dir(repo)
.output()
.unwrap();
assert_eq!(binary.stdout, vec![0, 159, 146, 150]);
let spaced_path = std::process::Command::new("git")
.args(["show", &format!("{commit_sha}:path with spaces.txt")])
.current_dir(repo)
.output()
.unwrap();
assert_eq!(spaced_path.stdout, b"quoted path\n");
let status = std::process::Command::new("git")
.args(["status", "--porcelain"])
.current_dir(repo)
.output()
.unwrap();
assert!(String::from_utf8(status.stdout).unwrap().trim().is_empty());
dump.add_file_bytes("second.txt", b"second\n".to_vec());
let second_expected_entries = dump.git_entries().unwrap();
let second_expected_entry_count = second_expected_entries.len();
let second_expected_bytes = second_expected_entries
.iter()
.map(|(_, bytes)| u64::try_from(bytes.len()).unwrap_or(u64::MAX))
.sum::<u64>();
let second_snapshot = writer.write_snapshot(&dump, "checkpoint 2").await.unwrap();
assert_eq!(second_snapshot.entry_count, second_expected_entry_count);
assert_eq!(second_snapshot.bytes, second_expected_bytes);
let second_commit_sha = second_snapshot.commit_sha;
let second_parent = std::process::Command::new("git")
.args(["rev-list", "--parents", "-n", "1", &second_commit_sha])
.current_dir(repo)
.output()
.unwrap();
assert!(second_parent.status.success());
let parent_line = String::from_utf8(second_parent.stdout).unwrap();
let parents: Vec<_> = parent_line.split_whitespace().collect();
assert_eq!(parents, vec![
second_commit_sha.as_str(),
commit_sha.as_str()
]);
let second_file = std::process::Command::new("git")
.args(["show", &format!("{second_commit_sha}:second.txt")])
.current_dir(repo)
.output()
.unwrap();
assert_eq!(second_file.stdout, b"second\n");
let metadata_commands = sandbox.commands_after_probe();
assert_eq!(
metadata_commands
.iter()
.filter(|command| command.contains(" fast-import "))
.count(),
2,
"metadata writer should use one fast-import per snapshot, got: {metadata_commands:?}"
);
for forbidden in [
"hash-object",
"update-index",
"write-tree",
"commit-tree",
"update-ref",
] {
assert!(
!metadata_commands
.iter()
.any(|command| command.contains(forbidden)),
"metadata writer should not run {forbidden} after probe, got: {metadata_commands:?}"
);
}
let refspec = format!("refs/heads/{branch}:refs/heads/{branch}");
assert_eq!(sandbox.pushes(), vec![refspec.clone(), refspec]);
}
#[tokio::test]
async fn sandbox_metadata_writer_records_log_safe_push_error() {
let repo_dir = tempfile::tempdir().unwrap();
let repo = repo_dir.path();
init_git_repo(repo);
std::fs::write(repo.join("tracked.txt"), "seed\n").unwrap();
git_commit_all(repo, "initial");
let missing_origin = repo_dir.path().join("missing-origin.git");
let add_remote = std::process::Command::new("git")
.args(["remote", "add", "origin", missing_origin.to_str().unwrap()])
.current_dir(repo)
.output()
.unwrap();
assert!(add_remote.status.success());
let sandbox = fabro_agent::LocalSandbox::new(repo.to_path_buf());
let run_id = fabro_types::fixtures::RUN_2.to_string();
let branch = crate::sandbox_metadata::metadata_branch_name(&run_id);
let mut projection = fabro_store::RunProjection::default();
projection.spec = Some(fabro_types::RunSpec {
run_id: fabro_types::fixtures::RUN_2,
settings: fabro_types::WorkflowSettings::default(),
graph: fabro_types::Graph::new("metadata"),
workflow_slug: Some("metadata".to_string()),
source_directory: Some("/Users/client/project".to_string()),
git: None,
labels: HashMap::new(),
provenance: None,
manifest_blob: None,
definition_blob: None,
fork_source_ref: None,
in_place: false,
});
let dump = crate::run_dump::RunDump::from_projection(&projection);
let expected_entries = dump.git_entries().unwrap();
let expected_entry_count = expected_entries.len();
let expected_bytes = expected_entries
.iter()
.map(|(_, bytes)| u64::try_from(bytes.len()).unwrap_or(u64::MAX))
.sum::<u64>();
let runtime = crate::sandbox_metadata::SandboxGitRuntime::new();
let writer = crate::sandbox_metadata::SandboxMetadataWriter::new(
&sandbox,
&runtime,
&run_id,
&branch,
crate::git::GitAuthor::default(),
);
let snapshot = writer.write_snapshot(&dump, "checkpoint").await.unwrap();
assert_eq!(snapshot.entry_count, expected_entry_count);
assert_eq!(snapshot.bytes, expected_bytes);
let push_error = snapshot.push_error.unwrap();
assert!(push_error.contains("git push origin"));
assert!(push_error.contains("hint:"));
assert!(
!push_error.contains("fatal:"),
"push error should be log-safe: {push_error}"
);
assert!(
!push_error.contains(missing_origin.to_str().unwrap()),
"push error should not include raw git stderr paths: {push_error}"
);
}
#[tokio::test]
async fn list_changed_files_raw_classifies_add_modify_delete() {
let repo_dir = tempfile::tempdir().unwrap();

89
lib/crates/fabro-workflow/src/sandbox_git_runtime.rs Normal file
View file

@ -0,0 +1,89 @@
use fabro_agent::Sandbox;
use fabro_sandbox::shell_quote;
use tokio::sync::OnceCell;
use crate::sandbox_git::GIT_REMOTE;
pub(crate) struct SandboxGitRuntime {
probe: OnceCell<Result<(), String>>,
}
impl SandboxGitRuntime {
pub(crate) fn new() -> Self {
Self {
probe: OnceCell::new(),
}
}
pub(crate) async fn ensure_git_available(&self, sandbox: &dyn Sandbox) -> Result<(), String> {
self.probe
.get_or_init(|| async { probe_sandbox_git(sandbox).await })
.await
.clone()
}
}
impl Default for SandboxGitRuntime {
fn default() -> Self {
Self::new()
}
}
async fn probe_sandbox_git(sandbox: &dyn Sandbox) -> Result<(), String> {
let temp = sandbox_temp_dir(sandbox, "probe", "git");
let index = format!("{temp}/index");
let probe_file = format!("{temp}/probe.txt");
let command = format!(
"set -e\n\
rm -rf {temp_q}\n\
mkdir -p {temp_q}\n\
printf probe > {probe_file_q}\n\
GIT_INDEX_FILE={index_q} {git} read-tree --empty\n\
blob=$({git} hash-object -w {probe_file_q})\n\
GIT_INDEX_FILE={index_q} {git} update-index --add --cacheinfo 100644,$blob,probe.txt\n\
GIT_INDEX_FILE={index_q} {git} write-tree >/dev/null\n\
rm -rf {temp_q}",
temp_q = shell_quote(&temp),
probe_file_q = shell_quote(&probe_file),
index_q = shell_quote(&index),
git = GIT_REMOTE,
);
exec_ok(sandbox, &command).await
}
fn sandbox_temp_dir(sandbox: &dyn Sandbox, run_id: &str, label: &str) -> String {
let cwd = sandbox.working_directory().trim_end_matches('/');
let id = uuid::Uuid::new_v4();
format!("{cwd}/.fabro/tmp/{label}-{run_id}-{id}")
}
async fn exec_ok(sandbox: &dyn Sandbox, command: &str) -> Result<(), String> {
let result = sandbox
.exec_command(command, 30_000, None, None, None)
.await
.map_err(|err| err.display_with_causes())?;
if result.is_success() {
Ok(())
} else {
Err(exec_err(command, &result))
}
}
fn exec_err(label: &str, result: &fabro_sandbox::ExecResult) -> String {
if result.is_timed_out() {
return format!("{label} timed out after {}ms", result.duration_ms);
}
if result.is_cancelled() {
return format!("{label} cancelled after {}ms", result.duration_ms);
}
let detail = format!("{}{}", result.stdout, result.stderr);
let detail = detail.trim();
if detail.is_empty() {
format!("{label} failed with exit {}", result.display_exit_code())
} else {
format!(
"{label} failed with exit {}: {detail}",
result.display_exit_code()
)
}
}

407
lib/crates/fabro-workflow/src/sandbox_metadata.rs
View file

@ -1,407 +0,0 @@
use std::collections::HashMap;
use std::fmt::Write as _;
use std::sync::atomic::{AtomicBool, Ordering};
use fabro_agent::Sandbox;
use fabro_sandbox::shell_quote;
use tokio::fs;
use tokio::sync::OnceCell;
use crate::git::{GitAuthor, META_BRANCH_PREFIX};
use crate::run_dump::RunDump;
use crate::sandbox_git::GIT_REMOTE;
#[derive(Debug, thiserror::Error)]
pub(crate) enum SandboxMetadataError {
#[error("sandbox git unavailable: {0}")]
GitUnavailable(String),
#[error("metadata dump serialization failed: {0}")]
Dump(#[from] anyhow::Error),
#[error("metadata temp file write failed: {0}")]
LocalTemp(std::io::Error),
#[error("{0}")]
Git(String),
#[error("{0}")]
Sandbox(String),
}
pub(crate) struct SandboxGitRuntime {
probe: OnceCell<Result<(), String>>,
metadata_degraded: AtomicBool,
metadata_warning_emitted: AtomicBool,
}
impl SandboxGitRuntime {
pub(crate) fn new() -> Self {
Self {
probe: OnceCell::new(),
metadata_degraded: AtomicBool::new(false),
metadata_warning_emitted: AtomicBool::new(false),
}
}
pub(crate) async fn ensure_git_available(&self, sandbox: &dyn Sandbox) -> Result<(), String> {
self.probe
.get_or_init(|| async { probe_sandbox_git(sandbox).await })
.await
.clone()
}
pub(crate) fn mark_metadata_degraded(&self) -> bool {
self.metadata_degraded.store(true, Ordering::SeqCst);
!self.metadata_warning_emitted.swap(true, Ordering::SeqCst)
}
pub(crate) fn metadata_degraded(&self) -> bool {
self.metadata_degraded.load(Ordering::SeqCst)
}
}
impl Default for SandboxGitRuntime {
fn default() -> Self {
Self::new()
}
}
pub(crate) fn metadata_branch_name(run_id: &str) -> String {
format!("{META_BRANCH_PREFIX}{run_id}")
}
pub(crate) struct SandboxMetadataWriter<'a> {
sandbox: &'a dyn Sandbox,
runtime: &'a SandboxGitRuntime,
run_id: &'a str,
branch: &'a str,
git_author: GitAuthor,
}
pub(crate) struct MetadataSnapshot {
pub commit_sha: String,
pub push_error: Option<String>,
pub entry_count: usize,
pub bytes: u64,
}
impl<'a> SandboxMetadataWriter<'a> {
pub(crate) fn new(
sandbox: &'a dyn Sandbox,
runtime: &'a SandboxGitRuntime,
run_id: &'a str,
branch: &'a str,
git_author: GitAuthor,
) -> Self {
Self {
sandbox,
runtime,
run_id,
branch,
git_author,
}
}
pub(crate) async fn write_snapshot(
&self,
dump: &RunDump,
message: &str,
) -> Result<MetadataSnapshot, SandboxMetadataError> {
self.runtime
.ensure_git_available(self.sandbox)
.await
.map_err(SandboxMetadataError::GitUnavailable)?;
let entries = dump.git_entries()?;
let entry_count = entries.len();
let bytes = metadata_entries_bytes(&entries);
let temp = sandbox_temp_dir(self.sandbox, self.run_id, "metadata");
exec_ok(
self.sandbox,
&format!(
"rm -rf {temp_q} && mkdir -p {temp_q}",
temp_q = shell_quote(&temp)
),
None,
)
.await?;
let result = self
.write_snapshot_in_temp(&entries, message, &temp, entry_count, bytes)
.await;
let _ = exec_ok(
self.sandbox,
&format!("rm -rf {}", shell_quote(&temp)),
None,
)
.await;
result
}
async fn write_snapshot_in_temp(
&self,
entries: &[(String, Vec<u8>)],
message: &str,
temp: &str,
entry_count: usize,
bytes: u64,
) -> Result<MetadataSnapshot, SandboxMetadataError> {
let full_ref = format!("refs/heads/{}", self.branch);
let old_commit = exec_stdout(
self.sandbox,
&format!(
"{GIT_REMOTE} rev-parse --verify -q {}^{{commit}} || true",
shell_quote(&full_ref)
),
None,
)
.await?;
let old_commit = (!old_commit.is_empty()).then_some(old_commit);
let mut commit_message = message.to_string();
self.git_author.append_footer(&mut commit_message);
let stream = fast_import_stream(
&full_ref,
old_commit.as_deref(),
&commit_message,
entries,
&self.git_author,
)?;
let local = tempfile::NamedTempFile::new().map_err(SandboxMetadataError::LocalTemp)?;
fs::write(local.path(), stream)
.await
.map_err(SandboxMetadataError::LocalTemp)?;
let remote = format!("{temp}/metadata.fi");
self.sandbox
.upload_file_from_local(local.path(), &remote)
.await
.map_err(|err| SandboxMetadataError::Sandbox(err.display_with_causes()))?;
let stdout = exec_stdout(
self.sandbox,
&format!(
"{GIT_REMOTE} fast-import --date-format=now < {}",
shell_quote(&remote)
),
None,
)
.await?;
let commit = parse_fast_import_mark(&stdout)?;
let refspec = format!("{full_ref}:{full_ref}");
let push_error = self
.sandbox
.git_push_ref(&refspec)
.await
.err()
.map(|err| err.to_string());
Ok(MetadataSnapshot {
commit_sha: commit,
push_error,
entry_count,
bytes,
})
}
}
fn metadata_entries_bytes(entries: &[(String, Vec<u8>)]) -> u64 {
entries.iter().fold(0, |total, (_, bytes)| {
total.saturating_add(u64::try_from(bytes.len()).unwrap_or(u64::MAX))
})
}
fn fast_import_stream(
full_ref: &str,
old_commit: Option<&str>,
commit_message: &str,
entries: &[(String, Vec<u8>)],
author: &GitAuthor,
) -> Result<Vec<u8>, SandboxMetadataError> {
let mut stream = Vec::new();
push_line(&mut stream, &format!("commit {full_ref}"));
push_line(&mut stream, "mark :1");
push_line(
&mut stream,
&format!("author {}", fast_import_ident(author)),
);
push_line(
&mut stream,
&format!("committer {}", fast_import_ident(author)),
);
push_data(&mut stream, commit_message.as_bytes());
if let Some(old_commit) = old_commit {
push_line(&mut stream, &format!("from {old_commit}"));
}
push_line(&mut stream, "deleteall");
for (path, bytes) in entries {
validate_metadata_path(path)?;
push_line(
&mut stream,
&format!("M 100644 inline {}", fast_import_quote_path(path)),
);
push_data(&mut stream, bytes);
}
push_line(&mut stream, "get-mark :1");
Ok(stream)
}
fn push_line(stream: &mut Vec<u8>, line: &str) {
stream.extend_from_slice(line.as_bytes());
stream.push(b'\n');
}
fn push_data(stream: &mut Vec<u8>, data: &[u8]) {
push_line(stream, &format!("data {}", data.len()));
stream.extend_from_slice(data);
stream.push(b'\n');
}
fn parse_fast_import_mark(stdout: &str) -> Result<String, SandboxMetadataError> {
stdout
.lines()
.rev()
.map(str::trim)
.find(|line| !line.is_empty() && line.bytes().all(|byte| byte.is_ascii_hexdigit()))
.map(ToString::to_string)
.ok_or_else(|| {
SandboxMetadataError::Git(format!(
"git fast-import did not report imported commit mark: {stdout:?}"
))
})
}
fn fast_import_ident(author: &GitAuthor) -> String {
let name = author
.name
.replace(['\n', '\r', '<', '>'], " ")
.trim()
.to_string();
let name = if name.is_empty() {
GitAuthor::default().name
} else {
name
};
let email = author
.email
.replace(['\n', '\r', '<', '>'], "")
.trim()
.to_string();
let email = if email.is_empty() {
GitAuthor::default().email
} else {
email
};
format!("{name} <{email}> now")
}
fn fast_import_quote_path(path: &str) -> String {
if path
.bytes()
.all(|byte| byte > b' ' && byte != b'"' && byte != b'\\')
{
return path.to_string();
}
let mut quoted = String::from("\"");
for byte in path.bytes() {
match byte {
b'\\' => quoted.push_str("\\\\"),
b'"' => quoted.push_str("\\\""),
b'\n' => quoted.push_str("\\n"),
b'\r' => quoted.push_str("\\r"),
b'\t' => quoted.push_str("\\t"),
b' '..=b'~' => quoted.push(byte as char),
_ => {
let _ = write!(quoted, "\\{byte:03o}");
}
}
}
quoted.push('"');
quoted
}
async fn probe_sandbox_git(sandbox: &dyn Sandbox) -> Result<(), String> {
let temp = sandbox_temp_dir(sandbox, "probe", "git");
let index = format!("{temp}/index");
let probe_file = format!("{temp}/probe.txt");
let command = format!(
"set -e\n\
rm -rf {temp_q}\n\
mkdir -p {temp_q}\n\
printf probe > {probe_file_q}\n\
GIT_INDEX_FILE={index_q} {git} read-tree --empty\n\
blob=$({git} hash-object -w {probe_file_q})\n\
GIT_INDEX_FILE={index_q} {git} update-index --add --cacheinfo 100644,$blob,probe.txt\n\
GIT_INDEX_FILE={index_q} {git} write-tree >/dev/null\n\
rm -rf {temp_q}",
temp_q = shell_quote(&temp),
probe_file_q = shell_quote(&probe_file),
index_q = shell_quote(&index),
git = GIT_REMOTE,
);
exec_ok(sandbox, &command, None)
.await
.map_err(|err| err.to_string())
}
fn sandbox_temp_dir(sandbox: &dyn Sandbox, run_id: &str, label: &str) -> String {
let cwd = sandbox.working_directory().trim_end_matches('/');
let id = uuid::Uuid::new_v4();
format!("{cwd}/.fabro/tmp/{label}-{run_id}-{id}")
}
async fn exec_stdout(
sandbox: &dyn Sandbox,
command: &str,
env: Option<&HashMap<String, String>>,
) -> Result<String, SandboxMetadataError> {
let result = sandbox
.exec_command(command, 30_000, None, env, None)
.await
.map_err(|err| SandboxMetadataError::Sandbox(err.display_with_causes()))?;
if result.is_success() {
Ok(result.stdout.trim().to_string())
} else {
Err(SandboxMetadataError::Git(exec_err(command, &result)))
}
}
async fn exec_ok(
sandbox: &dyn Sandbox,
command: &str,
env: Option<&HashMap<String, String>>,
) -> Result<(), SandboxMetadataError> {
exec_stdout(sandbox, command, env).await.map(|_| ())
}
fn exec_err(label: &str, result: &fabro_sandbox::ExecResult) -> String {
if result.is_timed_out() {
return format!("{label} timed out after {}ms", result.duration_ms);
}
if result.is_cancelled() {
return format!("{label} cancelled after {}ms", result.duration_ms);
}
let detail = format!("{}{}", result.stdout, result.stderr);
let detail = detail.trim();
if detail.is_empty() {
format!("{label} failed with exit {}", result.display_exit_code())
} else {
format!(
"{label} failed with exit {}: {detail}",
result.display_exit_code()
)
}
}
fn validate_metadata_path(path: &str) -> Result<(), SandboxMetadataError> {
let invalid = path.is_empty()
|| path.starts_with('/')
|| path
.split('/')
.any(|segment| segment.is_empty() || segment == "." || segment == "..");
if invalid {
return Err(SandboxMetadataError::Git(format!(
"invalid metadata path: {path}"
)));
}
Ok(())
}

15
lib/crates/fabro-workflow/src/services.rs
View file

@ -17,9 +17,10 @@ use tokio_util::sync::CancellationToken;
use crate::ManifestPath;
use crate::event::Emitter;
use crate::handler::HandlerRegistry;
use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle};
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git::GitState;
use crate::sandbox_metadata::SandboxGitRuntime;
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::workflow_bundle::WorkflowBundle;
/// Services shared across workflow phases.
@ -32,7 +33,9 @@ pub struct RunServices {
pub cancel_requested: Option<Arc<AtomicBool>>,
pub provider: Provider,
pub llm_source: Arc<dyn CredentialSource>,
pub(crate) metadata_runtime: Arc<SandboxGitRuntime>,
pub(crate) sandbox_git: Arc<SandboxGitRuntime>,
pub(crate) metadata_runtime: Arc<RunMetadataRuntime>,
pub(crate) metadata_writer: Option<RunMetadataWriterHandle>,
}
impl RunServices {
@ -45,7 +48,9 @@ impl RunServices {
cancel_requested: Option<Arc<AtomicBool>>,
provider: Provider,
llm_source: Arc<dyn CredentialSource>,
metadata_runtime: Arc<SandboxGitRuntime>,
sandbox_git: Arc<SandboxGitRuntime>,
metadata_runtime: Arc<RunMetadataRuntime>,
metadata_writer: Option<RunMetadataWriterHandle>,
) -> Arc<Self> {
Arc::new(Self {
run_store,
@ -55,7 +60,9 @@ impl RunServices {
cancel_requested,
provider,
llm_source,
sandbox_git,
metadata_runtime,
metadata_writer,
})
}
@ -206,6 +213,8 @@ impl EngineServices {
Provider::Anthropic,
Arc::new(StubCredentialSource),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
None,
),
registry: Arc::new(HandlerRegistry::new(Box::new(start::StartHandler))),
git_state: std::sync::RwLock::new(None),

5
lib/crates/fabro-workflow/src/test_support.rs
View file

@ -19,8 +19,9 @@ use crate::pipeline;
use crate::pipeline::types::{Executed, Initialized};
use crate::pipeline::{billing_from_checkpoint, build_terminal_event};
use crate::records::Checkpoint;
use crate::run_metadata::RunMetadataRuntime;
use crate::run_options::RunOptions;
use crate::sandbox_metadata::SandboxGitRuntime;
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::services::{EngineServices, RunServices};
/// These helpers stop at EXECUTE, so they emit the terminal event here to
@ -162,6 +163,8 @@ async fn initialized(
.llm_source
.unwrap_or_else(|| Arc::new(EnvCredentialSource::new())),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
None,
),
registry: Arc::new(registry),
git_state: std::sync::RwLock::new(None),

43
lib/crates/fabro-workflow/tests/it/integration.rs
View file

@ -10704,10 +10704,10 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() {
}
/// End-to-end test: pipeline with git checkpointing enabled + `meta_branch`
/// writes shadow branch with checkpoint data and includes `Fabro-Checkpoint`
/// trailer in run-branch commits.
/// but no worker-side GitHub credentials still writes run-branch checkpoint
/// commits and skips metadata-branch snapshots.
#[tokio::test]
async fn git_checkpoint_host_writes_shadow_branch() {
async fn git_checkpoint_host_skips_metadata_branch_without_writer_prereqs() {
// 1. Create a temporary git repo with an initial commit
let repo = tempfile::tempdir().unwrap();
std::process::Command::new("git")
@ -10821,43 +10821,24 @@ async fn git_checkpoint_host_writes_shadow_branch() {
.expect("pipeline should succeed");
assert_eq!(outcome.status, StageOutcome::Succeeded);
// 6. Assert shadow branch has checkpoint data in run.json
// 6. Without pre-run GitHub credentials, metadata snapshots are disabled.
let run_json = std::process::Command::new("git")
.args(["show", &format!("refs/heads/{meta_branch}:run.json")])
.current_dir(repo.path())
.output()
.expect("git show should run");
assert!(
run_json.status.success(),
"metadata run.json should exist: {}",
String::from_utf8_lossy(&run_json.stderr)
);
let projection: fabro_store::RunProjection =
serde_json::from_slice(&run_json.stdout).expect("run.json should parse");
let checkpoint = projection
.checkpoint
.expect("shadow branch should contain checkpoint data");
assert!(
!checkpoint.completed_nodes.is_empty(),
"checkpoint should have completed nodes"
);
assert!(
checkpoint.completed_nodes.contains(&"work".to_string()),
"checkpoint should contain the 'work' node"
!run_json.status.success(),
"metadata run.json should not exist without writer prerequisites"
);
// 7. Assert run-branch commit has Fabro-Checkpoint trailer pointing to shadow
// SHA
// 7. Assert run-branch commit still has the run checkpoint trailers.
let output = std::process::Command::new("git")
.args(["log", "--format=%B", "-1"])
.current_dir(&worktree_path)
.output()
.unwrap();
let commit_msg = String::from_utf8_lossy(&output.stdout).trim().to_string();
assert!(
commit_msg.contains("Fabro-Checkpoint:"),
"run-branch commit should have Fabro-Checkpoint trailer, got:\n{commit_msg}"
);
assert!(
commit_msg.contains("Fabro-Run:"),
"run-branch commit should have Fabro-Run trailer, got:\n{commit_msg}"
@ -10866,12 +10847,10 @@ async fn git_checkpoint_host_writes_shadow_branch() {
commit_msg.contains("Fabro-Completed:"),
"run-branch commit should have Fabro-Completed trailer, got:\n{commit_msg}"
);
// 8. Verify round-trip: shadow run.json contains the run spec
let run_spec = projection
.spec
.expect("shadow branch should contain run spec");
assert_eq!(run_spec.run_id, run_id);
assert!(
!commit_msg.contains("Fabro-Checkpoint:"),
"run-branch commit should not have Fabro-Checkpoint trailer without metadata snapshot, got:\n{commit_msg}"
);
// Cleanup worktree
let _ = std::process::Command::new("git")