diff --git a/Cargo.lock b/Cargo.lock index e71612fb2d..038fdc4dff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2652,6 +2652,7 @@ dependencies = [ "tokio", "tracing", "tracing-subscriber", + "url", ] [[package]] diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index b028d89c04..def4ec3750 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -2575,7 +2575,11 @@ paths: operationId: createRunPullRequest tags: [Runs] summary: Create Run Pull Request - description: Creates a pull request for a completed run on GitHub and persists the record on the server. + description: >- + Creates a pull request on GitHub and persists the record on the + server. By default the run must have a successful conclusion. With + force=true, failed conclusions are accepted and running, blocked, or + paused runs use their latest committed snapshot. parameters: - $ref: "#/components/parameters/RunId" requestBody: @@ -2586,7 +2590,7 @@ paths: $ref: "#/components/schemas/CreateRunPullRequestRequest" responses: "200": - description: Pull request created + description: Pull request created or an existing matching pull request linked content: application/json: schema: @@ -2611,8 +2615,9 @@ paths: $ref: "#/components/schemas/ErrorResponse" "409": description: >- - Pull request already exists for this run. Clients can GET - /runs/{id}/pull_request to retrieve the stored record. + Pull request already exists for this run, or the requested active + snapshot is not ready, eligible, or available on the remote run + branch. headers: x-request-id: $ref: "#/components/headers/XRequestId" @@ -2630,7 +2635,7 @@ paths: schema: $ref: "#/components/schemas/ErrorResponse" "503": - description: GitHub integration is unavailable on the server + description: GitHub integration or the active run snapshot is unavailable headers: x-request-id: $ref: "#/components/headers/XRequestId" @@ -12168,7 +12173,9 @@ components: properties: force: type: boolean - description: Create the pull request even if the run did not finish with succeeded or partially_succeeded. + description: >- + Create from a non-successful conclusion, or from the latest + committed snapshot when the run is running, blocked, or paused. example: false model: type: ["string", "null"] @@ -12592,6 +12599,7 @@ components: - agent - prompt - command + - pull_request - human - conditional - parallel diff --git a/docs/public/execution/context.mdx b/docs/public/execution/context.mdx index 6f25c39c4b..d8c418d505 100644 --- a/docs/public/execution/context.mdx +++ b/docs/public/execution/context.mdx @@ -46,6 +46,18 @@ Agents can also emit arbitrary context updates by including a JSON object with a |---|---| | `command.output` | The command's ordered output stream. Durable context stores this as a `blob://sha256/...` ref after the command completes; downstream prompts resolve it back to text. | +### Pull request nodes + +| Key | Value | +|---|---| +| `pull_request.url` | Public GitHub pull request URL | +| `pull_request.number` | Pull request number | +| `pull_request.owner` | Repository owner | +| `pull_request.repo` | Repository name | +| `pull_request.base_branch` | PR base branch | +| `pull_request.head_branch` | Managed run branch | +| `pull_request.draft` | `true` for PRs created by this node | + ### Human gates | Key | Value | diff --git a/docs/public/execution/run-configuration.mdx b/docs/public/execution/run-configuration.mdx index 300f0a8020..87505cd899 100644 --- a/docs/public/execution/run-configuration.mdx +++ b/docs/public/execution/run-configuration.mdx @@ -226,6 +226,10 @@ push = true | `enabled` | When `false`, Fabro does not create the managed run branch or checkpoint commits. This also disables metadata branch writes. | | `push` | When `false`, Fabro creates local checkpoint commits but does not push `fabro/run/` to the remote. | +The built-in `type="pull_request"` workflow node requires both fields to be +`true`, because it creates the PR from the latest committed and pushed run +snapshot. + ### `[run.meta_branch]` Configure Fabro's managed `fabro/meta/` metadata branch. @@ -497,6 +501,10 @@ The `sandbox` transport runs the MCP server inside the workflow's sandbox. This Automatically open a GitHub pull request when the workflow run completes successfully. Requires a [GitHub App](/integrations/github) to be configured. +To open a draft PR earlier and use it in later workflow stages, add an explicit +`type="pull_request"` node instead. That node is independently opt-in and does +not require this section. + ```toml title="run.toml" [run.pull_request] enabled = true diff --git a/docs/public/reference/cli.mdx b/docs/public/reference/cli.mdx index f264f7bab4..a89cdc3d8b 100644 --- a/docs/public/reference/cli.mdx +++ b/docs/public/reference/cli.mdx @@ -766,7 +766,7 @@ fabro pr [OPTIONS] | Command | Description | | --- | --- | | `fabro pr close` | Close a pull request | -| `fabro pr create` | Create a pull request from a completed run | +| `fabro pr create` | Create a pull request from a run | | `fabro pr link` | Link or replace the GitHub pull request associated with a run | | `fabro pr merge` | Merge a pull request | | `fabro pr unlink` | Unlink the pull request associated with a run | @@ -794,7 +794,9 @@ fabro pr close [OPTIONS] #### `fabro pr create` -Create a pull request from a completed run +Create a pull request from a run. By default, the run must have completed +successfully. Pass `--force` to use a failed conclusion or the latest committed +snapshot of a running, blocked, or paused run. ```bash fabro pr create [OPTIONS] @@ -810,7 +812,7 @@ fabro pr create [OPTIONS] | Option | Description | | --- | --- | -| `-f, --force` | Create PR even if the run status is not succeeded/partially_succeeded | +| `-f, --force` | Use a non-successful conclusion or an active run's latest committed snapshot | | `--model ` | LLM model for generating PR description | | `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | diff --git a/docs/public/reference/dot-language.mdx b/docs/public/reference/dot-language.mdx index 8c350d1ba5..7bc25d2f7e 100644 --- a/docs/public/reference/dot-language.mdx +++ b/docs/public/reference/dot-language.mdx @@ -162,6 +162,7 @@ Each node's `shape` attribute determines its execution behavior. See [Nodes & St | `box` (default) | agent | Multi-turn LLM with tool access | | `tab` | prompt | Single LLM call, no tools | | `parallelogram` | command | Execute a shell script | +| *(explicit `type` only)* | pull_request | Create or adopt a draft pull request during the run | | `hexagon` | human | Human-in-the-loop decision gate | | `diamond` | conditional | Route based on conditions | | `component` | parallel | Fan-out to concurrent branches | @@ -248,6 +249,14 @@ audit [ | `language` | String | `"shell"` (default) or `"python"` | | `output_schema` | String | Optional structured output validation. Accepts `routing`, `@path/to/schema.json`, or an inline JSON Schema object string. See [Structured output validation](#structured-output-validation). | +### Pull request nodes + +Set `type="pull_request"` explicitly; there is no shape alias. The node creates +or adopts a draft GitHub PR from the committed `base_sha..HEAD` snapshot and +stores its coordinates in `pull_request.*` context keys. It requires an enabled, +pushed run branch and cannot execute inside a parallel branch. Dirty and +untracked files are not included. + ### Parallel (fan-out) nodes | Attribute | Type | Description | diff --git a/docs/public/workflows/stages-and-nodes.mdx b/docs/public/workflows/stages-and-nodes.mdx index 819befe7eb..1772ca59a3 100644 --- a/docs/public/workflows/stages-and-nodes.mdx +++ b/docs/public/workflows/stages-and-nodes.mdx @@ -11,7 +11,9 @@ This distinction matters for observability and debugging: the workflow graph sho ## Node types -Every node's Graphviz `shape` attribute determines its execution behavior. If no shape is specified, the node defaults to an agent. +Most nodes use a Graphviz `shape` to select their execution behavior. A node +can instead set `type` explicitly. If neither selects a handler, the node +defaults to an agent. ### Start @@ -107,6 +109,39 @@ test [label="Run Tests", shape=parallelogram, script="cargo test 2>&1 || true"] | `script` | The shell command to execute | | `language` | `"shell"` (default) or `"python"` | +### Pull request + +**Type:** `pull_request` (explicit only) + +Creates or adopts a draft GitHub pull request during the run, so downstream +nodes can work with it: + +```dot +implement [label="Implement"] +create_pr [label="Open draft PR", type="pull_request"] +comment [label="Comment on PR"] + +implement -> create_pr -> comment +``` + +The node snapshots committed work from the run's base SHA through one captured +`HEAD`, pushes that exact commit to the managed run branch without force, and +opens a draft PR. Dirty and untracked files are excluded. Later checkpoint +pushes update the same PR branch. Retries are idempotent: Fabro reuses the +stored PR or adopts a matching open GitHub PR instead of creating a duplicate. + +This node requires a GitHub origin, configured GitHub credentials, and +`[run.run_branch] enabled = true, push = true`. It cannot run inside a parallel +branch. Dry runs are side-effect free and return placeholder context values. +Set the optional `model` attribute to override the run model used to generate +the PR title and body. + +Use this node rather than invoking `fabro pr create` from a command node. Fabro +does not expose the worker API token inside the sandbox. The draft remains open +if a later node fails. It does not use `[run.pull_request]`; that setting remains +the separate end-of-run PR mode. In this initial behavior, end-of-run processing +does not refresh the in-run PR body, mark it ready, or enable auto-merge. + ### Human **Shape:** `hexagon` diff --git a/lib/apps/fabro-cli/src/args.rs b/lib/apps/fabro-cli/src/args.rs index f560b62e66..983a807ec7 100644 --- a/lib/apps/fabro-cli/src/args.rs +++ b/lib/apps/fabro-cli/src/args.rs @@ -929,7 +929,7 @@ pub(crate) struct PrCreateArgs { /// LLM model for generating PR description #[arg(long)] pub(crate) model: Option, - /// Create PR even if the run status is not succeeded/partially_succeeded + /// Use a non-successful conclusion or an active run's committed snapshot #[arg(short, long)] pub(crate) force: bool, } @@ -1453,7 +1453,7 @@ pub(crate) struct PrNamespace { #[derive(Subcommand)] pub(crate) enum PrCommand { - /// Create a pull request from a completed run + /// Create a pull request from a run Create(PrCreateArgs), /// Link or replace the GitHub pull request associated with a run Link(PrLinkArgs), diff --git a/lib/apps/fabro-cli/src/commands/pr/create.rs b/lib/apps/fabro-cli/src/commands/pr/create.rs index 2706c09911..255f1556a6 100644 --- a/lib/apps/fabro-cli/src/commands/pr/create.rs +++ b/lib/apps/fabro-cli/src/commands/pr/create.rs @@ -16,7 +16,7 @@ pub(super) async fn create_command(args: PrCreateArgs, base_ctx: &CommandContext number = record.number, owner = %record.owner, repo = %record.repo, - "Created pull request" + "Created or linked pull request" ); if ctx.json_output() { diff --git a/lib/apps/fabro-cli/src/commands/run/runner.rs b/lib/apps/fabro-cli/src/commands/run/runner.rs index 4e6b72b8bd..492f3560ca 100644 --- a/lib/apps/fabro-cli/src/commands/run/runner.rs +++ b/lib/apps/fabro-cli/src/commands/run/runner.rs @@ -21,7 +21,7 @@ use fabro_store::{EventEnvelope, RunProjection, RunProjectionReducer}; use fabro_tool::fabro_client::ClientBackend; use fabro_types::settings::run::{RunMode, RunNamespace}; use fabro_types::{ - ArtifactUpload, EventBody, FailureReason, Principal, RunBlobId, RunEvent, RunId, + ArtifactUpload, EventBody, FailureReason, Graph, Principal, RunBlobId, RunEvent, RunId, WorkflowSettings, }; use fabro_vault::{SecretStore, Vault}; @@ -143,7 +143,7 @@ pub(crate) async fn execute( Some(arc) => Some(arc.read().await), None => None, }; - maybe_build_github_credentials(&run_spec.settings, vault_guard.as_deref())? + maybe_build_github_credentials(&run_spec.settings, &run_spec.graph, vault_guard.as_deref())? }; let services = StartServices { run_id, @@ -1112,6 +1112,7 @@ fn stamp_system_worker(mut event: RunEvent) -> RunEvent { fn maybe_build_github_credentials( settings: &WorkflowSettings, + graph: &Graph, vault: Option<&fabro_vault::Vault>, ) -> Result> { let resolved_run = &settings.run; @@ -1127,8 +1128,8 @@ fn maybe_build_github_credentials( return build_github_credentials(strategy, app_id.as_deref(), app_slug.as_deref(), vault); } - let pull_request_enabled = - resolved_run.execution.mode != RunMode::DryRun && resolved_run.pull_request.is_some(); + let pull_request_enabled = resolved_run.execution.mode != RunMode::DryRun + && (resolved_run.pull_request.is_some() || graph_has_pull_request_node(graph)); if pull_request_enabled { return Ok(build_github_credentials( strategy, @@ -1143,6 +1144,13 @@ fn maybe_build_github_credentials( Ok(None) } +fn graph_has_pull_request_node(graph: &Graph) -> bool { + graph + .nodes + .values() + .any(|node| node.node_type() == Some("pull_request")) +} + #[expect( clippy::disallowed_methods, reason = "CLI worker InterpString resolution facade for {{ env.* }} values." @@ -1766,8 +1774,9 @@ mod tests { EnvironmentProvider, RunIntegrationsGithubSettings, RunIntegrationsSettings, RunMode, RunNamespace, }; + use fabro_types::{AttrValue, Graph, Node}; - use super::super::requires_github_credentials; + use super::super::{graph_has_pull_request_node, requires_github_credentials}; fn run_with( permissions: HashMap, @@ -1814,5 +1823,18 @@ mod tests { let run = run_with(HashMap::new(), "docker", RunMode::DryRun); assert!(!requires_github_credentials(&run)); } + + #[test] + fn detects_explicit_pull_request_node_for_soft_credential_loading() { + let mut graph = Graph::new("test"); + let mut node = Node::new("create_pr"); + node.attrs.insert( + "type".to_string(), + AttrValue::String("pull_request".to_string()), + ); + graph.nodes.insert(node.id.clone(), node); + + assert!(graph_has_pull_request_node(&graph)); + } } } diff --git a/lib/apps/fabro-cli/tests/it/cmd/pr.rs b/lib/apps/fabro-cli/tests/it/cmd/pr.rs index de97b712d3..c9d13d5c39 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/pr.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/pr.rs @@ -14,7 +14,7 @@ fn help() { Usage: fabro pr [OPTIONS] Commands: - create Create a pull request from a completed run + create Create a pull request from a run link Link or replace the GitHub pull request associated with a run unlink Unlink the pull request associated with a run view View pull request details diff --git a/lib/apps/fabro-cli/tests/it/cmd/pr_create.rs b/lib/apps/fabro-cli/tests/it/cmd/pr_create.rs index 50b6153093..82d448ce44 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/pr_create.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/pr_create.rs @@ -18,7 +18,7 @@ fn help() { success: true exit_code: 0 ----- stdout ----- - Create a pull request from a completed run + Create a pull request from a run Usage: fabro pr create [OPTIONS] @@ -30,7 +30,7 @@ fn help() { --server Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=] --debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=] --model LLM model for generating PR description - -f, --force Create PR even if the run status is not succeeded/partially_succeeded + -f, --force Use a non-successful conclusion or an active run's committed snapshot --no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true] --quiet Suppress non-essential output [env: FABRO_QUIET=] --verbose Enable verbose output [env: FABRO_VERBOSE=] diff --git a/lib/apps/fabro-server/src/run_files.rs b/lib/apps/fabro-server/src/run_files.rs index 126408e431..2256bf9560 100644 --- a/lib/apps/fabro-server/src/run_files.rs +++ b/lib/apps/fabro-server/src/run_files.rs @@ -1198,7 +1198,7 @@ async fn load_projection( Ok(state.cached_run(run_id).await?.projection) } -async fn reconnect_run_sandbox( +pub(crate) async fn reconnect_run_sandbox( state: &Arc, run_id: &RunId, projection: &fabro_store::RunProjection, 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 f826c813f3..3f8c9ef78e 100644 --- a/lib/apps/fabro-server/src/server/handler/pull_requests.rs +++ b/lib/apps/fabro-server/src/server/handler/pull_requests.rs @@ -1,3 +1,4 @@ +use std::borrow::Cow; use std::sync::Arc; use super::super::{ @@ -6,6 +7,7 @@ use super::super::{ PullRequestLink, RequireRunScoped, Response, Router, RunId, State, StatusCode, get, lock_pull_request_create, post, pull_request, warn, workflow_event, }; +use crate::run_files; pub(super) fn routes() -> Router> { Router::new() @@ -165,11 +167,20 @@ struct RunPrInputs<'a> { goal: &'a str, base_branch: &'a str, run_branch: &'a str, - diff: &'a str, - conclusion: &'a fabro_types::Conclusion, + snapshot: PullRequestSnapshotPolicy<'a>, normalized_origin: String, } +enum PullRequestSnapshotPolicy<'a> { + Conclusion { + diff: &'a str, + conclusion: &'a fabro_types::Conclusion, + }, + LatestCommitted { + base_sha: &'a str, + }, +} + impl<'a> RunPrInputs<'a> { fn extract(run_state: &'a fabro_store::RunProjection, force: bool) -> Result { if let Some(record) = run_state.pull_request.as_ref() { @@ -180,6 +191,67 @@ impl<'a> RunPrInputs<'a> { )); } let run_spec = &run_state.spec; + let active_base_sha = if run_state.conclusion.is_none() { + if !force { + return Err(ApiError::with_code( + StatusCode::BAD_REQUEST, + "Run is not finished yet. Pass --force to use its latest committed snapshot.", + "run_not_finished", + )); + } + match run_state.status { + fabro_types::RunStatus::Running + | fabro_types::RunStatus::Blocked { .. } + | fabro_types::RunStatus::Paused { .. } => {} + fabro_types::RunStatus::Submitted + | fabro_types::RunStatus::Pending { .. } + | fabro_types::RunStatus::Runnable + | fabro_types::RunStatus::Starting => { + return Err(ApiError::with_code( + StatusCode::CONFLICT, + format!( + "Run status is '{}'; no committed active snapshot is ready yet.", + run_state.status + ), + "run_not_ready", + )); + } + fabro_types::RunStatus::Removing + | fabro_types::RunStatus::Succeeded { .. } + | fabro_types::RunStatus::Failed { .. } + | fabro_types::RunStatus::Dead => { + return Err(ApiError::with_code( + StatusCode::CONFLICT, + format!( + "Run status is '{}'; active pull request creation is not available.", + run_state.status + ), + "run_not_eligible", + )); + } + } + if !run_spec.settings.run.run_branch.push { + return Err(ApiError::with_code( + StatusCode::CONFLICT, + "Active pull request creation requires run.run_branch.push = true.", + "run_branch_not_pushed", + )); + } + let base_sha = run_state + .start + .as_ref() + .and_then(|start| start.base_sha.as_deref()) + .ok_or_else(|| { + ApiError::with_code( + StatusCode::BAD_REQUEST, + "Run has no committed base SHA.", + "missing_base_sha", + ) + })?; + Some(base_sha) + } else { + None + }; let origin_url = run_spec.repo_origin_url().ok_or_else(|| { ApiError::with_code( StatusCode::BAD_REQUEST, @@ -205,43 +277,47 @@ impl<'a> RunPrInputs<'a> { "missing_run_branch", ) })?; - let diff = run_state - .conclusion - .as_ref() - .and_then(|conclusion| conclusion.diff.patch.as_deref()) - .filter(|d| !d.trim().is_empty()) - .ok_or_else(|| { + let snapshot = if let Some(conclusion) = run_state.conclusion.as_ref() { + if !force && !conclusion.status.is_successful() { + return Err(ApiError::with_code( + StatusCode::BAD_REQUEST, + format!( + "Run status is '{}', expected succeeded or partially_succeeded", + conclusion.status + ), + "run_not_successful", + )); + } + let diff = conclusion + .diff + .patch + .as_deref() + .filter(|diff| !diff.trim().is_empty()) + .ok_or_else(|| { + ApiError::with_code( + StatusCode::BAD_REQUEST, + "Stored diff is empty — nothing to create a PR for", + "empty_diff", + ) + })?; + PullRequestSnapshotPolicy::Conclusion { diff, conclusion } + } else { + let base_sha = active_base_sha.ok_or_else(|| { ApiError::with_code( StatusCode::BAD_REQUEST, - "Stored diff is empty — nothing to create a PR for", - "empty_diff", + "Run has no committed base SHA.", + "missing_base_sha", ) })?; - let conclusion = run_state.conclusion.as_ref().ok_or_else(|| { - ApiError::with_code( - StatusCode::BAD_REQUEST, - "Run is not finished yet.", - "run_not_finished", - ) - })?; - if !force && !conclusion.status.is_successful() { - return Err(ApiError::with_code( - StatusCode::BAD_REQUEST, - format!( - "Run status is '{}', expected succeeded or partially_succeeded", - conclusion.status - ), - "run_not_successful", - )); - } + PullRequestSnapshotPolicy::LatestCommitted { base_sha } + }; let normalized_origin = fabro_github::normalize_repo_origin_url(origin_url); parse_github_owner_repo_from_url(&normalized_origin, "repo origin URL")?; Ok(Self { goal: run_spec.graph.goal(), base_branch, run_branch, - diff, - conclusion, + snapshot, normalized_origin, }) } @@ -297,6 +373,97 @@ async fn create_run_pull_request( Ok(inputs) => inputs, Err(err) => return err.into_response(), }; + let snapshot_policy = match &inputs.snapshot { + PullRequestSnapshotPolicy::Conclusion { .. } => "conclusion", + PullRequestSnapshotPolicy::LatestCommitted { .. } => "latest_committed", + }; + tracing::debug!( + force = body.force, + run_status = %run_state.status, + snapshot_policy, + "Preparing pull request snapshot" + ); + let (diff, conclusion) = match inputs.snapshot { + PullRequestSnapshotPolicy::Conclusion { diff, conclusion } => { + (Cow::Borrowed(diff), Some(conclusion)) + } + PullRequestSnapshotPolicy::LatestCommitted { base_sha } => { + let sandbox = match run_files::reconnect_run_sandbox(&state, &id, run_state).await { + Ok(sandbox) => sandbox, + Err(err) => { + warn!( + status = %err.status(), + "Failed to reconnect active run sandbox for pull request creation" + ); + let response = if err.status() == StatusCode::NOT_FOUND { + ApiError::with_code( + StatusCode::CONFLICT, + "The active run sandbox is not ready.", + "run_not_ready", + ) + } else { + ApiError::with_code( + StatusCode::SERVICE_UNAVAILABLE, + "The active run sandbox is unavailable.", + "snapshot_unavailable", + ) + }; + return response.into_response(); + } + }; + let snapshot = match pull_request::prepare_committed_pull_request_snapshot( + sandbox.as_ref(), + base_sha, + inputs.run_branch, + ) + .await + { + Ok(snapshot) => snapshot, + Err(err) => { + warn!( + error = %err, + "Failed to prepare active pull request snapshot" + ); + let response = match err { + pull_request::CommittedPullRequestSnapshotError::Push { .. } + | pull_request::CommittedPullRequestSnapshotError::RemoteBranchMissingCommit => { + ApiError::with_code( + StatusCode::CONFLICT, + "The captured commit is not available on the remote run branch.", + "run_branch_not_pushed", + ) + } + pull_request::CommittedPullRequestSnapshotError::Sandbox { .. } + | pull_request::CommittedPullRequestSnapshotError::Command { .. } + | pull_request::CommittedPullRequestSnapshotError::InvalidHead => { + ApiError::with_code( + StatusCode::SERVICE_UNAVAILABLE, + "The latest committed run snapshot could not be prepared.", + "snapshot_unavailable", + ) + } + pull_request::CommittedPullRequestSnapshotError::InvalidBase => { + ApiError::with_code( + StatusCode::BAD_REQUEST, + "Run has an invalid committed base SHA.", + "invalid_base_sha", + ) + } + }; + return response.into_response(); + } + }; + if snapshot.diff.trim().is_empty() { + return ApiError::with_code( + StatusCode::BAD_REQUEST, + "Committed diff is empty — nothing to create a PR for", + "empty_committed_diff", + ) + .into_response(); + } + (Cow::Owned(snapshot.diff), None) + } + }; let creds = match load_server_github_credentials(state.as_ref()).await { Ok(creds) => creds, Err(err) => return err.into_response(), @@ -324,14 +491,14 @@ async fn create_run_pull_request( base_branch: inputs.base_branch, head_branch: inputs.run_branch, goal: inputs.goal, - diff: inputs.diff, + diff: diff.as_ref(), model: &model, draft: true, auto_merge: None, run_store: &run_store_handle, llm_source: state.llm_source.as_ref(), catalog, - conclusion: Some(inputs.conclusion), + conclusion, run_state: Some(run_state), }; let created_pull_request = match pull_request::maybe_open_pull_request(request).await { @@ -343,16 +510,33 @@ async fn create_run_pull_request( ) .into_response(); } - Err(err) => return ApiError::new(StatusCode::BAD_GATEWAY, err).into_response(), + Err(err) => { + warn!( + error = %err, + "Pull request creation failed" + ); + return ApiError::new( + StatusCode::BAD_GATEWAY, + "GitHub pull request creation failed.", + ) + .into_response(); + } }; - let event = workflow_event::Event::pull_request_created( - &created_pull_request.link, - &created_pull_request.base_branch, - &created_pull_request.head_branch, - &created_pull_request.title, - true, - ); + let event = match created_pull_request.disposition { + pull_request::PullRequestDisposition::Created => { + workflow_event::Event::pull_request_created( + &created_pull_request.link, + &created_pull_request.base_branch, + &created_pull_request.head_branch, + &created_pull_request.title, + true, + ) + } + pull_request::PullRequestDisposition::Linked => workflow_event::Event::PullRequestLinked { + pull_request: created_pull_request.link.clone(), + }, + }; if let Err(err) = workflow_event::append_event(&run_store, &id, &event).await { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(); } @@ -522,3 +706,95 @@ async fn close_run_pull_request( Err(err) => ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response(), } } + +#[cfg(test)] +mod tests { + use super::*; + + fn submitted_projection() -> fabro_store::RunProjection { + fabro_store::RunProjection::new( + "Test run".to_string(), + fabro_types::RunSpec { + run_id: fabro_types::fixtures::RUN_1, + settings: fabro_types::WorkflowSettings::default(), + graph: fabro_types::Graph::new("test"), + graph_source: None, + workflow_slug: None, + automation: None, + source_directory: None, + labels: std::collections::HashMap::new(), + provenance: fabro_types::test_support::test_run_provenance(), + manifest_blob: None, + definition_blob: None, + git: None, + fork_source_ref: None, + }, + chrono::Utc::now(), + ) + } + + #[test] + fn active_run_without_force_is_classified_before_git_metadata() { + let projection = submitted_projection(); + let Err(error) = RunPrInputs::extract(&projection, false) else { + panic!("submitted run should not be eligible"); + }; + + assert_eq!(error.code(), Some("run_not_finished")); + } + + #[test] + fn force_distinguishes_not_ready_and_ineligible_states() { + let mut projection = submitted_projection(); + let Err(not_ready) = RunPrInputs::extract(&projection, true) else { + panic!("submitted run should not have a ready snapshot"); + }; + assert_eq!(not_ready.code(), Some("run_not_ready")); + + projection.status = fabro_types::RunStatus::Dead; + let Err(not_eligible) = RunPrInputs::extract(&projection, true) else { + panic!("dead run should not be eligible"); + }; + assert_eq!(not_eligible.code(), Some("run_not_eligible")); + } + + #[test] + fn force_accepts_running_blocked_and_paused_snapshots() { + let eligible = [ + fabro_types::RunStatus::Running, + fabro_types::RunStatus::Blocked { + blocked_reason: fabro_types::BlockedReason::HumanInputRequired, + }, + fabro_types::RunStatus::Paused { + prior_block: Some(fabro_types::BlockedReason::HumanInputRequired), + }, + ]; + + for status in eligible { + let mut projection = submitted_projection(); + projection.status = status; + projection.spec.git = Some(fabro_types::GitContext { + origin_url: "https://github.com/acme/widgets.git".to_string(), + branch: "main".to_string(), + sha: Some("0123456789abcdef".to_string()), + dirty: fabro_types::DirtyStatus::Clean, + push_outcome: fabro_types::PreRunPushOutcome::NotAttempted, + }); + projection.start = Some(fabro_types::StartRecord { + start_time: chrono::Utc::now(), + run_branch: Some("fabro/run/test".to_string()), + base_sha: Some("0123456789abcdef".to_string()), + }); + + let Ok(inputs) = RunPrInputs::extract(&projection, true) else { + panic!("{status} should be eligible"); + }; + assert!(matches!( + inputs.snapshot, + PullRequestSnapshotPolicy::LatestCommitted { + base_sha: "0123456789abcdef", + } + )); + } + } +} diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 71b4dfebf0..41c93a93b0 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -6528,6 +6528,27 @@ async fn create_completed_run_ready_for_pull_request( .await; } +#[expect( + clippy::disallowed_methods, + reason = "Temporary git fixture setup is intentionally synchronous." +)] +fn pr_test_git(repo: &Path, args: &[&str]) -> String { + let output = std::process::Command::new("git") + .args(args) + .current_dir(repo) + .output() + .expect("git should execute"); + assert!( + output.status.success(), + "git {args:?} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout) + .expect("git output should be UTF-8") + .trim() + .to_string() +} + fn test_event_envelope(seq: u32, run_id: RunId, body: EventBody) -> EventEnvelope { EventEnvelope { seq, @@ -9058,6 +9079,18 @@ 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 find_mock = github.mock(|when, then| { + when.method("GET") + .path("/repos/acme/widgets/pulls") + .query_param("state", "open") + .query_param("head", "acme:fabro/run/42") + .query_param("base", "main") + .query_param("per_page", "2") + .header("authorization", "Bearer ghu_test"); + then.status(200) + .header("content-type", "application/json") + .body("[]"); + }); let create_mock = github.mock(|when, then| { when.method("POST") .path("/repos/acme/widgets/pulls") @@ -9155,9 +9188,262 @@ async fn create_run_pull_request_creates_and_persists_record() { assert_eq!(state_body["pull_request"]["repo"], "widgets"); response_mock.assert_async().await; + find_mock.assert(); create_mock.assert(); } +#[tokio::test] +async fn create_run_pull_request_adopts_matching_open_pull_request() { + let github = MockServer::start(); + let find_mock = github.mock(|when, then| { + when.method("GET") + .path("/repos/acme/widgets/pulls") + .query_param("state", "open") + .query_param("head", "acme:fabro/run/42") + .query_param("base", "main") + .query_param("per_page", "2") + .header("authorization", "Bearer ghu_test"); + then.status(200) + .header("content-type", "application/json") + .body( + json!([{ + "html_url": "https://github.com/acme/widgets/pull/42", + "number": 42 + }]) + .to_string(), + ); + }); + let (state, app, run_id) = Box::pin(pr_test_app_with_completed_run( + Some("ghu_test"), + Some(github.base_url()), + Some("https://github.com/acme/widgets.git"), + )) + .await; + + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/pull_request"))) + .header("content-type", "application/json") + .body(Body::from( + json!({ "force": false, "model": null }).to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + let body = response_json!(response, StatusCode::OK).await; + + assert_eq!(body["number"], 42); + let projection = state.cached_run(&run_id).await.unwrap(); + assert_eq!( + projection.projection.pull_request, + Some(PullRequestLink { + owner: "acme".to_string(), + repo: "widgets".to_string(), + number: 42, + }) + ); + find_mock.assert(); +} + +#[tokio::test] +#[expect( + clippy::disallowed_methods, + reason = "Temporary git fixture setup is intentionally synchronous." +)] +async fn create_run_pull_request_force_uses_running_runs_latest_committed_snapshot() { + let repo_dir = tempfile::tempdir().unwrap(); + let remote_dir = tempfile::tempdir().unwrap(); + pr_test_git(repo_dir.path(), &["init", "-b", "main"]); + pr_test_git(repo_dir.path(), &[ + "config", + "user.email", + "fabro@example.test", + ]); + pr_test_git(repo_dir.path(), &["config", "user.name", "Fabro Test"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\n").unwrap(); + pr_test_git(repo_dir.path(), &["add", "tracked.txt"]); + pr_test_git(repo_dir.path(), &["commit", "-m", "base"]); + let base_sha = pr_test_git(repo_dir.path(), &["rev-parse", "HEAD"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\ncommitted\n").unwrap(); + pr_test_git(repo_dir.path(), &["add", "tracked.txt"]); + pr_test_git(repo_dir.path(), &["commit", "-m", "committed"]); + let head_sha = pr_test_git(repo_dir.path(), &["rev-parse", "HEAD"]); + std::fs::write( + repo_dir.path().join("tracked.txt"), + "base\ncommitted\ndirty\n", + ) + .unwrap(); + pr_test_git(remote_dir.path(), &["init", "--bare"]); + pr_test_git(repo_dir.path(), &[ + "remote", + "add", + "origin", + remote_dir.path().to_str().unwrap(), + ]); + + let github = MockServer::start(); + let find_mock = github.mock(|when, then| { + when.method("GET") + .path("/repos/acme/widgets/pulls") + .query_param("state", "open") + .query_param("head", "acme:fabro/run/active") + .query_param("base", "main") + .query_param("per_page", "2") + .header("authorization", "Bearer ghu_test"); + then.status(200) + .header("content-type", "application/json") + .body("[]"); + }); + let create_mock = github.mock(|when, then| { + when.method("POST") + .path("/repos/acme/widgets/pulls") + .header("authorization", "Bearer ghu_test"); + then.status(201) + .header("content-type", "application/json") + .body( + json!({ + "html_url": "https://github.com/acme/widgets/pull/77", + "number": 77, + "node_id": "PR_kwDOActive" + }) + .to_string(), + ); + }); + let llm = MockServer::start_async().await; + let response_mock = llm + .mock_async(|when, then| { + when.method(POST) + .path("/v1/responses") + .header("authorization", "Bearer openai-key"); + then.status(200) + .header("content-type", "application/json") + .json_body(openai_responses_payload( + &serde_json::to_string(&json!({ + "title": "Active snapshot", + "body": "Created while the run is active.", + })) + .unwrap(), + )); + }) + .await; + let state = create_github_token_app_state_with_env_lookup_and_llm_catalog_settings( + Some("ghu_test"), + Some(github.base_url()), + |_| None, + llm_catalog_settings_with_provider_base_url("openai", llm.url("/v1")), + ); + state + .stores + .vault + .set("OPENAI_API_KEY", "openai-key", SecretType::Token, None) + .await + .unwrap(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = fixtures::RUN_1; + let mut settings = WorkflowSettings::default(); + settings.run.run_branch.enabled = true; + settings.run.run_branch.push = true; + let mut graph = Graph::new("active-pr"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("Open an active pull request".to_string()), + ); + let git = fabro_types::GitContext { + origin_url: "https://github.com/acme/widgets.git".to_string(), + branch: "main".to_string(), + sha: Some(base_sha.clone()), + dirty: fabro_types::DirtyStatus::Clean, + push_outcome: fabro_types::PreRunPushOutcome::NotAttempted, + }; + create_durable_run_with_events(&state, run_id, &[ + workflow_event::Event::RunCreated { + run_id, + title: None, + settings: serde_json::to_value(&settings).unwrap(), + graph: serde_json::to_value(&graph).unwrap(), + workflow_source: None, + workflow_config: None, + labels: std::collections::BTreeMap::default(), + run_dir: repo_dir.path().display().to_string(), + source_directory: Some(repo_dir.path().display().to_string()), + workflow_slug: Some("active-pr".to_string()), + automation: None, + db_prefix: None, + provenance: test_support::test_run_provenance(), + manifest_blob: None, + git: Some(git), + fork_source_ref: None, + retried_from: None, + parent_id: None, + web_url: None, + }, + workflow_event::Event::RunSubmitted { + definition_blob: None, + }, + workflow_event::Event::RunRunnable { + source: fabro_types::RunRunnableSource::StartRequested, + actor: None, + }, + workflow_event::Event::RunStarting, + workflow_event::Event::WorkflowRunStarted { + name: "active-pr".to_string(), + run_id, + base_branch: Some("main".to_string()), + base_sha: Some(base_sha.clone()), + run_branch: Some("fabro/run/active".to_string()), + worktree_dir: None, + goal: Some("Open an active pull request".to_string()), + }, + workflow_event::Event::RunRunning, + workflow_event::Event::SandboxInitialized { + working_directory: repo_dir.path().display().to_string(), + provider: SandboxProviderKind::Local, + id: "active-pr-local".to_string(), + image: None, + snapshot: None, + repo_cloned: None, + clone_origin_url: None, + clone_branch: None, + workspace_root: None, + repos_root: None, + primary_repo_path: None, + primary_repo_link: None, + }, + ]) + .await; + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/pull_request"))) + .header("content-type", "application/json") + .body(Body::from( + json!({ "force": true, "model": "gpt-5.4" }).to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + let body = response_json!(response, StatusCode::OK).await; + + assert_eq!(body["number"], 77); + assert_eq!( + pr_test_git(remote_dir.path(), &[ + "rev-parse", + "refs/heads/fabro/run/active" + ]), + head_sha + ); + find_mock.assert(); + create_mock.assert(); + response_mock.assert_async().await; +} + #[tokio::test] async fn create_run_pull_request_returns_conflict_when_record_exists() { let (state, app, run_id) = pr_test_app(None, None); diff --git a/lib/components/fabro-github/Cargo.toml b/lib/components/fabro-github/Cargo.toml index 9db73130e9..d77b759222 100644 --- a/lib/components/fabro-github/Cargo.toml +++ b/lib/components/fabro-github/Cargo.toml @@ -26,6 +26,7 @@ tracing.workspace = true tokio = { workspace = true } base64.workspace = true thiserror.workspace = true +url.workspace = true [dev-dependencies] fabro-macros = { path = "../../foundation/fabro-macros" } diff --git a/lib/components/fabro-github/src/lib.rs b/lib/components/fabro-github/src/lib.rs index bac1f5fbc0..5f660cdf42 100644 --- a/lib/components/fabro-github/src/lib.rs +++ b/lib/components/fabro-github/src/lib.rs @@ -67,6 +67,30 @@ pub enum PullRequestApiError { repo: String, number: u64, }, + #[error( + "Found {count} open pull requests in {owner}/{repo} for {head} into {base}; expected at most one" + )] + AmbiguousMatch { + owner: String, + repo: String, + base: String, + head: String, + count: usize, + }, + #[error(transparent)] + Other(#[from] anyhow::Error), +} + +/// Errors returned while creating a pull request. +/// +/// GitHub uses `422 Unprocessable Entity` both for validation failures and +/// for the "a pull request already exists for this branch" race. Callers that +/// reconcile by branch need to distinguish that status from transport and +/// authentication failures without parsing an error string. +#[derive(Debug, thiserror::Error)] +pub enum CreatePullRequestError { + #[error("GitHub rejected pull request creation (422): {details}")] + UnprocessableEntity { details: String }, #[error(transparent)] Other(#[from] anyhow::Error), } @@ -627,12 +651,20 @@ pub async fn create_installation_access_token_for_pr( } /// Result of a successful pull request creation. +#[derive(Debug, Clone, PartialEq, Eq)] pub struct CreatedPullRequest { pub html_url: String, pub number: u64, pub node_id: String, } +/// Minimal result for an existing open pull request matched by head and base. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MatchingPullRequest { + pub html_url: String, + pub number: u64, +} + /// Create a pull request on GitHub. /// /// Signs a JWT, obtains a PR-scoped installation token, and POSTs to the @@ -650,8 +682,8 @@ pub async fn create_pull_request( title: &str, body: &str, draft: bool, -) -> anyhow::Result { - let client = ctx.http_client()?; +) -> Result { + let client = ctx.http_client().map_err(CreatePullRequestError::Other)?; create_pull_request_with_client(&client, ctx, owner, repo, base, head, title, body, draft).await } @@ -669,7 +701,7 @@ pub async fn create_pull_request_with_client( title: &str, body: &str, draft: bool, -) -> anyhow::Result { +) -> Result { #[derive(Deserialize)] struct PullRequestResponse { html_url: String, @@ -713,20 +745,24 @@ pub async fn create_pull_request_with_client( match resp.status { 201 => {} 422 => { - bail!("Pull request could not be created (422): {}", resp.text()); + return Err(CreatePullRequestError::UnprocessableEntity { + details: resp.text().to_string(), + }); } 401 | 403 => { - bail!( + return Err(anyhow!( "Authentication failed creating pull request ({})", resp.status - ); + ) + .into()); } _ => { - bail!( + return Err(anyhow!( "Unexpected status {} creating pull request: {}", resp.status, resp.text() - ); + ) + .into()); } } @@ -741,6 +777,91 @@ pub async fn create_pull_request_with_client( }) } +/// Find an open pull request whose head and base branches match exactly. +pub async fn find_open_pull_request( + ctx: &GitHubContext<'_>, + owner: &str, + repo: &str, + base: &str, + head: &str, +) -> Result, PullRequestApiError> { + let client = ctx.http_client()?; + find_open_pull_request_with_client(&client, ctx, owner, repo, base, head).await +} + +pub async fn find_open_pull_request_with_client( + client: &impl HttpClient, + ctx: &GitHubContext<'_>, + owner: &str, + repo: &str, + base: &str, + head: &str, +) -> Result, PullRequestApiError> { + #[derive(Deserialize)] + struct PullRequestResponse { + html_url: String, + number: u64, + } + + let token = ctx + .creds + .resolve_bearer_token( + client, + owner, + repo, + ctx.base_url, + serde_json::json!({ "contents": "write", "pull_requests": "write" }), + ) + .await?; + let query = url::form_urlencoded::Serializer::new(String::new()) + .append_pair("state", "open") + .append_pair("head", &format!("{owner}:{head}")) + .append_pair("base", base) + .append_pair("per_page", "2") + .finish(); + let url = format!("{}/repos/{owner}/{repo}/pulls?{query}", ctx.base_url); + let auth = format!("Bearer {token}"); + let resp = client + .request(HttpMethod::Get, &url, &github_headers(&auth), None) + .await + .context("Failed to find matching pull request")?; + + match resp.status { + 200 => {} + 401 | 403 => { + return Err(anyhow!( + "Authentication failed finding matching pull request ({})", + resp.status + ) + .into()); + } + status => { + return Err(anyhow!( + "Unexpected status {status} finding matching pull request: {}", + resp.text() + ) + .into()); + } + } + + let mut pull_requests = resp + .json::>() + .context("Failed to parse matching pull request response")?; + if pull_requests.len() > 1 { + return Err(PullRequestApiError::AmbiguousMatch { + owner: owner.to_string(), + repo: repo.to_string(), + base: base.to_string(), + head: head.to_string(), + count: pull_requests.len(), + }); + } + Ok(pull_requests.pop().map(|pull_request| MatchingPullRequest { + html_url: pull_request.html_url, + number: pull_request.number, + })) +} + fn merge_method_as_graphql_value(method: MergeStrategy) -> &'static str { match method { MergeStrategy::Merge => "MERGE", @@ -1710,6 +1831,122 @@ mod tests { } } + #[tokio::test] + async fn find_open_pull_request_matches_encoded_head_and_base() { + let mock = MockHttpClient::new() + .on( + HttpMethod::Get, + "/repos/owner/repo/pulls?state=open&head=owner%3Afabro%2Frun%2F123&base=main&per_page=2", + 200, + r#"[{"html_url":"https://github.com/owner/repo/pull/42","number":42}]"#, + ) + .with_req_header("Authorization", "Bearer ghu_test"); + let creds = GitHubCredentials::Pat("ghu_test".to_string()); + + let pull_request = find_open_pull_request_with_client( + &mock, + &GitHubContext::new(&creds, ""), + "owner", + "repo", + "main", + "fabro/run/123", + ) + .await + .unwrap(); + + assert_eq!( + pull_request, + Some(MatchingPullRequest { + html_url: "https://github.com/owner/repo/pull/42".to_string(), + number: 42, + }) + ); + } + + #[tokio::test] + async fn find_open_pull_request_returns_none_for_empty_list() { + let mock = MockHttpClient::new().on( + HttpMethod::Get, + "/repos/owner/repo/pulls?state=open&head=owner%3Afeature&base=main&per_page=2", + 200, + "[]", + ); + let creds = GitHubCredentials::Pat("ghu_test".to_string()); + + let pull_request = find_open_pull_request_with_client( + &mock, + &GitHubContext::new(&creds, ""), + "owner", + "repo", + "main", + "feature", + ) + .await + .unwrap(); + + assert_eq!(pull_request, None); + } + + #[tokio::test] + async fn find_open_pull_request_rejects_ambiguous_matches() { + let mock = MockHttpClient::new().on( + HttpMethod::Get, + "/repos/owner/repo/pulls?state=open&head=owner%3Afeature&base=main&per_page=2", + 200, + r#"[ + {"html_url":"https://github.com/owner/repo/pull/41","number":41}, + {"html_url":"https://github.com/owner/repo/pull/42","number":42} + ]"#, + ); + let creds = GitHubCredentials::Pat("ghu_test".to_string()); + + let error = find_open_pull_request_with_client( + &mock, + &GitHubContext::new(&creds, ""), + "owner", + "repo", + "main", + "feature", + ) + .await + .unwrap_err(); + + assert!(matches!(error, PullRequestApiError::AmbiguousMatch { + count: 2, + .. + })); + } + + #[tokio::test] + async fn create_pull_request_exposes_unprocessable_entity_status() { + let mock = MockHttpClient::new().on( + HttpMethod::Post, + "/repos/owner/repo/pulls", + 422, + r#"{"message":"A pull request already exists"}"#, + ); + let creds = GitHubCredentials::Pat("ghu_test".to_string()); + + let error = create_pull_request_with_client( + &mock, + &GitHubContext::new(&creds, ""), + "owner", + "repo", + "main", + "feature", + "Title", + "Body", + true, + ) + .await + .unwrap_err(); + + assert!(matches!( + error, + CreatePullRequestError::UnprocessableEntity { .. } + )); + } + // ----------------------------------------------------------------------- // create_installation_access_token — success // ----------------------------------------------------------------------- diff --git a/lib/components/fabro-workflow/src/context.rs b/lib/components/fabro-workflow/src/context.rs index 6e94368b24..f486cce4b0 100644 --- a/lib/components/fabro-workflow/src/context.rs +++ b/lib/components/fabro-workflow/src/context.rs @@ -48,6 +48,15 @@ pub mod keys { pub const HUMAN_GATE_LABEL: &str = "human.gate.label"; pub const HUMAN_GATE_TEXT: &str = "human.gate.text"; + // --- pull_request.* keys --- + pub const PULL_REQUEST_URL: &str = "pull_request.url"; + pub const PULL_REQUEST_NUMBER: &str = "pull_request.number"; + pub const PULL_REQUEST_OWNER: &str = "pull_request.owner"; + pub const PULL_REQUEST_REPO: &str = "pull_request.repo"; + pub const PULL_REQUEST_BASE_BRANCH: &str = "pull_request.base_branch"; + pub const PULL_REQUEST_HEAD_BRANCH: &str = "pull_request.head_branch"; + pub const PULL_REQUEST_DRAFT: &str = "pull_request.draft"; + // --- parallel.* keys --- pub const PARALLEL_RESULTS: &str = "parallel.results"; pub const PARALLEL_BRANCH_COUNT: &str = "parallel.branch_count"; diff --git a/lib/components/fabro-workflow/src/handler/mod.rs b/lib/components/fabro-workflow/src/handler/mod.rs index bc94af3b88..ac0cdeba3e 100644 --- a/lib/components/fabro-workflow/src/handler/mod.rs +++ b/lib/components/fabro-workflow/src/handler/mod.rs @@ -8,6 +8,7 @@ pub mod llm; pub mod manager_loop; pub mod parallel; pub mod prompt; +pub mod pull_request; pub mod start; pub mod structured_output; pub mod wait; @@ -178,6 +179,7 @@ pub fn default_registry( registry.register("human", Box::new(human::HumanHandler::new(interviewer))); registry.register("command", Box::new(command::CommandHandler)); registry.register("tool", Box::new(command::CommandHandler)); + registry.register("pull_request", Box::new(pull_request::PullRequestHandler)); registry.register("parallel", Box::new(parallel::ParallelHandler)); registry.register( "parallel.fan_in", diff --git a/lib/components/fabro-workflow/src/handler/pull_request.rs b/lib/components/fabro-workflow/src/handler/pull_request.rs new file mode 100644 index 0000000000..31657dceee --- /dev/null +++ b/lib/components/fabro-workflow/src/handler/pull_request.rs @@ -0,0 +1,555 @@ +use std::path::Path; + +use async_trait::async_trait; +use fabro_graphviz::graph::{Graph, Node}; +use fabro_types::PullRequestLink; + +use super::{EngineServices, Handler}; +use crate::context::{Context, WorkflowContext, keys}; +use crate::error::Error; +use crate::event::{Event, StageScope}; +use crate::outcome::Outcome; +use crate::pull_request::{ + AutoMergeOptions, CommittedPullRequestSnapshotError, OpenPullRequestRequest, + PullRequestDisposition, maybe_open_pull_request, prepare_committed_pull_request_snapshot, +}; + +pub struct PullRequestHandler; + +fn outcome_for_pull_request( + link: &PullRequestLink, + base_branch: &str, + head_branch: &str, +) -> Outcome { + let mut outcome = Outcome::success(); + outcome.context_updates.insert( + keys::PULL_REQUEST_URL.to_string(), + serde_json::json!(link.html_url()), + ); + outcome.context_updates.insert( + keys::PULL_REQUEST_NUMBER.to_string(), + serde_json::json!(link.number), + ); + outcome.context_updates.insert( + keys::PULL_REQUEST_OWNER.to_string(), + serde_json::json!(link.owner), + ); + outcome.context_updates.insert( + keys::PULL_REQUEST_REPO.to_string(), + serde_json::json!(link.repo), + ); + outcome.context_updates.insert( + keys::PULL_REQUEST_BASE_BRANCH.to_string(), + serde_json::json!(base_branch), + ); + outcome.context_updates.insert( + keys::PULL_REQUEST_HEAD_BRANCH.to_string(), + serde_json::json!(head_branch), + ); + outcome.context_updates.insert( + keys::PULL_REQUEST_DRAFT.to_string(), + serde_json::json!(true), + ); + outcome +} + +fn required<'a>(value: Option<&'a str>, message: &str) -> Result<&'a str, Error> { + value.ok_or_else(|| Error::Precondition(message.to_string())) +} + +fn model_for_node<'a>(node: &'a Node, run_model: &'a str) -> &'a str { + node.model().unwrap_or(run_model) +} + +fn validate_github_origin(origin_url: &str) -> Result<(), Error> { + let https_url = fabro_github::ssh_url_to_https(origin_url); + fabro_github::parse_github_owner_repo(&https_url) + .map(|_| ()) + .map_err(|_| { + Error::Precondition( + "pull_request nodes require a valid github.com repo origin".to_string(), + ) + }) +} + +#[async_trait] +impl Handler for PullRequestHandler { + async fn simulate( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _run_dir: &Path, + _services: &EngineServices, + ) -> Result { + Ok(outcome_for_pull_request( + &PullRequestLink { + owner: "fabro".to_string(), + repo: "dry-run".to_string(), + number: 1, + }, + "main", + "fabro/run/dry-run", + )) + } + + async fn execute( + &self, + node: &Node, + context: &Context, + graph: &Graph, + _run_dir: &Path, + services: &EngineServices, + ) -> Result { + let scope = StageScope::for_handler(context, &node.id); + let result = async { + if context.parallel_branch_id().is_some() { + return Err(Error::Precondition( + "pull_request nodes cannot execute inside a parallel branch".to_string(), + )); + } + + let run_state = services + .run + .run_store + .state() + .await + .map_err(|err| Error::handler_with_source("Failed to load run state", err))?; + if let Some(link) = run_state.pull_request.as_ref() { + let runtime = services.run.pull_request.as_ref(); + let base_branch = runtime + .and_then(|runtime| runtime.base_branch.as_deref()) + .unwrap_or("main"); + let head_branch = runtime + .and_then(|runtime| runtime.head_branch.as_deref()) + .unwrap_or(""); + return Ok(outcome_for_pull_request(link, base_branch, head_branch)); + } + + let runtime = services.run.pull_request.as_ref().ok_or_else(|| { + Error::Precondition("pull request runtime is unavailable".to_string()) + })?; + if !runtime.push_enabled { + return Err(Error::Precondition( + "pull_request nodes require run.run_branch.push = true".to_string(), + )); + } + let credentials = runtime.github.as_ref().ok_or_else(|| { + Error::Precondition( + "GitHub credentials are required for pull_request nodes".to_string(), + ) + })?; + let origin_url = required( + runtime.origin_url.as_deref(), + "pull_request nodes require a GitHub repo origin", + )?; + let base_branch = required( + runtime.base_branch.as_deref(), + "pull_request nodes require a base branch", + )?; + let head_branch = required( + runtime.head_branch.as_deref(), + "pull_request nodes require an enabled run branch", + )?; + let base_sha = required( + runtime.base_sha.as_deref(), + "pull_request nodes require a committed base SHA", + )?; + validate_github_origin(origin_url)?; + + let snapshot = prepare_committed_pull_request_snapshot( + services.run.sandbox.as_ref(), + base_sha, + head_branch, + ) + .await + .map_err(|err| match err { + CommittedPullRequestSnapshotError::InvalidBase => Error::Precondition( + "pull_request nodes require a valid committed base SHA".to_string(), + ), + err => Error::handler_with_source("Failed to prepare pull request snapshot", err), + })?; + if snapshot.diff.trim().is_empty() { + return Err(Error::Precondition( + "pull_request node found no committed changes to open".to_string(), + )); + } + + let model = model_for_node(node, &services.run.model); + let opened = maybe_open_pull_request(OpenPullRequestRequest { + github: fabro_github::GitHubContext::new(credentials, &runtime.github_base_url), + origin_url, + base_branch, + head_branch, + goal: graph.goal(), + diff: &snapshot.diff, + model, + draft: true, + auto_merge: None::, + run_store: &services.run.run_store, + llm_source: services.run.llm_source.as_ref(), + catalog: services.run.catalog.clone(), + conclusion: run_state.conclusion.as_ref(), + run_state: Some(&run_state), + }) + .await + .map_err(|err| Error::handler_with_anyhow("Pull request creation failed", err))? + .ok_or_else(|| { + Error::Precondition( + "pull_request node found no committed changes to open".to_string(), + ) + })?; + + match opened.disposition { + PullRequestDisposition::Created => { + services.run.emitter.emit_scoped( + &Event::pull_request_created( + &opened.link, + &opened.base_branch, + &opened.head_branch, + &opened.title, + true, + ), + &scope, + ); + } + PullRequestDisposition::Linked => { + services.run.emitter.emit_scoped( + &Event::PullRequestLinked { + pull_request: opened.link.clone(), + }, + &scope, + ); + } + } + + Ok(outcome_for_pull_request( + &opened.link, + &opened.base_branch, + &opened.head_branch, + )) + } + .await; + + if let Err(error) = &result { + services.run.emitter.emit_scoped( + &Event::PullRequestFailed { + error: error.to_string(), + }, + &scope, + ); + } + result + } +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use fabro_graphviz::graph::{AttrValue, Graph, Node}; + use fabro_store::Database; + use fabro_types::{WorkflowSettings, fixtures, test_support}; + use httpmock::MockServer; + use object_store::memory::InMemory; + + use super::*; + use crate::event::{Emitter, append_event}; + use crate::runtime_store::RunStoreHandle; + use crate::services::PullRequestRuntime; + + #[expect( + clippy::disallowed_methods, + reason = "Temporary git fixture setup is intentionally synchronous." + )] + fn git(repo: &Path, args: &[&str]) -> String { + let output = std::process::Command::new("git") + .args(args) + .current_dir(repo) + .output() + .expect("git should execute"); + assert!( + output.status.success(), + "git {args:?} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout) + .expect("git output should be UTF-8") + .trim() + .to_string() + } + + async fn test_run_store(graph: &Graph) -> RunStoreHandle { + let store = Arc::new(Database::new( + Arc::new(InMemory::new()), + "", + Duration::from_millis(1), + None, + )); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); + append_event(&run_store, &fixtures::RUN_1, &Event::RunCreated { + run_id: fixtures::RUN_1, + title: None, + settings: serde_json::to_value(WorkflowSettings::default()).unwrap(), + graph: serde_json::to_value(graph).unwrap(), + workflow_source: None, + workflow_config: None, + labels: std::collections::BTreeMap::new(), + run_dir: "/tmp/test".to_string(), + source_directory: None, + workflow_slug: Some("test".to_string()), + automation: None, + db_prefix: None, + provenance: test_support::test_run_provenance(), + manifest_blob: None, + git: None, + fork_source_ref: None, + retried_from: None, + parent_id: None, + web_url: None, + }) + .await + .unwrap(); + run_store.into() + } + + #[tokio::test] + async fn dry_run_returns_deterministic_pull_request_context() { + let outcome = PullRequestHandler + .simulate( + &Node::new("create_pr"), + &Context::new(), + &Graph::new("test"), + Path::new("."), + &EngineServices::test_default(), + ) + .await + .unwrap(); + + assert_eq!( + outcome.context_updates[keys::PULL_REQUEST_URL], + serde_json::json!("https://github.com/fabro/dry-run/pull/1") + ); + assert_eq!( + outcome.context_updates[keys::PULL_REQUEST_DRAFT], + serde_json::json!(true) + ); + } + + #[tokio::test] + async fn pull_request_node_rejects_parallel_branch_context() { + let context = Context::new(); + context.set( + keys::INTERNAL_PARALLEL_BRANCH_ID, + serde_json::json!("fanout@1:0"), + ); + + let error = PullRequestHandler + .execute( + &Node::new("create_pr"), + &context, + &Graph::new("test"), + Path::new("."), + &EngineServices::test_default(), + ) + .await + .unwrap_err(); + + assert!(matches!(error, Error::Precondition(_))); + assert!(error.to_string().contains("parallel branch")); + } + + #[test] + fn node_model_overrides_run_model() { + let mut node = Node::new("create_pr"); + node.attrs.insert( + "model".to_string(), + AttrValue::String("node-model".to_string()), + ); + + assert_eq!(model_for_node(&node, "run-model"), "node-model"); + assert_eq!( + model_for_node(&Node::new("create_pr"), "run-model"), + "run-model" + ); + } + + #[test] + fn github_origin_validation_accepts_https_and_scp_syntax() { + assert!(validate_github_origin("https://github.com/acme/widgets.git").is_ok()); + assert!(validate_github_origin("git@github.com:acme/widgets.git").is_ok()); + assert!(validate_github_origin("https://gitlab.com/acme/widgets.git").is_err()); + } + + #[tokio::test] + #[expect( + clippy::disallowed_methods, + reason = "Temporary git fixture setup is intentionally synchronous." + )] + async fn creates_context_and_links_matching_pull_request_from_committed_snapshot() { + let repo_dir = tempfile::tempdir().unwrap(); + let remote_dir = tempfile::tempdir().unwrap(); + git(repo_dir.path(), &["init", "-b", "main"]); + git(repo_dir.path(), &[ + "config", + "user.email", + "fabro@example.test", + ]); + git(repo_dir.path(), &["config", "user.name", "Fabro Test"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "base"]); + let base_sha = git(repo_dir.path(), &["rev-parse", "HEAD"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\ncommitted\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "committed"]); + let head_sha = git(repo_dir.path(), &["rev-parse", "HEAD"]); + git(remote_dir.path(), &["init", "--bare"]); + git(repo_dir.path(), &[ + "remote", + "add", + "origin", + remote_dir.path().to_str().unwrap(), + ]); + + let github = MockServer::start(); + let find_mock = github.mock(|when, then| { + when.method("GET") + .path("/repos/acme/widgets/pulls") + .query_param("state", "open") + .query_param("head", "acme:fabro/run/test") + .query_param("base", "main") + .query_param("per_page", "2") + .header("authorization", "Bearer ghu_test"); + then.status(200) + .header("content-type", "application/json") + .body( + serde_json::json!([{ + "html_url": "https://github.com/acme/widgets/pull/42", + "number": 42 + }]) + .to_string(), + ); + }); + + let mut graph = Graph::new("test"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("Open a pull request".to_string()), + ); + let run_store = test_run_store(&graph).await; + let emitter = Arc::new(Emitter::new(fixtures::RUN_1)); + let received = Arc::new(Mutex::new(Vec::new())); + let received_for_listener = Arc::clone(&received); + emitter.on_event(move |event| { + received_for_listener.lock().unwrap().push(event.clone()); + }); + let mut services = EngineServices::test_default(); + services.run = services + .run + .with_run_store(run_store) + .with_sandbox(Arc::new(fabro_agent::LocalSandbox::new( + repo_dir.path().to_path_buf(), + ))) + .with_emitter(emitter) + .with_pull_request(PullRequestRuntime { + github: Some(fabro_github::GitHubCredentials::Pat("ghu_test".to_string())), + github_base_url: github.base_url(), + origin_url: Some("https://github.com/acme/widgets.git".to_string()), + base_branch: Some("main".to_string()), + head_branch: Some("fabro/run/test".to_string()), + base_sha: Some(base_sha), + push_enabled: true, + }); + + let outcome = PullRequestHandler + .execute( + &Node::new("create_pr"), + &Context::new(), + &graph, + repo_dir.path(), + &services, + ) + .await + .unwrap(); + + assert_eq!( + outcome.context_updates[keys::PULL_REQUEST_URL], + serde_json::json!("https://github.com/acme/widgets/pull/42") + ); + assert_eq!( + outcome.context_updates[keys::PULL_REQUEST_NUMBER], + serde_json::json!(42) + ); + assert_eq!( + outcome.context_updates[keys::PULL_REQUEST_OWNER], + serde_json::json!("acme") + ); + assert_eq!( + outcome.context_updates[keys::PULL_REQUEST_REPO], + serde_json::json!("widgets") + ); + assert_eq!( + outcome.context_updates[keys::PULL_REQUEST_BASE_BRANCH], + serde_json::json!("main") + ); + assert_eq!( + outcome.context_updates[keys::PULL_REQUEST_HEAD_BRANCH], + serde_json::json!("fabro/run/test") + ); + assert_eq!( + outcome.context_updates[keys::PULL_REQUEST_DRAFT], + serde_json::json!(true) + ); + assert_eq!( + git(remote_dir.path(), &[ + "rev-parse", + "refs/heads/fabro/run/test" + ]), + head_sha + ); + let received = received.lock().unwrap(); + assert_eq!(received.len(), 1); + assert_eq!(received[0].event_name(), "pull_request.linked"); + find_mock.assert(); + } + + #[tokio::test] + async fn succeeds_from_stored_pull_request_without_runtime_capability() { + let graph = Graph::new("test"); + let run_store = test_run_store(&graph).await; + let link = PullRequestLink { + owner: "acme".to_string(), + repo: "widgets".to_string(), + number: 42, + }; + run_store + .append_run_event(&crate::event::to_run_event( + &fixtures::RUN_1, + &Event::PullRequestLinked { + pull_request: link.clone(), + }, + )) + .await + .unwrap(); + let mut services = EngineServices::test_default(); + services.run = services.run.with_run_store(run_store); + + let outcome = PullRequestHandler + .execute( + &Node::new("create_pr"), + &Context::new(), + &graph, + Path::new("."), + &services, + ) + .await + .unwrap(); + + assert_eq!( + outcome.context_updates[keys::PULL_REQUEST_URL], + serde_json::json!(link.html_url()) + ); + } +} diff --git a/lib/components/fabro-workflow/src/pipeline/initialize.rs b/lib/components/fabro-workflow/src/pipeline/initialize.rs index 149a9ee594..4c61d6e160 100644 --- a/lib/components/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/components/fabro-workflow/src/pipeline/initialize.rs @@ -30,7 +30,8 @@ use crate::run_metadata::{RunMetadataRuntime, build_metadata_writer, metadata_br use crate::run_options::{GitCheckpointOptions, RunOptions}; use crate::sandbox_git_runtime::SandboxGitRuntime; use crate::services::{ - EngineServices, FabroRunToolServices, RunLocations, RunServices, WorkflowToolEnvProvider, + EngineServices, FabroRunToolServices, PullRequestRuntime, RunLocations, RunServices, + WorkflowToolEnvProvider, }; use crate::stage_execution::{StageExecutionSeed, StageExecutionTracker}; use crate::steering_hub::SteeringHub; @@ -610,6 +611,19 @@ pub async fn initialize( } }; + let pull_request_runtime = PullRequestRuntime { + github: options.run_options.github_app.clone(), + github_base_url: fabro_github::github_api_base_url(), + origin_url: options.sandbox_env.origin_url.clone(), + base_branch: options.run_options.base_branch.clone(), + head_branch: options.run_options.run_branch().map(str::to_string), + base_sha: options + .run_options + .git + .as_ref() + .and_then(|git| git.base_sha.clone()), + push_enabled: options.run_options.settings.run.run_branch.push, + }; let run_services = RunServices::new( options.run_store.clone(), Arc::clone(&options.emitter), @@ -625,7 +639,8 @@ pub async fn initialize( metadata_runtime, metadata_writer, StageExecutionTracker::seeded(stage_executions), - ); + ) + .with_pull_request(pull_request_runtime); let engine = Arc::new(EngineServices { run: Arc::clone(&run_services), registry, diff --git a/lib/components/fabro-workflow/src/pipeline/mod.rs b/lib/components/fabro-workflow/src/pipeline/mod.rs index d0ba5ae1ba..7a0cb6d658 100644 --- a/lib/components/fabro-workflow/src/pipeline/mod.rs +++ b/lib/components/fabro-workflow/src/pipeline/mod.rs @@ -17,8 +17,10 @@ pub use initialize::initialize; pub use parse::parse; pub(crate) use persist::persist; pub use pull_request::{ - AutoMergeOptions, CreatedPullRequest, OpenPullRequestRequest, PrContent, build_pr_content, - maybe_open_pull_request, pull_request, + AutoMergeOptions, CommittedPullRequestSnapshot, CommittedPullRequestSnapshotError, + CreatedPullRequest, OpenPullRequestRequest, PrContent, PullRequestDisposition, + build_pr_content, maybe_open_pull_request, prepare_committed_pull_request_snapshot, + pull_request, }; pub use transform::transform; pub use types::{ diff --git a/lib/components/fabro-workflow/src/pipeline/pull_request.rs b/lib/components/fabro-workflow/src/pipeline/pull_request.rs index 0dcc7745c2..f18f2402b4 100644 --- a/lib/components/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/components/fabro-workflow/src/pipeline/pull_request.rs @@ -1,12 +1,15 @@ use std::collections::HashSet; use std::sync::{Arc, LazyLock}; +use anyhow::Context as _; +use fabro_agent::Sandbox; use fabro_auth::CredentialSource; use fabro_github::{self as github_app, ssh_url_to_https}; use fabro_graphviz::parser; use fabro_llm::client::Client; use fabro_llm::generate::{GenerateParams, generate_object}; use fabro_model::{Catalog, ProviderId}; +use fabro_sandbox::shell_quote; use fabro_store::RunProjection; use fabro_types::PullRequestLink; use fabro_types::settings::run::MergeStrategy; @@ -483,6 +486,233 @@ pub struct CreatedPullRequest { pub title: String, pub base_branch: String, pub head_branch: String, + pub disposition: PullRequestDisposition, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PullRequestDisposition { + Created, + Linked, +} + +/// A stable, committed snapshot used to create a pull request while a run is +/// still active. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CommittedPullRequestSnapshot { + pub head_sha: String, + pub diff: String, +} + +#[derive(Debug, thiserror::Error)] +pub enum CommittedPullRequestSnapshotError { + #[error("failed to execute {operation}")] + Sandbox { + operation: &'static str, + #[source] + source: fabro_sandbox::Error, + }, + #[error("{operation} failed")] + Command { + operation: &'static str, + #[source] + source: fabro_sandbox::Error, + }, + #[error("run base SHA is not a full git object ID")] + InvalidBase, + #[error("sandbox HEAD did not resolve to a commit SHA")] + InvalidHead, + #[error("failed to push committed snapshot to the run branch")] + Push { + #[source] + source: fabro_sandbox::Error, + }, + #[error("remote run branch does not contain the captured commit")] + RemoteBranchMissingCommit, +} + +fn is_full_git_object_id(value: &str) -> bool { + matches!(value.len(), 40 | 64) && value.bytes().all(|byte| byte.is_ascii_hexdigit()) +} + +async fn remote_branch_contains_commit( + sandbox: &dyn Sandbox, + head_sha: &str, + remote_ref: &str, +) -> bool { + const GIT: &str = "git -c maintenance.auto=0 -c gc.auto=0 -c core.hooksPath=/dev/null -c core.fsmonitor=false"; + let remote_ref = shell_quote(remote_ref); + let Ok(remote_result) = sandbox + .exec_command( + &format!("{GIT} ls-remote --heads origin {remote_ref}"), + 30_000, + None, + None, + None, + ) + .await + else { + return false; + }; + if !remote_result.is_success() { + return false; + } + let Some(remote_sha) = remote_result + .stdout + .split_whitespace() + .next() + .filter(|sha| is_full_git_object_id(sha)) + else { + return false; + }; + if remote_sha == head_sha { + return true; + } + + let remote_commit = shell_quote(&format!("{remote_sha}^{{commit}}")); + let remote_is_available = sandbox + .exec_command( + &format!("{GIT} cat-file -e {remote_commit}"), + 10_000, + None, + None, + None, + ) + .await + .is_ok_and(|result| result.is_success()); + if !remote_is_available { + let Ok(fetch_result) = sandbox + .exec_command( + &format!( + "{GIT} fetch --no-tags --no-recurse-submodules --no-write-fetch-head origin {remote_ref}" + ), + 30_000, + None, + None, + None, + ) + .await + else { + return false; + }; + if !fetch_result.is_success() { + return false; + } + } + + let head_sha = shell_quote(head_sha); + let remote_sha = shell_quote(remote_sha); + sandbox + .exec_command( + &format!("{GIT} merge-base --is-ancestor {head_sha} {remote_sha}"), + 10_000, + None, + None, + None, + ) + .await + .is_ok_and(|result| result.is_success()) +} + +/// Capture the sandbox's committed `HEAD`, compute the cumulative diff from +/// `base_sha` to that exact commit, and ensure that commit is present on the +/// remote run branch. Dirty and untracked work is intentionally excluded. +pub async fn prepare_committed_pull_request_snapshot( + sandbox: &dyn Sandbox, + base_sha: &str, + head_branch: &str, +) -> Result { + const GIT: &str = "git -c maintenance.auto=0 -c gc.auto=0 -c core.hooksPath=/dev/null -c core.fsmonitor=false -c protocol.file.allow=never -c core.quotePath=false"; + + if !is_full_git_object_id(base_sha) { + return Err(CommittedPullRequestSnapshotError::InvalidBase); + } + + let head_result = sandbox + .exec_command( + &format!("{GIT} rev-parse --verify HEAD^{{commit}}"), + 10_000, + None, + None, + None, + ) + .await + .map_err(|source| CommittedPullRequestSnapshotError::Sandbox { + operation: "git rev-parse HEAD", + source, + })?; + if !head_result.is_success() { + return Err(CommittedPullRequestSnapshotError::Command { + operation: "git rev-parse HEAD", + source: fabro_sandbox::Error::exec("git rev-parse HEAD", head_result), + }); + } + let head_sha = head_result.stdout.trim().to_ascii_lowercase(); + if !is_full_git_object_id(&head_sha) { + return Err(CommittedPullRequestSnapshotError::InvalidHead); + } + debug!( + head_sha = %head_sha.get(..12).unwrap_or(&head_sha), + head_branch, + "Captured committed pull request snapshot" + ); + + let base_sha = shell_quote(base_sha); + let quoted_head = shell_quote(&head_sha); + let diff_result = sandbox + .exec_command( + &format!("{GIT} diff --binary --full-index {base_sha}..{quoted_head}"), + 30_000, + None, + None, + None, + ) + .await + .map_err(|source| CommittedPullRequestSnapshotError::Sandbox { + operation: "git diff for pull request", + source, + })?; + if !diff_result.is_success() { + return Err(CommittedPullRequestSnapshotError::Command { + operation: "git diff for pull request", + source: fabro_sandbox::Error::exec("git diff for pull request", diff_result), + }); + } + if diff_result.stdout.trim().is_empty() { + debug!( + head_sha = %head_sha.get(..12).unwrap_or(&head_sha), + "Committed pull request snapshot has no diff" + ); + return Ok(CommittedPullRequestSnapshot { + head_sha, + diff: diff_result.stdout, + }); + } + + let remote_ref = format!("refs/heads/{head_branch}"); + let push_error = sandbox + .git_push_ref(&format!("{head_sha}:{remote_ref}")) + .await + .err(); + let remote_contains_commit = + remote_branch_contains_commit(sandbox, &head_sha, &remote_ref).await; + debug!( + head_sha = %head_sha.get(..12).unwrap_or(&head_sha), + head_branch, + push_failed = push_error.is_some(), + remote_contains_commit, + "Verified committed pull request snapshot on remote branch" + ); + if !remote_contains_commit { + if let Some(source) = push_error { + return Err(CommittedPullRequestSnapshotError::Push { source }); + } + return Err(CommittedPullRequestSnapshotError::RemoteBranchMissingCommit); + } + + Ok(CommittedPullRequestSnapshot { + head_sha, + diff: diff_result.stdout, + }) } /// Optionally open a pull request after a successful workflow run. @@ -491,7 +721,7 @@ pub struct CreatedPullRequest { /// the diff was empty, or `Err` on failure. pub async fn maybe_open_pull_request( req: OpenPullRequestRequest<'_>, -) -> Result, String> { +) -> anyhow::Result> { if req.diff.is_empty() { debug!("Empty diff, skipping pull request creation"); return Ok(None); @@ -499,7 +729,36 @@ pub async fn maybe_open_pull_request( 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:#}"))?; + github_app::parse_github_owner_repo(&https_url).context("Failed to parse GitHub origin")?; + + if let Some(existing) = github_app::find_open_pull_request( + &req.github, + &owner, + &repo, + req.base_branch, + req.head_branch, + ) + .await + .context("Failed to reconcile an existing pull request")? + { + info!( + owner, + repo, + pr_number = existing.number, + "Existing pull request linked" + ); + return Ok(Some(CreatedPullRequest { + link: PullRequestLink { + owner, + repo, + number: existing.number, + }, + title: String::new(), + base_branch: req.base_branch.to_string(), + head_branch: req.head_branch.to_string(), + disposition: PullRequestDisposition::Linked, + })); + } let content = build_pr_content( req.diff, @@ -512,11 +771,12 @@ pub async fn maybe_open_pull_request( req.run_state, ) .await - .map_err(|err| format!("{err:#}"))?; + .map_err(anyhow::Error::msg) + .context("Failed to build pull request content")?; let body = truncate_pr_body(&content.body); let title = content.title; - let created = github_app::create_pull_request( + let created = match github_app::create_pull_request( &req.github, &owner, &repo, @@ -527,7 +787,39 @@ pub async fn maybe_open_pull_request( req.draft, ) .await - .map_err(|err| format!("{err:#}"))?; + { + Ok(created) => created, + Err(github_app::CreatePullRequestError::UnprocessableEntity { .. }) => { + let existing = github_app::find_open_pull_request( + &req.github, + &owner, + &repo, + req.base_branch, + req.head_branch, + ) + .await + .context("Failed to reconcile pull request after GitHub returned 422")? + .ok_or_else(|| anyhow::anyhow!("GitHub rejected pull request creation (422)"))?; + info!( + owner, + repo, + pr_number = existing.number, + "Concurrent pull request creation reconciled" + ); + return Ok(Some(CreatedPullRequest { + link: PullRequestLink { + owner, + repo, + number: existing.number, + }, + title, + base_branch: req.base_branch.to_string(), + head_branch: req.head_branch.to_string(), + disposition: PullRequestDisposition::Linked, + })); + } + Err(err) => return Err(anyhow::Error::new(err).context("Failed to create pull request")), + }; info!(pr_url = %created.html_url, created.number, "Pull request created"); @@ -565,6 +857,7 @@ pub async fn maybe_open_pull_request( title, base_branch: req.base_branch.to_string(), head_branch: req.head_branch.to_string(), + disposition: PullRequestDisposition::Created, })) } @@ -581,73 +874,90 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) -> 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 mut pr_url = services + .run_store + .state() + .await + .ok() + .and_then(|state| state.pull_request.as_ref().map(PullRequestLink::html_url)); + if pr_url.is_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 auto_merge = if pr_cfg.auto_merge { - Some(AutoMergeOptions { - merge_strategy: pr_cfg.merge_strategy, + 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, }) - } 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}"), - ); + .await + { + Ok(Some(created)) => { + match created.disposition { + PullRequestDisposition::Created => { + services.emitter.emit(&Event::pull_request_created( + &created.link, + &created.base_branch, + &created.head_branch, + &created.title, + pr_cfg.draft, + )); + } + PullRequestDisposition::Linked => { + services.emitter.emit(&Event::PullRequestLinked { + pull_request: created.link.clone(), + }); + } + } + pr_url = Some(created.link.html_url()); + } + Ok(None) => {} + Err(e) => { + let error = e.to_string(); + services.emitter.emit(&Event::PullRequestFailed { + error: error.clone(), + }); + services.emitter.notice( + RunNoticeLevel::Warn, + RunNoticeCode::PullRequestFailed, + format!("PR creation failed: {error}"), + ); + } } } } @@ -691,7 +1001,7 @@ 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; @@ -822,6 +1132,27 @@ mod tests { Arc::new(EnvCredentialSource::new()) } + #[expect( + clippy::disallowed_methods, + reason = "Temporary git fixture setup is intentionally synchronous." + )] + fn git(repo: &std::path::Path, args: &[&str]) -> String { + let output = std::process::Command::new("git") + .args(args) + .current_dir(repo) + .output() + .expect("git should execute"); + assert!( + output.status.success(), + "git {args:?} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout) + .expect("git output should be UTF-8") + .trim() + .to_string() + } + fn test_projection() -> RunProjection { RunProjection::new( "Test run".to_string(), @@ -959,6 +1290,96 @@ mod tests { assert_eq!(finalized.pushed_branch, None); } + #[tokio::test] + async fn post_run_phase_reuses_pull_request_created_inside_run() { + let store = test_store(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); + let settings = WorkflowSettings::default(); + let graph = Graph::new("test"); + append_event(&run_store, &fixtures::RUN_1, &Event::RunCreated { + run_id: fixtures::RUN_1, + title: None, + settings: serde_json::to_value(&settings).unwrap(), + graph: serde_json::to_value(&graph).unwrap(), + workflow_source: None, + workflow_config: None, + labels: std::collections::BTreeMap::new(), + run_dir: "/tmp/test".to_string(), + source_directory: None, + workflow_slug: Some("test".to_string()), + automation: None, + db_prefix: None, + provenance: test_support::test_run_provenance(), + manifest_blob: None, + git: None, + fork_source_ref: None, + retried_from: None, + parent_id: None, + web_url: None, + }) + .await + .unwrap(); + let link = PullRequestLink { + owner: "acme".to_string(), + repo: "widgets".to_string(), + number: 42, + }; + append_event( + &run_store, + &fixtures::RUN_1, + &Event::pull_request_created( + &link, + "main", + "fabro/run/test", + "Draft from workflow", + true, + ), + ) + .await + .unwrap(); + let event_count = run_store.list_events().await.unwrap().len(); + let services = EngineServices::test_default() + .run + .with_run_store(run_store.clone().into()); + let temp = tempfile::tempdir().unwrap(); + 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: Some("main".to_string()), + 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, + run_options, + services, + }; + + let finalized = pull_request(concluded, &PullRequestOptions { + pr_config: None, + github_app: None, + origin_url: None, + model: "test-model".to_string(), + }) + .await; + + assert_eq!(finalized.pr_url, Some(link.html_url())); + assert_eq!(run_store.list_events().await.unwrap().len(), event_count); + } + // ── format_arc_details_section tests ──────────────────────────────── #[test] @@ -1575,6 +1996,289 @@ mod tests { assert!(result.unwrap().is_none()); } + #[tokio::test] + #[expect( + clippy::disallowed_methods, + reason = "Temporary git fixture setup is intentionally synchronous." + )] + async fn committed_snapshot_excludes_dirty_and_untracked_work_and_pushes_captured_head() { + let repo_dir = tempfile::tempdir().unwrap(); + let remote_dir = tempfile::tempdir().unwrap(); + git(repo_dir.path(), &["init", "-b", "main"]); + git(repo_dir.path(), &[ + "config", + "user.email", + "fabro@example.test", + ]); + git(repo_dir.path(), &["config", "user.name", "Fabro Test"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "base"]); + let base_sha = git(repo_dir.path(), &["rev-parse", "HEAD"]); + + std::fs::write(repo_dir.path().join("tracked.txt"), "base\ncommitted\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "committed"]); + let head_sha = git(repo_dir.path(), &["rev-parse", "HEAD"]); + + std::fs::write( + repo_dir.path().join("tracked.txt"), + "base\ncommitted\ndirty\n", + ) + .unwrap(); + std::fs::write(repo_dir.path().join("untracked.txt"), "untracked\n").unwrap(); + git(remote_dir.path(), &["init", "--bare"]); + git(repo_dir.path(), &[ + "remote", + "add", + "origin", + remote_dir.path().to_str().unwrap(), + ]); + + let sandbox = fabro_agent::LocalSandbox::new(repo_dir.path().to_path_buf()); + let snapshot = + prepare_committed_pull_request_snapshot(&sandbox, &base_sha, "fabro/run/snapshot-test") + .await + .unwrap(); + + assert_eq!(snapshot.head_sha, head_sha); + assert!(snapshot.diff.contains("+committed")); + assert!(!snapshot.diff.contains("+dirty")); + assert!(!snapshot.diff.contains("untracked.txt")); + assert_eq!( + git(remote_dir.path(), &[ + "rev-parse", + "refs/heads/fabro/run/snapshot-test" + ]), + head_sha + ); + } + + #[tokio::test] + #[expect( + clippy::disallowed_methods, + reason = "Temporary git fixture setup is intentionally synchronous." + )] + async fn committed_snapshot_accepts_remote_branch_that_already_contains_captured_head() { + let repo_dir = tempfile::tempdir().unwrap(); + let remote_dir = tempfile::tempdir().unwrap(); + let advance_dir = tempfile::tempdir().unwrap(); + git(repo_dir.path(), &["init", "-b", "main"]); + git(repo_dir.path(), &[ + "config", + "user.email", + "fabro@example.test", + ]); + git(repo_dir.path(), &["config", "user.name", "Fabro Test"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "base"]); + let base_sha = git(repo_dir.path(), &["rev-parse", "HEAD"]); + + std::fs::write(repo_dir.path().join("tracked.txt"), "base\ncaptured\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "captured"]); + let captured_sha = git(repo_dir.path(), &["rev-parse", "HEAD"]); + + git(remote_dir.path(), &["init", "--bare"]); + git(repo_dir.path(), &[ + "remote", + "add", + "origin", + remote_dir.path().to_str().unwrap(), + ]); + git(repo_dir.path(), &[ + "push", + "origin", + "HEAD:refs/heads/fabro/run/snapshot-test", + ]); + + git(advance_dir.path(), &[ + "clone", + remote_dir.path().to_str().unwrap(), + ".", + ]); + git(advance_dir.path(), &[ + "checkout", + "-b", + "advance", + "origin/fabro/run/snapshot-test", + ]); + git(advance_dir.path(), &[ + "config", + "user.email", + "fabro@example.test", + ]); + git(advance_dir.path(), &["config", "user.name", "Fabro Test"]); + std::fs::write( + advance_dir.path().join("tracked.txt"), + "base\ncaptured\nnewer\n", + ) + .unwrap(); + git(advance_dir.path(), &["add", "tracked.txt"]); + git(advance_dir.path(), &["commit", "-m", "newer"]); + let remote_head = git(advance_dir.path(), &["rev-parse", "HEAD"]); + git(advance_dir.path(), &[ + "push", + "origin", + "HEAD:refs/heads/fabro/run/snapshot-test", + ]); + + let sandbox = fabro_agent::LocalSandbox::new(repo_dir.path().to_path_buf()); + let snapshot = + prepare_committed_pull_request_snapshot(&sandbox, &base_sha, "fabro/run/snapshot-test") + .await + .unwrap(); + + assert_eq!(snapshot.head_sha, captured_sha); + assert_eq!( + git(remote_dir.path(), &[ + "rev-parse", + "refs/heads/fabro/run/snapshot-test" + ]), + remote_head + ); + } + + #[tokio::test] + #[expect( + clippy::disallowed_methods, + reason = "Temporary git fixture setup is intentionally synchronous." + )] + async fn committed_snapshot_rejects_divergent_remote_branch() { + let repo_dir = tempfile::tempdir().unwrap(); + let remote_dir = tempfile::tempdir().unwrap(); + git(repo_dir.path(), &["init", "-b", "main"]); + git(repo_dir.path(), &[ + "config", + "user.email", + "fabro@example.test", + ]); + git(repo_dir.path(), &["config", "user.name", "Fabro Test"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "base"]); + let base_sha = git(repo_dir.path(), &["rev-parse", "HEAD"]); + + std::fs::write(repo_dir.path().join("tracked.txt"), "base\ncaptured\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "captured"]); + let captured_sha = git(repo_dir.path(), &["rev-parse", "HEAD"]); + + git(repo_dir.path(), &["checkout", "-b", "divergent", &base_sha]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\ndivergent\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "divergent"]); + git(remote_dir.path(), &["init", "--bare"]); + git(repo_dir.path(), &[ + "remote", + "add", + "origin", + remote_dir.path().to_str().unwrap(), + ]); + git(repo_dir.path(), &[ + "push", + "origin", + "HEAD:refs/heads/fabro/run/snapshot-test", + ]); + git(repo_dir.path(), &["checkout", "--detach", &captured_sha]); + + let sandbox = fabro_agent::LocalSandbox::new(repo_dir.path().to_path_buf()); + let error = + prepare_committed_pull_request_snapshot(&sandbox, &base_sha, "fabro/run/snapshot-test") + .await + .unwrap_err(); + + assert!(matches!( + error, + CommittedPullRequestSnapshotError::Push { .. } + )); + } + + #[tokio::test] + #[expect( + clippy::disallowed_methods, + reason = "Temporary git fixture setup is intentionally synchronous." + )] + async fn committed_snapshot_rejects_missing_remote_branch() { + let repo_dir = tempfile::tempdir().unwrap(); + git(repo_dir.path(), &["init", "-b", "main"]); + git(repo_dir.path(), &[ + "config", + "user.email", + "fabro@example.test", + ]); + git(repo_dir.path(), &["config", "user.name", "Fabro Test"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "base"]); + let base_sha = git(repo_dir.path(), &["rev-parse", "HEAD"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\ncommitted\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "committed"]); + + let sandbox = fabro_agent::LocalSandbox::new(repo_dir.path().to_path_buf()); + let error = + prepare_committed_pull_request_snapshot(&sandbox, &base_sha, "fabro/run/snapshot-test") + .await + .unwrap_err(); + + assert!(matches!( + error, + CommittedPullRequestSnapshotError::RemoteBranchMissingCommit + )); + } + + #[tokio::test] + #[expect( + clippy::disallowed_methods, + reason = "Temporary git fixture setup is intentionally synchronous." + )] + async fn committed_snapshot_ignores_dirty_only_changes_without_pushing() { + let repo_dir = tempfile::tempdir().unwrap(); + git(repo_dir.path(), &["init", "-b", "main"]); + git(repo_dir.path(), &[ + "config", + "user.email", + "fabro@example.test", + ]); + git(repo_dir.path(), &["config", "user.name", "Fabro Test"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\n").unwrap(); + git(repo_dir.path(), &["add", "tracked.txt"]); + git(repo_dir.path(), &["commit", "-m", "base"]); + let base_sha = git(repo_dir.path(), &["rev-parse", "HEAD"]); + std::fs::write(repo_dir.path().join("tracked.txt"), "base\ndirty\n").unwrap(); + std::fs::write(repo_dir.path().join("untracked.txt"), "untracked\n").unwrap(); + + let sandbox = fabro_agent::LocalSandbox::new(repo_dir.path().to_path_buf()); + let snapshot = + prepare_committed_pull_request_snapshot(&sandbox, &base_sha, "fabro/run/snapshot-test") + .await + .unwrap(); + + assert_eq!(snapshot.head_sha, base_sha); + assert!(snapshot.diff.is_empty()); + } + + #[tokio::test] + async fn committed_snapshot_rejects_non_oid_base_before_running_git() { + let repo_dir = tempfile::tempdir().unwrap(); + let sandbox = fabro_agent::LocalSandbox::new(repo_dir.path().to_path_buf()); + + let error = prepare_committed_pull_request_snapshot( + &sandbox, + "--output=/tmp/not-allowed", + "fabro/run/snapshot-test", + ) + .await + .unwrap_err(); + + assert!(matches!( + error, + CommittedPullRequestSnapshotError::InvalidBase + )); + } + #[tokio::test] async fn load_pull_request_diff_uses_store_without_disk_patch() { let tmp = tempfile::tempdir().unwrap(); @@ -1800,18 +2504,19 @@ mod tests { /// client from the credential source, so the in-process MockProvider /// cannot intercept — we mock the OpenAI HTTP endpoint instead. struct FallbackHarness { - _vault_dir: tempfile::TempDir, + _vault_dir: tempfile::TempDir, // Held to keep the mock listener alive for the duration of the test; // the test interacts with it via `Client::from_source` (which goes // out via HTTP to the mock URL stored in `llm_source`). - openai_server: MockServer, - github_server: MockServer, - openai_mock_id: usize, - github_mock_id: usize, - llm_source: Arc, - catalog: Arc, - creds: fabro_github::GitHubCredentials, - run_store: RunStoreHandle, + openai_server: MockServer, + github_server: MockServer, + openai_mock_id: usize, + github_find_mock_id: usize, + github_mock_id: usize, + llm_source: Arc, + catalog: Arc, + creds: fabro_github::GitHubCredentials, + run_store: RunStoreHandle, } impl FallbackHarness { @@ -1819,6 +2524,9 @@ mod tests { httpmock::Mock::new(self.openai_mock_id, &self.openai_server) .assert_async() .await; + httpmock::Mock::new(self.github_find_mock_id, &self.github_server) + .assert_async() + .await; httpmock::Mock::new(self.github_mock_id, &self.github_server) .assert_async() .await; @@ -1843,6 +2551,20 @@ mod tests { .await; let github_server = MockServer::start_async().await; + let github_find_mock = github_server + .mock_async(|when, then| { + when.method(GET) + .path("/repos/owner/repo/pulls") + .query_param("state", "open") + .query_param("head", "owner:fabro/run/123") + .query_param("base", "main") + .query_param("per_page", "2") + .header("authorization", "Bearer test-token"); + then.status(200) + .header("content-type", "application/json") + .json_body(serde_json::json!([])); + }) + .await; let github_mock = github_server .mock_async(|when, then| { when.method(POST) @@ -1947,6 +2669,7 @@ mod tests { .unwrap(); let openai_mock_id = openai_mock.id; + let github_find_mock_id = github_find_mock.id; let github_mock_id = github_mock.id; FallbackHarness { @@ -1954,6 +2677,7 @@ mod tests { openai_server, github_server, openai_mock_id, + github_find_mock_id, github_mock_id, llm_source, catalog, diff --git a/lib/components/fabro-workflow/src/pull_request.rs b/lib/components/fabro-workflow/src/pull_request.rs index bb9a747449..9f4c8995cb 100644 --- a/lib/components/fabro-workflow/src/pull_request.rs +++ b/lib/components/fabro-workflow/src/pull_request.rs @@ -1,4 +1,5 @@ pub use crate::pipeline::{ - AutoMergeOptions, CreatedPullRequest, OpenPullRequestRequest, PrContent, build_pr_content, - maybe_open_pull_request, + AutoMergeOptions, CommittedPullRequestSnapshot, CommittedPullRequestSnapshotError, + CreatedPullRequest, OpenPullRequestRequest, PrContent, PullRequestDisposition, + build_pr_content, maybe_open_pull_request, prepare_committed_pull_request_snapshot, }; diff --git a/lib/components/fabro-workflow/src/services.rs b/lib/components/fabro-workflow/src/services.rs index 2af76a9c72..c653b3f036 100644 --- a/lib/components/fabro-workflow/src/services.rs +++ b/lib/components/fabro-workflow/src/services.rs @@ -83,6 +83,18 @@ pub struct FabroRunToolServices { pub user_settings_path: PathBuf, } +/// Run-scoped inputs used by the built-in `pull_request` node. +#[derive(Clone)] +pub(crate) struct PullRequestRuntime { + pub github: Option, + pub github_base_url: String, + pub origin_url: Option, + pub base_branch: Option, + pub head_branch: Option, + pub base_sha: Option, + pub push_enabled: bool, +} + /// Services shared across workflow phases. /// /// Production construction is expected to happen from pipeline initialization @@ -107,6 +119,7 @@ pub struct RunServices { pub(crate) metadata_runtime: Arc, pub(crate) metadata_writer: Option, pub(crate) interview_blocker: Arc, + pub(crate) pull_request: Option, /// Run-scoped stage execution allocator, shared between the core /// lifecycle and direct-dispatch handlers such as parallel branches. pub(crate) stage_executions: StageExecutionTracker, @@ -145,6 +158,7 @@ impl RunServices { metadata_runtime, metadata_writer, interview_blocker: Arc::new(RunInterviewBlocker::new()), + pull_request: None, stage_executions, }) } @@ -212,6 +226,17 @@ impl RunServices { }) } + #[must_use] + pub(crate) fn with_pull_request( + self: &Arc, + pull_request: PullRequestRuntime, + ) -> Arc { + Arc::new(Self { + pull_request: Some(pull_request), + ..self.as_ref().clone() + }) + } + #[cfg(test)] #[must_use] pub(crate) fn with_catalog_context( diff --git a/lib/foundation/fabro-api/tests/stage_handler_round_trip.rs b/lib/foundation/fabro-api/tests/stage_handler_round_trip.rs index e824b4869f..f45f455f43 100644 --- a/lib/foundation/fabro-api/tests/stage_handler_round_trip.rs +++ b/lib/foundation/fabro-api/tests/stage_handler_round_trip.rs @@ -17,6 +17,7 @@ fn stage_handler_serializes_openapi_wire_values() { (StageHandler::Agent, "agent"), (StageHandler::Prompt, "prompt"), (StageHandler::Command, "command"), + (StageHandler::PullRequest, "pull_request"), (StageHandler::Human, "human"), (StageHandler::Conditional, "conditional"), (StageHandler::Parallel, "parallel"), diff --git a/lib/foundation/fabro-types/src/graph.rs b/lib/foundation/fabro-types/src/graph.rs index 7ef9bad991..863636f3ab 100644 --- a/lib/foundation/fabro-types/src/graph.rs +++ b/lib/foundation/fabro-types/src/graph.rs @@ -86,6 +86,7 @@ pub const KNOWN_HANDLER_TYPES: &[&str] = &[ "parallel.fan_in", "command", "tool", + "pull_request", "stack.manager_loop", "wait", ]; diff --git a/lib/foundation/fabro-types/src/stage_handler.rs b/lib/foundation/fabro-types/src/stage_handler.rs index eb75eb0b9f..99f0a4a8c5 100644 --- a/lib/foundation/fabro-types/src/stage_handler.rs +++ b/lib/foundation/fabro-types/src/stage_handler.rs @@ -22,6 +22,7 @@ pub enum StageHandler { Agent, Prompt, Command, + PullRequest, Human, Conditional, Parallel, @@ -42,6 +43,7 @@ impl StageHandler { "exit" => Self::Exit, "prompt" => Self::Prompt, "command" | "tool" => Self::Command, + "pull_request" => Self::PullRequest, "human" => Self::Human, "conditional" => Self::Conditional, "parallel" => Self::Parallel, diff --git a/lib/foundation/fabro-types/tests/stage_handler.rs b/lib/foundation/fabro-types/tests/stage_handler.rs index 9abc208cd1..8a54b4638d 100644 --- a/lib/foundation/fabro-types/tests/stage_handler.rs +++ b/lib/foundation/fabro-types/tests/stage_handler.rs @@ -9,6 +9,7 @@ fn stage_handler_serializes_canonical_wire_values() { (StageHandler::Agent, "agent"), (StageHandler::Prompt, "prompt"), (StageHandler::Command, "command"), + (StageHandler::PullRequest, "pull_request"), (StageHandler::Human, "human"), (StageHandler::Conditional, "conditional"), (StageHandler::Parallel, "parallel"), @@ -40,6 +41,10 @@ fn stage_handler_maps_current_handler_types_and_defaults_to_agent() { StageHandler::from_handler_type(Some("tool")), StageHandler::Command ); + assert_eq!( + StageHandler::from_handler_type(Some("pull_request")), + StageHandler::PullRequest + ); assert_eq!( StageHandler::from_handler_type(Some("parallel.fan_in")), StageHandler::ParallelFanIn diff --git a/lib/packages/fabro-api-client/src/api/runs-api.ts b/lib/packages/fabro-api-client/src/api/runs-api.ts index c0717d0661..26c66101b5 100644 --- a/lib/packages/fabro-api-client/src/api/runs-api.ts +++ b/lib/packages/fabro-api-client/src/api/runs-api.ts @@ -409,7 +409,7 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration) }; }, /** - * Creates a pull request for a completed run on GitHub and persists the record on the server. + * Creates a pull request on GitHub and persists the record on the server. By default the run must have a successful conclusion. With force=true, failed conclusions are accepted and running, blocked, or paused runs use their latest committed snapshot. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -1646,7 +1646,7 @@ export const RunsApiFp = function(configuration?: Configuration) { return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, /** - * Creates a pull request for a completed run on GitHub and persists the record on the server. + * Creates a pull request on GitHub and persists the record on the server. By default the run must have a successful conclusion. With force=true, failed conclusions are accepted and running, blocked, or paused runs use their latest committed snapshot. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -2090,7 +2090,7 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath? return localVarFp.createRun(runManifest, options).then((request) => request(axios, basePath)); }, /** - * Creates a pull request for a completed run on GitHub and persists the record on the server. + * Creates a pull request on GitHub and persists the record on the server. By default the run must have a successful conclusion. With force=true, failed conclusions are accepted and running, blocked, or paused runs use their latest committed snapshot. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -2462,7 +2462,7 @@ export class RunsApi extends BaseAPI { } /** - * Creates a pull request for a completed run on GitHub and persists the record on the server. + * Creates a pull request on GitHub and persists the record on the server. By default the run must have a successful conclusion. With force=true, failed conclusions are accepted and running, blocked, or paused runs use their latest committed snapshot. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest diff --git a/lib/packages/fabro-api-client/src/models/create-run-pull-request-request.ts b/lib/packages/fabro-api-client/src/models/create-run-pull-request-request.ts index 7ec1a3020a..c8575c8b16 100644 --- a/lib/packages/fabro-api-client/src/models/create-run-pull-request-request.ts +++ b/lib/packages/fabro-api-client/src/models/create-run-pull-request-request.ts @@ -19,7 +19,7 @@ */ export interface CreateRunPullRequestRequest { /** - * Create the pull request even if the run did not finish with succeeded or partially_succeeded. + * Create from a non-successful conclusion, or from the latest committed snapshot when the run is running, blocked, or paused. */ 'force': boolean; /** diff --git a/lib/packages/fabro-api-client/src/models/stage-handler.ts b/lib/packages/fabro-api-client/src/models/stage-handler.ts index 1398d3e342..44c877871f 100644 --- a/lib/packages/fabro-api-client/src/models/stage-handler.ts +++ b/lib/packages/fabro-api-client/src/models/stage-handler.ts @@ -24,6 +24,7 @@ export const StageHandler = { AGENT: 'agent', PROMPT: 'prompt', COMMAND: 'command', + PULL_REQUEST: 'pull_request', HUMAN: 'human', CONDITIONAL: 'conditional', PARALLEL: 'parallel',