diff --git a/Cargo.lock b/Cargo.lock index ed214b58b..fc33387bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/Cargo.toml b/Cargo.toml index 15594fded..2346c7cb8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 48b0aa634..383cb005f 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -186,7 +186,7 @@ impl Handler for ParallelHandler { // --- Git isolation: checkpoint "parallel base" before fan-out --- let base_sha: Option = 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, diff --git a/lib/crates/fabro-workflow/src/lib.rs b/lib/crates/fabro-workflow/src/lib.rs index 6baba63ba..e350021e4 100644 --- a/lib/crates/fabro-workflow/src/lib.rs +++ b/lib/crates/fabro-workflow/src/lib.rs @@ -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; diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index b2cfaffe0..ce146cc84 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -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>; type WfNodeResult = NodeResult>; @@ -70,7 +71,9 @@ pub(crate) struct GitLifecycle { pub run_id: RunId, pub run_store: RunStoreHandle, pub run_options: Arc, - pub metadata_runtime: Arc, + pub sandbox_git: Arc, + pub metadata_runtime: Arc, + pub metadata_writer: Option, pub start_node_id: Option, // Cross-lifecycle data (shared with EventLifecycle) pub checkpoint_git_result: Arc>>, @@ -84,7 +87,7 @@ impl RunLifecycle 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 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 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, run_store: RunStoreHandle, run_options: Arc, - metadata_runtime: Arc, + metadata_runtime: Arc, + ) -> 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, + run_store: RunStoreHandle, + run_options: Arc, + metadata_runtime: Arc, + metadata_writer: Option, ) -> 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::(); 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(), diff --git a/lib/crates/fabro-workflow/src/lifecycle/mod.rs b/lib/crates/fabro-workflow/src/lifecycle/mod.rs index c6770dd9d..5c622247e 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/mod.rs @@ -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>; type WfNodeResult = NodeResult>; @@ -88,7 +89,9 @@ impl WorkflowLifecycle { run_store: &RunStoreHandle, artifact_sink: Option, run_options: &Arc, - metadata_runtime: Arc, + sandbox_git: Arc, + metadata_runtime: Arc, + metadata_writer: Option, is_resume: bool, on_node: crate::OnNodeCallback, run_control: Option>, @@ -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, diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index afe46a02f..298f5cdf0 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -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 { diff --git a/lib/crates/fabro-workflow/src/pipeline/execute.rs b/lib/crates/fabro-workflow/src/pipeline/execute.rs index 99343830d..f7912548a 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute.rs @@ -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, diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index 9f036cced..23d1ca502 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -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, sandbox: Arc, - metadata_runtime: Arc, + metadata_runtime: Arc, + metadata_writer: Option, ) -> Arc { 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"), diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index 8cce99cb8..78c8ba6be 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -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, ) -> Result { - 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), diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index b99e4fc27..86cd5dd9d 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -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( diff --git a/lib/crates/fabro-workflow/src/run_dump.rs b/lib/crates/fabro-workflow/src/run_dump.rs index f67591530..5b13306f8 100644 --- a/lib/crates/fabro-workflow/src/run_dump.rs +++ b/lib/crates/fabro-workflow/src/run_dump.rs @@ -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)>) -> 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, diff --git a/lib/crates/fabro-workflow/src/run_metadata.rs b/lib/crates/fabro-workflow/src/run_metadata.rs new file mode 100644 index 000000000..b4fd71537 --- /dev/null +++ b/lib/crates/fabro-workflow/src/run_metadata.rs @@ -0,0 +1,1038 @@ +use std::collections::{BTreeMap, HashMap}; +use std::path::Path; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use git2::{ + Cred, Direction, ErrorClass, ErrorCode, FetchOptions, FileMode, Oid, PushOptions, + RemoteCallbacks, Repository, Signature, +}; +use tokio::task::{self, JoinError}; + +use crate::git::{GitAuthor, META_BRANCH_PREFIX}; +use crate::run_dump::RunDump; +use crate::run_options::RunOptions; + +pub(crate) const METADATA_PERMISSIONS: &[(&str, &str)] = &[("contents", "write")]; + +pub(crate) fn metadata_branch_name(run_id: &str) -> String { + format!("{META_BRANCH_PREFIX}{run_id}") +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum RunMetadataError { + #[error("metadata writer initialization failed: {0}")] + Init(String), + #[error("metadata token mint failed: {0}")] + TokenMint(String), + #[error("metadata remote discovery failed: {0}")] + Discovery(String), + #[error("metadata remote fetch failed: {0}")] + Fetch(String), + #[error("invalid metadata path: {0}")] + InvalidPath(String), + #[error("metadata dump serialization failed: {0}")] + DumpSerialize(anyhow::Error), + #[error("metadata tree build failed: {0}")] + Tree(String), + #[error("metadata commit failed: {0}")] + Commit(String), + #[error("metadata writer task failed: {0}")] + Join(JoinError), +} + +#[derive(Debug)] +pub(crate) struct MetadataSnapshot { + pub commit_sha: String, + pub push_error: Option, + pub entry_count: usize, + pub bytes: u64, +} + +pub(crate) struct RunMetadataRuntime { + metadata_degraded: AtomicBool, + metadata_warning_emitted: AtomicBool, +} + +impl RunMetadataRuntime { + pub(crate) fn new() -> Self { + Self { + metadata_degraded: AtomicBool::new(false), + metadata_warning_emitted: AtomicBool::new(false), + } + } + + 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 RunMetadataRuntime { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +pub(crate) trait AuthProvider: Send + Sync { + async fn token(&self) -> Result, RunMetadataError>; +} + +struct GitHubAuthProvider { + creds: fabro_github::GitHubCredentials, + origin_url: String, + permissions: HashMap, +} + +impl GitHubAuthProvider { + fn new( + creds: fabro_github::GitHubCredentials, + origin_url: String, + permissions: HashMap, + ) -> Self { + Self { + creds, + origin_url, + permissions, + } + } +} + +#[async_trait] +impl AuthProvider for GitHubAuthProvider { + async fn token(&self) -> Result, RunMetadataError> { + mint_token(&self.creds, &self.origin_url, &self.permissions) + .await + .map(Some) + .map_err(RunMetadataError::TokenMint) + } +} + +#[cfg(test)] +struct NoAuth; + +#[cfg(test)] +#[async_trait] +impl AuthProvider for NoAuth { + async fn token(&self) -> Result, RunMetadataError> { + Ok(None) + } +} + +#[derive(Clone)] +pub(crate) struct RunMetadataWriterHandle { + writer: Arc>, + auth: Arc, +} + +impl RunMetadataWriterHandle { + pub(crate) fn new(writer: RunMetadataWriter, auth: Arc) -> Self { + Self { + writer: Arc::new(Mutex::new(writer)), + auth, + } + } + + #[cfg(test)] + fn remote_url_for_test(&self) -> String { + self.writer + .lock() + .expect("metadata writer mutex poisoned") + .remote_url + .clone() + } + + #[cfg(test)] + pub(crate) fn new_for_test( + remote_url: String, + branch: String, + author: GitAuthor, + fetch_depth: Option, + ) -> Result { + let writer = RunMetadataWriter::new(remote_url, branch, author, fetch_depth)?; + Ok(Self::new(writer, Arc::new(NoAuth))) + } + + pub(crate) async fn write_snapshot( + &self, + dump: &RunDump, + message: &str, + ) -> Result { + let token = self.auth.token().await?; + let entries = dump + .git_entries() + .map_err(RunMetadataError::DumpSerialize)?; + let entry_count = entries.len(); + let bytes = metadata_entries_bytes(&entries); + let message = message.to_string(); + let writer = Arc::clone(&self.writer); + + task::spawn_blocking(move || { + let mut guard = writer.lock().expect("metadata writer mutex poisoned"); + guard.write_snapshot_blocking(&entries, entry_count, bytes, &message, token.as_deref()) + }) + .await + .map_err(RunMetadataError::Join)? + } +} + +pub(crate) fn build_metadata_writer( + run_options: &RunOptions, +) -> Result, RunMetadataError> { + let Some(git) = run_options.pre_run_git.as_ref() else { + return Ok(None); + }; + let Some(meta_branch) = run_options + .git + .as_ref() + .and_then(|git| git.meta_branch.as_ref()) + else { + return Ok(None); + }; + let Some(creds) = run_options.github_app.as_ref() else { + return Ok(None); + }; + + let normalized_url = fabro_github::normalize_repo_origin_url(&git.origin_url); + if !normalized_url.starts_with("https://") { + return Ok(None); + } + if fabro_github::parse_github_owner_repo(&normalized_url).is_err() { + return Ok(None); + } + + let auth = Arc::new(GitHubAuthProvider::new( + creds.clone(), + normalized_url.clone(), + metadata_permissions(), + )); + let writer = RunMetadataWriter::new( + normalized_url, + meta_branch.clone(), + run_options.git_author(), + Some(1), + )?; + Ok(Some(RunMetadataWriterHandle::new(writer, auth))) +} + +fn metadata_permissions() -> HashMap { + METADATA_PERMISSIONS + .iter() + .map(|(key, value)| ((*key).to_string(), (*value).to_string())) + .collect() +} + +pub(crate) async fn mint_token( + creds: &fabro_github::GitHubCredentials, + origin_url: &str, + permissions: &HashMap, +) -> Result { + if let fabro_github::GitHubCredentials::Token(token) = creds { + return Ok(token.clone()); + } + + let normalized_url = fabro_github::normalize_repo_origin_url(origin_url); + let (owner, repo) = fabro_github::parse_github_owner_repo(&normalized_url)?; + 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)?; + let client = fabro_http::http_client().map_err(|err| err.to_string())?; + let permissions = serde_json::to_value(permissions).map_err(|err| err.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(), + permissions, + install_url.as_deref(), + ) + .await +} + +pub(crate) struct RunMetadataWriter { + repo: Repository, + tempdir: tempfile::TempDir, + remote_url: String, + branch: String, + author: GitAuthor, + fetch_depth: Option, + parent_oid: Option, + discovered: bool, +} + +impl RunMetadataWriter { + pub(crate) fn new( + remote_url: String, + branch: String, + author: GitAuthor, + fetch_depth: Option, + ) -> Result { + let tempdir = tempfile::tempdir() + .map_err(|_| RunMetadataError::Init("failed to create writer tempdir".to_string()))?; + let repo = Repository::init_bare(tempdir.path()) + .map_err(|err| RunMetadataError::Init(redact_metadata_error(&err, tempdir.path())))?; + repo.odb() + .and_then(|odb| odb.add_new_mempack_backend(999).map(|_| ())) + .map_err(|err| RunMetadataError::Init(redact_metadata_error(&err, tempdir.path())))?; + + Ok(Self { + repo, + tempdir, + remote_url, + branch, + author, + fetch_depth, + parent_oid: None, + discovered: false, + }) + } + + fn write_snapshot_blocking( + &mut self, + entries: &[(String, Vec)], + entry_count: usize, + bytes: u64, + message: &str, + token: Option<&str>, + ) -> Result { + self.discover_parent(token)?; + for (path, _) in entries { + validate_metadata_path(path)?; + } + let tree_oid = build_tree(&self.repo, entries) + .map_err(|err| RunMetadataError::Tree(self.redact_tree_error(err)))?; + let tree = self + .repo + .find_tree(tree_oid) + .map_err(|err| RunMetadataError::Tree(self.redact(&err)))?; + let sig = Signature::now(&self.author.name, &self.author.email) + .map_err(|err| RunMetadataError::Commit(self.redact(&err)))?; + let mut full_message = message.to_string(); + self.author.append_footer(&mut full_message); + let parents = self + .parent_oid + .iter() + .map(|oid| self.repo.find_commit(*oid)) + .collect::, _>>() + .map_err(|err| RunMetadataError::Commit(self.redact(&err)))?; + let parent_refs = parents.iter().collect::>(); + let full_ref = self.full_ref(); + let commit_oid = self + .repo + .commit( + Some(&full_ref), + &sig, + &sig, + &full_message, + &tree, + &parent_refs, + ) + .map_err(|err| RunMetadataError::Commit(self.redact(&err)))?; + self.parent_oid = Some(commit_oid); + + let push_error = self.push(token).err(); + Ok(MetadataSnapshot { + commit_sha: commit_oid.to_string(), + push_error, + entry_count, + bytes, + }) + } + + fn discover_parent(&mut self, token: Option<&str>) -> Result<(), RunMetadataError> { + if self.discovered { + return Ok(()); + } + + let full_ref = self.full_ref(); + if let Some(path) = file_remote_path(&self.remote_url) { + let remote_repo = Repository::open(path) + .map_err(|err| RunMetadataError::Discovery(self.redact(&err)))?; + if remote_repo.find_reference(&full_ref).is_err() { + self.discovered = true; + return Ok(()); + } + drop(remote_repo); + self.fetch_parent(token, &full_ref)?; + self.discovered = true; + return Ok(()); + } + + let head_match = (|| -> Result, git2::Error> { + let mut remote = self.repo.remote_anonymous(&self.remote_url)?; + let connection = + remote.connect_auth(Direction::Fetch, Some(make_callbacks(token)), None)?; + let head = connection + .list()? + .iter() + .find(|head| head.name() == full_ref) + .map(git2::RemoteHead::oid); + Ok(head) + })() + .map_err(|err| RunMetadataError::Discovery(self.redact(&err)))?; + + let Some(_) = head_match else { + self.discovered = true; + return Ok(()); + }; + + self.fetch_parent(token, &full_ref)?; + self.discovered = true; + Ok(()) + } + + fn fetch_parent( + &mut self, + token: Option<&str>, + full_ref: &str, + ) -> Result<(), RunMetadataError> { + (|| -> Result<(), git2::Error> { + let mut remote = self.repo.remote_anonymous(&self.remote_url)?; + let mut fetch_opts = FetchOptions::new(); + if let Some(depth) = self.fetch_depth { + fetch_opts.depth(depth); + } + fetch_opts.remote_callbacks(make_callbacks(token)); + let refspec = format!("+{full_ref}:{full_ref}"); + remote.fetch(&[refspec.as_str()], Some(&mut fetch_opts), None)?; + let tip = self.repo.find_reference(full_ref)?.peel_to_commit()?.id(); + self.parent_oid = Some(tip); + Ok(()) + })() + .map_err(|err| RunMetadataError::Fetch(self.redact(&err))) + } + + fn push(&self, token: Option<&str>) -> Result<(), String> { + (|| -> Result<(), git2::Error> { + let mut remote = self.repo.remote_anonymous(&self.remote_url)?; + let mut push_opts = PushOptions::new(); + push_opts.remote_callbacks(make_callbacks(token)); + let full_ref = self.full_ref(); + let refspec = format!("{full_ref}:{full_ref}"); + remote.push(&[refspec.as_str()], Some(&mut push_opts)) + })() + .map_err(|err| self.redact(&err)) + } + + fn full_ref(&self) -> String { + format!("refs/heads/{}", self.branch) + } + + fn redact(&self, err: &git2::Error) -> String { + redact_metadata_error(err, self.tempdir.path()) + } + + fn redact_tree_error(&self, err: BuildTreeError) -> String { + match err { + BuildTreeError::Git(err) => self.redact(&err), + BuildTreeError::Conflict(message) => message, + } + } +} + +fn make_callbacks(token: Option<&str>) -> RemoteCallbacks<'_> { + let mut callbacks = RemoteCallbacks::new(); + if let Some(token) = token { + callbacks.credentials(move |_, _, _| Cred::userpass_plaintext("x-access-token", token)); + } + callbacks +} + +fn file_remote_path(remote_url: &str) -> Option<&Path> { + remote_url.strip_prefix("file://").map(Path::new) +} + +fn metadata_entries_bytes(entries: &[(String, Vec)]) -> u64 { + entries.iter().fold(0, |total, (_, bytes)| { + total.saturating_add(u64::try_from(bytes.len()).unwrap_or(u64::MAX)) + }) +} + +fn validate_metadata_path(path: &str) -> Result<(), RunMetadataError> { + let invalid = path.is_empty() + || path.starts_with('/') + || path + .split('/') + .any(|segment| segment.is_empty() || segment == "." || segment == ".."); + if invalid { + return Err(RunMetadataError::InvalidPath(path.to_string())); + } + Ok(()) +} + +enum BuildTreeError { + Git(git2::Error), + Conflict(String), +} + +impl From for BuildTreeError { + fn from(value: git2::Error) -> Self { + Self::Git(value) + } +} + +enum TreeNode { + File(Vec), + Dir(BTreeMap), +} + +fn build_tree(repo: &Repository, entries: &[(String, Vec)]) -> Result { + let mut root = BTreeMap::new(); + for (path, bytes) in entries { + insert_tree_node(&mut root, path, bytes.clone())?; + } + write_tree_node(repo, &root) +} + +fn insert_tree_node( + root: &mut BTreeMap, + path: &str, + bytes: Vec, +) -> Result<(), BuildTreeError> { + let segments = path.split('/').collect::>(); + let mut current = root; + for segment in &segments[..segments.len() - 1] { + let node = current + .entry((*segment).to_string()) + .or_insert_with(|| TreeNode::Dir(BTreeMap::new())); + match node { + TreeNode::Dir(children) => current = children, + TreeNode::File(_) => { + return Err(BuildTreeError::Conflict(format!( + "metadata path conflicts with file: {path}" + ))); + } + } + } + let filename = segments + .last() + .expect("validated path should have a filename"); + if matches!(current.get(*filename), Some(TreeNode::Dir(_))) { + return Err(BuildTreeError::Conflict(format!( + "metadata path conflicts with directory: {path}" + ))); + } + current.insert((*filename).to_string(), TreeNode::File(bytes)); + Ok(()) +} + +fn write_tree_node( + repo: &Repository, + children: &BTreeMap, +) -> Result { + let mut builder = repo.treebuilder(None)?; + for (name, node) in children { + match node { + TreeNode::File(bytes) => { + let oid = repo.blob(bytes)?; + builder.insert(name, oid, i32::from(FileMode::Blob))?; + } + TreeNode::Dir(children) => { + let oid = write_tree_node(repo, children)?; + builder.insert(name, oid, i32::from(FileMode::Tree))?; + } + } + } + Ok(builder.write()?) +} + +pub(crate) fn redact_metadata_error(err: &git2::Error, tempdir: &Path) -> String { + const MAX_ERROR_LEN: usize = 500; + + let mut message = err.message().to_string(); + message = redact_url_userinfo(&message); + if let Some(tempdir) = tempdir.to_str() { + message = message.replace(tempdir, ""); + for part in tempdir.split('/').filter(|part| !part.is_empty()) { + if part.starts_with(".tmp") || part.starts_with("tmp") { + message = message.replace(part, ""); + } + } + } + let prefix = match (err.code(), err.class()) { + (ErrorCode::Auth, _) => "github authentication failed: ", + (ErrorCode::NotFastForward, _) => "non-fast-forward push rejected: ", + (_, ErrorClass::Net) => "network failure: ", + _ => "git operation failed: ", + }; + let mut redacted = format!("{prefix}{message}"); + if redacted.len() > MAX_ERROR_LEN { + redacted.truncate(MAX_ERROR_LEN); + redacted.push_str("..."); + } + redacted +} + +fn redact_url_userinfo(input: &str) -> String { + let mut output = String::with_capacity(input.len()); + let mut rest = input; + while let Some(scheme_pos) = rest.find("://") { + let prefix_end = scheme_pos + 3; + output.push_str(&rest[..prefix_end]); + let after_scheme = &rest[prefix_end..]; + let host_end = after_scheme + .find(|ch: char| ch == '/' || ch.is_whitespace()) + .unwrap_or(after_scheme.len()); + let authority = &after_scheme[..host_end]; + if let Some((_, host)) = authority.rsplit_once('@') { + output.push_str("***@"); + output.push_str(host); + } else { + output.push_str(authority); + } + rest = &after_scheme[host_end..]; + } + output.push_str(rest); + output +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::path::Path; + use std::sync::Arc; + + use fabro_store::RunProjection; + use fabro_types::{DirtyStatus, GitContext, PreRunPushOutcome, RunSpec, WorkflowSettings}; + use git2::{ErrorClass, ErrorCode}; + + use super::*; + use crate::git::GitAuthor; + use crate::run_dump::RunDump; + use crate::run_options::{GitCheckpointOptions, RunOptions}; + + fn run_git(repo: &Path, args: &[&str]) -> String { + String::from_utf8(run_git_bytes(repo, args)) + .unwrap() + .trim() + .to_string() + } + + #[expect( + clippy::disallowed_methods, + reason = "metadata writer tests use synchronous git commands to inspect temporary repositories" + )] + fn run_git_bytes(repo: &Path, args: &[&str]) -> Vec { + let output = std::process::Command::new("git") + .args(args) + .current_dir(repo) + .output() + .unwrap(); + assert!( + output.status.success(), + "git {args:?} failed: stdout={} stderr={}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + output.stdout + } + + #[expect( + clippy::disallowed_methods, + reason = "metadata writer tests use synchronous git commands to set up local fake remotes" + )] + fn init_bare_remote() -> tempfile::TempDir { + let remote = tempfile::tempdir().unwrap(); + let output = std::process::Command::new("git") + .args(["init", "--bare"]) + .current_dir(remote.path()) + .output() + .unwrap(); + assert!( + output.status.success(), + "git init --bare failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + remote + } + + #[expect( + clippy::disallowed_methods, + reason = "metadata writer tests use synchronous git commands to set up temporary repositories" + )] + fn init_git_repo(repo: &Path) { + let init = std::process::Command::new("git") + .args(["init", "-b", "main"]) + .current_dir(repo) + .output() + .unwrap(); + assert!(init.status.success()); + for (key, value) in [("user.name", "Test"), ("user.email", "test@test.com")] { + let config = std::process::Command::new("git") + .args(["config", key, value]) + .current_dir(repo) + .output() + .unwrap(); + assert!(config.status.success()); + } + let commit = std::process::Command::new("git") + .args(["commit", "--allow-empty", "-m", "initial"]) + .current_dir(repo) + .output() + .unwrap(); + assert!(commit.status.success()); + } + + fn file_url(path: &Path) -> String { + format!("file://{}", path.display()) + } + + fn metadata_dump() -> RunDump { + let mut projection = RunProjection::default(); + projection.spec = Some(RunSpec { + run_id: fabro_types::fixtures::RUN_1, + settings: WorkflowSettings::default(), + graph: fabro_types::Graph::new("metadata"), + workflow_slug: Some("metadata".to_string()), + source_directory: Some("/Users/client/project".to_string()), + git: Some(GitContext { + origin_url: "https://github.com/fabro-sh/fabro.git".to_string(), + branch: "main".to_string(), + sha: None, + dirty: DirtyStatus::Clean, + push_outcome: PreRunPushOutcome::NotAttempted, + }), + labels: HashMap::new(), + provenance: None, + manifest_blob: None, + definition_blob: None, + fork_source_ref: None, + in_place: false, + }); + + let mut 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()); + dump + } + + fn run_options_for_origin(origin_url: &str) -> RunOptions { + RunOptions { + settings: WorkflowSettings::default(), + run_dir: tempfile::tempdir().unwrap().path().to_path_buf(), + cancel_token: None, + run_id: fabro_types::fixtures::RUN_1, + labels: HashMap::new(), + workflow_slug: Some("metadata".to_string()), + github_app: Some(fabro_github::GitHubCredentials::Token( + "ghs_token".to_string(), + )), + pre_run_git: Some(GitContext { + origin_url: origin_url.to_string(), + branch: "main".to_string(), + sha: None, + dirty: DirtyStatus::Clean, + push_outcome: PreRunPushOutcome::NotAttempted, + }), + fork_source_ref: None, + base_branch: None, + display_base_sha: None, + git: Some(GitCheckpointOptions { + base_sha: None, + run_branch: None, + meta_branch: Some("fabro/meta/test-run".to_string()), + }), + } + } + + #[tokio::test] + async fn metadata_writer_writes_binary_snapshot_to_fake_remote() { + let remote = init_bare_remote(); + let branch = "fabro/meta/test-run"; + let dump = metadata_dump(); + 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::(); + let writer = RunMetadataWriter::new( + file_url(remote.path()), + branch.to_string(), + GitAuthor::default(), + None, + ) + .unwrap(); + let handle = RunMetadataWriterHandle::new(writer, Arc::new(NoAuth)); + + let snapshot = handle.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 = run_git(remote.path(), &["rev-parse", branch]); + assert_eq!(commit_sha, snapshot.commit_sha); + assert_eq!( + run_git_bytes(remote.path(), &[ + "show", + &format!("{commit_sha}:binary/payload.bin") + ]), + vec![0, 159, 146, 150] + ); + assert_eq!( + run_git(remote.path(), &[ + "show", + &format!("{commit_sha}:path with spaces.txt") + ]), + "quoted path" + ); + } + + #[tokio::test] + async fn metadata_writer_preserves_linear_history_across_snapshots() { + let remote = init_bare_remote(); + let branch = "fabro/meta/test-run"; + let dump = metadata_dump(); + let handle = RunMetadataWriterHandle::new_for_test( + file_url(remote.path()), + branch.to_string(), + GitAuthor::default(), + None, + ) + .unwrap(); + + let first = handle.write_snapshot(&dump, "checkpoint 1").await.unwrap(); + let mut second_dump = dump.clone(); + second_dump.add_file_bytes("second.txt", b"second\n".to_vec()); + let second = handle + .write_snapshot(&second_dump, "checkpoint 2") + .await + .unwrap(); + let mut third_dump = second_dump.clone(); + third_dump.add_file_bytes("third.txt", b"third\n".to_vec()); + let third = handle + .write_snapshot(&third_dump, "checkpoint 3") + .await + .unwrap(); + + let history = run_git(remote.path(), &["rev-list", "--parents", branch]); + let lines = history.lines().collect::>(); + assert_eq!(lines.len(), 3); + assert_eq!( + lines[0], + format!("{} {}", third.commit_sha, second.commit_sha) + ); + assert_eq!( + lines[1], + format!("{} {}", second.commit_sha, first.commit_sha) + ); + assert_eq!(lines[2], first.commit_sha); + } + + #[tokio::test] + async fn metadata_writer_resumes_from_existing_remote_tip() { + let remote = init_bare_remote(); + let branch = "fabro/meta/test-run"; + let first_handle = RunMetadataWriterHandle::new_for_test( + file_url(remote.path()), + branch.to_string(), + GitAuthor::default(), + None, + ) + .unwrap(); + let first = first_handle + .write_snapshot(&metadata_dump(), "checkpoint 1") + .await + .unwrap(); + let mut second_dump = metadata_dump(); + second_dump.add_file_bytes("second.txt", b"second\n".to_vec()); + let second = first_handle + .write_snapshot(&second_dump, "checkpoint 2") + .await + .unwrap(); + drop(first_handle); + + let resumed_handle = RunMetadataWriterHandle::new_for_test( + file_url(remote.path()), + branch.to_string(), + GitAuthor::default(), + None, + ) + .unwrap(); + let mut third_dump = metadata_dump(); + third_dump.add_file_bytes("third.txt", b"third\n".to_vec()); + let third = resumed_handle + .write_snapshot(&third_dump, "checkpoint 3") + .await + .unwrap(); + + let parent_line = run_git(remote.path(), &[ + "rev-list", + "--parents", + "-n", + "1", + &third.commit_sha, + ]); + assert_eq!( + parent_line, + format!("{} {}", third.commit_sha, second.commit_sha) + ); + let root_line = run_git(remote.path(), &[ + "rev-list", + "--parents", + "-n", + "1", + &first.commit_sha, + ]); + assert_eq!(root_line, first.commit_sha); + } + + #[tokio::test] + async fn metadata_writer_returns_push_error_after_local_commit() { + let remote = tempfile::tempdir().unwrap(); + init_git_repo(remote.path()); + let handle = RunMetadataWriterHandle::new_for_test( + file_url(remote.path()), + "main".to_string(), + GitAuthor::default(), + None, + ) + .unwrap(); + + let snapshot = handle + .write_snapshot(&metadata_dump(), "checkpoint") + .await + .unwrap(); + + assert!(!snapshot.commit_sha.is_empty()); + let push_error = snapshot.push_error.unwrap(); + assert!(push_error.contains("git operation failed")); + assert!(push_error.contains("non-bare repos"), "{push_error}"); + } + + #[tokio::test] + #[expect( + clippy::disallowed_methods, + reason = "metadata writer test uses a synchronous git command to inspect a temporary remote" + )] + async fn metadata_writer_rejects_invalid_paths_before_commit() { + let remote = init_bare_remote(); + let handle = RunMetadataWriterHandle::new_for_test( + file_url(remote.path()), + "fabro/meta/test-run".to_string(), + GitAuthor::default(), + None, + ) + .unwrap(); + let dump = RunDump::from_raw_entries(vec![("../escape.txt".to_string(), b"x".to_vec())]); + + let err = handle + .write_snapshot(&dump, "checkpoint") + .await + .unwrap_err(); + + assert!(matches!(err, RunMetadataError::InvalidPath(_))); + let missing_ref = std::process::Command::new("git") + .args(["rev-parse", "--verify", "fabro/meta/test-run"]) + .current_dir(remote.path()) + .output() + .unwrap(); + assert!(!missing_ref.status.success()); + } + + #[tokio::test] + async fn metadata_writer_discovery_failure_is_pre_commit_error() { + let missing = tempfile::tempdir().unwrap().path().join("missing.git"); + let handle = RunMetadataWriterHandle::new_for_test( + file_url(&missing), + "fabro/meta/test-run".to_string(), + GitAuthor::default(), + None, + ) + .unwrap(); + + let err = handle + .write_snapshot(&metadata_dump(), "checkpoint") + .await + .unwrap_err(); + + assert!(matches!(err, RunMetadataError::Discovery(_))); + } + + #[test] + fn metadata_error_redaction_strips_credentials_and_tempdir() { + let tempdir = Path::new("/var/folders/fake/.tmpXYZ"); + let err = git2::Error::from_str( + "authenticated request to https://x-access-token:ghs_aaaaaa@github.com/owner/repo.git in /var/folders/fake/.tmpXYZ/objects failed", + ); + + let redacted = redact_metadata_error(&err, tempdir); + + assert!(!redacted.contains("ghs_aaaaaa")); + assert!(!redacted.contains("/var/folders/fake/.tmpXYZ")); + assert!(!redacted.contains(".tmpXYZ")); + assert!(redacted.contains("https://***@github.com/owner/repo.git")); + assert!(redacted.starts_with("git operation failed:")); + } + + #[test] + fn metadata_error_redaction_maps_error_codes_to_stable_hints() { + let tempdir = Path::new("/tmp/fabro"); + + let auth = redact_metadata_error( + &git2::Error::new(ErrorCode::Auth, ErrorClass::Net, "bad credentials"), + tempdir, + ); + let non_fast_forward = redact_metadata_error( + &git2::Error::new(ErrorCode::NotFastForward, ErrorClass::Reference, "rejected"), + tempdir, + ); + let network = redact_metadata_error( + &git2::Error::new(ErrorCode::GenericError, ErrorClass::Net, "offline"), + tempdir, + ); + + assert!(auth.starts_with("github authentication failed:")); + assert!(non_fast_forward.starts_with("non-fast-forward push rejected:")); + assert!(network.starts_with("network failure:")); + } + + #[test] + fn metadata_writer_factory_normalizes_github_urls_and_skips_non_github() { + let cases = [ + ( + "git@github.com:owner/repo.git", + "https://github.com/owner/repo", + ), + ( + "ssh://git@github.com/owner/repo.git", + "https://github.com/owner/repo", + ), + ( + "https://github.com/owner/repo.git", + "https://github.com/owner/repo", + ), + ( + "https://ghs_aaaaaa@github.com/owner/repo.git", + "https://github.com/owner/repo", + ), + ( + "https://x-access-token:ghs_aaaaaa@github.com/owner/repo.git", + "https://github.com/owner/repo", + ), + ]; + + for (origin, expected) in cases { + let options = run_options_for_origin(origin); + let handle = build_metadata_writer(&options).unwrap().unwrap(); + assert_eq!(handle.remote_url_for_test(), expected); + assert!(!handle.remote_url_for_test().contains("ghs_aaaaaa")); + assert!(!handle.remote_url_for_test().contains('@')); + } + + assert!( + build_metadata_writer(&run_options_for_origin("https://gitlab.com/owner/repo.git")) + .unwrap() + .is_none() + ); + } +} diff --git a/lib/crates/fabro-workflow/src/sandbox_git.rs b/lib/crates/fabro-workflow/src/sandbox_git.rs index 5b6c1e3b2..5fcb1f9e7 100644 --- a/lib/crates/fabro-workflow/src/sandbox_git.rs +++ b/lib/crates/fabro-workflow/src/sandbox_git.rs @@ -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, - commands: Arc>>, - pushes: Arc>>, - } - - impl RecordingSandbox { - fn new(inner: Arc) -> Self { - Self { - inner, - commands: Arc::new(Mutex::new(Vec::new())), - pushes: Arc::new(Mutex::new(Vec::new())), - } - } - - fn commands_after_probe(&self) -> Vec { - self.commands - .lock() - .unwrap() - .iter() - .filter(|command| !command.contains("probe.txt")) - .cloned() - .collect() - } - - fn pushes(&self) -> Vec { - self.pushes.lock().unwrap().clone() - } - } - - #[async_trait] - impl Sandbox for RecordingSandbox { - async fn read_file( - &self, - path: &str, - offset: Option, - limit: Option, - ) -> fabro_sandbox::Result { - 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 { - self.inner.file_exists(path).await - } - - async fn list_directory( - &self, - path: &str, - depth: Option, - ) -> fabro_sandbox::Result> { - 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>, - cancel_token: Option, - ) -> fabro_sandbox::Result { - 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> { - self.inner.grep(pattern, path, options).await - } - - async fn glob( - &self, - pattern: &str, - path: Option<&str>, - ) -> fabro_sandbox::Result> { - 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>, } @@ -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::(); - 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::(); - 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::(); - 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(); diff --git a/lib/crates/fabro-workflow/src/sandbox_git_runtime.rs b/lib/crates/fabro-workflow/src/sandbox_git_runtime.rs new file mode 100644 index 000000000..2cc00c7c3 --- /dev/null +++ b/lib/crates/fabro-workflow/src/sandbox_git_runtime.rs @@ -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>, +} + +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() + ) + } +} diff --git a/lib/crates/fabro-workflow/src/sandbox_metadata.rs b/lib/crates/fabro-workflow/src/sandbox_metadata.rs deleted file mode 100644 index 1d29051b8..000000000 --- a/lib/crates/fabro-workflow/src/sandbox_metadata.rs +++ /dev/null @@ -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>, - 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, - 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 { - 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)], - message: &str, - temp: &str, - entry_count: usize, - bytes: u64, - ) -> Result { - 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)]) -> 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)], - author: &GitAuthor, -) -> Result, 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, line: &str) { - stream.extend_from_slice(line.as_bytes()); - stream.push(b'\n'); -} - -fn push_data(stream: &mut Vec, 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 { - 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>, -) -> Result { - 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>, -) -> 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(()) -} diff --git a/lib/crates/fabro-workflow/src/services.rs b/lib/crates/fabro-workflow/src/services.rs index 47c1e727d..ae3912790 100644 --- a/lib/crates/fabro-workflow/src/services.rs +++ b/lib/crates/fabro-workflow/src/services.rs @@ -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>, pub provider: Provider, pub llm_source: Arc, - pub(crate) metadata_runtime: Arc, + pub(crate) sandbox_git: Arc, + pub(crate) metadata_runtime: Arc, + pub(crate) metadata_writer: Option, } impl RunServices { @@ -45,7 +48,9 @@ impl RunServices { cancel_requested: Option>, provider: Provider, llm_source: Arc, - metadata_runtime: Arc, + sandbox_git: Arc, + metadata_runtime: Arc, + metadata_writer: Option, ) -> Arc { 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), diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index 074845e58..3f6cf5e47 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -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), diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index edfce5503..1c522911e 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -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")