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',