scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-08 22:21:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

feat(workflow): add pull request stage

Browse source
This commit is contained in:
Bryan Helmkamp 2026-07-26 08:20:11 -04:00
parent 2bcf94fed8
commit cfcf45f195
No known key found for this signature in database
32 changed files with 2400 additions and 160 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -2652,6 +2652,7 @@ dependencies = [
"tokio",
"tracing",
"tracing-subscriber",
"url",
]
[[package]]

20
docs/public/api-reference/fabro-api.yaml
View file

@ -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

12
docs/public/execution/context.mdx
View file

@ -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 |

8
docs/public/execution/run-configuration.mdx
View file

@ -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/<id>` 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/<id>` 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

8
docs/public/reference/cli.mdx
View file

@ -766,7 +766,7 @@ fabro pr [OPTIONS] <COMMAND>
| 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] <RUN_ID>
#### `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] <RUN_ID>
@ -810,7 +812,7 @@ fabro pr create [OPTIONS] <RUN_ID>
| 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 <model>` | LLM model for generating PR description |
| `--server <server>` | Fabro server target: http(s) URL or absolute Unix socket path |

9
docs/public/reference/dot-language.mdx
View file

@ -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 |

37
docs/public/workflows/stages-and-nodes.mdx
View file

@ -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`

4
lib/apps/fabro-cli/src/args.rs
View file

@ -929,7 +929,7 @@ pub(crate) struct PrCreateArgs {
/// LLM model for generating PR description
#[arg(long)]
pub(crate) model: Option<String>,
/// 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),

2
lib/apps/fabro-cli/src/commands/pr/create.rs
View file

@ -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() {

32
lib/apps/fabro-cli/src/commands/run/runner.rs
View file

@ -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<Option<fabro_github::GitHubCredentials>> {
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<String, InterpString>,
@ -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));
}
}
}

2
lib/apps/fabro-cli/tests/it/cmd/pr.rs
View file

@ -14,7 +14,7 @@ fn help() {
Usage: fabro pr [OPTIONS] <COMMAND>
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

4
lib/apps/fabro-cli/tests/it/cmd/pr_create.rs
View file

@ -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] <RUN_ID>
@ -30,7 +30,7 @@ fn help() {
--server <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 <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=]

2
lib/apps/fabro-server/src/run_files.rs
View file

@ -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<AppState>,
run_id: &RunId,
projection: &fabro_store::RunProjection,

354
lib/apps/fabro-server/src/server/handler/pull_requests.rs
View file

@ -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<Arc<AppState>> {
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<Self, ApiError> {
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",
}
));
}
}
}

286
lib/apps/fabro-server/src/server/tests.rs
View file

@ -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);

1
lib/components/fabro-github/Cargo.toml
View file

@ -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" }

253
lib/components/fabro-github/src/lib.rs
View file

@ -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<CreatedPullRequest> {
let client = ctx.http_client()?;
) -> Result<CreatedPullRequest, CreatePullRequestError> {
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<CreatedPullRequest> {
) -> Result<CreatedPullRequest, CreatePullRequestError> {
#[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<Option<MatchingPullRequest>, 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<Option<MatchingPullRequest>, 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::<Vec<PullRequestResponse>>()
.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
// -----------------------------------------------------------------------

9
lib/components/fabro-workflow/src/context.rs
View file

@ -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";

2
lib/components/fabro-workflow/src/handler/mod.rs
View file

@ -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",

555
lib/components/fabro-workflow/src/handler/pull_request.rs Normal file
View file

@ -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<Outcome, Error> {
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<Outcome, Error> {
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::<AutoMergeOptions>,
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())
);
}
}

19
lib/components/fabro-workflow/src/pipeline/initialize.rs
View file

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

6
lib/components/fabro-workflow/src/pipeline/mod.rs
View file

@ -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::{

884
lib/components/fabro-workflow/src/pipeline/pull_request.rs
View file

File diff suppressed because it is too large Load diff

5
lib/components/fabro-workflow/src/pull_request.rs
View file

@ -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,
};

25
lib/components/fabro-workflow/src/services.rs
View file

@ -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<fabro_github::GitHubCredentials>,
pub github_base_url: String,
pub origin_url: Option<String>,
pub base_branch: Option<String>,
pub head_branch: Option<String>,
pub base_sha: Option<String>,
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<RunMetadataRuntime>,
pub(crate) metadata_writer: Option<RunMetadataWriterHandle>,
pub(crate) interview_blocker: Arc<RunInterviewBlocker>,
pub(crate) pull_request: Option<PullRequestRuntime>,
/// 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<Self>,
pull_request: PullRequestRuntime,
) -> Arc<Self> {
Arc::new(Self {
pull_request: Some(pull_request),
..self.as_ref().clone()
})
}
#[cfg(test)]
#[must_use]
pub(crate) fn with_catalog_context(

1
lib/foundation/fabro-api/tests/stage_handler_round_trip.rs
View file

@ -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"),

1
lib/foundation/fabro-types/src/graph.rs
View file

@ -86,6 +86,7 @@ pub const KNOWN_HANDLER_TYPES: &[&str] = &[
"parallel.fan_in",
"command",
"tool",
"pull_request",
"stack.manager_loop",
"wait",
];

2
lib/foundation/fabro-types/src/stage_handler.rs
View file

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

5
lib/foundation/fabro-types/tests/stage_handler.rs
View file

@ -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

8
lib/packages/fabro-api-client/src/api/runs-api.ts generated
View file

@ -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

2
lib/packages/fabro-api-client/src/models/create-run-pull-request-request.ts generated
View file

@ -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;
/**

1
lib/packages/fabro-api-client/src/models/stage-handler.ts generated
View file

@ -24,6 +24,7 @@ export const StageHandler = {
AGENT: 'agent',
PROMPT: 'prompt',
COMMAND: 'command',
PULL_REQUEST: 'pull_request',
HUMAN: 'human',
CONDITIONAL: 'conditional',
PARALLEL: 'parallel',