From 1c82bd90086fa31687223b35b829048b59775950 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 27 Jul 2026 11:25:18 -0400 Subject: [PATCH 1/4] fix(workflow): make publish failures terminal --- docs/internal/events.md | 2 + docs/public/api-reference/fabro-api.yaml | 1 + docs/public/integrations/github.mdx | 7 +- .../src/commands/run/run_progress/event.rs | 1 + .../src/commands/run/run_progress/mod.rs | 1 + lib/apps/fabro-cli/tests/it/cmd/pr_view.rs | 1 + lib/apps/fabro-server/src/demo/mod.rs | 1 + lib/apps/fabro-server/src/install.rs | 23 +- lib/apps/fabro-server/src/server.rs | 16 + .../src/server/handler/pull_requests.rs | 15 + lib/apps/fabro-server/src/server/tests.rs | 14 +- lib/components/fabro-github/src/lib.rs | 128 +++++- lib/components/fabro-store/src/run_state.rs | 2 + lib/components/fabro-workflow/README.md | 3 +- lib/components/fabro-workflow/src/error.rs | 82 +++- .../fabro-workflow/src/event/convert.rs | 2 + .../fabro-workflow/src/event/events.rs | 4 + .../fabro-workflow/src/operations/start.rs | 20 +- .../fabro-workflow/src/pipeline/finalize.rs | 344 ++++++++++++++-- .../fabro-workflow/src/pipeline/mod.rs | 10 +- .../fabro-workflow/src/pipeline/publish.rs | 201 +++++++++ .../src/pipeline/pull_request.rs | 384 ++++++------------ .../fabro-workflow/src/pipeline/types.rs | 60 ++- .../fabro-api/tests/status_round_trip.rs | 1 + .../fabro-types/src/run_event/misc.rs | 2 + lib/foundation/fabro-types/src/status.rs | 1 + .../src/models/failure-reason.ts | 1 + 27 files changed, 989 insertions(+), 338 deletions(-) create mode 100644 lib/components/fabro-workflow/src/pipeline/publish.rs diff --git a/docs/internal/events.md b/docs/internal/events.md index 196925e94..63948d376 100644 --- a/docs/internal/events.md +++ b/docs/internal/events.md @@ -2116,6 +2116,7 @@ These legacy events may appear in older run logs. Current CLI backend runs do no "properties": { "pr_url": "https://github.com/org/repo/pull/42", "pr_number": 42, + "head_sha": "d34db33f", "draft": true } } @@ -2125,6 +2126,7 @@ These legacy events may appear in older run logs. Current CLI backend runs do no |----------|------|-------------| | `pr_url` | string | Pull request URL | | `pr_number` | number | Pull request number | +| `head_sha` | string (optional) | Verified commit SHA at the remote PR head; absent on older events | | `draft` | boolean | Whether the PR is a draft | ### `pull_request.linked` diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index b028d89c0..2c3a0358a 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -8894,6 +8894,7 @@ components: type: string enum: - workflow_error + - publish_failed - cancelled - approval_denied - terminated diff --git a/docs/public/integrations/github.mdx b/docs/public/integrations/github.mdx index e70fe5428..741c19e47 100644 --- a/docs/public/integrations/github.mdx +++ b/docs/public/integrations/github.mdx @@ -55,6 +55,7 @@ When you choose the GitHub App strategy, the CLI opens GitHub with a pre-filled | Permission | Level | Purpose | |---|---|---| | Contents | Write | Clone repos, push run branches and checkpoints | + | Workflows | Write | Push changes under `.github/workflows/` | | Metadata | Read | Look up repository installation status | | Pull requests | Write | Create and update PRs from workflows | | Checks | Write | Report workflow status on commits | @@ -219,7 +220,7 @@ When a workflow runs in a remote sandbox (Daytona or Docker), Fabro clones the c 2. SSH URLs (e.g. `git@github.com:owner/repo.git`) are converted to HTTPS 3. Fabro signs a short-lived JWT using the App ID and private key (RS256, 10-minute validity) 4. Using the JWT, Fabro looks up the GitHub App installation for the repository (`GET /repos/\{owner\}/\{repo\}/installation`) -5. Fabro requests a scoped Installation Access Token with `contents: write` permission on the specific repository +5. Fabro requests a scoped Installation Access Token with `contents: write` and `workflows: write` permissions on the specific repository 6. The sandbox clones via HTTPS using `x-access-token` as the username and the token as the password For public repositories, the clone works without credentials. The token is still generated because it's needed for pushing checkpoints. @@ -248,7 +249,9 @@ The upper bound on what Fabro will mint is whatever permissions the GitHub App i ### Checkpoint pushing -After each workflow stage, Fabro [checkpoints](/execution/checkpoints) by pushing the run branch and metadata branch to origin. Inside remote sandboxes, the git remote URL is configured with the Installation Access Token for authenticated pushing. +After each workflow stage, Fabro [checkpoints](/execution/checkpoints) by pushing the run branch and metadata branch to origin. Before a successful run becomes terminal, the publish stage pushes the final commit again and treats failure as a run failure. Inside remote sandboxes, the git remote URL is configured with the Installation Access Token for authenticated pushing. + +When pull request creation is enabled, Fabro then checks that GitHub reports the run branch at the exact final commit before opening the PR. A failed final push, branch check, or PR creation marks the run as failed with `publish_failed`; the terminal run event is emitted only after this step finishes. For long-running workflows, Fabro refreshes the token before each push since Installation Access Tokens are short-lived (typically 1 hour). diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs index 1febfc5b7..d53a591ff 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs @@ -614,6 +614,7 @@ mod tests { repo: "widgets".into(), base_branch: "main".into(), head_branch: "fabro/run/42".into(), + head_sha: "final-sha".into(), title: "Ship the server-side PR".into(), draft: true, }; diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs index b18683354..4b7a04cbf 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs @@ -1383,6 +1383,7 @@ mod tests { repo: "fabro".into(), base_branch: "main".into(), head_branch: "fabro/run/42".into(), + head_sha: "final-sha".into(), title: "Ship the change".into(), draft: true, }); diff --git a/lib/apps/fabro-cli/tests/it/cmd/pr_view.rs b/lib/apps/fabro-cli/tests/it/cmd/pr_view.rs index 4c716c8ea..35a68e4d9 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/pr_view.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/pr_view.rs @@ -85,6 +85,7 @@ fn pr_view_reads_pull_request_from_store_without_pull_request_json() { repo: "fabro".to_string(), base_branch: "main".to_string(), head_branch: "fabro/run/demo".to_string(), + head_sha: Some("final-sha".to_string()), title: "Map the constellations".to_string(), draft: false, }), diff --git a/lib/apps/fabro-server/src/demo/mod.rs b/lib/apps/fabro-server/src/demo/mod.rs index b9d51dcef..554b73672 100644 --- a/lib/apps/fabro-server/src/demo/mod.rs +++ b/lib/apps/fabro-server/src/demo/mod.rs @@ -1280,6 +1280,7 @@ mod runs { fn parse_failure_reason(reason: &str) -> Option { match reason { "workflow_error" => Some(FailureReason::WorkflowError), + "publish_failed" => Some(FailureReason::PublishFailed), "cancelled" => Some(FailureReason::Cancelled), "approval_denied" => Some(FailureReason::ApprovalDenied), "terminated" => Some(FailureReason::Terminated), diff --git a/lib/apps/fabro-server/src/install.rs b/lib/apps/fabro-server/src/install.rs index 12bb5fbb7..ded7f9216 100644 --- a/lib/apps/fabro-server/src/install.rs +++ b/lib/apps/fabro-server/src/install.rs @@ -2053,6 +2053,7 @@ fn build_github_app_manifest( "public": false, "default_permissions": { "contents": "write", + "workflows": "write", "metadata": "read", "pull_requests": "write", "checks": "write", @@ -2354,11 +2355,27 @@ mod tests { InstallObjectStoreCredentialMode, InstallObjectStoreInput, InstallObjectStoreProvider, InstallObjectStoreState, InstallSandboxProviderState, InstallSandboxState, InstallTokenQuery, LlmProvidersInput, PendingInstall, ServerConfigInput, ServerSecrets, - classify_object_store_validation_error, detect_canonical_url, install_object_store_lookup, - lock_unpoisoned, post_install_finish, provider_base_url_override, - resolve_install_object_store_state, token_is_valid, write_artifact_store_metadata, + build_github_app_manifest, classify_object_store_validation_error, detect_canonical_url, + install_object_store_lookup, lock_unpoisoned, post_install_finish, + provider_base_url_override, resolve_install_object_store_state, token_is_valid, + write_artifact_store_metadata, }; + #[test] + fn github_app_manifest_allows_workflow_file_writes() { + let manifest = build_github_app_manifest( + "Fabro Test", + "https://fabro.example/setup", + "https://fabro.example/auth/callback/github", + "https://fabro.example/setup", + ); + + assert_eq!( + manifest["default_permissions"]["workflows"], + serde_json::Value::String("write".to_string()) + ); + } + #[test] fn token_validation_accepts_any_matching_source() { let state = InstallAppState::for_test("expected"); diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 11a9a0948..c3d3113af 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -4183,6 +4183,14 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { reason: FailureReason::Cancelled, }; } + Err(e @ WorkflowError::Publish { .. }) => { + let detail = e.display_with_causes(); + error!(run_id = %run_id, error = %detail, "Run publish failed"); + managed_run.status = RunStatus::Failed { + reason: FailureReason::PublishFailed, + }; + managed_run.error = Some(detail); + } Err(e) => { error!(run_id = %run_id, error = %e, "Run failed"); managed_run.status = RunStatus::Failed { @@ -4197,6 +4205,14 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { reason: FailureReason::Cancelled, }; } + Err(e @ WorkflowError::Publish { .. }) => { + let detail = e.display_with_causes(); + error!(run_id = %run_id, error = %detail, "Run publish failed"); + managed_run.status = RunStatus::Failed { + reason: FailureReason::PublishFailed, + }; + managed_run.error = Some(detail); + } Err(e) => { error!(run_id = %run_id, error = %e, "Run failed"); managed_run.status = RunStatus::Failed { 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 f826c813f..a30a55c03 100644 --- a/lib/apps/fabro-server/src/server/handler/pull_requests.rs +++ b/lib/apps/fabro-server/src/server/handler/pull_requests.rs @@ -165,6 +165,7 @@ struct RunPrInputs<'a> { goal: &'a str, base_branch: &'a str, run_branch: &'a str, + final_git_sha: &'a str, diff: &'a str, conclusion: &'a fabro_types::Conclusion, normalized_origin: String, @@ -224,6 +225,17 @@ impl<'a> RunPrInputs<'a> { "run_not_finished", ) })?; + let final_git_sha = conclusion + .final_git_commit_sha + .as_deref() + .filter(|sha| !sha.trim().is_empty()) + .ok_or_else(|| { + ApiError::with_code( + StatusCode::BAD_REQUEST, + "Run has no final git commit SHA — the remote branch cannot be verified.", + "missing_final_git_commit", + ) + })?; if !force && !conclusion.status.is_successful() { return Err(ApiError::with_code( StatusCode::BAD_REQUEST, @@ -240,6 +252,7 @@ impl<'a> RunPrInputs<'a> { goal: run_spec.graph.goal(), base_branch, run_branch, + final_git_sha, diff, conclusion, normalized_origin, @@ -323,6 +336,7 @@ async fn create_run_pull_request( origin_url: &inputs.normalized_origin, base_branch: inputs.base_branch, head_branch: inputs.run_branch, + expected_head_sha: inputs.final_git_sha, goal: inputs.goal, diff: inputs.diff, model: &model, @@ -350,6 +364,7 @@ async fn create_run_pull_request( &created_pull_request.link, &created_pull_request.base_branch, &created_pull_request.head_branch, + &created_pull_request.head_sha, &created_pull_request.title, true, ); diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 71b4dfebf..73c9fcd34 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -4685,6 +4685,7 @@ channel = "#deploys" repo: "fabro".to_string(), base_branch: "main".to_string(), head_branch: "fabro/run/test".to_string(), + head_sha: "final-sha".to_string(), title: "Ship & notify".to_string(), draft: false, }, @@ -6425,6 +6426,7 @@ async fn create_run_with_pull_request_record( repo: "widgets".to_string(), base_branch: "main".to_string(), head_branch: "feature".to_string(), + head_sha: "final-sha".to_string(), title: title.to_string(), draft: false, }, @@ -6519,7 +6521,7 @@ async fn create_completed_run_ready_for_pull_request( status: "succeeded".to_string(), reason: SuccessReason::Completed, total_usd_micros: None, - final_git_commit_sha: None, + final_git_commit_sha: Some("final-sha".to_string()), final_patch: Some(final_patch.to_string()), diff_summary: None, billing: None, @@ -9058,6 +9060,14 @@ async fn get_run_pull_request_returns_stored_github_association_when_github_pr_i #[tokio::test] async fn create_run_pull_request_creates_and_persists_record() { let github = MockServer::start(); + let branch_mock = github.mock(|when, then| { + when.method("GET") + .path("/repos/acme/widgets/branches/fabro/run/42") + .header("authorization", "Bearer ghu_test"); + then.status(200) + .header("content-type", "application/json") + .body(json!({ "commit": { "sha": "final-sha" } }).to_string()); + }); let create_mock = github.mock(|when, then| { when.method("POST") .path("/repos/acme/widgets/pulls") @@ -9155,6 +9165,7 @@ async fn create_run_pull_request_creates_and_persists_record() { assert_eq!(state_body["pull_request"]["repo"], "widgets"); response_mock.assert_async().await; + branch_mock.assert(); create_mock.assert(); } @@ -16477,6 +16488,7 @@ async fn list_runs_includes_live_metadata_from_run_state() { repo: "repo".to_string(), base_branch: "main".to_string(), head_branch: "fabro/run".to_string(), + head_sha: "final-sha".to_string(), title: "Fix board metadata".to_string(), draft: false, }, diff --git a/lib/components/fabro-github/src/lib.rs b/lib/components/fabro-github/src/lib.rs index bac1f5fbc..d150c4759 100644 --- a/lib/components/fabro-github/src/lib.rs +++ b/lib/components/fabro-github/src/lib.rs @@ -587,7 +587,10 @@ async fn mint_installation_token_with_jwt( }) } -/// Request a scoped Installation Access Token with `contents: write`. +/// Request a scoped Installation Access Token for git writes. +/// +/// The `workflows` permission is required when a pushed commit creates or +/// updates files under `.github/workflows/`. pub async fn create_installation_access_token( client: &impl HttpClient, jwt: &str, @@ -601,7 +604,7 @@ pub async fn create_installation_access_token( owner, repo, base_url, - serde_json::json!({ "contents": "write" }), + serde_json::json!({ "contents": "write", "workflows": "write" }), ) .await } @@ -899,6 +902,55 @@ pub async fn branch_exists( branch_exists_with_client(&client, ctx, owner, repo, branch).await } +/// Return the commit SHA at the head of a GitHub branch. +/// +/// Returns `None` when the branch does not exist. +pub async fn branch_head_sha( + ctx: &GitHubContext<'_>, + owner: &str, + repo: &str, + branch: &str, +) -> anyhow::Result> { + #[derive(Deserialize)] + struct BranchResponse { + commit: BranchCommit, + } + + #[derive(Deserialize)] + struct BranchCommit { + sha: String, + } + + let client = ctx.http_client()?; + let token = ctx + .creds + .resolve_bearer_token( + &client, + owner, + repo, + ctx.base_url, + serde_json::json!({ "contents": "read" }), + ) + .await?; + + let url = format!("{}/repos/{owner}/{repo}/branches/{branch}", ctx.base_url); + let auth = format!("Bearer {token}"); + let resp = HttpClient::request(&client, HttpMethod::Get, &url, &github_headers(&auth), None) + .await + .context("Failed to read remote branch head")?; + + match resp.status { + 200 => { + let branch: BranchResponse = resp + .json() + .context("Failed to parse remote branch response")?; + Ok(Some(branch.commit.sha)) + } + 404 => Ok(None), + status => bail!("Unexpected status {status} reading branch '{branch}'"), + } +} + async fn branch_exists_with_client( client: &impl HttpClient, ctx: &GitHubContext<'_>, @@ -1032,26 +1084,47 @@ pub async fn update_app_webhook_config( /// Resolve git clone credentials for a GitHub repository. /// -/// Returns `(username, password)` for authenticated cloning. +/// Returns `(username, password)` for authenticated cloning and pushing. /// Always generates a token regardless of repo visibility, since the token -/// is needed for pushing from the sandbox. +/// is needed for pushing from the sandbox. The token includes `workflows: +/// write` so a run can publish workflow-file changes. pub async fn resolve_clone_credentials( ctx: &GitHubContext<'_>, owner: &str, repo: &str, +) -> anyhow::Result<(Option, Option)> { + match ctx.creds { + GitHubCredentials::Pat(token) => { + Ok((Some("x-access-token".to_string()), Some(token.clone()))) + } + GitHubCredentials::Installation(token) => Ok(( + Some("x-access-token".to_string()), + Some(token.valid_token()?.to_string()), + )), + GitHubCredentials::App(_) => { + let client = ctx.http_client()?; + resolve_clone_credentials_with_client(&client, ctx, owner, repo).await + } + } +} + +async fn resolve_clone_credentials_with_client( + client: &impl HttpClient, + ctx: &GitHubContext<'_>, + owner: &str, + repo: &str, ) -> anyhow::Result<(Option, Option)> { let token = match ctx.creds { GitHubCredentials::Pat(token) => token.clone(), GitHubCredentials::Installation(token) => token.valid_token()?.to_string(), GitHubCredentials::App(_) => { - let client = ctx.http_client()?; ctx.creds .resolve_bearer_token( - &client, + client, owner, repo, ctx.base_url, - serde_json::json!({ "contents": "write" }), + serde_json::json!({ "contents": "write", "workflows": "write" }), ) .await? } @@ -1775,7 +1848,9 @@ mod tests { r#"{"token": "ghs_xxx", "expires_at": "2099-01-01T00:00:00Z"}"#, ) .with_req_header("Authorization", "Bearer test-jwt") - .with_req_body(r#"{"permissions":{"contents":"write"},"repositories":["repo"]}"#); + .with_req_body( + r#"{"permissions":{"contents":"write","workflows":"write"},"repositories":["repo"]}"#, + ); let token = create_installation_access_token(&mock, "test-jwt", "owner", "repo", "") .await @@ -2296,6 +2371,43 @@ mod tests { ); } + #[tokio::test] + async fn resolve_clone_credentials_requests_workflow_write_permission() { + let mock = MockHttpClient::new() + .on( + HttpMethod::Get, + "/repos/owner/repo/installation", + 200, + r#"{"id": 123}"#, + ) + .on( + HttpMethod::Post, + "/app/installations/123/access_tokens", + 201, + r#"{"token": "ghs_xxx", "expires_at": "2099-01-01T00:00:00Z"}"#, + ) + .with_req_body( + r#"{"permissions":{"contents":"write","workflows":"write"},"repositories":["repo"]}"#, + ); + let credentials = GitHubCredentials::App(GitHubAppCredentials { + app_id: "test".to_string(), + private_key_pem: test_rsa_key().to_string(), + slug: None, + }); + let context = GitHubContext::new(&credentials, ""); + let resolved = resolve_clone_credentials_with_client(&mock, &context, "owner", "repo") + .await + .unwrap(); + + assert_eq!( + resolved, + ( + Some("x-access-token".to_string()), + Some("ghs_xxx".to_string()) + ) + ); + } + #[test] fn installation_token_valid_token_rejects_expired_tokens() { let expired = InstallationToken { diff --git a/lib/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index 9ab1e2b2d..4f076b41c 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -3791,6 +3791,7 @@ mod tests { repo: "fabro".to_string(), base_branch: "main".to_string(), head_branch: "fabro/run/demo".to_string(), + head_sha: Some("final-sha".to_string()), title: "Add run PR chip".to_string(), draft: false, }), @@ -3840,6 +3841,7 @@ mod tests { repo: github_pull_request.repo.clone(), base_branch: "main".to_string(), head_branch: "fabro/run/demo".to_string(), + head_sha: Some("final-sha".to_string()), title: "Add run PR chip".to_string(), draft: false, }), diff --git a/lib/components/fabro-workflow/README.md b/lib/components/fabro-workflow/README.md index 21deba55b..007051592 100644 --- a/lib/components/fabro-workflow/README.md +++ b/lib/components/fabro-workflow/README.md @@ -67,7 +67,8 @@ assert_eq!(graph.goal(), "Run tests"); use fabro_workflow::operations::start; use fabro_workflow::pipeline; -// Use `operations::start(...)` for the full initialize -> execute -> finalize flow. +// Use `operations::start(...)` for the full +// initialize -> execute -> conclude -> publish -> finalize flow. // Use `pipeline::initialize(...)` + `pipeline::execute(...)` when you need partial lifecycle control. ``` diff --git a/lib/components/fabro-workflow/src/error.rs b/lib/components/fabro-workflow/src/error.rs index f52079a29..c134a5e70 100644 --- a/lib/components/fabro-workflow/src/error.rs +++ b/lib/components/fabro-workflow/src/error.rs @@ -285,6 +285,15 @@ pub enum Error { source: Option, }, + #[error("Publish error: {message}")] + Publish { + message: String, + failure_class: FailureCategory, + exec_output_tail: Option, + #[source] + source: Option, + }, + #[error("Handler error: {message}")] Handler { message: String, @@ -420,10 +429,49 @@ impl Error { Self::engine_with_source(message, source) } + /// Build an error for the required publish stage. + pub fn publish(message: impl Into) -> Self { + let message = message.into(); + let failure_class = classify_failure_reason(&message); + Self::Publish { + message, + failure_class, + exec_output_tail: None, + source: None, + } + } + + pub fn publish_with_source( + message: impl Into, + source: impl Into, + ) -> Self { + Self::publish_with_source_and_exec_output_tail(message, source, None) + } + + pub fn publish_with_source_and_exec_output_tail( + message: impl Into, + source: impl Into, + exec_output_tail: Option, + ) -> Self { + let message = message.into(); + let source = SharedError::new(source.into()); + let causes = collect_chain(&source); + let rendered = render_with_causes(&message, &causes); + let failure_class = classify_failure_reason(&rendered); + Self::Publish { + message, + failure_class, + exec_output_tail, + source: Some(source), + } + } + #[must_use] pub fn causes(&self) -> Vec { match self { - Self::Engine { source, .. } | Self::Handler { source, .. } => source + Self::Engine { source, .. } + | Self::Publish { source, .. } + | Self::Handler { source, .. } => source .as_ref() .map_or_else(Vec::new, |source| collect_chain(source)), Self::Template { source, .. } => collect_chain(source), @@ -439,15 +487,17 @@ impl Error { /// Whether this error category is retryable (transient) or terminal. /// - /// Retryable: Handler (transient handler failures), Engine (could be - /// transient), Io (network/disk issues are often transient), - /// Llm (delegates to SdkError). Terminal: Parse, Validation, - /// OutputSchemaValidation, Stylesheet (configuration errors), Checkpoint - /// (storage integrity), Cancelled (explicit cancellation). + /// Retryable: Handler and Engine, I/O, LLM errors when the SDK marks them + /// retryable, and Publish errors classified as transient infrastructure + /// failures. Terminal: Parse, Validation, OutputSchemaValidation, + /// Stylesheet, Checkpoint, and Cancelled. #[must_use] pub fn is_retryable(&self) -> bool { match self { Self::Handler { .. } | Self::Engine { .. } | Self::Io(_) => true, + Self::Publish { failure_class, .. } => { + matches!(failure_class, FailureCategory::TransientInfra) + } Self::Llm(sdk_err) => sdk_err.retryable(), Self::Parse(_) | Self::Validation(_) @@ -483,9 +533,9 @@ impl Error { | Self::Unsupported(_) | Self::OutputSchemaValidation(_) => FailureCategory::Deterministic, Self::Precondition(_) | Self::RunNotFound(_) => FailureCategory::Structural, - Self::Handler { failure_class, .. } | Self::Engine { failure_class, .. } => { - *failure_class - } + Self::Handler { failure_class, .. } + | Self::Engine { failure_class, .. } + | Self::Publish { failure_class, .. } => *failure_class, } } @@ -502,13 +552,18 @@ impl Error { #[must_use] pub fn to_failure_detail(&self) -> FailureDetail { let message = match self { - Self::Engine { message, .. } | Self::Handler { message, .. } => message.clone(), + Self::Engine { message, .. } + | Self::Publish { message, .. } + | Self::Handler { message, .. } => message.clone(), _ => self.to_string(), }; let explicit_exec_output_tail = match self { Self::Engine { exec_output_tail, .. } + | Self::Publish { + exec_output_tail, .. + } | Self::Handler { exec_output_tail, .. } => exec_output_tail.clone(), @@ -1974,6 +2029,7 @@ mod tests { }], }, Error::engine("engine err"), + Error::publish("publish err"), Error::handler("handler err"), Error::Llm(SdkError::Network { message: "refused".into(), @@ -2005,6 +2061,12 @@ mod tests { ); } + #[test] + fn publish_error_is_only_retryable_for_transient_failures() { + assert!(Error::publish("connection timed out").is_retryable()); + assert!(!Error::publish("permission denied").is_retryable()); + } + #[test] fn failure_class_stability() { let messages = [ diff --git a/lib/components/fabro-workflow/src/event/convert.rs b/lib/components/fabro-workflow/src/event/convert.rs index 5cef4e581..42a4ffbc4 100644 --- a/lib/components/fabro-workflow/src/event/convert.rs +++ b/lib/components/fabro-workflow/src/event/convert.rs @@ -1311,6 +1311,7 @@ fn event_body_from_event(event: &Event) -> EventBody { repo, base_branch, head_branch, + head_sha, title, draft, } => EventBody::PullRequestCreated(fabro_types::PullRequestCreatedProps { @@ -1320,6 +1321,7 @@ fn event_body_from_event(event: &Event) -> EventBody { repo: repo.clone(), base_branch: base_branch.clone(), head_branch: head_branch.clone(), + head_sha: (!head_sha.is_empty()).then(|| head_sha.clone()), title: title.clone(), draft: *draft, }), diff --git a/lib/components/fabro-workflow/src/event/events.rs b/lib/components/fabro-workflow/src/event/events.rs index 20803b1fa..f512e80c6 100644 --- a/lib/components/fabro-workflow/src/event/events.rs +++ b/lib/components/fabro-workflow/src/event/events.rs @@ -730,6 +730,8 @@ pub enum Event { repo: String, base_branch: String, head_branch: String, + #[serde(default)] + head_sha: String, title: String, draft: bool, }, @@ -769,6 +771,7 @@ impl Event { record: &PullRequestLink, base_branch: &str, head_branch: &str, + head_sha: &str, title: &str, draft: bool, ) -> Self { @@ -779,6 +782,7 @@ impl Event { repo: record.repo.clone(), base_branch: base_branch.to_string(), head_branch: head_branch.to_string(), + head_sha: head_sha.to_string(), title: title.to_string(), draft, } diff --git a/lib/components/fabro-workflow/src/operations/start.rs b/lib/components/fabro-workflow/src/operations/start.rs index 3b560c7cb..58c9e1c20 100644 --- a/lib/components/fabro-workflow/src/operations/start.rs +++ b/lib/components/fabro-workflow/src/operations/start.rs @@ -37,8 +37,8 @@ use crate::event::{ use crate::handler::HandlerRegistry; use crate::outcome::{Outcome, StageOutcome}; use crate::pipeline::{ - self, FinalizeOptions, Finalized, InitOptions, LlmSpec, Persisted, PullRequestOptions, - ResumeState, SandboxEnvSpec, build_conclusion_from_store, classify_engine_result, + self, FinalizeOptions, Finalized, InitOptions, LlmSpec, Persisted, PublishOptions, ResumeState, + SandboxEnvSpec, build_conclusion_from_store, classify_engine_result, }; #[cfg(test)] use crate::records::Checkpoint; @@ -794,7 +794,7 @@ fn runtime_setup_commands( } impl RunSession { - /// Shared engine: initialize, execute, finalize, pull_request. + /// Shared engine: initialize, execute, conclude, publish, finalize. async fn run( self, persisted: Persisted, @@ -921,14 +921,14 @@ impl RunSession { .expect("last_git_sha mutex should not be poisoned: no code panics while holding this lock") .clone(), }; - let pr_opts = PullRequestOptions { + let publish_opts = PublishOptions { pr_config: self.pr_config, github_app: self.pr_github_app, origin_url: self.pr_origin_url, model: self.pr_model, }; - let concluded = match Box::pin(pipeline::finalize(executed, &finalize_opts)).await { + let concluded = match Box::pin(pipeline::conclude(executed, &finalize_opts)).await { Ok(concluded) => concluded, Err(err) => { self.steering_hub.drain_pending_at_run_end(); @@ -936,7 +936,15 @@ impl RunSession { return Err(err); } }; - let finalized = Box::pin(pipeline::pull_request(concluded, &pr_opts)).await; + let published = Box::pin(pipeline::publish(concluded, &publish_opts)).await; + let finalized = match Box::pin(pipeline::finalize(published, &finalize_opts)).await { + Ok(finalized) => finalized, + Err(err) => { + self.steering_hub.drain_pending_at_run_end(); + store_progress_logger.flush().await; + return Err(err); + } + }; // Emit `agent.steer.dropped { reason: run_ended }` for any // unconsumed pending steers on the success path, then flush. The // scopeguard above re-runs as a no-op (drain is idempotent on an diff --git a/lib/components/fabro-workflow/src/pipeline/finalize.rs b/lib/components/fabro-workflow/src/pipeline/finalize.rs index 949578880..673989a5c 100644 --- a/lib/components/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/components/fabro-workflow/src/pipeline/finalize.rs @@ -9,7 +9,7 @@ use fabro_types::{BilledTokenCounts, DiffSummary, EventBody, RunFailure, RunProj use fabro_util::error::collect_causes; use fabro_util::time::elapsed_ms; -use super::types::{Concluded, Executed, FinalizeOptions}; +use super::types::{Concluded, Executed, FinalizeOptions, Finalized, Published}; use crate::error::{Error, run_failure_from_error, run_failure_from_outcome_failure}; use crate::event::{Event, RunNoticeCode, RunNoticeLevel}; use crate::outcome::{Outcome, StageOutcome}; @@ -56,6 +56,15 @@ pub fn classify_engine_result( reason: FailureReason::Cancelled, }, ), + Err(err @ Error::Publish { .. }) => ( + StageOutcome::Failed { + retry_requested: false, + }, + Some(run_failure_from_error(err, FailureReason::PublishFailed)), + RunStatus::Failed { + reason: FailureReason::PublishFailed, + }, + ), Err(err) => ( StageOutcome::Failed { retry_requested: false, @@ -483,6 +492,9 @@ pub(crate) fn build_terminal_event( Err(Error::Cancelled) => { run_failure_from_error(&Error::Cancelled, FailureReason::Cancelled) } + Err(err @ Error::Publish { .. }) => { + run_failure_from_error(err, FailureReason::PublishFailed) + } Err(err) => run_failure_from_error(err, FailureReason::WorkflowError), Ok(outcome) => { if let Some(failure) = outcome.failure.as_ref() { @@ -521,16 +533,13 @@ async fn stop_sandbox_on_terminal( Ok(()) } -/// FINALIZE phase: build conclusion, write the meta branch, emit the terminal -/// `WorkflowRunCompleted`/`WorkflowRunFailed` event. -/// -/// The terminal event is emitted here (not from `on_run_end`) so observers -/// can't act on "done" before the meta branch writes are flushed. +/// CONCLUDE phase: collect the execution result, final commit, and diff. /// /// # Errors /// -/// Returns `Error` if persisting terminal state fails. -pub async fn finalize(executed: Executed, options: &FinalizeOptions) -> Result { +/// Returns `Error` if the run state needed to build the conclusion cannot be +/// collected. +pub async fn conclude(executed: Executed, options: &FinalizeOptions) -> Result { let Executed { graph, outcome, @@ -561,20 +570,78 @@ pub async fn finalize(executed: Executed, options: &FinalizeOptions) -> Result Result { + let Published { + execution_outcome, + publish_outcome, + mut conclusion, + artifact_count, + run_options, + services, + } = published; + + let pushed_branch = publish_outcome + .as_ref() + .ok() + .and_then(|outcome| outcome.pushed_branch()) + .map(str::to_string); + let pr_url = publish_outcome + .as_ref() + .ok() + .and_then(|outcome| outcome.pr_url()) + .map(str::to_string); + let outcome = match (execution_outcome, publish_outcome) { + (Err(error), _) | (Ok(_), Err(error)) => Err(error), + (Ok(outcome), Ok(_)) => Ok(outcome), + }; + + let (final_status, failure, _run_status) = classify_engine_result(&outcome); + conclusion.status = final_status; + conclusion.failure = failure; + + write_finalize_commit(&run_options, &services, &conclusion).await; if services.metadata_runtime.metadata_degraded() { services.emitter.notice( @@ -588,9 +655,9 @@ pub async fn finalize(executed: Executed, options: &FinalizeOptions) -> Result Result Result { + let concluded = conclude(executed, options).await?; + let published = crate::pipeline::publish(concluded, &crate::pipeline::PublishOptions { + pr_config: None, + github_app: None, + origin_url: None, + model: "test-model".to_string(), + }) + .await; + finalize(published, options).await + } + fn test_store() -> Arc { Arc::new(Database::new( Arc::new(InMemory::new()), @@ -869,6 +951,26 @@ mod tests { use crate::test_support::test_usage; + #[test] + fn publish_error_builds_publish_failed_terminal_event() { + let event = build_terminal_event( + &Err(Error::publish("GitHub rejected pull request creation")), + fabro_types::RunTiming::wall_only(10), + 0, + Some("final-sha".to_string()), + Some("diff".to_string()), + None, + None, + ); + + match event { + Event::WorkflowRunFailed { failure, .. } => { + assert_eq!(failure.reason, FailureReason::PublishFailed); + } + other => panic!("expected run failure, got {other:?}"), + } + } + #[test] fn conclusion_stage_order_follows_projection_first_event_order() { let mut projection = test_projection(); @@ -1068,7 +1170,7 @@ mod tests { services, ); - let concluded = finalize(executed, &FinalizeOptions { + let concluded = finalize_executed(executed, &FinalizeOptions { run_dir: run_dir.clone(), run_id: test_run_id(), workflow_name: "test".to_string(), @@ -1249,7 +1351,7 @@ mod tests { services, ); - finalize(executed, &FinalizeOptions { + finalize_executed(executed, &FinalizeOptions { run_dir: repo_dir.path().to_path_buf(), run_id: test_run_id(), workflow_name: "test".to_string(), @@ -1273,6 +1375,196 @@ mod tests { ]); } + #[tokio::test] + async fn configured_run_branch_without_remote_is_not_reported_as_pushed() { + let repo_dir = tempfile::tempdir().unwrap(); + let emitter = Arc::new(Emitter::new(test_run_id())); + let events = record_events(&emitter); + let services = test_services( + RunStoreHandle::local(seeded_run_store().await), + emitter, + Arc::new(MockSandbox::linux()), + Arc::new(RunMetadataRuntime::new()), + None, + ); + let mut run_options = test_run_options(repo_dir.path()); + run_options.git = Some(GitCheckpointOptions { + base_sha: None, + run_branch: Some("fabro/run/test".to_string()), + meta_branch: None, + }); + let executed = test_executed( + Graph::new("test"), + Ok(Outcome::success()), + run_options, + 5, + services, + ); + let options = FinalizeOptions { + run_dir: repo_dir.path().to_path_buf(), + run_id: test_run_id(), + workflow_name: "test".to_string(), + preserve_sandbox: false, + stop_on_terminal: true, + last_git_sha: Some("final-sha".to_string()), + }; + let concluded = conclude(executed, &options).await.unwrap(); + let published = crate::pipeline::publish(concluded, &crate::pipeline::PublishOptions { + pr_config: None, + github_app: None, + origin_url: None, + model: "test-model".to_string(), + }) + .await; + + assert!(matches!( + &published.publish_outcome, + Ok(crate::pipeline::PublishOutcome::NotRequested) + )); + let finalized = finalize(published, &options).await.unwrap(); + + assert!(finalized.outcome.is_ok()); + assert_eq!(finalized.pushed_branch, None); + let events = events.lock().unwrap(); + let names = events.iter().map(RunEvent::event_name).collect::>(); + assert_eq!(names, vec!["run.completed"]); + } + + #[tokio::test] + async fn final_push_failure_becomes_terminal_publish_failure() { + let repo_dir = tempfile::tempdir().unwrap(); + let sandbox = Arc::new(MockSandbox::linux()); + let emitter = Arc::new(Emitter::new(test_run_id())); + let events = record_events(&emitter); + let services = test_services( + RunStoreHandle::local(seeded_run_store().await), + emitter, + sandbox, + Arc::new(RunMetadataRuntime::new()), + None, + ); + let mut run_options = test_run_options(repo_dir.path()); + run_options.git = Some(GitCheckpointOptions { + base_sha: None, + run_branch: Some("fabro/run/test".to_string()), + meta_branch: None, + }); + let executed = test_executed( + Graph::new("test"), + Ok(Outcome::success()), + run_options, + 5, + services, + ); + let options = FinalizeOptions { + run_dir: repo_dir.path().to_path_buf(), + run_id: test_run_id(), + workflow_name: "test".to_string(), + preserve_sandbox: false, + stop_on_terminal: true, + last_git_sha: Some("final-sha".to_string()), + }; + let concluded = conclude(executed, &options).await.unwrap(); + let published = crate::pipeline::publish(concluded, &crate::pipeline::PublishOptions { + pr_config: None, + github_app: None, + origin_url: Some("https://github.com/owner/repo.git".to_string()), + model: "test-model".to_string(), + }) + .await; + + assert!(matches!( + &published.publish_outcome, + Err(Error::Publish { .. }) + )); + let finalized = finalize(published, &options).await.unwrap(); + + assert!(matches!(finalized.outcome, Err(Error::Publish { .. }))); + assert_eq!( + finalized + .conclusion + .failure + .as_ref() + .map(|failure| failure.reason), + Some(FailureReason::PublishFailed) + ); + let events = events.lock().unwrap(); + let names = events.iter().map(RunEvent::event_name).collect::>(); + assert_eq!(names, vec!["git.push", "run.failed"]); + match &events.last().unwrap().body { + EventBody::RunFailed(props) => { + assert_eq!(props.failure.reason, FailureReason::PublishFailed); + } + other => panic!("expected run.failed, got {other:?}"), + } + } + + #[tokio::test] + async fn pull_request_failure_precedes_terminal_publish_failure() { + let repo_dir = tempfile::tempdir().unwrap(); + init_git_repo(repo_dir.path()); + let emitter = Arc::new(Emitter::new(test_run_id())); + let events = record_events(&emitter); + let services = test_services( + RunStoreHandle::local(seeded_run_store().await), + emitter, + Arc::new(fabro_agent::LocalSandbox::new( + repo_dir.path().to_path_buf(), + )), + Arc::new(RunMetadataRuntime::new()), + None, + ); + let mut run_options = test_run_options(repo_dir.path()); + run_options.base_branch = Some("main".to_string()); + run_options.git = Some(GitCheckpointOptions { + base_sha: None, + run_branch: Some("fabro/run/test".to_string()), + meta_branch: None, + }); + let executed = test_executed( + Graph::new("test"), + Ok(Outcome::success()), + run_options, + 5, + services, + ); + let options = FinalizeOptions { + run_dir: repo_dir.path().to_path_buf(), + run_id: test_run_id(), + workflow_name: "test".to_string(), + preserve_sandbox: false, + stop_on_terminal: true, + last_git_sha: Some("final-sha".to_string()), + }; + let mut concluded = conclude(executed, &options).await.unwrap(); + concluded.conclusion.diff.patch = + Some("diff --git a/a b/a\n+published change\n".to_string()); + let published = crate::pipeline::publish(concluded, &crate::pipeline::PublishOptions { + pr_config: Some(fabro_types::settings::run::PullRequestSettings { + enabled: true, + draft: true, + auto_merge: false, + merge_strategy: fabro_types::settings::run::MergeStrategy::Squash, + }), + github_app: None, + origin_url: Some("https://github.com/owner/repo.git".to_string()), + model: "test-model".to_string(), + }) + .await; + let finalized = finalize(published, &options).await.unwrap(); + + assert!(matches!(finalized.outcome, Err(Error::Publish { .. }))); + let events = events.lock().unwrap(); + let names = events.iter().map(RunEvent::event_name).collect::>(); + assert_eq!(names, vec!["git.push", "pull_request.failed", "run.failed"]); + match &events.last().unwrap().body { + EventBody::RunFailed(props) => { + assert_eq!(props.failure.reason, FailureReason::PublishFailed); + } + other => panic!("expected run.failed, got {other:?}"), + } + } + #[tokio::test] async fn finalize_stops_sandbox_on_terminal_without_deleting() { let repo_dir = tempfile::tempdir().unwrap(); @@ -1292,7 +1584,7 @@ mod tests { services, ); - finalize(executed, &FinalizeOptions { + finalize_executed(executed, &FinalizeOptions { run_dir: repo_dir.path().to_path_buf(), run_id: test_run_id(), workflow_name: "test".to_string(), @@ -1326,7 +1618,7 @@ mod tests { services, ); - finalize(executed, &FinalizeOptions { + finalize_executed(executed, &FinalizeOptions { run_dir: repo_dir.path().to_path_buf(), run_id: test_run_id(), workflow_name: "test".to_string(), @@ -1379,7 +1671,7 @@ mod tests { services, ); - finalize(executed, &FinalizeOptions { + finalize_executed(executed, &FinalizeOptions { run_dir: repo.to_path_buf(), run_id: test_run_id(), workflow_name: "test".to_string(), diff --git a/lib/components/fabro-workflow/src/pipeline/mod.rs b/lib/components/fabro-workflow/src/pipeline/mod.rs index d0ba5ae1b..57fd59e15 100644 --- a/lib/components/fabro-workflow/src/pipeline/mod.rs +++ b/lib/components/fabro-workflow/src/pipeline/mod.rs @@ -3,6 +3,7 @@ mod finalize; mod initialize; mod parse; mod persist; +mod publish; mod pull_request; mod transform; pub(crate) mod types; @@ -12,18 +13,19 @@ pub use execute::execute; pub(crate) use finalize::build_conclusion_from_store; #[cfg(any(test, feature = "test-support"))] pub(crate) use finalize::{billing_from_projection, build_terminal_event}; -pub use finalize::{classify_engine_result, finalize, write_finalize_commit}; +pub use finalize::{classify_engine_result, conclude, finalize, write_finalize_commit}; pub use initialize::initialize; pub use parse::parse; pub(crate) use persist::persist; +pub use publish::publish; pub use pull_request::{ AutoMergeOptions, CreatedPullRequest, OpenPullRequestRequest, PrContent, build_pr_content, - maybe_open_pull_request, pull_request, + maybe_open_pull_request, }; pub use transform::transform; pub use types::{ Concluded, Executed, FinalizeOptions, Finalized, InitOptions, Initialized, LlmSpec, Parsed, - Persisted, PullRequestOptions, ResumeState, SandboxEnvSpec, TEMPLATE_UNDEFINED_VARIABLE_RULE, - TransformOptions, Transformed, Validated, + Persisted, PublishOptions, PublishOutcome, Published, ResumeState, SandboxEnvSpec, + TEMPLATE_UNDEFINED_VARIABLE_RULE, TransformOptions, Transformed, Validated, }; pub use validate::validate; diff --git a/lib/components/fabro-workflow/src/pipeline/publish.rs b/lib/components/fabro-workflow/src/pipeline/publish.rs new file mode 100644 index 000000000..d4ba2fd61 --- /dev/null +++ b/lib/components/fabro-workflow/src/pipeline/publish.rs @@ -0,0 +1,201 @@ +use std::sync::Arc; + +use super::pull_request::{AutoMergeOptions, OpenPullRequestRequest, maybe_open_pull_request}; +use super::types::{Concluded, PublishOptions, PublishOutcome, Published}; +use crate::error::Error; +use crate::event::Event; +use crate::outcome::StageOutcome; + +/// PUBLISH phase: push the final run commit and, when configured, open a pull +/// request. +/// +/// Publish is always present in the pipeline. It becomes a no-op when the run +/// did not succeed, is a dry run, or has no remote branch configured. +pub async fn publish(concluded: Concluded, options: &PublishOptions) -> Published { + let publish_outcome = publish_inner(&concluded, options).await; + let Concluded { + outcome, + conclusion, + artifact_count, + graph: _, + run_options, + services, + } = concluded; + + Published { + execution_outcome: outcome, + publish_outcome, + conclusion, + artifact_count, + run_options, + services, + } +} + +async fn publish_inner( + concluded: &Concluded, + options: &PublishOptions, +) -> Result { + let successful_execution = concluded.outcome.as_ref().is_ok_and(|outcome| { + matches!( + outcome.status, + StageOutcome::Succeeded | StageOutcome::PartiallySucceeded + ) + }); + if !successful_execution || concluded.run_options.dry_run_enabled() { + return Ok(PublishOutcome::NotRequested); + } + + let pull_request_requested = options.pr_config.is_some(); + let Some(origin_url) = options + .origin_url + .as_deref() + .filter(|origin| !origin.trim().is_empty()) + else { + if pull_request_requested { + return Err(pull_request_error( + concluded, + "pull request creation requires a GitHub origin URL", + )); + } + return Ok(PublishOutcome::NotRequested); + }; + let Some(run_branch) = concluded.run_options.run_branch() else { + if pull_request_requested { + return Err(pull_request_error( + concluded, + "pull request creation requires a run branch", + )); + } + return Ok(PublishOutcome::NotRequested); + }; + if !concluded.run_options.settings.run.run_branch.push { + if pull_request_requested { + return Err(pull_request_error( + concluded, + "pull request creation requires run branch pushing", + )); + } + return Ok(PublishOutcome::NotRequested); + } + + let final_sha = concluded + .conclusion + .final_git_commit_sha + .as_deref() + .ok_or_else(|| Error::publish("cannot publish a run without a final git commit SHA"))?; + let refspec = format!("refs/heads/{run_branch}:refs/heads/{run_branch}"); + match concluded.services.sandbox.git_push_ref(&refspec).await { + Ok(()) => { + concluded.services.emitter.emit(&Event::GitPush { + branch: run_branch.to_string(), + success: true, + exec_output_tail: None, + }); + } + Err(error) => { + let exec_output_tail = fabro_sandbox::default_redacted_output_tail(&error); + concluded.services.emitter.emit(&Event::GitPush { + branch: run_branch.to_string(), + success: false, + exec_output_tail: exec_output_tail.clone(), + }); + return Err(Error::publish_with_source_and_exec_output_tail( + format!("failed to push final commit {final_sha} to branch '{run_branch}'"), + error, + exec_output_tail, + )); + } + } + + let diff = concluded + .conclusion + .diff + .patch + .as_deref() + .unwrap_or_default(); + let Some(pr_config) = options.pr_config.as_ref() else { + return Ok(PublishOutcome::Published { + pushed_branch: run_branch.to_string(), + pr_url: None, + }); + }; + if diff.trim().is_empty() { + return Ok(PublishOutcome::NoChanges { + pushed_branch: run_branch.to_string(), + }); + } + + let base_branch = concluded + .run_options + .base_branch + .as_deref() + .ok_or_else(|| { + pull_request_error(concluded, "pull request creation requires a base branch") + })?; + let credentials = options.github_app.as_ref().ok_or_else(|| { + pull_request_error( + concluded, + "pull request creation requires GitHub credentials", + ) + })?; + let auto_merge = pr_config.auto_merge.then_some(AutoMergeOptions { + merge_strategy: pr_config.merge_strategy, + }); + let github_base_url = fabro_github::github_api_base_url(); + + let created = maybe_open_pull_request(OpenPullRequestRequest { + github: fabro_github::GitHubContext::new(credentials, &github_base_url), + origin_url, + base_branch, + head_branch: run_branch, + expected_head_sha: final_sha, + goal: concluded.graph.goal(), + diff, + model: &options.model, + draft: pr_config.draft, + auto_merge, + run_store: &concluded.services.run_store, + llm_source: concluded.services.llm_source.as_ref(), + catalog: Arc::clone(&concluded.services.catalog), + conclusion: Some(&concluded.conclusion), + run_state: None, + }) + .await + .map_err(|error| { + concluded.services.emitter.emit(&Event::PullRequestFailed { + error: error.clone(), + }); + Error::publish_with_source("failed to create pull request", anyhow::anyhow!(error)) + })? + .ok_or_else(|| { + pull_request_error( + concluded, + "pull request creation found no changes after the stored diff was checked", + ) + })?; + + concluded + .services + .emitter + .emit(&Event::pull_request_created( + &created.link, + &created.base_branch, + &created.head_branch, + &created.head_sha, + &created.title, + pr_config.draft, + )); + + Ok(PublishOutcome::Published { + pushed_branch: run_branch.to_string(), + pr_url: Some(created.link.html_url()), + }) +} + +fn pull_request_error(concluded: &Concluded, message: &str) -> Error { + concluded.services.emitter.emit(&Event::PullRequestFailed { + error: message.to_string(), + }); + Error::publish(message) +} diff --git a/lib/components/fabro-workflow/src/pipeline/pull_request.rs b/lib/components/fabro-workflow/src/pipeline/pull_request.rs index 0dcc7745c..638c87847 100644 --- a/lib/components/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/components/fabro-workflow/src/pipeline/pull_request.rs @@ -13,9 +13,7 @@ use fabro_types::settings::run::MergeStrategy; use fabro_util::text::strip_goal_decoration; use tracing::{debug, info, warn}; -use super::types::{Concluded, Finalized, PullRequestOptions}; -use crate::event::{Event, RunNoticeCode, RunNoticeLevel}; -use crate::outcome::{StageOutcome, format_cost as outcome_format_cost}; +use crate::outcome::format_cost as outcome_format_cost; use crate::records::{Conclusion, RunSpec}; use crate::runtime_store::RunStoreHandle; @@ -327,22 +325,6 @@ fn assemble_pr_body( parts.join("\n") } -async fn load_pull_request_diff(run_store: &RunStoreHandle) -> String { - run_store - .state() - .await - .inspect_err(|err| { - tracing::warn!(error = %err, "Failed to load final patch from store for PR"); - }) - .ok() - .and_then(|state| { - state - .conclusion - .and_then(|conclusion| conclusion.diff.patch) - }) - .unwrap_or_default() -} - /// Build complete PR content by combining LLM-generated narrative with /// deterministic fallbacks and programmatic sections. pub async fn build_pr_content( @@ -461,20 +443,23 @@ pub struct AutoMergeOptions { /// Inputs for [`maybe_open_pull_request`]. pub struct OpenPullRequestRequest<'a> { - pub github: github_app::GitHubContext<'a>, - pub origin_url: &'a str, - pub base_branch: &'a str, - pub head_branch: &'a str, - pub goal: &'a str, - pub diff: &'a str, - pub model: &'a str, - pub draft: bool, - pub auto_merge: Option, - pub run_store: &'a RunStoreHandle, - pub llm_source: &'a dyn CredentialSource, - pub catalog: Arc, - pub conclusion: Option<&'a Conclusion>, - pub run_state: Option<&'a RunProjection>, + pub github: github_app::GitHubContext<'a>, + pub origin_url: &'a str, + pub base_branch: &'a str, + pub head_branch: &'a str, + /// Commit that must be visible at the remote branch before the PR is + /// opened. + pub expected_head_sha: &'a str, + pub goal: &'a str, + pub diff: &'a str, + pub model: &'a str, + pub draft: bool, + pub auto_merge: Option, + pub run_store: &'a RunStoreHandle, + pub llm_source: &'a dyn CredentialSource, + pub catalog: Arc, + pub conclusion: Option<&'a Conclusion>, + pub run_state: Option<&'a RunProjection>, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -483,6 +468,7 @@ pub struct CreatedPullRequest { pub title: String, pub base_branch: String, pub head_branch: String, + pub head_sha: String, } /// Optionally open a pull request after a successful workflow run. @@ -516,6 +502,25 @@ pub async fn maybe_open_pull_request( let body = truncate_pr_body(&content.body); let title = content.title; + let remote_head = github_app::branch_head_sha(&req.github, &owner, &repo, req.head_branch) + .await + .map_err(|err| format!("failed to verify remote branch head: {err:#}"))?; + match remote_head { + Some(remote_head) if remote_head == req.expected_head_sha => {} + Some(remote_head) => { + return Err(format!( + "remote branch '{}' points to commit {remote_head}, expected final commit {}", + req.head_branch, req.expected_head_sha + )); + } + None => { + return Err(format!( + "remote branch '{}' does not exist; expected final commit {}", + req.head_branch, req.expected_head_sha + )); + } + } + let created = github_app::create_pull_request( &req.github, &owner, @@ -565,111 +570,10 @@ pub async fn maybe_open_pull_request( title, base_branch: req.base_branch.to_string(), head_branch: req.head_branch.to_string(), + head_sha: req.expected_head_sha.to_string(), })) } -/// PULL_REQUEST phase: optionally create a pull request after finalize. -/// -/// This stage is infallible: failures are emitted and logged, but the pipeline -/// completes. -pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) -> Finalized { - let Concluded { - outcome, - conclusion, - graph, - run_options, - services, - } = concluded; - - let mut pr_url = None; - if let Some(pr_cfg) = &options.pr_config { - if run_options.dry_run_enabled() { - tracing::debug!("Skipping PR creation: run is in dry-run mode"); - } else if let Err(ref e) = outcome { - tracing::debug!(error = %e, "Skipping PR creation: engine returned an error"); - } else if let Ok(ref result) = outcome { - if matches!( - result.status, - StageOutcome::Succeeded | StageOutcome::PartiallySucceeded - ) { - let diff = load_pull_request_diff(&services.run_store).await; - if let (Some(base_branch), Some(run_branch), Some(creds), Some(origin)) = ( - &run_options.base_branch, - run_options.run_branch(), - &options.github_app, - &options.origin_url, - ) { - let auto_merge = if pr_cfg.auto_merge { - Some(AutoMergeOptions { - merge_strategy: pr_cfg.merge_strategy, - }) - } else { - None - }; - - match maybe_open_pull_request(OpenPullRequestRequest { - github: github_app::GitHubContext::new( - creds, - &github_app::github_api_base_url(), - ), - origin_url: origin, - base_branch, - head_branch: run_branch, - goal: graph.goal(), - diff: &diff, - model: &options.model, - draft: pr_cfg.draft, - auto_merge, - run_store: &services.run_store, - llm_source: services.llm_source.as_ref(), - catalog: Arc::clone(&services.catalog), - conclusion: Some(&conclusion), - run_state: None, - }) - .await - { - Ok(Some(created)) => { - services.emitter.emit(&Event::pull_request_created( - &created.link, - &created.base_branch, - &created.head_branch, - &created.title, - pr_cfg.draft, - )); - pr_url = Some(created.link.html_url()); - } - Ok(None) => {} - Err(e) => { - services - .emitter - .emit(&Event::PullRequestFailed { error: e.clone() }); - services.emitter.notice( - RunNoticeLevel::Warn, - RunNoticeCode::PullRequestFailed, - format!("PR creation failed: {e}"), - ); - } - } - } - } - } - } - - Finalized { - run_id: run_options.run_id, - outcome, - conclusion, - pushed_branch: run_options - .settings - .run - .run_branch - .push - .then(|| run_options.run_branch().map(str::to_string)) - .flatten(), - pr_url, - } -} - #[cfg(test)] mod tests { use std::collections::HashMap; @@ -691,18 +595,14 @@ mod tests { }; use fabro_vault::{SecretType, Vault}; use futures::stream; - use httpmock::Method::POST; + use httpmock::Method::{GET, POST}; use httpmock::MockServer; use object_store::memory::InMemory; use tokio::sync::RwLock as AsyncRwLock; - use tokio_util::sync::CancellationToken; use super::*; use crate::event::{Event, append_event}; - use crate::outcome::Outcome; use crate::records::StageSummary; - use crate::run_options::{GitCheckpointOptions, RunOptions}; - use crate::services::EngineServices; struct MockProvider { name: String, @@ -917,48 +817,6 @@ mod tests { } } - #[tokio::test] - async fn pull_request_omits_pushed_branch_when_run_branch_push_disabled() { - let temp = tempfile::tempdir().unwrap(); - let mut settings = WorkflowSettings::default(); - settings.run.run_branch.push = false; - let run_options = RunOptions { - settings, - run_dir: temp.path().to_path_buf(), - cancel_token: CancellationToken::new(), - run_id: fixtures::RUN_1, - labels: HashMap::new(), - workflow_slug: None, - github_app: None, - pre_run_git: None, - fork_source_ref: None, - base_branch: None, - display_base_sha: None, - git: Some(GitCheckpointOptions { - base_sha: None, - run_branch: Some("fabro/run/test".to_string()), - meta_branch: None, - }), - }; - let concluded = Concluded { - outcome: Ok(Outcome::success()), - conclusion: make_test_conclusion(), - graph: Graph::new("test"), - run_options, - services: EngineServices::test_default().run, - }; - - let finalized = pull_request(concluded, &PullRequestOptions { - pr_config: None, - github_app: None, - origin_url: None, - model: "test-model".to_string(), - }) - .await; - - assert_eq!(finalized.pushed_branch, None); - } - // ── format_arc_details_section tests ──────────────────────────────── #[test] @@ -1555,20 +1413,21 @@ mod tests { }); let base_url = github_app::github_api_base_url(); let result = maybe_open_pull_request(OpenPullRequestRequest { - github: github_app::GitHubContext::new(&creds, &base_url), - origin_url: "https://github.com/owner/repo.git", - base_branch: "main", - head_branch: "fabro/run/123", - goal: "Fix bug", - diff: "", - model: "claude-sonnet-4-20250514", - draft: false, - auto_merge: None, - run_store: &run_store_handle, - llm_source: llm_source.as_ref(), - catalog: test_catalog(), - conclusion: None, - run_state: None, + github: github_app::GitHubContext::new(&creds, &base_url), + origin_url: "https://github.com/owner/repo.git", + base_branch: "main", + head_branch: "fabro/run/123", + expected_head_sha: "final-sha", + goal: "Fix bug", + diff: "", + model: "claude-sonnet-4-20250514", + draft: false, + auto_merge: None, + run_store: &run_store_handle, + llm_source: llm_source.as_ref(), + catalog: test_catalog(), + conclusion: None, + run_state: None, }) .await; assert!(result.is_ok()); @@ -1576,79 +1435,41 @@ mod tests { } #[tokio::test] - async fn load_pull_request_diff_uses_store_without_disk_patch() { - let tmp = tempfile::tempdir().unwrap(); - let store = test_store(); - let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); - let run_spec = RunSpec { - run_id: fixtures::RUN_1, - settings: fabro_types::WorkflowSettings::default(), - graph: Graph::new("test"), - graph_source: None, - workflow_slug: None, - automation: None, - source_directory: Some(tmp.path().display().to_string()), - git: None, - labels: std::collections::HashMap::new(), - provenance: test_support::test_run_provenance(), - manifest_blob: None, - definition_blob: None, - fork_source_ref: None, - }; - append_event(&run_store, &fixtures::RUN_1, &Event::RunCreated { - run_id: fixtures::RUN_1, - title: None, - settings: serde_json::to_value(&run_spec.settings).unwrap(), - graph: serde_json::to_value(&run_spec.graph).unwrap(), - workflow_source: None, - workflow_config: None, - labels: run_spec.labels.clone().into_iter().collect(), - run_dir: tmp.path().display().to_string(), - source_directory: run_spec.source_directory.clone(), - workflow_slug: None, - automation: None, - db_prefix: None, - provenance: run_spec.provenance.clone(), - manifest_blob: None, - git: None, - fork_source_ref: None, - retried_from: None, - parent_id: None, - web_url: None, + async fn stale_remote_branch_is_rejected_before_pull_request_creation() { + let payload = pr_content_json("Fix bug", "Narrative."); + let harness = setup_fallback_test_harness_with_branch_sha(&payload, "stale-sha").await; + let github_base_url = harness.github_server.url(""); + let error = maybe_open_pull_request(OpenPullRequestRequest { + github: fabro_github::GitHubContext::new(&harness.creds, &github_base_url), + origin_url: "https://github.com/owner/repo.git", + base_branch: "main", + head_branch: "fabro/run/123", + expected_head_sha: "final-sha", + goal: "Fix bug", + diff: "diff --git a/src/lib.rs b/src/lib.rs\n+fn x() {}\n", + model: "claude-sonnet-4-20250514", + draft: false, + auto_merge: None, + run_store: &harness.run_store, + llm_source: harness.llm_source.as_ref(), + catalog: harness.catalog.clone(), + conclusion: None, + run_state: None, }) .await - .unwrap(); - append_event(&run_store, &fixtures::RUN_1, &Event::RunRunnable { - source: fabro_types::RunRunnableSource::StartRequested, - actor: None, - }) - .await - .unwrap(); - append_event(&run_store, &fixtures::RUN_1, &Event::RunStarting) - .await - .unwrap(); - append_event(&run_store, &fixtures::RUN_1, &Event::RunRunning) - .await - .unwrap(); - append_event(&run_store, &fixtures::RUN_1, &Event::WorkflowRunCompleted { - timing: fabro_types::RunTiming::wall_only(1), - artifact_count: 0, - status: "succeeded".to_string(), - reason: SuccessReason::Completed, - total_usd_micros: None, - final_git_commit_sha: None, - final_patch: Some( - "diff --git a/src/lib.rs b/src/lib.rs\n+fn from_store() {}\n".to_string(), - ), - diff_summary: None, - billing: None, - }) - .await - .unwrap(); + .expect_err("stale remote branch must prevent PR creation"); - let diff = load_pull_request_diff(&run_store.clone().into()).await; - - assert!(diff.contains("from_store")); + assert!(error.contains("stale-sha")); + assert!(error.contains("final-sha")); + httpmock::Mock::new(harness.openai_mock_id, &harness.openai_server) + .assert_async() + .await; + httpmock::Mock::new(harness.branch_mock_id, &harness.github_server) + .assert_async() + .await; + httpmock::Mock::new(harness.github_mock_id, &harness.github_server) + .assert_calls_async(0) + .await; } // ── Structured-output PR content tests ────────────────────────────── @@ -1807,6 +1628,7 @@ mod tests { openai_server: MockServer, github_server: MockServer, openai_mock_id: usize, + branch_mock_id: usize, github_mock_id: usize, llm_source: Arc, catalog: Arc, @@ -1819,6 +1641,9 @@ mod tests { httpmock::Mock::new(self.openai_mock_id, &self.openai_server) .assert_async() .await; + httpmock::Mock::new(self.branch_mock_id, &self.github_server) + .assert_async() + .await; httpmock::Mock::new(self.github_mock_id, &self.github_server) .assert_async() .await; @@ -1830,6 +1655,13 @@ mod tests { /// credential source, and a run store seeded with a non-empty /// `final_patch`. async fn setup_fallback_test_harness(openai_payload_text: &str) -> FallbackHarness { + setup_fallback_test_harness_with_branch_sha(openai_payload_text, "final-sha").await + } + + async fn setup_fallback_test_harness_with_branch_sha( + openai_payload_text: &str, + branch_sha: &str, + ) -> FallbackHarness { let openai_server = MockServer::start_async().await; let openai_mock = openai_server .mock_async(|when, then| { @@ -1843,6 +1675,19 @@ mod tests { .await; let github_server = MockServer::start_async().await; + let branch_sha = branch_sha.to_string(); + let branch_mock = github_server + .mock_async(move |when, then| { + when.method(GET) + .path("/repos/owner/repo/branches/fabro/run/123") + .header("authorization", "Bearer test-token"); + then.status(200) + .header("content-type", "application/json") + .json_body(serde_json::json!({ + "commit": { "sha": branch_sha } + })); + }) + .await; let github_mock = github_server .mock_async(|when, then| { when.method(POST) @@ -1878,8 +1723,7 @@ mod tests { let store = test_store(); let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); - // Seed a non-empty `final_patch` so `load_pull_request_diff` returns - // diff content and the early-return for empty diffs does not fire. + // Seed a completed run so the PR body can include run details. let run_spec = RunSpec { run_id: fixtures::RUN_1, settings: fabro_types::WorkflowSettings::default(), @@ -1947,6 +1791,7 @@ mod tests { .unwrap(); let openai_mock_id = openai_mock.id; + let branch_mock_id = branch_mock.id; let github_mock_id = github_mock.id; FallbackHarness { @@ -1954,6 +1799,7 @@ mod tests { openai_server, github_server, openai_mock_id, + branch_mock_id, github_mock_id, llm_source, catalog, @@ -1978,6 +1824,7 @@ mod tests { origin_url: "https://github.com/owner/repo.git", base_branch: "main", head_branch: "fabro/run/123", + expected_head_sha: "final-sha", goal: "Fix telemetry leak\n\ndetails...", diff: "diff --git a/src/lib.rs b/src/lib.rs\n+fn x() {}\n", model: "gpt-5.4", @@ -2015,6 +1862,7 @@ mod tests { origin_url: "https://github.com/owner/repo.git", base_branch: "main", head_branch: "fabro/run/123", + expected_head_sha: "final-sha", goal: &goal, diff: "diff --git a/src/lib.rs b/src/lib.rs\n+fn x() {}\n", model: "gpt-5.4", diff --git a/lib/components/fabro-workflow/src/pipeline/types.rs b/lib/components/fabro-workflow/src/pipeline/types.rs index 425cba9aa..c7cfcc815 100644 --- a/lib/components/fabro-workflow/src/pipeline/types.rs +++ b/lib/components/fabro-workflow/src/pipeline/types.rs @@ -337,17 +337,59 @@ pub struct Executed { pub model: String, } -/// Output of the FINALIZE phase. +/// Output of the CONCLUDE phase. #[non_exhaustive] pub struct Concluded { - pub outcome: Result, - pub conclusion: Conclusion, - pub graph: Graph, - pub run_options: RunOptions, - pub services: Arc, + pub outcome: Result, + pub conclusion: Conclusion, + pub artifact_count: usize, + pub graph: Graph, + pub run_options: RunOptions, + pub services: Arc, } -/// Output of the PULL_REQUEST phase. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PublishOutcome { + NotRequested, + NoChanges { + pushed_branch: String, + }, + Published { + pushed_branch: String, + pr_url: Option, + }, +} + +impl PublishOutcome { + pub fn pushed_branch(&self) -> Option<&str> { + match self { + Self::NotRequested => None, + Self::NoChanges { pushed_branch } | Self::Published { pushed_branch, .. } => { + Some(pushed_branch) + } + } + } + + pub fn pr_url(&self) -> Option<&str> { + match self { + Self::Published { pr_url, .. } => pr_url.as_deref(), + Self::NotRequested | Self::NoChanges { .. } => None, + } + } +} + +/// Output of the PUBLISH phase. +#[non_exhaustive] +pub struct Published { + pub execution_outcome: Result, + pub publish_outcome: Result, + pub conclusion: Conclusion, + pub artifact_count: usize, + pub run_options: RunOptions, + pub services: Arc, +} + +/// Output of the FINALIZE phase. #[non_exhaustive] pub struct Finalized { pub run_id: RunId, @@ -383,8 +425,8 @@ pub struct FinalizeOptions { pub last_git_sha: Option, } -/// Options for the PULL_REQUEST phase. -pub struct PullRequestOptions { +/// Options for the PUBLISH phase. +pub struct PublishOptions { pub pr_config: Option, pub github_app: Option, pub origin_url: Option, diff --git a/lib/foundation/fabro-api/tests/status_round_trip.rs b/lib/foundation/fabro-api/tests/status_round_trip.rs index efaa5e7b0..7308b26fc 100644 --- a/lib/foundation/fabro-api/tests/status_round_trip.rs +++ b/lib/foundation/fabro-api/tests/status_round_trip.rs @@ -113,6 +113,7 @@ fn success_reason_json_tokens_match_openapi() { #[test] fn failure_reason_json_tokens_match_openapi() { assert_string_json(FailureReason::WorkflowError, "workflow_error"); + assert_string_json(FailureReason::PublishFailed, "publish_failed"); assert_string_json(FailureReason::Cancelled, "cancelled"); assert_string_json(FailureReason::ApprovalDenied, "approval_denied"); assert_string_json(FailureReason::Terminated, "terminated"); diff --git a/lib/foundation/fabro-types/src/run_event/misc.rs b/lib/foundation/fabro-types/src/run_event/misc.rs index d3bf9c1c0..41bf2cbbb 100644 --- a/lib/foundation/fabro-types/src/run_event/misc.rs +++ b/lib/foundation/fabro-types/src/run_event/misc.rs @@ -247,6 +247,8 @@ pub struct PullRequestCreatedProps { pub repo: String, pub base_branch: String, pub head_branch: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub head_sha: Option, pub title: String, pub draft: bool, } diff --git a/lib/foundation/fabro-types/src/status.rs b/lib/foundation/fabro-types/src/status.rs index 770f6aa5d..2224d0708 100644 --- a/lib/foundation/fabro-types/src/status.rs +++ b/lib/foundation/fabro-types/src/status.rs @@ -290,6 +290,7 @@ pub enum SuccessReason { #[strum(serialize_all = "snake_case")] pub enum FailureReason { WorkflowError, + PublishFailed, Cancelled, ApprovalDenied, Terminated, diff --git a/lib/packages/fabro-api-client/src/models/failure-reason.ts b/lib/packages/fabro-api-client/src/models/failure-reason.ts index b11a85248..79172e887 100644 --- a/lib/packages/fabro-api-client/src/models/failure-reason.ts +++ b/lib/packages/fabro-api-client/src/models/failure-reason.ts @@ -20,6 +20,7 @@ export const FailureReason = { WORKFLOW_ERROR: 'workflow_error', + PUBLISH_FAILED: 'publish_failed', CANCELLED: 'cancelled', APPROVAL_DENIED: 'approval_denied', TERMINATED: 'terminated', From 73f48eeddb31f9fa83438a0da3cd078a774d8b5d Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 28 Jul 2026 15:09:10 -0400 Subject: [PATCH 2/4] refactor: simplify publish pipeline and collapse duplicated stage errors Follow-up cleanup on the publish-failures change. Error model: - Collapse `Error::{Engine, Publish, Handler}` into one `Error::Stage` with an `ErrorStage` discriminator. The three shared a field shape and had to be edited together in four match groups; nine near-identical constructors become two private helpers. - Add `Error::failure_reason()`, replacing the same error -> FailureReason mapping written out in four places. - Publish errors are now terminal. Publish runs once, after execution, so no caller could ever act on the retryable classification. Publish phase: - Fix: a branch that was pushed is now still reported when pull request creation fails afterwards. `PublishOutcome` records what happened and carries the error separately, instead of hiding both behind a `Result`. - Drop `PublishOutcome::NoChanges`, which no consumer distinguished from `Published { pr_url: None }`. - Move publish onto `Concluded` as methods and replace three near-identical precondition guards with one `publish_target()`. Pull requests: - `maybe_open_pull_request` -> `open_pull_request` returning the record directly. Both callers already reject empty diffs, so the `Ok(None)` path was unreachable. - Drop `CreatedPullRequest.head_sha`, which echoed back its own input. GitHub client: - Delete `branch_exists`, which had no callers and duplicated `branch_head_sha`. Give `branch_head_sha` the `_with_client` split every sibling has and port the tests to `MockHttpClient`. - Collapse the copy-pasted credential match in `resolve_clone_credentials`. Events: - `PullRequestCreated.head_sha` is `Option` instead of using an empty string to mean absent. - Centralize the run-branch refspec in `lifecycle::git::push_run_branch`, so `git.push` reports a branch name from both emitters as documented. Co-Authored-By: Claude Opus 5 (1M context) --- Cargo.lock | 1 + .../src/commands/run/run_progress/event.rs | 2 +- .../src/commands/run/run_progress/mod.rs | 2 +- lib/apps/fabro-server/src/server.rs | 48 +-- .../src/server/handler/pull_requests.rs | 13 +- lib/apps/fabro-server/src/server/tests.rs | 6 +- lib/components/fabro-github/src/lib.rs | 252 ++++++--------- lib/components/fabro-workflow/Cargo.toml | 1 + lib/components/fabro-workflow/src/error.rs | 228 +++++++------- .../fabro-workflow/src/event/convert.rs | 2 +- .../fabro-workflow/src/event/events.rs | 7 +- .../fabro-workflow/src/lifecycle/event.rs | 2 +- .../fabro-workflow/src/lifecycle/git.rs | 22 +- .../fabro-workflow/src/operations/start.rs | 14 +- .../fabro-workflow/src/pipeline/finalize.rs | 171 ++++++---- .../fabro-workflow/src/pipeline/mod.rs | 2 +- .../fabro-workflow/src/pipeline/publish.rs | 294 +++++++++--------- .../src/pipeline/pull_request.rs | 87 ++---- .../fabro-workflow/src/pipeline/types.rs | 40 +-- .../fabro-workflow/src/pull_request.rs | 2 +- 20 files changed, 535 insertions(+), 661 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e71612fb2..2b5fa3d2a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3393,6 +3393,7 @@ dependencies = [ "serde_json", "sha2 0.10.9", "shlex", + "strum 0.28.0", "tempfile", "thiserror 2.0.18", "tokio", diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs index d53a591ff..af050d073 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs @@ -614,7 +614,7 @@ mod tests { repo: "widgets".into(), base_branch: "main".into(), head_branch: "fabro/run/42".into(), - head_sha: "final-sha".into(), + head_sha: Some("final-sha".to_string()), title: "Ship the server-side PR".into(), draft: true, }; diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs index 4b7a04cbf..38ba83bab 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs @@ -1383,7 +1383,7 @@ mod tests { repo: "fabro".into(), base_branch: "main".into(), head_branch: "fabro/run/42".into(), - head_sha: "final-sha".into(), + head_sha: Some("final-sha".to_string()), title: "Ship the change".into(), draft: true, }); diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index c3d3113af..0fc2cd014 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -4169,9 +4169,15 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { let mut runs = state.runs.lock().expect("runs lock poisoned"); if let Some(managed_run) = runs.get_mut(&run_id) { match &result { - ExecutionResult::Completed(result) => match result.as_ref() { - Ok(started) => match &started.finalized.outcome { - Ok(_) => { + ExecutionResult::Completed(result) => { + // A run can fail either before it produces a `Started` or in + // its own outcome; both carry the same `WorkflowError`. + let outcome = match result.as_ref() { + Ok(started) => started.finalized.outcome.as_ref().map(|_| ()), + Err(e) => Err(e), + }; + match outcome { + Ok(()) => { info!(run_id = %run_id, "Run completed"); managed_run.status = RunStatus::Succeeded { reason: SuccessReason::Completed, @@ -4183,44 +4189,16 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { reason: FailureReason::Cancelled, }; } - Err(e @ WorkflowError::Publish { .. }) => { + Err(e) => { let detail = e.display_with_causes(); - error!(run_id = %run_id, error = %detail, "Run publish failed"); + error!(run_id = %run_id, error = %detail, "Run failed"); managed_run.status = RunStatus::Failed { - reason: FailureReason::PublishFailed, + reason: e.failure_reason(), }; managed_run.error = Some(detail); } - Err(e) => { - error!(run_id = %run_id, error = %e, "Run failed"); - managed_run.status = RunStatus::Failed { - reason: FailureReason::WorkflowError, - }; - managed_run.error = Some(e.to_string()); - } - }, - Err(WorkflowError::Cancelled) => { - info!(run_id = %run_id, "Run cancelled"); - managed_run.status = RunStatus::Failed { - reason: FailureReason::Cancelled, - }; } - Err(e @ WorkflowError::Publish { .. }) => { - let detail = e.display_with_causes(); - error!(run_id = %run_id, error = %detail, "Run publish failed"); - managed_run.status = RunStatus::Failed { - reason: FailureReason::PublishFailed, - }; - managed_run.error = Some(detail); - } - Err(e) => { - error!(run_id = %run_id, error = %e, "Run failed"); - managed_run.status = RunStatus::Failed { - reason: FailureReason::WorkflowError, - }; - managed_run.error = Some(e.to_string()); - } - }, + } ExecutionResult::CancelledBySignal => { info!(run_id = %run_id, "Run cancelled"); managed_run.status = RunStatus::Failed { 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 a30a55c03..8bf5fa898 100644 --- a/lib/apps/fabro-server/src/server/handler/pull_requests.rs +++ b/lib/apps/fabro-server/src/server/handler/pull_requests.rs @@ -348,15 +348,8 @@ async fn create_run_pull_request( conclusion: Some(inputs.conclusion), run_state: Some(run_state), }; - let created_pull_request = match pull_request::maybe_open_pull_request(request).await { - Ok(Some(created)) => created, - Ok(None) => { - return ApiError::new( - StatusCode::INTERNAL_SERVER_ERROR, - "Pull request creation returned no record unexpectedly.", - ) - .into_response(); - } + let created_pull_request = match pull_request::open_pull_request(request).await { + Ok(created) => created, Err(err) => return ApiError::new(StatusCode::BAD_GATEWAY, err).into_response(), }; @@ -364,7 +357,7 @@ async fn create_run_pull_request( &created_pull_request.link, &created_pull_request.base_branch, &created_pull_request.head_branch, - &created_pull_request.head_sha, + inputs.final_git_sha, &created_pull_request.title, true, ); diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 73c9fcd34..40e7461e4 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -4685,7 +4685,7 @@ channel = "#deploys" repo: "fabro".to_string(), base_branch: "main".to_string(), head_branch: "fabro/run/test".to_string(), - head_sha: "final-sha".to_string(), + head_sha: Some("final-sha".to_string()), title: "Ship & notify".to_string(), draft: false, }, @@ -6426,7 +6426,7 @@ async fn create_run_with_pull_request_record( repo: "widgets".to_string(), base_branch: "main".to_string(), head_branch: "feature".to_string(), - head_sha: "final-sha".to_string(), + head_sha: Some("final-sha".to_string()), title: title.to_string(), draft: false, }, @@ -16488,7 +16488,7 @@ async fn list_runs_includes_live_metadata_from_run_state() { repo: "repo".to_string(), base_branch: "main".to_string(), head_branch: "fabro/run".to_string(), - head_sha: "final-sha".to_string(), + head_sha: Some("final-sha".to_string()), title: "Fix board metadata".to_string(), draft: false, }, diff --git a/lib/components/fabro-github/src/lib.rs b/lib/components/fabro-github/src/lib.rs index d150c4759..df45eca9d 100644 --- a/lib/components/fabro-github/src/lib.rs +++ b/lib/components/fabro-github/src/lib.rs @@ -888,20 +888,6 @@ fn normalize_https_host_path(url: &str) -> String { } } -/// Check whether a branch exists in a GitHub repository. -/// -/// Uses a GitHub App installation token to query the branches API. -/// Returns `true` if the branch exists, `false` if it doesn't (404). -pub async fn branch_exists( - ctx: &GitHubContext<'_>, - owner: &str, - repo: &str, - branch: &str, -) -> anyhow::Result { - let client = ctx.http_client()?; - branch_exists_with_client(&client, ctx, owner, repo, branch).await -} - /// Return the commit SHA at the head of a GitHub branch. /// /// Returns `None` when the branch does not exist. @@ -910,6 +896,17 @@ pub async fn branch_head_sha( owner: &str, repo: &str, branch: &str, +) -> anyhow::Result> { + let client = ctx.http_client()?; + branch_head_sha_with_client(&client, ctx, owner, repo, branch).await +} + +async fn branch_head_sha_with_client( + client: &impl HttpClient, + ctx: &GitHubContext<'_>, + owner: &str, + repo: &str, + branch: &str, ) -> anyhow::Result> { #[derive(Deserialize)] struct BranchResponse { @@ -921,11 +918,10 @@ pub async fn branch_head_sha( sha: String, } - let client = ctx.http_client()?; let token = ctx .creds .resolve_bearer_token( - &client, + client, owner, repo, ctx.base_url, @@ -935,7 +931,8 @@ pub async fn branch_head_sha( let url = format!("{}/repos/{owner}/{repo}/branches/{branch}", ctx.base_url); let auth = format!("Bearer {token}"); - let resp = HttpClient::request(&client, HttpMethod::Get, &url, &github_headers(&auth), None) + let resp = client + .request(HttpMethod::Get, &url, &github_headers(&auth), None) .await .context("Failed to read remote branch head")?; @@ -951,38 +948,6 @@ pub async fn branch_head_sha( } } -async fn branch_exists_with_client( - client: &impl HttpClient, - ctx: &GitHubContext<'_>, - owner: &str, - repo: &str, - branch: &str, -) -> anyhow::Result { - let token = ctx - .creds - .resolve_bearer_token( - client, - owner, - repo, - ctx.base_url, - serde_json::json!({ "contents": "write" }), - ) - .await?; - - let url = format!("{}/repos/{owner}/{repo}/branches/{branch}", ctx.base_url); - let auth = format!("Bearer {token}"); - let resp = client - .request(HttpMethod::Get, &url, &github_headers(&auth), None) - .await - .context("Failed to check branch existence")?; - - match resp.status { - 200 => Ok(true), - 404 => Ok(false), - status => bail!("Unexpected status {status} checking branch '{branch}'"), - } -} - /// Check whether a GitHub App is installed for a specific repository. /// /// Uses the App JWT to query `GET /repos/{owner}/{repo}/installation`. @@ -1092,46 +1057,36 @@ pub async fn resolve_clone_credentials( ctx: &GitHubContext<'_>, owner: &str, repo: &str, -) -> anyhow::Result<(Option, Option)> { - match ctx.creds { - GitHubCredentials::Pat(token) => { - Ok((Some("x-access-token".to_string()), Some(token.clone()))) - } - GitHubCredentials::Installation(token) => Ok(( - Some("x-access-token".to_string()), - Some(token.valid_token()?.to_string()), - )), - GitHubCredentials::App(_) => { - let client = ctx.http_client()?; - resolve_clone_credentials_with_client(&client, ctx, owner, repo).await - } - } -} - -async fn resolve_clone_credentials_with_client( - client: &impl HttpClient, - ctx: &GitHubContext<'_>, - owner: &str, - repo: &str, ) -> anyhow::Result<(Option, Option)> { let token = match ctx.creds { GitHubCredentials::Pat(token) => token.clone(), GitHubCredentials::Installation(token) => token.valid_token()?.to_string(), GitHubCredentials::App(_) => { - ctx.creds - .resolve_bearer_token( - client, - owner, - repo, - ctx.base_url, - serde_json::json!({ "contents": "write", "workflows": "write" }), - ) - .await? + let client = ctx.http_client()?; + mint_git_write_token(&client, ctx, owner, repo).await? } }; Ok((Some("x-access-token".to_string()), Some(token))) } +/// Mint an installation token scoped for git writes, including workflow files. +async fn mint_git_write_token( + client: &impl HttpClient, + ctx: &GitHubContext<'_>, + owner: &str, + repo: &str, +) -> anyhow::Result { + ctx.creds + .resolve_bearer_token( + client, + owner, + repo, + ctx.base_url, + serde_json::json!({ "contents": "write", "workflows": "write" }), + ) + .await +} + /// Embed a token into an HTTPS URL for authenticated git operations. /// /// Converts `https://github.com/owner/repo` to @@ -1983,12 +1938,21 @@ mod tests { } // ----------------------------------------------------------------------- - // branch_exists + // branch_head_sha // ----------------------------------------------------------------------- - #[tokio::test] - async fn branch_exists_returns_true_on_200() { - let mock = MockHttpClient::new() + fn app_creds() -> GitHubCredentials { + GitHubCredentials::App(GitHubAppCredentials { + app_id: "test".to_string(), + private_key_pem: test_rsa_key().to_string(), + slug: None, + }) + } + + /// Mock the installation-token exchange every App-credentialed call makes + /// before it reaches the endpoint under test. + fn mock_with_installation_token() -> MockHttpClient { + MockHttpClient::new() .on( HttpMethod::Get, "/repos/owner/repo/installation", @@ -2001,20 +1965,19 @@ mod tests { 201, r#"{"token": "ghs_test", "expires_at": "2099-01-01T00:00:00Z"}"#, ) - .on( - HttpMethod::Get, - "/repos/owner/repo/branches/my-branch", - 200, - r#"{"name": "my-branch"}"#, - ); + } - let pem = test_rsa_key(); - let creds = GitHubCredentials::App(GitHubAppCredentials { - app_id: "test".to_string(), - private_key_pem: pem.to_string(), - slug: None, - }); - let result = branch_exists_with_client( + #[tokio::test] + async fn branch_head_sha_returns_commit_on_200() { + let mock = mock_with_installation_token().on( + HttpMethod::Get, + "/repos/owner/repo/branches/my-branch", + 200, + r#"{"name": "my-branch", "commit": {"sha": "abc123"}}"#, + ); + + let creds = app_creds(); + let result = branch_head_sha_with_client( &mock, &GitHubContext::new(&creds, ""), "owner", @@ -2022,38 +1985,21 @@ mod tests { "my-branch", ) .await; - assert!(result.unwrap()); + + assert_eq!(result.unwrap(), Some("abc123".to_string())); } #[tokio::test] - async fn branch_exists_returns_false_on_404() { - let mock = MockHttpClient::new() - .on( - HttpMethod::Get, - "/repos/owner/repo/installation", - 200, - r#"{"id": 1}"#, - ) - .on( - HttpMethod::Post, - "/app/installations/1/access_tokens", - 201, - r#"{"token": "ghs_test", "expires_at": "2099-01-01T00:00:00Z"}"#, - ) - .on( - HttpMethod::Get, - "/repos/owner/repo/branches/no-such-branch", - 404, - "", - ); + async fn branch_head_sha_returns_none_on_404() { + let mock = mock_with_installation_token().on( + HttpMethod::Get, + "/repos/owner/repo/branches/no-such-branch", + 404, + "", + ); - let pem = test_rsa_key(); - let creds = GitHubCredentials::App(GitHubAppCredentials { - app_id: "test".to_string(), - private_key_pem: pem.to_string(), - slug: None, - }); - let result = branch_exists_with_client( + let creds = app_creds(); + let result = branch_head_sha_with_client( &mock, &GitHubContext::new(&creds, ""), "owner", @@ -2061,38 +2007,21 @@ mod tests { "no-such-branch", ) .await; - assert!(!result.unwrap()); + + assert_eq!(result.unwrap(), None); } #[tokio::test] - async fn branch_exists_returns_error_on_500() { - let mock = MockHttpClient::new() - .on( - HttpMethod::Get, - "/repos/owner/repo/installation", - 200, - r#"{"id": 1}"#, - ) - .on( - HttpMethod::Post, - "/app/installations/1/access_tokens", - 201, - r#"{"token": "ghs_test", "expires_at": "2099-01-01T00:00:00Z"}"#, - ) - .on( - HttpMethod::Get, - "/repos/owner/repo/branches/broken", - 500, - "", - ); + async fn branch_head_sha_returns_error_on_500() { + let mock = mock_with_installation_token().on( + HttpMethod::Get, + "/repos/owner/repo/branches/broken", + 500, + "", + ); - let pem = test_rsa_key(); - let creds = GitHubCredentials::App(GitHubAppCredentials { - app_id: "test".to_string(), - private_key_pem: pem.to_string(), - slug: None, - }); - let result = branch_exists_with_client( + let creds = app_creds(); + let result = branch_head_sha_with_client( &mock, &GitHubContext::new(&creds, ""), "owner", @@ -2100,22 +2029,23 @@ mod tests { "broken", ) .await; + assert!(result.is_err()); } #[tokio::test] - async fn branch_exists_with_token_uses_direct_bearer_token() { + async fn branch_head_sha_with_token_uses_direct_bearer_token() { let mock = MockHttpClient::new() .on( HttpMethod::Get, "/repos/owner/repo/branches/my-branch", 200, - r#"{"name": "my-branch"}"#, + r#"{"name": "my-branch", "commit": {"sha": "abc123"}}"#, ) .with_req_header("Authorization", "Bearer ghu_test"); let creds = GitHubCredentials::Pat("ghu_test".to_string()); - let result = branch_exists_with_client( + let result = branch_head_sha_with_client( &mock, &GitHubContext::new(&creds, ""), "owner", @@ -2124,7 +2054,7 @@ mod tests { ) .await; - assert!(result.unwrap()); + assert_eq!(result.unwrap(), Some("abc123".to_string())); } // ----------------------------------------------------------------------- @@ -2395,17 +2325,11 @@ mod tests { slug: None, }); let context = GitHubContext::new(&credentials, ""); - let resolved = resolve_clone_credentials_with_client(&mock, &context, "owner", "repo") + let token = mint_git_write_token(&mock, &context, "owner", "repo") .await .unwrap(); - assert_eq!( - resolved, - ( - Some("x-access-token".to_string()), - Some("ghs_xxx".to_string()) - ) - ); + assert_eq!(token, "ghs_xxx"); } #[test] diff --git a/lib/components/fabro-workflow/Cargo.toml b/lib/components/fabro-workflow/Cargo.toml index 932bb0071..535a9f6b5 100644 --- a/lib/components/fabro-workflow/Cargo.toml +++ b/lib/components/fabro-workflow/Cargo.toml @@ -46,6 +46,7 @@ fabro-static.workspace = true fabro-types = { path = "../../foundation/fabro-types" } fabro-http.workspace = true thiserror.workspace = true +strum.workspace = true serde.workspace = true serde_json.workspace = true jsonschema.workspace = true diff --git a/lib/components/fabro-workflow/src/error.rs b/lib/components/fabro-workflow/src/error.rs index c134a5e70..ac7eed9c5 100644 --- a/lib/components/fabro-workflow/src/error.rs +++ b/lib/components/fabro-workflow/src/error.rs @@ -252,6 +252,21 @@ impl FailureSignatureExt for FailureSignature { } } +/// Pipeline stage that produced an [`Error::Stage`]. +/// +/// The three stages share a failure shape — a message, an eagerly classified +/// [`FailureCategory`], an optional command output tail, and an optional +/// source — and differ only in where they run and whether a retry is possible. +#[derive(Debug, Clone, Copy, PartialEq, Eq, strum::Display)] +pub enum ErrorStage { + /// A node handler failed. Retryable: the engine can re-run the node. + Handler, + /// The engine itself failed while driving the graph. Retryable. + Engine, + /// The publish stage failed. Terminal: publish runs once, after execution. + Publish, +} + #[derive(ThisError, Debug, Clone)] pub enum Error { #[error("Parse error: {0}")] @@ -276,26 +291,9 @@ pub enum Error { source: SharedTemplateError, }, - #[error("Engine error: {message}")] - Engine { - message: String, - failure_class: FailureCategory, - exec_output_tail: Option, - #[source] - source: Option, - }, - - #[error("Publish error: {message}")] - Publish { - message: String, - failure_class: FailureCategory, - exec_output_tail: Option, - #[source] - source: Option, - }, - - #[error("Handler error: {message}")] - Handler { + #[error("{stage} error: {message}")] + Stage { + stage: ErrorStage, message: String, failure_class: FailureCategory, exec_output_tail: Option, @@ -334,17 +332,48 @@ pub enum Error { impl Error { /// Smart constructor for Handler errors. Classifies the failure reason /// eagerly. - pub fn handler(message: impl Into) -> Self { + /// Build a stage error, classifying the message eagerly. + fn stage( + stage: ErrorStage, + message: impl Into, + exec_output_tail: Option, + ) -> Self { let message = message.into(); let failure_class = classify_failure_reason(&message); - Self::Handler { + Self::Stage { + stage, message, failure_class, - exec_output_tail: None, + exec_output_tail, source: None, } } + /// Build a stage error from a source, classifying the rendered chain so + /// hints buried in the causes still reach [`Self::failure_category`]. + fn stage_with_source( + stage: ErrorStage, + message: impl Into, + source: impl Into, + exec_output_tail: Option, + ) -> Self { + let message = message.into(); + let source = SharedError::new(source.into()); + let failure_class = + classify_failure_reason(&render_with_causes(&message, &collect_chain(&source))); + Self::Stage { + stage, + message, + failure_class, + exec_output_tail, + source: Some(source), + } + } + + pub fn handler(message: impl Into) -> Self { + Self::stage(ErrorStage::Handler, message, None) + } + pub fn template(message: impl Into, source: TemplateError) -> Self { Self::Template { message: message.into(), @@ -356,14 +385,7 @@ impl Error { message: impl Into, exec_output_tail: Option, ) -> Self { - let message = message.into(); - let failure_class = classify_failure_reason(&message); - Self::Handler { - message, - failure_class, - exec_output_tail, - source: None, - } + Self::stage(ErrorStage::Handler, message, exec_output_tail) } pub fn handler_with_source( @@ -378,51 +400,22 @@ impl Error { source: impl Into, exec_output_tail: Option, ) -> Self { - let message = message.into(); - let source = SharedError::new(source.into()); - let causes = collect_chain(&source); - let rendered = render_with_causes(&message, &causes); - let failure_class = classify_failure_reason(&rendered); - Self::Handler { - message, - failure_class, - exec_output_tail, - source: Some(source), - } + Self::stage_with_source(ErrorStage::Handler, message, source, exec_output_tail) } pub fn handler_with_anyhow(message: impl Into, source: anyhow::Error) -> Self { Self::handler_with_source(message, source) } - /// Smart constructor for Engine errors. Classifies the failure reason - /// eagerly. pub fn engine(message: impl Into) -> Self { - let message = message.into(); - let failure_class = classify_failure_reason(&message); - Self::Engine { - message, - failure_class, - exec_output_tail: None, - source: None, - } + Self::stage(ErrorStage::Engine, message, None) } pub fn engine_with_source( message: impl Into, source: impl Into, ) -> Self { - let message = message.into(); - let source = SharedError::new(source.into()); - let causes = collect_chain(&source); - let rendered = render_with_causes(&message, &causes); - let failure_class = classify_failure_reason(&rendered); - Self::Engine { - message, - failure_class, - exec_output_tail: None, - source: Some(source), - } + Self::stage_with_source(ErrorStage::Engine, message, source, None) } pub fn engine_with_anyhow(message: impl Into, source: anyhow::Error) -> Self { @@ -431,14 +424,7 @@ impl Error { /// Build an error for the required publish stage. pub fn publish(message: impl Into) -> Self { - let message = message.into(); - let failure_class = classify_failure_reason(&message); - Self::Publish { - message, - failure_class, - exec_output_tail: None, - source: None, - } + Self::stage(ErrorStage::Publish, message, None) } pub fn publish_with_source( @@ -453,25 +439,13 @@ impl Error { source: impl Into, exec_output_tail: Option, ) -> Self { - let message = message.into(); - let source = SharedError::new(source.into()); - let causes = collect_chain(&source); - let rendered = render_with_causes(&message, &causes); - let failure_class = classify_failure_reason(&rendered); - Self::Publish { - message, - failure_class, - exec_output_tail, - source: Some(source), - } + Self::stage_with_source(ErrorStage::Publish, message, source, exec_output_tail) } #[must_use] pub fn causes(&self) -> Vec { match self { - Self::Engine { source, .. } - | Self::Publish { source, .. } - | Self::Handler { source, .. } => source + Self::Stage { source, .. } => source .as_ref() .map_or_else(Vec::new, |source| collect_chain(source)), Self::Template { source, .. } => collect_chain(source), @@ -487,16 +461,16 @@ impl Error { /// Whether this error category is retryable (transient) or terminal. /// - /// Retryable: Handler and Engine, I/O, LLM errors when the SDK marks them - /// retryable, and Publish errors classified as transient infrastructure - /// failures. Terminal: Parse, Validation, OutputSchemaValidation, - /// Stylesheet, Checkpoint, and Cancelled. + /// Retryable: Handler and Engine stages (the engine can re-run the node), + /// I/O, and LLM errors the SDK marks retryable. Terminal: the Publish + /// stage (it runs once, after execution), Parse, Validation, + /// OutputSchemaValidation, Stylesheet, Checkpoint, and Cancelled. #[must_use] pub fn is_retryable(&self) -> bool { match self { - Self::Handler { .. } | Self::Engine { .. } | Self::Io(_) => true, - Self::Publish { failure_class, .. } => { - matches!(failure_class, FailureCategory::TransientInfra) + Self::Io(_) => true, + Self::Stage { stage, .. } => { + matches!(stage, ErrorStage::Handler | ErrorStage::Engine) } Self::Llm(sdk_err) => sdk_err.retryable(), Self::Parse(_) @@ -533,9 +507,20 @@ impl Error { | Self::Unsupported(_) | Self::OutputSchemaValidation(_) => FailureCategory::Deterministic, Self::Precondition(_) | Self::RunNotFound(_) => FailureCategory::Structural, - Self::Handler { failure_class, .. } - | Self::Engine { failure_class, .. } - | Self::Publish { failure_class, .. } => *failure_class, + Self::Stage { failure_class, .. } => *failure_class, + } + } + + /// The terminal [`FailureReason`] this error maps to on a run. + #[must_use] + pub fn failure_reason(&self) -> FailureReason { + match self { + Self::Cancelled => FailureReason::Cancelled, + Self::Stage { + stage: ErrorStage::Publish, + .. + } => FailureReason::PublishFailed, + _ => FailureReason::WorkflowError, } } @@ -551,23 +536,13 @@ impl Error { #[must_use] pub fn to_failure_detail(&self) -> FailureDetail { - let message = match self { - Self::Engine { message, .. } - | Self::Publish { message, .. } - | Self::Handler { message, .. } => message.clone(), - _ => self.to_string(), - }; - let explicit_exec_output_tail = match self { - Self::Engine { - exec_output_tail, .. - } - | Self::Publish { - exec_output_tail, .. - } - | Self::Handler { - exec_output_tail, .. - } => exec_output_tail.clone(), - _ => None, + let (message, explicit_exec_output_tail) = match self { + Self::Stage { + message, + exec_output_tail, + .. + } => (message.clone(), exec_output_tail.clone()), + _ => (self.to_string(), None), }; FailureDetail { message, @@ -880,7 +855,10 @@ mod tests { source, }); - assert!(matches!(fabro_error, Error::Engine { .. })); + assert!(matches!(fabro_error, Error::Stage { + stage: ErrorStage::Engine, + .. + })); let message = fabro_error.to_string(); assert!(message.contains("deserialize run spec on branch fabro/meta/run-1")); assert!(message.contains(&source_message)); @@ -2061,10 +2039,30 @@ mod tests { ); } + /// Publish runs once, after execution, so no caller can retry it — even + /// when the message looks transient. The failure category is still + /// classified for reporting. #[test] - fn publish_error_is_only_retryable_for_transient_failures() { - assert!(Error::publish("connection timed out").is_retryable()); + fn publish_errors_are_terminal() { + assert!(!Error::publish("connection timed out").is_retryable()); assert!(!Error::publish("permission denied").is_retryable()); + assert_eq!( + Error::publish("connection timed out").failure_category(), + FailureCategory::TransientInfra + ); + } + + #[test] + fn failure_reason_distinguishes_publish_and_cancelled() { + assert_eq!( + Error::publish("nope").failure_reason(), + FailureReason::PublishFailed + ); + assert_eq!(Error::Cancelled.failure_reason(), FailureReason::Cancelled); + assert_eq!( + Error::engine("boom").failure_reason(), + FailureReason::WorkflowError + ); } #[test] diff --git a/lib/components/fabro-workflow/src/event/convert.rs b/lib/components/fabro-workflow/src/event/convert.rs index 42a4ffbc4..094f36dd4 100644 --- a/lib/components/fabro-workflow/src/event/convert.rs +++ b/lib/components/fabro-workflow/src/event/convert.rs @@ -1321,7 +1321,7 @@ fn event_body_from_event(event: &Event) -> EventBody { repo: repo.clone(), base_branch: base_branch.clone(), head_branch: head_branch.clone(), - head_sha: (!head_sha.is_empty()).then(|| head_sha.clone()), + head_sha: head_sha.clone(), title: title.clone(), draft: *draft, }), diff --git a/lib/components/fabro-workflow/src/event/events.rs b/lib/components/fabro-workflow/src/event/events.rs index f512e80c6..4bc7bca4f 100644 --- a/lib/components/fabro-workflow/src/event/events.rs +++ b/lib/components/fabro-workflow/src/event/events.rs @@ -730,8 +730,9 @@ pub enum Event { repo: String, base_branch: String, head_branch: String, - #[serde(default)] - head_sha: String, + /// Absent on events written before the head SHA was recorded. + #[serde(default, skip_serializing_if = "Option::is_none")] + head_sha: Option, title: String, draft: bool, }, @@ -782,7 +783,7 @@ impl Event { repo: record.repo.clone(), base_branch: base_branch.to_string(), head_branch: head_branch.to_string(), - head_sha: head_sha.to_string(), + head_sha: Some(head_sha.to_string()), title: title.to_string(), draft, } diff --git a/lib/components/fabro-workflow/src/lifecycle/event.rs b/lib/components/fabro-workflow/src/lifecycle/event.rs index c4af83250..9a6ca4b0e 100644 --- a/lib/components/fabro-workflow/src/lifecycle/event.rs +++ b/lib/components/fabro-workflow/src/lifecycle/event.rs @@ -496,7 +496,7 @@ impl RunLifecycle for EventLifecycle { } for push in &result.push_results { self.emitter.emit(&Event::GitPush { - branch: push.refspec.clone(), + branch: push.branch.clone(), success: push.success, exec_output_tail: push.exec_output_tail.clone(), }); diff --git a/lib/components/fabro-workflow/src/lifecycle/git.rs b/lib/components/fabro-workflow/src/lifecycle/git.rs index 4db74c904..656a298a7 100644 --- a/lib/components/fabro-workflow/src/lifecycle/git.rs +++ b/lib/components/fabro-workflow/src/lifecycle/git.rs @@ -69,11 +69,24 @@ pub(crate) struct GitCheckpointResult { #[derive(Debug, Clone)] pub(crate) struct PushResult { - pub refspec: String, + pub branch: String, pub success: bool, pub exec_output_tail: Option, } +/// Push a run branch to its remote counterpart. +/// +/// Owns the refspec convention so the checkpoint push and the terminal publish +/// push cannot drift apart. +pub(crate) async fn push_run_branch( + sandbox: &dyn fabro_sandbox::Sandbox, + branch: &str, +) -> fabro_sandbox::Result<()> { + sandbox + .git_push_ref(&format!("refs/heads/{branch}:refs/heads/{branch}")) + .await +} + /// Sub-lifecycle responsible for git operations (checkpoint commits, pushes, /// diffs). pub(crate) struct GitLifecycle { @@ -307,15 +320,14 @@ impl RunLifecycle for GitLifecycle { .as_ref() .and_then(|g| g.run_branch.as_ref()) { - let refspec = format!("refs/heads/{branch}:refs/heads/{branch}"); let (push_ok, exec_output_tail) = - match self.sandbox.git_push_ref(&refspec).await { + match push_run_branch(self.sandbox.as_ref(), branch).await { Ok(()) => (true, None), Err(err) => { let exec_output_tail = fabro_sandbox::default_redacted_output_tail(&err); tracing::warn!( - refspec = %refspec, + branch = %branch, error = %fabro_sandbox::display_for_log(&err), "git push from run lifecycle failed" ); @@ -329,7 +341,7 @@ impl RunLifecycle for GitLifecycle { } }; git_result.push_results.push(PushResult { - refspec, + branch: branch.clone(), success: push_ok, exec_output_tail, }); diff --git a/lib/components/fabro-workflow/src/operations/start.rs b/lib/components/fabro-workflow/src/operations/start.rs index 58c9e1c20..67143dc4b 100644 --- a/lib/components/fabro-workflow/src/operations/start.rs +++ b/lib/components/fabro-workflow/src/operations/start.rs @@ -928,16 +928,12 @@ impl RunSession { model: self.pr_model, }; - let concluded = match Box::pin(pipeline::conclude(executed, &finalize_opts)).await { - Ok(concluded) => concluded, - Err(err) => { - self.steering_hub.drain_pending_at_run_end(); - store_progress_logger.flush().await; - return Err(err); - } + let concluding = async { + let concluded = Box::pin(pipeline::conclude(executed, &finalize_opts)).await?; + let published = Box::pin(pipeline::publish(concluded, &publish_opts)).await; + Box::pin(pipeline::finalize(published, &finalize_opts)).await }; - let published = Box::pin(pipeline::publish(concluded, &publish_opts)).await; - let finalized = match Box::pin(pipeline::finalize(published, &finalize_opts)).await { + let finalized = match concluding.await { Ok(finalized) => finalized, Err(err) => { self.steering_hub.drain_pending_at_run_end(); diff --git a/lib/components/fabro-workflow/src/pipeline/finalize.rs b/lib/components/fabro-workflow/src/pipeline/finalize.rs index 673989a5c..73f01568d 100644 --- a/lib/components/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/components/fabro-workflow/src/pipeline/finalize.rs @@ -9,7 +9,7 @@ use fabro_types::{BilledTokenCounts, DiffSummary, EventBody, RunFailure, RunProj use fabro_util::error::collect_causes; use fabro_util::time::elapsed_ms; -use super::types::{Concluded, Executed, FinalizeOptions, Finalized, Published}; +use super::types::{Concluded, Executed, FinalizeOptions, Finalized, PublishOutcome, Published}; use crate::error::{Error, run_failure_from_error, run_failure_from_outcome_failure}; use crate::event::{Event, RunNoticeCode, RunNoticeLevel}; use crate::outcome::{Outcome, StageOutcome}; @@ -44,36 +44,16 @@ pub fn classify_engine_result( }; (status, failure, run_status) } - Err(Error::Cancelled) => ( - StageOutcome::Failed { - retry_requested: false, - }, - Some(run_failure_from_error( - &Error::Cancelled, - FailureReason::Cancelled, - )), - RunStatus::Failed { - reason: FailureReason::Cancelled, - }, - ), - Err(err @ Error::Publish { .. }) => ( - StageOutcome::Failed { - retry_requested: false, - }, - Some(run_failure_from_error(err, FailureReason::PublishFailed)), - RunStatus::Failed { - reason: FailureReason::PublishFailed, - }, - ), - Err(err) => ( - StageOutcome::Failed { - retry_requested: false, - }, - Some(run_failure_from_error(err, FailureReason::WorkflowError)), - RunStatus::Failed { - reason: FailureReason::WorkflowError, - }, - ), + Err(err) => { + let reason = err.failure_reason(); + ( + StageOutcome::Failed { + retry_requested: false, + }, + Some(run_failure_from_error(err, reason)), + RunStatus::Failed { reason }, + ) + } } } @@ -489,13 +469,7 @@ pub(crate) fn build_terminal_event( } let failure = match outcome { - Err(Error::Cancelled) => { - run_failure_from_error(&Error::Cancelled, FailureReason::Cancelled) - } - Err(err @ Error::Publish { .. }) => { - run_failure_from_error(err, FailureReason::PublishFailed) - } - Err(err) => run_failure_from_error(err, FailureReason::WorkflowError), + Err(err) => run_failure_from_error(err, err.failure_reason()), Ok(outcome) => { if let Some(failure) = outcome.failure.as_ref() { run_failure_from_outcome_failure(failure, FailureReason::WorkflowError) @@ -616,25 +590,22 @@ pub async fn finalize(published: Published, options: &FinalizeOptions) -> Result let Published { execution_outcome, publish_outcome, + publish_error, mut conclusion, artifact_count, run_options, services, } = published; - let pushed_branch = publish_outcome - .as_ref() - .ok() - .and_then(|outcome| outcome.pushed_branch()) - .map(str::to_string); - let pr_url = publish_outcome - .as_ref() - .ok() - .and_then(|outcome| outcome.pr_url()) - .map(str::to_string); - let outcome = match (execution_outcome, publish_outcome) { - (Err(error), _) | (Ok(_), Err(error)) => Err(error), - (Ok(outcome), Ok(_)) => Ok(outcome), + let PublishOutcome { + pushed_branch, + pr_url, + } = publish_outcome; + // An execution failure outranks a publish failure: publish only runs after + // a successful execution, so the two are never both set. + let outcome = match (execution_outcome, publish_error) { + (Err(error), _) | (Ok(_), Some(error)) => Err(error), + (Ok(outcome), None) => Ok(outcome), }; let (final_status, failure, _run_status) = classify_engine_result(&outcome); @@ -725,6 +696,7 @@ mod tests { use super::*; use crate::context::Context; + use crate::error::ErrorStage; use crate::event::{Emitter, StoreProgressLogger, append_event}; use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle}; use crate::run_options::{GitCheckpointOptions, RunOptions}; @@ -1417,10 +1389,8 @@ mod tests { }) .await; - assert!(matches!( - &published.publish_outcome, - Ok(crate::pipeline::PublishOutcome::NotRequested) - )); + assert_eq!(published.publish_outcome, PublishOutcome::default()); + assert!(published.publish_error.is_none()); let finalized = finalize(published, &options).await.unwrap(); assert!(finalized.outcome.is_ok()); @@ -1474,12 +1444,21 @@ mod tests { .await; assert!(matches!( - &published.publish_outcome, - Err(Error::Publish { .. }) + &published.publish_error, + Some(Error::Stage { + stage: ErrorStage::Publish, + .. + }) )); let finalized = finalize(published, &options).await.unwrap(); - assert!(matches!(finalized.outcome, Err(Error::Publish { .. }))); + assert!(matches!( + finalized.outcome, + Err(Error::Stage { + stage: ErrorStage::Publish, + .. + }) + )); assert_eq!( finalized .conclusion @@ -1499,6 +1478,71 @@ mod tests { } } + /// An empty diff means there is nothing to open a pull request for. The + /// branch still gets pushed and the run still succeeds. + #[tokio::test] + async fn empty_diff_pushes_branch_without_opening_pull_request() { + let repo_dir = tempfile::tempdir().unwrap(); + init_git_repo(repo_dir.path()); + let emitter = Arc::new(Emitter::new(test_run_id())); + let events = record_events(&emitter); + let services = test_services( + RunStoreHandle::local(seeded_run_store().await), + emitter, + Arc::new(fabro_agent::LocalSandbox::new( + repo_dir.path().to_path_buf(), + )), + Arc::new(RunMetadataRuntime::new()), + None, + ); + let mut run_options = test_run_options(repo_dir.path()); + run_options.base_branch = Some("main".to_string()); + run_options.git = Some(GitCheckpointOptions { + base_sha: None, + run_branch: Some("fabro/run/test".to_string()), + meta_branch: None, + }); + let executed = test_executed( + Graph::new("test"), + Ok(Outcome::success()), + run_options, + 5, + services, + ); + let options = FinalizeOptions { + run_dir: repo_dir.path().to_path_buf(), + run_id: test_run_id(), + workflow_name: "test".to_string(), + preserve_sandbox: false, + stop_on_terminal: true, + last_git_sha: Some("final-sha".to_string()), + }; + let mut concluded = conclude(executed, &options).await.unwrap(); + concluded.conclusion.diff.patch = None; + let published = crate::pipeline::publish(concluded, &crate::pipeline::PublishOptions { + pr_config: Some(fabro_types::settings::run::PullRequestSettings { + enabled: true, + draft: true, + auto_merge: false, + merge_strategy: fabro_types::settings::run::MergeStrategy::Squash, + }), + github_app: None, + origin_url: Some("https://github.com/owner/repo.git".to_string()), + model: "test-model".to_string(), + }) + .await; + + assert!(published.publish_error.is_none()); + let finalized = finalize(published, &options).await.unwrap(); + + assert!(finalized.outcome.is_ok()); + assert_eq!(finalized.pushed_branch.as_deref(), Some("fabro/run/test")); + assert_eq!(finalized.pr_url, None); + let events = events.lock().unwrap(); + let names = events.iter().map(RunEvent::event_name).collect::>(); + assert_eq!(names, vec!["git.push", "run.completed"]); + } + #[tokio::test] async fn pull_request_failure_precedes_terminal_publish_failure() { let repo_dir = tempfile::tempdir().unwrap(); @@ -1553,7 +1597,16 @@ mod tests { .await; let finalized = finalize(published, &options).await.unwrap(); - assert!(matches!(finalized.outcome, Err(Error::Publish { .. }))); + assert!(matches!( + finalized.outcome, + Err(Error::Stage { + stage: ErrorStage::Publish, + .. + }) + )); + // The push landed before the pull request failed, so the branch is + // still reported — that is exactly the run where the user needs it. + assert_eq!(finalized.pushed_branch.as_deref(), Some("fabro/run/test")); let events = events.lock().unwrap(); let names = events.iter().map(RunEvent::event_name).collect::>(); assert_eq!(names, vec!["git.push", "pull_request.failed", "run.failed"]); diff --git a/lib/components/fabro-workflow/src/pipeline/mod.rs b/lib/components/fabro-workflow/src/pipeline/mod.rs index 57fd59e15..98b813b6a 100644 --- a/lib/components/fabro-workflow/src/pipeline/mod.rs +++ b/lib/components/fabro-workflow/src/pipeline/mod.rs @@ -20,7 +20,7 @@ pub(crate) use persist::persist; pub use publish::publish; pub use pull_request::{ AutoMergeOptions, CreatedPullRequest, OpenPullRequestRequest, PrContent, build_pr_content, - maybe_open_pull_request, + open_pull_request, }; pub use transform::transform; pub use types::{ diff --git a/lib/components/fabro-workflow/src/pipeline/publish.rs b/lib/components/fabro-workflow/src/pipeline/publish.rs index d4ba2fd61..10eca9448 100644 --- a/lib/components/fabro-workflow/src/pipeline/publish.rs +++ b/lib/components/fabro-workflow/src/pipeline/publish.rs @@ -1,10 +1,10 @@ use std::sync::Arc; -use super::pull_request::{AutoMergeOptions, OpenPullRequestRequest, maybe_open_pull_request}; +use super::pull_request::{AutoMergeOptions, OpenPullRequestRequest, open_pull_request}; use super::types::{Concluded, PublishOptions, PublishOutcome, Published}; use crate::error::Error; use crate::event::Event; -use crate::outcome::StageOutcome; +use crate::lifecycle::git::push_run_branch; /// PUBLISH phase: push the final run commit and, when configured, open a pull /// request. @@ -12,7 +12,9 @@ use crate::outcome::StageOutcome; /// Publish is always present in the pipeline. It becomes a no-op when the run /// did not succeed, is a dry run, or has no remote branch configured. pub async fn publish(concluded: Concluded, options: &PublishOptions) -> Published { - let publish_outcome = publish_inner(&concluded, options).await; + let mut publish_outcome = PublishOutcome::default(); + let publish_error = concluded.publish(options, &mut publish_outcome).await.err(); + let Concluded { outcome, conclusion, @@ -25,6 +27,7 @@ pub async fn publish(concluded: Concluded, options: &PublishOptions) -> Publishe Published { execution_outcome: outcome, publish_outcome, + publish_error, conclusion, artifact_count, run_options, @@ -32,170 +35,153 @@ pub async fn publish(concluded: Concluded, options: &PublishOptions) -> Publishe } } -async fn publish_inner( - concluded: &Concluded, - options: &PublishOptions, -) -> Result { - let successful_execution = concluded.outcome.as_ref().is_ok_and(|outcome| { - matches!( - outcome.status, - StageOutcome::Succeeded | StageOutcome::PartiallySucceeded - ) - }); - if !successful_execution || concluded.run_options.dry_run_enabled() { - return Ok(PublishOutcome::NotRequested); - } +impl Concluded { + /// Run the publish steps, recording each one into `outcome` as it lands. + /// + /// `outcome` accumulates what actually happened, so a branch that reached + /// the remote is still reported when pull request creation later fails. + async fn publish( + &self, + options: &PublishOptions, + outcome: &mut PublishOutcome, + ) -> Result<(), Error> { + // A run that did not succeed, or that never intended to touch the + // remote, has nothing to publish — even when a pull request was asked + // for. Only a run that got far enough to publish can fail publishing. + if !self + .outcome + .as_ref() + .is_ok_and(|o| o.status.is_successful()) + || self.run_options.dry_run_enabled() + { + return Ok(()); + } - let pull_request_requested = options.pr_config.is_some(); - let Some(origin_url) = options - .origin_url - .as_deref() - .filter(|origin| !origin.trim().is_empty()) - else { - if pull_request_requested { - return Err(pull_request_error( - concluded, - "pull request creation requires a GitHub origin URL", - )); - } - return Ok(PublishOutcome::NotRequested); - }; - let Some(run_branch) = concluded.run_options.run_branch() else { - if pull_request_requested { - return Err(pull_request_error( - concluded, - "pull request creation requires a run branch", - )); - } - return Ok(PublishOutcome::NotRequested); - }; - if !concluded.run_options.settings.run.run_branch.push { - if pull_request_requested { - return Err(pull_request_error( - concluded, - "pull request creation requires run branch pushing", - )); - } - return Ok(PublishOutcome::NotRequested); - } + let pull_request_requested = options.pr_config.is_some(); + let (origin_url, run_branch) = match self.publish_target(options) { + Ok(target) => target, + Err(_) if !pull_request_requested => return Ok(()), + Err(reason) => return Err(self.pull_request_error(reason)), + }; - let final_sha = concluded - .conclusion - .final_git_commit_sha - .as_deref() - .ok_or_else(|| Error::publish("cannot publish a run without a final git commit SHA"))?; - let refspec = format!("refs/heads/{run_branch}:refs/heads/{run_branch}"); - match concluded.services.sandbox.git_push_ref(&refspec).await { - Ok(()) => { - concluded.services.emitter.emit(&Event::GitPush { - branch: run_branch.to_string(), - success: true, - exec_output_tail: None, - }); - } - Err(error) => { - let exec_output_tail = fabro_sandbox::default_redacted_output_tail(&error); - concluded.services.emitter.emit(&Event::GitPush { - branch: run_branch.to_string(), - success: false, - exec_output_tail: exec_output_tail.clone(), - }); - return Err(Error::publish_with_source_and_exec_output_tail( - format!("failed to push final commit {final_sha} to branch '{run_branch}'"), - error, - exec_output_tail, - )); - } - } + let final_sha = self + .conclusion + .final_git_commit_sha + .as_deref() + .ok_or_else(|| Error::publish("cannot publish a run without a final git commit SHA"))?; - let diff = concluded - .conclusion - .diff - .patch - .as_deref() - .unwrap_or_default(); - let Some(pr_config) = options.pr_config.as_ref() else { - return Ok(PublishOutcome::Published { - pushed_branch: run_branch.to_string(), - pr_url: None, - }); - }; - if diff.trim().is_empty() { - return Ok(PublishOutcome::NoChanges { - pushed_branch: run_branch.to_string(), - }); - } + self.push_final_commit(run_branch, final_sha).await?; + outcome.pushed_branch = Some(run_branch.to_string()); - let base_branch = concluded - .run_options - .base_branch - .as_deref() - .ok_or_else(|| { - pull_request_error(concluded, "pull request creation requires a base branch") + let Some(pr_config) = options.pr_config.as_ref() else { + return Ok(()); + }; + let diff = self.conclusion.diff.patch.as_deref().unwrap_or_default(); + if diff.trim().is_empty() { + return Ok(()); + } + + let base_branch = self.run_options.base_branch.as_deref().ok_or_else(|| { + self.pull_request_error("pull request creation requires a base branch") })?; - let credentials = options.github_app.as_ref().ok_or_else(|| { - pull_request_error( - concluded, - "pull request creation requires GitHub credentials", - ) - })?; - let auto_merge = pr_config.auto_merge.then_some(AutoMergeOptions { - merge_strategy: pr_config.merge_strategy, - }); - let github_base_url = fabro_github::github_api_base_url(); + let credentials = options.github_app.as_ref().ok_or_else(|| { + self.pull_request_error("pull request creation requires GitHub credentials") + })?; + let github_base_url = fabro_github::github_api_base_url(); - let created = maybe_open_pull_request(OpenPullRequestRequest { - github: fabro_github::GitHubContext::new(credentials, &github_base_url), - origin_url, - base_branch, - head_branch: run_branch, - expected_head_sha: final_sha, - goal: concluded.graph.goal(), - diff, - model: &options.model, - draft: pr_config.draft, - auto_merge, - run_store: &concluded.services.run_store, - llm_source: concluded.services.llm_source.as_ref(), - catalog: Arc::clone(&concluded.services.catalog), - conclusion: Some(&concluded.conclusion), - run_state: None, - }) - .await - .map_err(|error| { - concluded.services.emitter.emit(&Event::PullRequestFailed { - error: error.clone(), - }); - Error::publish_with_source("failed to create pull request", anyhow::anyhow!(error)) - })? - .ok_or_else(|| { - pull_request_error( - concluded, - "pull request creation found no changes after the stored diff was checked", - ) - })?; + let created = open_pull_request(OpenPullRequestRequest { + github: fabro_github::GitHubContext::new(credentials, &github_base_url), + origin_url, + base_branch, + head_branch: run_branch, + expected_head_sha: final_sha, + goal: self.graph.goal(), + diff, + model: &options.model, + draft: pr_config.draft, + auto_merge: pr_config.auto_merge.then_some(AutoMergeOptions { + merge_strategy: pr_config.merge_strategy, + }), + run_store: &self.services.run_store, + llm_source: self.services.llm_source.as_ref(), + catalog: Arc::clone(&self.services.catalog), + conclusion: Some(&self.conclusion), + run_state: None, + }) + .await + .map_err(|error| { + self.services.emitter.emit(&Event::PullRequestFailed { + error: error.clone(), + }); + Error::publish_with_source("failed to create pull request", anyhow::anyhow!(error)) + })?; - concluded - .services - .emitter - .emit(&Event::pull_request_created( + self.services.emitter.emit(&Event::pull_request_created( &created.link, &created.base_branch, &created.head_branch, - &created.head_sha, + final_sha, &created.title, pr_config.draft, )); + outcome.pr_url = Some(created.link.html_url()); - Ok(PublishOutcome::Published { - pushed_branch: run_branch.to_string(), - pr_url: Some(created.link.html_url()), - }) -} + Ok(()) + } -fn pull_request_error(concluded: &Concluded, message: &str) -> Error { - concluded.services.emitter.emit(&Event::PullRequestFailed { - error: message.to_string(), - }); - Error::publish(message) + /// The origin and run branch to publish to. + /// + /// `Err` carries why there is no target. That is only a failure when a + /// pull request was requested; otherwise publish just has nothing to do. + fn publish_target<'a>( + &'a self, + options: &'a PublishOptions, + ) -> Result<(&'a str, &'a str), &'static str> { + let origin_url = options + .origin_url + .as_deref() + .filter(|origin| !origin.trim().is_empty()) + .ok_or("pull request creation requires a GitHub origin URL")?; + let run_branch = self + .run_options + .run_branch() + .ok_or("pull request creation requires a run branch")?; + if !self.run_options.settings.run.run_branch.push { + return Err("pull request creation requires run branch pushing"); + } + Ok((origin_url, run_branch)) + } + + async fn push_final_commit(&self, run_branch: &str, final_sha: &str) -> Result<(), Error> { + match push_run_branch(self.services.sandbox.as_ref(), run_branch).await { + Ok(()) => { + self.services.emitter.emit(&Event::GitPush { + branch: run_branch.to_string(), + success: true, + exec_output_tail: None, + }); + Ok(()) + } + Err(error) => { + let exec_output_tail = fabro_sandbox::default_redacted_output_tail(&error); + self.services.emitter.emit(&Event::GitPush { + branch: run_branch.to_string(), + success: false, + exec_output_tail: exec_output_tail.clone(), + }); + Err(Error::publish_with_source_and_exec_output_tail( + format!("failed to push final commit {final_sha} to branch '{run_branch}'"), + error, + exec_output_tail, + )) + } + } + } + + fn pull_request_error(&self, message: &str) -> Error { + self.services.emitter.emit(&Event::PullRequestFailed { + error: message.to_string(), + }); + Error::publish(message) + } } diff --git a/lib/components/fabro-workflow/src/pipeline/pull_request.rs b/lib/components/fabro-workflow/src/pipeline/pull_request.rs index 638c87847..d66072923 100644 --- a/lib/components/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/components/fabro-workflow/src/pipeline/pull_request.rs @@ -441,7 +441,7 @@ pub struct AutoMergeOptions { pub merge_strategy: MergeStrategy, } -/// Inputs for [`maybe_open_pull_request`]. +/// Inputs for [`open_pull_request`]. pub struct OpenPullRequestRequest<'a> { pub github: github_app::GitHubContext<'a>, pub origin_url: &'a str, @@ -468,21 +468,15 @@ pub struct CreatedPullRequest { pub title: String, pub base_branch: String, pub head_branch: String, - pub head_sha: String, } -/// Optionally open a pull request after a successful workflow run. +/// Open a pull request for a completed run. /// -/// Returns `Ok(Some(CreatedPullRequest))` if a PR was created, `Ok(None)` if -/// the diff was empty, or `Err` on failure. -pub async fn maybe_open_pull_request( +/// Callers are responsible for skipping runs with an empty diff; reaching here +/// means a pull request is expected, so every failure is an error. +pub async fn open_pull_request( req: OpenPullRequestRequest<'_>, -) -> Result, String> { - if req.diff.is_empty() { - debug!("Empty diff, skipping pull request creation"); - return Ok(None); - } - +) -> Result { let https_url = ssh_url_to_https(req.origin_url); let (owner, repo) = github_app::parse_github_owner_repo(&https_url).map_err(|err| format!("{err:#}"))?; @@ -565,13 +559,12 @@ pub async fn maybe_open_pull_request( number: created.number, }; - Ok(Some(CreatedPullRequest { + Ok(CreatedPullRequest { link, title, base_branch: req.base_branch.to_string(), head_branch: req.head_branch.to_string(), - head_sha: req.expected_head_sha.to_string(), - })) + }) } #[cfg(test)] @@ -581,7 +574,7 @@ mod tests { use std::time::Duration; use chrono::Utc; - use fabro_auth::{CredentialSource, EnvCredentialSource, VaultCredentialSource}; + use fabro_auth::{CredentialSource, VaultCredentialSource}; use fabro_graphviz::graph::Graph; use fabro_llm::Error as LlmError; use fabro_llm::client::Client; @@ -687,10 +680,6 @@ mod tests { )) } - fn test_catalog() -> Arc { - Arc::new(Catalog::from_builtin().expect("default catalog should build")) - } - fn test_catalog_with_provider_base_url(provider: &str, base_url: &str) -> Arc { let mut settings = LlmCatalogSettings::default(); settings @@ -718,10 +707,6 @@ mod tests { )) } - fn test_llm_source() -> Arc { - Arc::new(EnvCredentialSource::new()) - } - fn test_projection() -> RunProjection { RunProjection::new( "Test run".to_string(), @@ -1400,46 +1385,12 @@ mod tests { ); } - #[tokio::test] - async fn empty_diff_returns_none() { - let store = test_store(); - let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); - let run_store_handle: RunStoreHandle = run_store.into(); - let llm_source = test_llm_source(); - let creds = fabro_github::GitHubCredentials::App(fabro_github::GitHubAppCredentials { - app_id: "123".to_string(), - private_key_pem: "unused".to_string(), - slug: None, - }); - let base_url = github_app::github_api_base_url(); - let result = maybe_open_pull_request(OpenPullRequestRequest { - github: github_app::GitHubContext::new(&creds, &base_url), - origin_url: "https://github.com/owner/repo.git", - base_branch: "main", - head_branch: "fabro/run/123", - expected_head_sha: "final-sha", - goal: "Fix bug", - diff: "", - model: "claude-sonnet-4-20250514", - draft: false, - auto_merge: None, - run_store: &run_store_handle, - llm_source: llm_source.as_ref(), - catalog: test_catalog(), - conclusion: None, - run_state: None, - }) - .await; - assert!(result.is_ok()); - assert!(result.unwrap().is_none()); - } - #[tokio::test] async fn stale_remote_branch_is_rejected_before_pull_request_creation() { let payload = pr_content_json("Fix bug", "Narrative."); let harness = setup_fallback_test_harness_with_branch_sha(&payload, "stale-sha").await; let github_base_url = harness.github_server.url(""); - let error = maybe_open_pull_request(OpenPullRequestRequest { + let error = open_pull_request(OpenPullRequestRequest { github: fabro_github::GitHubContext::new(&harness.creds, &github_base_url), origin_url: "https://github.com/owner/repo.git", base_branch: "main", @@ -1614,9 +1565,9 @@ mod tests { assert!(body.contains("Generated with [Fabro](https://fabro.sh)")); } - // ── maybe_open_pull_request fallback tests ────────────────────────── + // ── open_pull_request fallback tests ────────────────────────── - /// Set of mock servers and credentials for the `maybe_open_pull_request` + /// Set of mock servers and credentials for the `open_pull_request` /// fallback path. The builder's `Client::from_source` rebuilds the LLM /// client from the credential source, so the in-process MockProvider /// cannot intercept — we mock the OpenAI HTTP endpoint instead. @@ -1812,14 +1763,14 @@ mod tests { /// falls back to `pr_title_from_goal` (first line, decoration stripped) /// and PR creation succeeds with that title. #[tokio::test] - async fn maybe_open_pull_request_falls_back_to_goal_title_when_llm_returns_empty_title() { + async fn open_pull_request_falls_back_to_goal_title_when_llm_returns_empty_title() { let payload = pr_content_json("", "Narrative."); let harness = setup_fallback_test_harness(&payload).await; let github_base_url = harness.github_server.url(""); let github = github_app::GitHubContext::new(&harness.creds, &github_base_url); - let result = maybe_open_pull_request(OpenPullRequestRequest { + let result = open_pull_request(OpenPullRequestRequest { github, origin_url: "https://github.com/owner/repo.git", base_branch: "main", @@ -1839,15 +1790,14 @@ mod tests { .await .expect("PR creation should succeed"); - let record = result.expect("PR record should be Some"); - assert_eq!(record.title, "Fix telemetry leak"); + assert_eq!(result.title, "Fix telemetry leak"); harness.assert_mocks_called_once().await; } /// LLM returns an empty title; the content builder fallback still caps /// the deterministic goal title at 72 chars ending with `…`. #[tokio::test] - async fn maybe_open_pull_request_caps_fallback_title_at_72_chars() { + async fn open_pull_request_caps_fallback_title_at_72_chars() { let payload = pr_content_json("", "Narrative."); let harness = setup_fallback_test_harness(&payload).await; @@ -1857,7 +1807,7 @@ mod tests { // Single ~200-char line, no `Plan:` / heading prefix, no newlines. let goal = "x".repeat(200); - let result = maybe_open_pull_request(OpenPullRequestRequest { + let result = open_pull_request(OpenPullRequestRequest { github, origin_url: "https://github.com/owner/repo.git", base_branch: "main", @@ -1877,8 +1827,7 @@ mod tests { .await .expect("PR creation should succeed"); - let record = result.expect("PR record should be Some"); - let title = record.title; + let title = result.title; assert_eq!(title.chars().count(), 72); assert!(title.ends_with('\u{2026}')); harness.assert_mocks_called_once().await; diff --git a/lib/components/fabro-workflow/src/pipeline/types.rs b/lib/components/fabro-workflow/src/pipeline/types.rs index c7cfcc815..19bbc42cf 100644 --- a/lib/components/fabro-workflow/src/pipeline/types.rs +++ b/lib/components/fabro-workflow/src/pipeline/types.rs @@ -348,41 +348,23 @@ pub struct Concluded { pub services: Arc, } -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum PublishOutcome { - NotRequested, - NoChanges { - pushed_branch: String, - }, - Published { - pushed_branch: String, - pr_url: Option, - }, -} - -impl PublishOutcome { - pub fn pushed_branch(&self) -> Option<&str> { - match self { - Self::NotRequested => None, - Self::NoChanges { pushed_branch } | Self::Published { pushed_branch, .. } => { - Some(pushed_branch) - } - } - } - - pub fn pr_url(&self) -> Option<&str> { - match self { - Self::Published { pr_url, .. } => pr_url.as_deref(), - Self::NotRequested | Self::NoChanges { .. } => None, - } - } +/// What the PUBLISH phase actually accomplished. +/// +/// Recorded separately from the phase's error so a branch that reached the +/// remote is still reported when a later step, such as pull request creation, +/// fails. An all-`None` value means publish had nothing to do. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct PublishOutcome { + pub pushed_branch: Option, + pub pr_url: Option, } /// Output of the PUBLISH phase. #[non_exhaustive] pub struct Published { pub execution_outcome: Result, - pub publish_outcome: Result, + pub publish_outcome: PublishOutcome, + pub publish_error: Option, pub conclusion: Conclusion, pub artifact_count: usize, pub run_options: RunOptions, diff --git a/lib/components/fabro-workflow/src/pull_request.rs b/lib/components/fabro-workflow/src/pull_request.rs index bb9a74744..78723914d 100644 --- a/lib/components/fabro-workflow/src/pull_request.rs +++ b/lib/components/fabro-workflow/src/pull_request.rs @@ -1,4 +1,4 @@ pub use crate::pipeline::{ AutoMergeOptions, CreatedPullRequest, OpenPullRequestRequest, PrContent, build_pr_content, - maybe_open_pull_request, + open_pull_request, }; From fa6f7e55586917a73fd2b190e6fb890911f11191 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 28 Jul 2026 15:34:18 -0400 Subject: [PATCH 3/4] fix(workflow): verify remote head before generating PR content, tolerate replica lag Three follow-ups from the efficiency review of the publish pipeline. Check the branch before spending an LLM call: `open_pull_request` generated the PR title and body first and only then verified the remote branch pointed at the run's final commit. Every stale branch therefore cost a full content generation before failing. The verification is the cheap check, so it now runs first. Tolerate GitHub read-after-write lag: `GET /repos/{owner}/{repo}/branches/{branch}` is replica-served and can briefly report the previous commit, or 404 for a branch that is new on the remote, right after the push publish just made. It was read once with no retry. Since publish failures are terminal, a replica that had not caught up yet would discard a fully successful run. It is now read up to three times. These two land together on purpose: the LLM call was the only thing buying slack against the race, so reordering without the retry would have made it more likely. Keep commit SHAs out of failure classification: `classify_failure_reason` substring-matches bare "500", "502", "503" and "504" as transient-infra hints. Both publish messages embed a commit SHA, and a 40-char hex string contains one of those often enough to matter, so a deterministic failure could be reported as transient. Long hex runs are now masked before matching; the three-digit status codes those hints look for are too short to be affected. The hex regex is shared with `normalize_failure_reason`, which already had its own copy. Co-Authored-By: Claude Opus 5 (1M context) --- lib/components/fabro-workflow/src/error.rs | 37 +++++++-- .../src/pipeline/pull_request.rs | 83 ++++++++++++++----- 2 files changed, 88 insertions(+), 32 deletions(-) diff --git a/lib/components/fabro-workflow/src/error.rs b/lib/components/fabro-workflow/src/error.rs index ac7eed9c5..1525c570c 100644 --- a/lib/components/fabro-workflow/src/error.rs +++ b/lib/components/fabro-workflow/src/error.rs @@ -1,5 +1,5 @@ use std::fmt; -use std::sync::Arc; +use std::sync::{Arc, LazyLock}; use fabro_graphviz::Error as GraphvizError; use fabro_llm::{Error as LlmError, ProviderErrorKind}; @@ -11,6 +11,7 @@ use fabro_types::settings::AmbiguousModelRef; use fabro_types::{ExecOutputTail, FailureReason, RunFailure}; use fabro_util::error::{SharedError, collect_causes, collect_chain, render_with_causes}; use fabro_validate::Diagnostic; +use regex::Regex; use thiserror::Error as ThisError; use crate::outcome::{FailureDetail, Outcome, StageOutcome}; @@ -153,13 +154,21 @@ impl miette::Diagnostic for SharedTemplateError { } } +/// Matches git SHAs and other long hex blobs. +static HEX_RE: LazyLock = + LazyLock::new(|| Regex::new(r"\b[0-9a-f]{7,64}\b").expect("hardcoded regex should compile")); + /// Classify a failure reason string using heuristics. /// /// This is the fallback when structured error information is not available /// (e.g. for `Handler(String)` or `Engine(String)` errors). #[must_use] pub fn classify_failure_reason(reason: &str) -> FailureCategory { - let lower = reason.to_lowercase(); + // Mask commit SHAs first. They are hex, so one contains "500" or "503" + // often enough to matter, which would read as a transient infra hint. The + // bare status codes those hints look for are too short to be masked. + let lowered = reason.to_lowercase(); + let lower = HEX_RE.replace_all(&lowered, ""); if lower.contains("interrupt") || (lower.contains("cancel") @@ -196,13 +205,6 @@ pub fn classify_failure_reason(reason: &str) -> FailureCategory { /// semantically identical errors produce the same signature regardless of /// line numbers, commit hashes, or timestamps. pub fn normalize_failure_reason(reason: &str) -> String { - use std::sync::LazyLock; - - use regex::Regex; - - static HEX_RE: LazyLock = LazyLock::new(|| { - Regex::new(r"\b[0-9a-f]{7,64}\b").expect("hardcoded regex should compile") - }); static DIGITS_RE: LazyLock = LazyLock::new(|| Regex::new(r"\b\d+\b").expect("hardcoded regex should compile")); static COMMA_SPACE_RE: LazyLock = @@ -2085,6 +2087,23 @@ mod tests { } } + /// Commit SHAs are hex, so they contain digit runs like "503" often enough + /// to matter. Masking them keeps a deterministic failure from being + /// reported as transient just because of the SHA it names. + #[test] + fn commit_shas_do_not_trip_transient_infra_hints() { + let sha = "a503b1c9d4e2f7a8b6c3d0e1f2a3b4c5d6e7f8a9"; + assert_eq!( + classify_failure_reason(&format!("failed to push final commit {sha} to branch 'x'")), + FailureCategory::Deterministic + ); + // A real status code is still a transient hint. + assert_eq!( + classify_failure_reason("push rejected with 503"), + FailureCategory::TransientInfra + ); + } + #[test] fn to_fail_outcome_preserves_class() { let err = Error::handler("timeout"); diff --git a/lib/components/fabro-workflow/src/pipeline/pull_request.rs b/lib/components/fabro-workflow/src/pipeline/pull_request.rs index d66072923..65c4a4308 100644 --- a/lib/components/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/components/fabro-workflow/src/pipeline/pull_request.rs @@ -1,5 +1,6 @@ use std::collections::HashSet; use std::sync::{Arc, LazyLock}; +use std::time::Duration; use fabro_auth::CredentialSource; use fabro_github::{self as github_app, ssh_url_to_https}; @@ -11,6 +12,7 @@ use fabro_store::RunProjection; use fabro_types::PullRequestLink; use fabro_types::settings::run::MergeStrategy; use fabro_util::text::strip_goal_decoration; +use tokio::time::sleep; use tracing::{debug, info, warn}; use crate::outcome::format_cost as outcome_format_cost; @@ -470,6 +472,54 @@ pub struct CreatedPullRequest { pub head_branch: String, } +/// How many times to read the remote branch head before giving up. +/// +/// `GET /repos/{owner}/{repo}/branches/{branch}` is replica-served, so shortly +/// after the push that publish just made it can still report the previous +/// commit — or 404 for a branch that is new on the remote. +const BRANCH_HEAD_ATTEMPTS: u32 = 3; +const BRANCH_HEAD_RETRY_DELAY: Duration = Duration::from_millis(500); + +/// Confirm the remote branch points at the run's final commit. +/// +/// Publish failures are terminal, so a replica that has not caught up yet must +/// not be mistaken for a genuinely stale branch. +async fn verify_remote_head( + req: &OpenPullRequestRequest<'_>, + owner: &str, + repo: &str, +) -> Result<(), String> { + let mut last_seen = Ok(None); + for attempt in 1..=BRANCH_HEAD_ATTEMPTS { + last_seen = github_app::branch_head_sha(&req.github, owner, repo, req.head_branch).await; + match &last_seen { + Ok(Some(head)) if head == req.expected_head_sha => return Ok(()), + Ok(head) => debug!( + attempt, + head = ?head, + expected = req.expected_head_sha, + "Remote branch head does not match the final commit yet" + ), + Err(err) => debug!(attempt, error = %err, "Failed to read remote branch head"), + } + if attempt < BRANCH_HEAD_ATTEMPTS { + sleep(BRANCH_HEAD_RETRY_DELAY).await; + } + } + + Err(match last_seen { + Ok(Some(head)) => format!( + "remote branch '{}' points to commit {head}, expected final commit {}", + req.head_branch, req.expected_head_sha + ), + Ok(None) => format!( + "remote branch '{}' does not exist; expected final commit {}", + req.head_branch, req.expected_head_sha + ), + Err(err) => format!("failed to verify remote branch head: {err:#}"), + }) +} + /// Open a pull request for a completed run. /// /// Callers are responsible for skipping runs with an empty diff; reaching here @@ -481,6 +531,10 @@ pub async fn open_pull_request( let (owner, repo) = github_app::parse_github_owner_repo(&https_url).map_err(|err| format!("{err:#}"))?; + // Verify before generating content: this is the cheap check, and a stale + // branch would otherwise cost a full LLM call before failing. + verify_remote_head(&req, &owner, &repo).await?; + let content = build_pr_content( req.diff, req.goal, @@ -496,25 +550,6 @@ pub async fn open_pull_request( let body = truncate_pr_body(&content.body); let title = content.title; - let remote_head = github_app::branch_head_sha(&req.github, &owner, &repo, req.head_branch) - .await - .map_err(|err| format!("failed to verify remote branch head: {err:#}"))?; - match remote_head { - Some(remote_head) if remote_head == req.expected_head_sha => {} - Some(remote_head) => { - return Err(format!( - "remote branch '{}' points to commit {remote_head}, expected final commit {}", - req.head_branch, req.expected_head_sha - )); - } - None => { - return Err(format!( - "remote branch '{}' does not exist; expected final commit {}", - req.head_branch, req.expected_head_sha - )); - } - } - let created = github_app::create_pull_request( &req.github, &owner, @@ -1412,11 +1447,13 @@ mod tests { assert!(error.contains("stale-sha")); assert!(error.contains("final-sha")); - httpmock::Mock::new(harness.openai_mock_id, &harness.openai_server) - .assert_async() - .await; + // The branch is re-read to ride out replica lag... httpmock::Mock::new(harness.branch_mock_id, &harness.github_server) - .assert_async() + .assert_calls_async(BRANCH_HEAD_ATTEMPTS as usize) + .await; + // ...but the check runs first, so no LLM call and no PR creation. + httpmock::Mock::new(harness.openai_mock_id, &harness.openai_server) + .assert_calls_async(0) .await; httpmock::Mock::new(harness.github_mock_id, &harness.github_server) .assert_calls_async(0) From 1af6a3d8be7b0c781ce9425a1eef037a6e8f37e8 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 28 Jul 2026 16:55:52 -0400 Subject: [PATCH 4/4] fix: stop inferring a final commit SHA, route slashed branches in the GitHub twin MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both from Copilot review feedback on #652. Do not fall back to `base_sha` for `final_git_commit_sha`: `base_sha` is where the run started, not what it produced. When a run made commits but no SHA was tracked, the conclusion reported the base commit as the run's final commit — a durable, API-exposed field — and publish then checked the pushed branch against it, failing a branch that was pushed correctly. The SHA is now only required where it is actually used: verifying the remote head before opening a pull request. Pushing never needed it, since the refspec sends whatever the branch points at. A run with no tracked SHA therefore still pushes its branch and succeeds; it fails only if a pull request is requested, where an unverifiable head is a real problem. Route branch names with slashes in the GitHub twin: Run branches are `fabro/run/`. GitHub routes the branch as the remainder of the path, but the twin declared a single-segment `{branch}` capture, so every real run branch 404'd against it. Now a wildcard, with a test covering the slashed case that the existing single-segment tests missed. Co-Authored-By: Claude Opus 5 (1M context) --- .../fabro-workflow/src/pipeline/finalize.rs | 68 +++++++++++++++++-- .../fabro-workflow/src/pipeline/publish.rs | 23 ++++--- test/twin/github/src/handlers/branches.rs | 45 ++++++++++++ test/twin/github/src/handlers/mod.rs | 4 +- 4 files changed, 123 insertions(+), 17 deletions(-) diff --git a/lib/components/fabro-workflow/src/pipeline/finalize.rs b/lib/components/fabro-workflow/src/pipeline/finalize.rs index 73f01568d..4b5c69dbf 100644 --- a/lib/components/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/components/fabro-workflow/src/pipeline/finalize.rs @@ -544,12 +544,6 @@ pub async fn conclude(executed: Executed, options: &FinalizeOptions) -> Result Result return Err(self.pull_request_error(reason)), }; - let final_sha = self - .conclusion - .final_git_commit_sha - .as_deref() - .ok_or_else(|| Error::publish("cannot publish a run without a final git commit SHA"))?; - - self.push_final_commit(run_branch, final_sha).await?; + self.push_final_commit(run_branch).await?; outcome.pushed_branch = Some(run_branch.to_string()); let Some(pr_config) = options.pr_config.as_ref() else { @@ -81,6 +75,17 @@ impl Concluded { return Ok(()); } + // Only pull request creation needs the SHA, to check that the remote + // branch really carries this run's work. Pushing does not: the refspec + // sends whatever the branch points at. + let final_sha = self + .conclusion + .final_git_commit_sha + .as_deref() + .ok_or_else(|| { + self.pull_request_error("pull request creation requires the run's final commit SHA") + })?; + let base_branch = self.run_options.base_branch.as_deref().ok_or_else(|| { self.pull_request_error("pull request creation requires a base branch") })?; @@ -152,7 +157,7 @@ impl Concluded { Ok((origin_url, run_branch)) } - async fn push_final_commit(&self, run_branch: &str, final_sha: &str) -> Result<(), Error> { + async fn push_final_commit(&self, run_branch: &str) -> Result<(), Error> { match push_run_branch(self.services.sandbox.as_ref(), run_branch).await { Ok(()) => { self.services.emitter.emit(&Event::GitPush { @@ -170,7 +175,7 @@ impl Concluded { exec_output_tail: exec_output_tail.clone(), }); Err(Error::publish_with_source_and_exec_output_tail( - format!("failed to push final commit {final_sha} to branch '{run_branch}'"), + format!("failed to push run branch '{run_branch}'"), error, exec_output_tail, )) diff --git a/test/twin/github/src/handlers/branches.rs b/test/twin/github/src/handlers/branches.rs index 5eca78d4f..1e0646d5f 100644 --- a/test/twin/github/src/handlers/branches.rs +++ b/test/twin/github/src/handlers/branches.rs @@ -172,6 +172,51 @@ mod tests { server.shutdown().await; } + /// Run branches contain slashes (`fabro/run/`). GitHub routes the + /// branch as the rest of the path, so the twin must too. + #[tokio::test] + async fn branch_with_slashes_returns_200() { + let pem = test_rsa_private_key(); + let mut state = AppState::new(); + state.register_app(AppOptions { + app_id: "100".to_string(), + slug: "test-app".to_string(), + owner_login: "owner".to_string(), + public: true, + private_key_pem: pem.to_string(), + webhook_secret: None, + }); + state.add_installation("100", "owner", vec!["repo".to_string()], false); + state.add_repository( + "owner", + "repo", + vec!["main".to_string(), "fabro/run/123".to_string()], + false, + ); + let server = TestServer::start(state).await; + + let jwt = sign_test_jwt("100", pem); + let client = test_http_client(); + let token = get_installation_token(&client, &jwt, "owner", "repo", server.url()).await; + + let resp = client + .get(format!( + "{}/repos/owner/repo/branches/fabro/run/123", + server.url() + )) + .header("Authorization", format!("Bearer {token}")) + .header("Accept", "application/vnd.github+json") + .send() + .await + .unwrap(); + + assert_eq!(resp.status(), 200); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["name"], "fabro/run/123"); + + server.shutdown().await; + } + #[tokio::test] async fn branch_not_found_returns_404() { let pem = test_rsa_private_key(); diff --git a/test/twin/github/src/handlers/mod.rs b/test/twin/github/src/handlers/mod.rs index c0a652008..30912c764 100644 --- a/test/twin/github/src/handlers/mod.rs +++ b/test/twin/github/src/handlers/mod.rs @@ -30,8 +30,10 @@ pub fn build_router(state: SharedState) -> Router { post(installations::create_access_token), ) // Branch endpoints + // Wildcard: branch names contain slashes (`fabro/run/`), and + // GitHub routes the branch as the remainder of the path. .route( - "/repos/{owner}/{repo}/branches/{branch}", + "/repos/{owner}/{repo}/branches/{*branch}", get(branches::get_branch), ) // Pull request endpoints