From 82156442a8c55596c4671c61d61e3d1602f7a414 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Thu, 1 Oct 2026 14:42:33 -0400 Subject: [PATCH 1/2] Resolve durable final patch blobs for Files Changed --- .../app/routes/run-files.render.test.tsx | 19 ++ lib/apps/fabro-server/src/run_files.rs | 175 +++++++++++++++--- .../fabro-server/tests/it/api/run_files.rs | 166 ++++++++++++++++- 3 files changed, 326 insertions(+), 34 deletions(-) diff --git a/apps/fabro-web/app/routes/run-files.render.test.tsx b/apps/fabro-web/app/routes/run-files.render.test.tsx index 6acba2739..11799af43 100644 --- a/apps/fabro-web/app/routes/run-files.render.test.tsx +++ b/apps/fabro-web/app/routes/run-files.render.test.tsx @@ -326,6 +326,25 @@ describe("RunFiles rendering", () => { expect(lastCall.mountId).not.toBe(firstMountId); }); + test("renders the saved final patch after the sandbox is gone", () => { + const patch = "diff --git a/docs/live.md b/docs/live.md\n--- a/docs/live.md\n+++ b/docs/live.md\n@@ -1 +1,2 @@\n original\n+saved change\n"; + currentFilesPayload = makePatchPayload(patch); + currentFilesPayload.meta = { + ...currentFilesPayload.meta, + source: "final_patch", + degraded: true, + degraded_reason: "sandbox_gone", + }; + + const renderer = renderRunFiles(); + + expect(renderer.root.findAllByProps({ "data-run-file-row": "true" })).toHaveLength(1); + expect(patchDiffCalls).toHaveLength(1); + expect(patchDiffCalls[0].patch).toBe(patch); + expect(multiFileDiffCalls).toHaveLength(0); + expect(treeText(renderer.toJSON())).not.toContain("Diff unavailable"); + }); + test("refreshing from a populated diff to an empty diff shows a no-changes toast", async () => { currentFilesPayload = makePayload(1); const renderer = renderRunFiles("/runs/run_1/files?scope=all"); diff --git a/lib/apps/fabro-server/src/run_files.rs b/lib/apps/fabro-server/src/run_files.rs index 69fc4b05f..d937e9279 100644 --- a/lib/apps/fabro-server/src/run_files.rs +++ b/lib/apps/fabro-server/src/run_files.rs @@ -35,8 +35,8 @@ use fabro_api::types::{ RunFilesMetaToSha, }; use fabro_redact::SecretRedactor; -use fabro_types::RunId; -use fabro_util::shell; +use fabro_types::{RunId, blob_ref}; +use fabro_util::{error, shell}; use fabro_workflow::sandbox_git::{ DiffError, DiffNumstat, RawDiffEntry, SubmoduleChange, SymlinkChange, list_changed_files_raw, list_diff_numstat, stream_blob_metadata, stream_blobs, @@ -250,9 +250,9 @@ where /// 4. On garbage-collected base commits for aggregate scopes, fall through to a /// degraded response built from the terminal conclusion diff. /// -/// All logging emits a single `tracing::info!` with an allowlisted field -/// set enforced by [`RunFilesMetrics::emit`] — no paths, contents, or raw -/// git stderr. +/// Success metrics emit a single `tracing::info!` with an allowlisted field +/// set enforced by [`RunFilesMetrics::emit`]. Logs exclude paths, contents, +/// and raw git stderr. pub async fn list_run_files( _auth: RequiredUser, State(state): State>, @@ -538,20 +538,29 @@ async fn materialize_sandbox_path( let sandbox = match reconnect_run_sandbox(state, run_id, &projection).await { Ok(sandbox) => sandbox, Err(err) if sandbox_read_error_should_fallback(&err) => { - return Ok(build_fallback_response( + return load_fallback_response( + state, &projection, RunFilesMetaDegradedReason::SandboxGone, run_id, start, - )); + ) + .await; } Err(err) => return Err(err), }; let materialized = match scope { ListRunFilesScope::Committed => { - materialize_committed_sandbox_path(&sandbox, &projection, &base_sha, run_id, start) - .await + materialize_committed_sandbox_path( + state, + &sandbox, + &projection, + &base_sha, + run_id, + start, + ) + .await } ListRunFilesScope::Uncommitted => { materialize_working_tree_sandbox_path( @@ -577,12 +586,16 @@ async fn materialize_sandbox_path( match materialized { Ok(body) => Ok(body), - Err(err) if sandbox_read_error_should_fallback(&err) => Ok(build_fallback_response( - &projection, - RunFilesMetaDegradedReason::SandboxGone, - run_id, - start, - )), + Err(err) if sandbox_read_error_should_fallback(&err) => { + load_fallback_response( + state, + &projection, + RunFilesMetaDegradedReason::SandboxGone, + run_id, + start, + ) + .await + } Err(err) => Err(err), } } @@ -595,6 +608,7 @@ fn sandbox_read_error_should_fallback(err: &ApiError) -> bool { } async fn materialize_committed_sandbox_path( + state: &AppState, sandbox: &SandboxCheckout, projection: &fabro_store::RunProjection, base_sha: &str, @@ -605,7 +619,7 @@ async fn materialize_committed_sandbox_path( let (to_sha, to_sha_committed_at) = resolve_head_sha_and_time(sandbox).await?; materialize_committed_range_sandbox_path( sandbox, - Some(projection), + Some((state, projection)), base_sha, &to_sha, to_sha_committed_at, @@ -618,7 +632,7 @@ async fn materialize_committed_sandbox_path( async fn materialize_committed_range_sandbox_path( sandbox: &SandboxCheckout, - fallback_projection: Option<&fabro_store::RunProjection>, + fallback: Option<(&AppState, &fabro_store::RunProjection)>, base_sha: &str, to_sha: &str, to_sha_committed_at: Option>, @@ -649,13 +663,15 @@ async fn materialize_committed_range_sandbox_path( let raw_entries = match raw_res { Ok(v) => v, Err(DiffError::Permanent { .. }) => { - if let Some(projection) = fallback_projection { - return Ok(build_fallback_response( + if let Some((state, projection)) = fallback { + return load_fallback_response( + state, projection, RunFilesMetaDegradedReason::SandboxGone, run_id, start, - )); + ) + .await; } return Err(ApiError::bad_request("Invalid git diff range.")); } @@ -800,24 +816,76 @@ fn sandbox_git_error(op: &str, error: &sandbox_driver::Error) -> ApiError { transient_503(op, &display_for_log(error, &SecretRedactor)) } -/// Build the degraded response from the stored terminal diff patch. -/// When `conclusion.diff.patch` is `None`, returns the empty envelope (UI maps -/// this to R4(c)). Keeps the same `FileDiff[]` shape as live responses, but -/// leaves contents unavailable because the server only has a unified patch. -fn build_fallback_response( +/// Resolve the terminal patch before parsing it: Petri's projection carries +/// a blob reference, while older projections may carry inline patch text. +async fn load_fallback_response( + state: &AppState, projection: &fabro_store::RunProjection, reason: RunFilesMetaDegradedReason, run_id: &RunId, start: Instant, -) -> PaginatedRunFileList { +) -> ListRunFilesResult { let Some(patch) = projection .conclusion .as_ref() .and_then(|conclusion| conclusion.diff.patch.as_deref()) else { - return empty_envelope(RunFilesMetaSource::FinalPatch, RunFilesMetaScope::Committed); + return Ok(empty_envelope( + RunFilesMetaSource::FinalPatch, + RunFilesMetaScope::Committed, + )); }; + let bytes; + let patch = if let Some(hash) = blob_ref::parse_blob_ref(patch) { + bytes = state + .store_ref() + .blobs() + .read(&hash) + .await + .map_err(|error| { + tracing::error!( + %run_id, + error = %error::collect_chain(&error).join(": "), + "Failed to read final patch blob" + ); + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "The saved final patch could not be read.", + ) + })? + .ok_or_else(|| { + tracing::error!(%run_id, "Final patch blob is missing"); + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "The saved final patch is missing.", + ) + })?; + std::str::from_utf8(&bytes).map_err(|error| { + tracing::error!(%run_id, %error, "Final patch blob is not UTF-8"); + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "The saved final patch is not valid UTF-8.", + ) + })? + } else { + patch + }; + + Ok(build_fallback_response( + projection, patch, reason, run_id, start, + )) +} + +/// Build the degraded response from resolved patch text. Full file contents +/// remain unavailable because the server only has a unified patch. +fn build_fallback_response( + projection: &fabro_store::RunProjection, + patch: &str, + reason: RunFilesMetaDegradedReason, + run_id: &RunId, + start: Instant, +) -> PaginatedRunFileList { let entries: Vec = split_patch_sections(patch) .into_iter() .map(|section| section.text.to_string()) @@ -2382,6 +2450,7 @@ index 1111111..2222222 160000 fn fallback_response_json(patch: &str) -> serde_json::Value { serde_json::to_value(build_fallback_response( &fallback_projection(patch), + patch, RunFilesMetaDegradedReason::SandboxGone, &RunId::new(), Instant::now(), @@ -2389,6 +2458,58 @@ index 1111111..2222222 160000 .expect("fallback response should serialize") } + #[tokio::test] + async fn fallback_loader_preserves_inline_and_absent_patches() { + let state = crate::test_support::test_app_state(); + let patch = simple_patch("README.md"); + let mut projection = fallback_projection(&patch); + let response = load_fallback_response( + &state, + &projection, + RunFilesMetaDegradedReason::SandboxGone, + &projection.spec.run_id, + Instant::now(), + ) + .await + .unwrap(); + assert_eq!(response.data.len(), 1); + assert_eq!( + response.data[0].unified_patch.as_deref(), + Some(patch.as_str()) + ); + + projection.conclusion.as_mut().unwrap().diff.patch = None; + let response = load_fallback_response( + &state, + &projection, + RunFilesMetaDegradedReason::SandboxGone, + &projection.spec.run_id, + Instant::now(), + ) + .await + .unwrap(); + assert!(response.data.is_empty()); + assert_eq!(response.meta.total_changed, 0); + } + + #[tokio::test] + async fn fallback_loader_rejects_non_utf8_patch_bytes() { + let state = crate::test_support::test_app_state(); + let hash = state.store_ref().blobs().write(&[0xff]).await.unwrap(); + let projection = fallback_projection(&blob_ref::format_blob_ref(&hash)); + let error = load_fallback_response( + &state, + &projection, + RunFilesMetaDegradedReason::SandboxGone, + &projection.spec.run_id, + Instant::now(), + ) + .await + .unwrap_err(); + assert_eq!(error.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!(error.detail(), "The saved final patch is not valid UTF-8."); + } + fn sandbox_patch_response_json(entries: &[String]) -> serde_json::Value { serde_json::to_value(build_patch_backed_response( entries, diff --git a/lib/apps/fabro-server/tests/it/api/run_files.rs b/lib/apps/fabro-server/tests/it/api/run_files.rs index 28f52757d..ba58149da 100644 --- a/lib/apps/fabro-server/tests/it/api/run_files.rs +++ b/lib/apps/fabro-server/tests/it/api/run_files.rs @@ -1,18 +1,21 @@ //! HTTP-level integration tests for `GET /api/v1/runs/{id}/files`. //! -//! These tests exercise the handler's request-plumbing branches — -//! authentication extractor, route matching, query validation, demo-mode -//! branching, and the empty-envelope / not-found responses — without -//! requiring a reconnected sandbox. The sandbox happy path is covered by -//! unit tests on the sandbox-git helpers and by `stitch_file_diff` tests -//! in `run_files.rs`. +//! Request plumbing, empty/not-found responses, and recovery from a durable +//! final patch after deleting an isolated local run's sandbox. Sandbox-git +//! helpers and file assembly also have unit coverage in `run_files.rs`. + +use std::sync::Arc; use axum::body::Body; use axum::http::{Request, StatusCode}; +use fabro_server::test_support::{TestAppStateBuilder, test_store_bundle}; +use fabro_store::{BlobStore, test_support as store_test_support}; +use fabro_types::blob_ref; use tower::ServiceExt; use crate::helpers::{ - MINIMAL_DOT, api, minimal_intent_json, response_json, response_status, test_app_state, + MINIMAL_DOT, api, create_and_start_run_from_intent, minimal_intent_json, response_json, + response_status, run_json, test_app_state, test_app_with_scheduler, wait_for_run_status, }; fn files_url(run_id: &str) -> String { @@ -27,6 +30,155 @@ fn files_url_with_scope(run_id: &str, scope: &str) -> String { format!("{}?scope={scope}", files_url(run_id)) } +async fn get_json(app: &axum::Router, url: &str, status: StatusCode) -> serde_json::Value { + let response = app + .clone() + .oneshot( + Request::builder() + .uri(url) + .body(Body::empty()) + .expect("GET request should build"), + ) + .await + .expect("test router should respond"); + response_json(response, status, format!("GET {url}")).await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn final_patch_survives_sandbox_removal_and_reports_unreadable_blobs() { + // Execute a real command-only run in an isolated local checkout. Keeping + // the blob pool lets us simulate storage damage after verifying recovery. + let workspace = tempfile::tempdir().unwrap(); + tokio::fs::write(workspace.path().join("README.md"), "original\n") + .await + .unwrap(); + { + let repo = git2::Repository::init(workspace.path()).unwrap(); + let mut index = repo.index().unwrap(); + index.add_path(std::path::Path::new("README.md")).unwrap(); + index.write().unwrap(); + let tree_id = index.write_tree().unwrap(); + let tree = repo.find_tree(tree_id).unwrap(); + let signature = git2::Signature::now("Fabro Test", "fabro@example.com").unwrap(); + repo.commit(Some("HEAD"), &signature, &signature, "initial", &tree, &[]) + .unwrap(); + } + let pool = store_test_support::in_memory_pool_with(&[fabro_db::BLOBS_MIGRATION_SQL]); + let blobs = Arc::new(BlobStore::new(pool.clone())); + let store = Arc::new(store_test_support::test_database_with_blobs(Arc::clone( + &blobs, + ))); + let (_, artifacts) = test_store_bundle(); + let state = TestAppStateBuilder::new() + .in_process_execution() + .store_bundle(store, artifacts) + .build(); + let app = test_app_with_scheduler(Arc::clone(&state)); + let workflow = r#"digraph Changes { + start [shape=Mdiamond] + edit [shape=parallelogram, script="printf 'updated\\n' >> README.md; mkdir -p artifacts; printf 'saved\\n' > artifacts/report.txt; printf '\\000\\001' > artifacts/data.bin"] + exit [shape=Msquare] + start -> edit -> exit + }"#; + let mut intent = minimal_intent_json(&app, workflow, workspace.path()).await; + intent["title"] = serde_json::json!("Saved final patch"); + let run_id = create_and_start_run_from_intent(&app, intent).await; + let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await; + assert_eq!(status, "succeeded", "{}", run_json(&app, &run_id).await); + state + .test_petri_projector() + .settle(run_id.parse().unwrap()) + .await; + let projection = get_json(&app, &api(&format!("/runs/{run_id}/state")), StatusCode::OK).await; + let reference = projection["conclusion"]["diff"]["patch"].as_str().unwrap(); + let hash = blob_ref::parse_blob_ref(reference).expect("the final patch is blob-backed"); + let patch = blobs + .read(&hash) + .await + .unwrap() + .expect("durable patch bytes"); + let live = get_json(&app, &files_url(&run_id), StatusCode::OK).await; + assert_eq!(live["meta"]["source"], "sandbox"); + assert_eq!(live["meta"]["total_changed"], 3); + + let sandbox_dir = projection["sandbox"]["instance"]["runtime"]["working_directory"] + .as_str() + .unwrap(); + assert_ne!(std::path::Path::new(sandbox_dir), workspace.path()); + tokio::fs::remove_dir_all(sandbox_dir).await.unwrap(); + assert!(!tokio::fs::try_exists(sandbox_dir).await.unwrap()); + + let saved = get_json(&app, &files_url(&run_id), StatusCode::OK).await; + assert_eq!(saved["meta"]["source"], "final_patch"); + assert_eq!(saved["meta"]["degraded"], true); + assert_eq!(saved["meta"]["degraded_reason"], "sandbox_gone"); + assert_eq!(saved["meta"]["scope"], "committed"); + assert_eq!(saved["meta"]["total_changed"], 3); + assert_eq!( + saved["meta"]["stats"], + serde_json::json!({"additions": 2, "deletions": 0}) + ); + let files = saved["data"].as_array().unwrap(); + assert_eq!(files.len(), 3); + let file = |name: &str| { + files + .iter() + .find(|f| f["new_file"]["name"] == name) + .unwrap() + }; + let readme = file("README.md"); + assert_eq!(readme["change_kind"], "modified"); + assert!(readme["old_file"]["contents"].is_null()); + assert!(readme["new_file"]["contents"].is_null()); + let readme_patch = readme["unified_patch"].as_str().unwrap(); + assert!(readme_patch.contains(" original\n+updated\n")); + let report = file("artifacts/report.txt"); + assert_eq!(report["change_kind"], "added"); + assert!( + report["unified_patch"] + .as_str() + .unwrap() + .contains("+saved\n") + ); + assert_eq!(file("artifacts/data.bin")["binary"], true); + assert!(file("artifacts/data.bin")["unified_patch"].is_null()); + assert_eq!(blobs.read(&hash).await.unwrap().unwrap(), patch); + + // A saved reference whose bytes are gone must not look like an empty diff. + sqlx::query("DELETE FROM blobs WHERE hash = ?") + .bind(hash.to_string()) + .execute(&pool) + .await + .unwrap(); + let missing = get_json(&app, &files_url(&run_id), StatusCode::INTERNAL_SERVER_ERROR).await; + assert_eq!( + missing["errors"][0]["detail"], + "The saved final patch is missing." + ); + sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind(hash.to_string()) + .bind(b"corrupt patch".as_slice()) + .execute(&pool) + .await + .unwrap(); + let corrupt = get_json(&app, &files_url(&run_id), StatusCode::INTERNAL_SERVER_ERROR).await; + assert_eq!( + corrupt["errors"][0]["detail"], + "The saved final patch could not be read." + ); + // Repair is visible on the next request; failed materializations aren't cached. + sqlx::query("UPDATE blobs SET data = ? WHERE hash = ?") + .bind(patch.as_ref()) + .bind(hash.to_string()) + .execute(&pool) + .await + .unwrap(); + assert_eq!( + get_json(&app, &files_url(&run_id), StatusCode::OK).await, + saved + ); +} + #[tokio::test] async fn invalid_run_id_returns_400() { let app = fabro_server::test_support::build_test_router(test_app_state()); From b6de8bf8d31d246ab531693ff4bbb2385ae8c123 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Fri, 2 Oct 2026 14:49:40 -0400 Subject: [PATCH 2/2] Resolve the saved final patch when creating a pull request afterwards Creating a pull request for a finished run read the projection's final patch field directly. Petri stores that patch in the blob table and the field holds a blob reference, so the description model was handed the reference string instead of the diff, and the empty-diff check could never fire. Move the reference resolution out of Files Changed into a shared final_patch::load helper and use it for both readers. Pull request input extraction now reads the real patch, judges emptiness by its text, and reports a missing or unreadable blob instead of describing a placeholder. Co-Authored-By: Claude Opus 5.5 --- lib/apps/fabro-server/src/final_patch.rs | 119 +++++++++++++++++ lib/apps/fabro-server/src/lib.rs | 1 + lib/apps/fabro-server/src/run_files.rs | 122 ++---------------- .../src/server/handler/pull_requests.rs | 89 ++++++++++++- .../src/server/pull_request_supervisor.rs | 4 +- lib/apps/fabro-server/src/test_support.rs | 48 +++++++ 6 files changed, 264 insertions(+), 119 deletions(-) create mode 100644 lib/apps/fabro-server/src/final_patch.rs diff --git a/lib/apps/fabro-server/src/final_patch.rs b/lib/apps/fabro-server/src/final_patch.rs new file mode 100644 index 000000000..d1a75ccb0 --- /dev/null +++ b/lib/apps/fabro-server/src/final_patch.rs @@ -0,0 +1,119 @@ +//! A run's final patch as text. +//! +//! Petri keeps the final patch in the blob table, and the projection's +//! `conclusion.diff.patch` carries a `blob://sha256/` reference to it; +//! older projections carry the patch text inline. Server readers resolve the +//! patch here so none of them mistakes the reference for patch text. + +use std::borrow::Cow; + +use axum::http::StatusCode; +use fabro_store::RunProjection; +use fabro_types::blob_ref; +use fabro_util::error; + +use crate::error::ApiError; +use crate::server::AppState; + +/// The run's final patch, or `None` when the run recorded none. A reference +/// whose bytes are missing, fail the blob store's integrity check, or are not +/// UTF-8 is an error, never an empty patch. +pub(crate) async fn load<'a>( + state: &AppState, + projection: &'a RunProjection, +) -> Result>, ApiError> { + let Some(patch) = projection + .conclusion + .as_ref() + .and_then(|conclusion| conclusion.diff.patch.as_deref()) + else { + return Ok(None); + }; + let Some(hash) = blob_ref::parse_blob_ref(patch.trim()) else { + return Ok(Some(Cow::Borrowed(patch))); + }; + + let run_id = &projection.spec.run_id; + let bytes = state + .store_ref() + .blobs() + .read(&hash) + .await + .map_err(|error| { + tracing::error!( + %run_id, + error = %error::collect_chain(&error).join(": "), + "Failed to read final patch blob" + ); + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "The saved final patch could not be read.", + ) + })? + .ok_or_else(|| { + tracing::error!(%run_id, "Final patch blob is missing"); + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "The saved final patch is missing.", + ) + })?; + let text = String::from_utf8(bytes.into()).map_err(|error| { + tracing::error!(%run_id, %error, "Final patch blob is not UTF-8"); + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "The saved final patch is not valid UTF-8.", + ) + })?; + Ok(Some(Cow::Owned(text))) +} + +#[cfg(test)] +mod tests { + use axum::http::StatusCode; + use fabro_types::blob_ref; + + use super::load; + use crate::test_support::{test_app_state, test_concluded_run_projection}; + + #[tokio::test] + async fn inline_and_absent_patches_pass_through() { + let state = test_app_state(); + let projection = test_concluded_run_projection(Some("diff --git a/x b/x\n")); + let patch = load(&state, &projection).await.unwrap(); + assert_eq!(patch.as_deref(), Some("diff --git a/x b/x\n")); + + let projection = test_concluded_run_projection(None); + assert!(load(&state, &projection).await.unwrap().is_none()); + } + + #[tokio::test] + async fn blob_reference_resolves_to_its_bytes() { + let state = test_app_state(); + let text = "diff --git a/x b/x\n+saved\n"; + let hash = state + .store_ref() + .blobs() + .write(text.as_bytes()) + .await + .unwrap(); + let projection = test_concluded_run_projection(Some(&blob_ref::format_blob_ref(&hash))); + let patch = load(&state, &projection).await.unwrap(); + assert_eq!(patch.as_deref(), Some(text)); + } + + #[tokio::test] + async fn missing_and_non_utf8_blobs_are_errors() { + let state = test_app_state(); + let absent = fabro_types::BlobHash::new(b"never written"); + let projection = test_concluded_run_projection(Some(&blob_ref::format_blob_ref(&absent))); + let error = load(&state, &projection).await.unwrap_err(); + assert_eq!(error.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!(error.detail(), "The saved final patch is missing."); + + let hash = state.store_ref().blobs().write(&[0xff]).await.unwrap(); + let projection = test_concluded_run_projection(Some(&blob_ref::format_blob_ref(&hash))); + let error = load(&state, &projection).await.unwrap_err(); + assert_eq!(error.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!(error.detail(), "The saved final patch is not valid UTF-8."); + } +} diff --git a/lib/apps/fabro-server/src/lib.rs b/lib/apps/fabro-server/src/lib.rs index 3584fd7e2..fe2361608 100644 --- a/lib/apps/fabro-server/src/lib.rs +++ b/lib/apps/fabro-server/src/lib.rs @@ -25,6 +25,7 @@ pub mod csp; mod demo; pub mod diagnostics; pub mod error; +mod final_patch; mod git_checkout; pub mod github_webhooks; pub mod install; diff --git a/lib/apps/fabro-server/src/run_files.rs b/lib/apps/fabro-server/src/run_files.rs index d937e9279..610c45ed2 100644 --- a/lib/apps/fabro-server/src/run_files.rs +++ b/lib/apps/fabro-server/src/run_files.rs @@ -35,8 +35,8 @@ use fabro_api::types::{ RunFilesMetaToSha, }; use fabro_redact::SecretRedactor; -use fabro_types::{RunId, blob_ref}; -use fabro_util::{error, shell}; +use fabro_types::RunId; +use fabro_util::shell; use fabro_workflow::sandbox_git::{ DiffError, DiffNumstat, RawDiffEntry, SubmoduleChange, SymlinkChange, list_changed_files_raw, list_diff_numstat, stream_blob_metadata, stream_blobs, @@ -53,8 +53,8 @@ use tokio::sync::{Mutex, watch}; use crate::error::ApiError; use crate::principal_middleware::RequiredUser; use crate::run_files_security::{RunFilesMetrics, is_sensitive}; -use crate::sandbox_access; use crate::server::{AppState, parse_run_id_path}; +use crate::{final_patch, sandbox_access}; /// Per-file cap: 256 KiB OR 20k lines (whichever comes first). pub(crate) const PER_FILE_BYTES_CAP: u64 = 256 * 1024; @@ -816,8 +816,8 @@ fn sandbox_git_error(op: &str, error: &sandbox_driver::Error) -> ApiError { transient_503(op, &display_for_log(error, &SecretRedactor)) } -/// Resolve the terminal patch before parsing it: Petri's projection carries -/// a blob reference, while older projections may carry inline patch text. +/// The degraded response from the run's final patch, resolved through +/// [`final_patch::load`] so a blob reference is read, not parsed as a patch. async fn load_fallback_response( state: &AppState, projection: &fabro_store::RunProjection, @@ -825,55 +825,14 @@ async fn load_fallback_response( run_id: &RunId, start: Instant, ) -> ListRunFilesResult { - let Some(patch) = projection - .conclusion - .as_ref() - .and_then(|conclusion| conclusion.diff.patch.as_deref()) - else { + let Some(patch) = final_patch::load(state, projection).await? else { return Ok(empty_envelope( RunFilesMetaSource::FinalPatch, RunFilesMetaScope::Committed, )); }; - - let bytes; - let patch = if let Some(hash) = blob_ref::parse_blob_ref(patch) { - bytes = state - .store_ref() - .blobs() - .read(&hash) - .await - .map_err(|error| { - tracing::error!( - %run_id, - error = %error::collect_chain(&error).join(": "), - "Failed to read final patch blob" - ); - ApiError::new( - StatusCode::INTERNAL_SERVER_ERROR, - "The saved final patch could not be read.", - ) - })? - .ok_or_else(|| { - tracing::error!(%run_id, "Final patch blob is missing"); - ApiError::new( - StatusCode::INTERNAL_SERVER_ERROR, - "The saved final patch is missing.", - ) - })?; - std::str::from_utf8(&bytes).map_err(|error| { - tracing::error!(%run_id, %error, "Final patch blob is not UTF-8"); - ApiError::new( - StatusCode::INTERNAL_SERVER_ERROR, - "The saved final patch is not valid UTF-8.", - ) - })? - } else { - patch - }; - Ok(build_fallback_response( - projection, patch, reason, run_id, start, + projection, &patch, reason, run_id, start, )) } @@ -1805,12 +1764,13 @@ fn count_flags(data: &[FileDiff]) -> (u64, u64, u64, u64) { mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; - use fabro_types::{PetriAdmission, RunId, test_support}; + use fabro_types::RunId; use pebble_coding_agent::sandbox_driver::test_support::{MockSandbox, exec_result}; use sandbox_driver::ExecResult; use tokio::time::{Duration, sleep}; use super::*; + use crate::test_support::{test_app_state, test_concluded_run_projection}; /// The mock as the Run Files endpoints hold a sandbox. fn checkout(mock: &MockSandbox) -> SandboxCheckout { @@ -2407,49 +2367,9 @@ index 1111111..2222222 160000 } } - fn fallback_projection(patch: &str) -> fabro_store::RunProjection { - let mut projection = fabro_store::RunProjection::new( - "Test run".to_string(), - fabro_types::RunSpec { - run_id: fabro_types::fixtures::RUN_1, - settings: fabro_types::WorkflowSettings::default(), - graph: fabro_types::RunGraph::new("test"), - graph_source: None, - workflow_slug: None, - workflow_version_id: None, - target: None, - automation: None, - source_directory: None, - labels: HashMap::default(), - provenance: test_support::test_run_provenance(), - definition_blob: None, - spec_blob: None, - git: None, - fork_source_ref: None, - admission: PetriAdmission::default(), - }, - chrono::Utc::now(), - ); - projection.conclusion = Some(fabro_types::Conclusion { - timestamp: chrono::Utc::now(), - status: fabro_types::StageOutcome::Succeeded, - timing: fabro_types::RunTiming::wall_only(1), - failure: None, - final_git_commit_sha: None, - stages: Vec::new(), - usage: None, - total_retries: 0, - diff: fabro_types::RunDiff { - patch: Some(patch.to_string()), - summary: None, - }, - }); - projection - } - fn fallback_response_json(patch: &str) -> serde_json::Value { serde_json::to_value(build_fallback_response( - &fallback_projection(patch), + &test_concluded_run_projection(Some(patch)), patch, RunFilesMetaDegradedReason::SandboxGone, &RunId::new(), @@ -2460,9 +2380,9 @@ index 1111111..2222222 160000 #[tokio::test] async fn fallback_loader_preserves_inline_and_absent_patches() { - let state = crate::test_support::test_app_state(); + let state = test_app_state(); let patch = simple_patch("README.md"); - let mut projection = fallback_projection(&patch); + let mut projection = test_concluded_run_projection(Some(&patch)); let response = load_fallback_response( &state, &projection, @@ -2492,24 +2412,6 @@ index 1111111..2222222 160000 assert_eq!(response.meta.total_changed, 0); } - #[tokio::test] - async fn fallback_loader_rejects_non_utf8_patch_bytes() { - let state = crate::test_support::test_app_state(); - let hash = state.store_ref().blobs().write(&[0xff]).await.unwrap(); - let projection = fallback_projection(&blob_ref::format_blob_ref(&hash)); - let error = load_fallback_response( - &state, - &projection, - RunFilesMetaDegradedReason::SandboxGone, - &projection.spec.run_id, - Instant::now(), - ) - .await - .unwrap_err(); - assert_eq!(error.status(), StatusCode::INTERNAL_SERVER_ERROR); - assert_eq!(error.detail(), "The saved final patch is not valid UTF-8."); - } - fn sandbox_patch_response_json(entries: &[String]) -> serde_json::Value { serde_json::to_value(build_patch_backed_response( entries, diff --git a/lib/apps/fabro-server/src/server/handler/pull_requests.rs b/lib/apps/fabro-server/src/server/handler/pull_requests.rs index 650a8d09c..76b2efabf 100644 --- a/lib/apps/fabro-server/src/server/handler/pull_requests.rs +++ b/lib/apps/fabro-server/src/server/handler/pull_requests.rs @@ -1,3 +1,4 @@ +use std::borrow::Cow; use std::sync::Arc; use std::time::Duration; @@ -12,6 +13,7 @@ use super::super::{ PullRequestLink, RequireRunScoped, Response, Router, RunId, State, StatusCode, get, post, run_records, warn, }; +use crate::final_patch; pub(super) fn routes() -> Router> { Router::new() @@ -189,13 +191,16 @@ pub(in crate::server) struct RunPrInputs<'a> { pub(in crate::server) base_branch: &'a str, pub(in crate::server) run_branch: &'a str, pub(in crate::server) final_git_sha: &'a str, - pub(in crate::server) diff: &'a str, + pub(in crate::server) diff: Cow<'a, str>, pub(in crate::server) conclusion: &'a fabro_types::Conclusion, pub(in crate::server) normalized_origin: String, } impl<'a> RunPrInputs<'a> { - pub(in crate::server) fn extract( + /// What a pull request for the run needs, with its final patch resolved + /// from the blob table so the description is written from the real diff. + pub(in crate::server) async fn extract( + state: &AppState, run_state: &'a fabro_store::RunProjection, force: bool, ) -> Result { @@ -228,10 +233,8 @@ impl<'a> RunPrInputs<'a> { "missing_run_branch", ) })?; - let diff = run_state - .conclusion - .as_ref() - .and_then(|conclusion| conclusion.diff.patch.as_deref()) + let diff = final_patch::load(state, run_state) + .await? .filter(|d| !d.trim().is_empty()) .ok_or_else(|| { ApiError::with_code( @@ -333,7 +336,7 @@ async fn create_run_pull_request( { return accepted_pull_request_creation_response(&id, creation.clone()); } - if let Err(err) = RunPrInputs::extract(&run_state, body.force) { + if let Err(err) = RunPrInputs::extract(&state, &run_state, body.force).await { return err.into_response(); } if let Err(err) = load_server_github_credentials(state.as_ref()).await { @@ -605,3 +608,75 @@ async fn close_run_pull_request( Err(err) => ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response(), } } + +#[cfg(test)] +mod tests { + use axum::http::StatusCode; + use fabro_types::{DirtyStatus, GitContext, StartRecord, blob_ref}; + + use super::RunPrInputs; + use crate::test_support::{test_app_state, test_concluded_run_projection}; + + /// A finished run with everything a pull request needs and `patch` as + /// its final diff. + fn pull_request_ready_projection(patch: &str) -> fabro_store::RunProjection { + let mut projection = test_concluded_run_projection(Some(patch)); + projection.spec.git = Some(GitContext { + origin_url: "https://github.com/acme/widgets.git".to_string(), + branch: "main".to_string(), + sha: None, + dirty: DirtyStatus::Clean, + }); + projection.start = Some(StartRecord { + start_time: chrono::Utc::now(), + run_branch: Some("fabro/run/test".to_string()), + base_sha: None, + }); + projection.conclusion.as_mut().unwrap().final_git_commit_sha = Some("abc123".to_string()); + projection + } + + #[tokio::test] + async fn extract_reads_a_blob_backed_final_patch() { + let state = test_app_state(); + let text = "diff --git a/src/lib.rs b/src/lib.rs\n+fn x() {}\n"; + let hash = state + .store_ref() + .blobs() + .write(text.as_bytes()) + .await + .unwrap(); + let projection = pull_request_ready_projection(&blob_ref::format_blob_ref(&hash)); + + let inputs = RunPrInputs::extract(&state, &projection, false) + .await + .unwrap(); + assert_eq!(inputs.diff, text); + } + + #[tokio::test] + async fn extract_judges_emptiness_by_the_resolved_patch() { + let state = test_app_state(); + let hash = state.store_ref().blobs().write(b"\n").await.unwrap(); + let projection = pull_request_ready_projection(&blob_ref::format_blob_ref(&hash)); + + let Err(error) = RunPrInputs::extract(&state, &projection, false).await else { + panic!("a whitespace-only saved patch should be empty"); + }; + assert_eq!(error.status(), StatusCode::BAD_REQUEST); + assert_eq!(error.code(), Some("empty_diff")); + } + + #[tokio::test] + async fn extract_reports_a_missing_patch_blob() { + let state = test_app_state(); + let absent = fabro_types::BlobHash::new(b"never written"); + let projection = pull_request_ready_projection(&blob_ref::format_blob_ref(&absent)); + + let Err(error) = RunPrInputs::extract(&state, &projection, false).await else { + panic!("a missing patch blob should fail extraction"); + }; + assert_eq!(error.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!(error.detail(), "The saved final patch is missing."); + } +} diff --git a/lib/apps/fabro-server/src/server/pull_request_supervisor.rs b/lib/apps/fabro-server/src/server/pull_request_supervisor.rs index 3b266b673..52160002c 100644 --- a/lib/apps/fabro-server/src/server/pull_request_supervisor.rs +++ b/lib/apps/fabro-server/src/server/pull_request_supervisor.rs @@ -161,7 +161,7 @@ async fn attempt_pull_request_creation( run_state: &fabro_store::RunProjection, creation: &PullRequestCreation, ) -> anyhow::Result> { - let inputs = match RunPrInputs::extract(run_state, creation.force) { + let inputs = match RunPrInputs::extract(state, run_state, creation.force).await { Ok(inputs) => inputs, Err(err) => return Ok(Err(err.detail().to_string())), }; @@ -181,7 +181,7 @@ async fn attempt_pull_request_creation( head_branch: inputs.run_branch, expected_head_sha: inputs.final_git_sha, goal: inputs.goal, - diff: inputs.diff, + diff: &inputs.diff, model: &creation.model, draft: true, auto_merge: None, diff --git a/lib/apps/fabro-server/src/test_support.rs b/lib/apps/fabro-server/src/test_support.rs index dc3350fc3..6374fba43 100644 --- a/lib/apps/fabro-server/src/test_support.rs +++ b/lib/apps/fabro-server/src/test_support.rs @@ -381,6 +381,54 @@ pub fn test_app_state() -> Arc { ready_test_app_state_builder().build() } +/// A succeeded run's projection whose conclusion carries `patch` as its final +/// diff: patch text, a blob reference, or nothing. +#[cfg(test)] +pub(crate) fn test_concluded_run_projection(patch: Option<&str>) -> fabro_store::RunProjection { + use fabro_types::{ + Conclusion, PetriAdmission, RunDiff, RunGraph, RunSpec, RunTiming, StageOutcome, + WorkflowSettings, fixtures, + }; + + let mut projection = fabro_store::RunProjection::new( + "Test run".to_string(), + RunSpec { + run_id: fixtures::RUN_1, + settings: WorkflowSettings::default(), + graph: RunGraph::new("test"), + graph_source: None, + workflow_slug: None, + workflow_version_id: None, + target: None, + automation: None, + source_directory: None, + labels: HashMap::default(), + provenance: fabro_types::test_support::test_run_provenance(), + definition_blob: None, + spec_blob: None, + git: None, + fork_source_ref: None, + admission: PetriAdmission::default(), + }, + chrono::Utc::now(), + ); + projection.conclusion = Some(Conclusion { + timestamp: chrono::Utc::now(), + status: StageOutcome::Succeeded, + timing: RunTiming::wall_only(1), + failure: None, + final_git_commit_sha: None, + stages: Vec::new(), + usage: None, + total_retries: 0, + diff: RunDiff { + patch: patch.map(str::to_string), + summary: None, + }, + }); + projection +} + pub fn test_app_state_in_process() -> Arc { ready_test_app_state_builder() .in_process_execution()