From 30ce02f2a834a330937b158f78c59cf739f6b54b Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 1 Aug 2026 19:28:52 -0400 Subject: [PATCH 1/2] Make pull request creation durable and asynchronous --- docs/internal/events.md | 20 + docs/public/api-reference/fabro-api.yaml | 109 +++++- lib/apps/fabro-cli/tests/it/cmd/pr_create.rs | 62 ++- lib/apps/fabro-server/src/error.rs | 4 + lib/apps/fabro-server/src/serve.rs | 17 +- lib/apps/fabro-server/src/server.rs | 14 + .../fabro-server/src/server/handler/mod.rs | 2 +- .../src/server/handler/pull_requests.rs | 360 +++++++++++++++++- lib/apps/fabro-server/src/server/tests.rs | 205 +++++++++- lib/components/fabro-github/src/lib.rs | 148 +++++++ lib/components/fabro-store/src/run_state.rs | 144 ++++++- .../fabro-workflow/src/event/convert.rs | 11 + .../fabro-workflow/src/event/events.rs | 18 +- .../fabro-workflow/src/event/names.rs | 1 + .../src/pipeline/pull_request.rs | 186 +++++++-- lib/foundation/fabro-api/build.rs | 15 + lib/foundation/fabro-api/src/lib.rs | 3 +- .../tests/pull_request_round_trip.rs | 41 +- lib/foundation/fabro-client/src/client.rs | 63 ++- lib/foundation/fabro-types/src/lib.rs | 3 +- .../fabro-types/src/pull_request.rs | 39 ++ .../fabro-types/src/run_event/misc.rs | 10 +- .../fabro-types/src/run_event/mod.rs | 3 + .../fabro-types/src/run_projection.rs | 49 +-- .../src/.openapi-generator/FILES | 2 + .../fabro-api-client/src/api/runs-api.ts | 88 ++++- .../fabro-api-client/src/models/index.ts | 2 + .../models/pull-request-creation-status.ts | 27 ++ .../src/models/pull-request-creation.ts | 44 +++ .../src/models/run-projection.ts | 4 + 30 files changed, 1560 insertions(+), 134 deletions(-) create mode 100644 lib/packages/fabro-api-client/src/models/pull-request-creation-status.ts create mode 100644 lib/packages/fabro-api-client/src/models/pull-request-creation.ts diff --git a/docs/internal/events.md b/docs/internal/events.md index 2a2274766d..d01c0991b6 100644 --- a/docs/internal/events.md +++ b/docs/internal/events.md @@ -2107,6 +2107,26 @@ These legacy events may appear in older run logs. Current CLI backend runs do no ## Pull request events +### `pull_request.creation_requested` + +```json +{ + "id": "...", "ts": "...", "run_id": "...", + "event": "pull_request.creation_requested", + "properties": { + "creation_id": "01KYYK70WTZT2E551P3H5P0059", + "model": "gpt-5.4", + "force": false + } +} +``` + +| Property | Type | Description | +|----------|------|-------------| +| `creation_id` | string | Stable identifier for this pull request creation request | +| `model` | string | Resolved model identifier used to generate the pull request content | +| `force` | boolean | Whether creation is allowed for a run without a successful conclusion | + ### `pull_request.created` ```json diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index 3851689de4..b151695cc9 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: | + Durably requests creation of a pull request for a completed run. The + server generates the pull request content and creates the GitHub pull + request after this request returns. Poll the URL in the Location + response header until the creation succeeds or fails. parameters: - $ref: "#/components/parameters/RunId" requestBody: @@ -2585,12 +2589,23 @@ paths: schema: $ref: "#/components/schemas/CreateRunPullRequestRequest" responses: - "200": - description: Pull request created + "202": + description: Pull request creation was durably accepted + headers: + Location: + description: URL for the latest pull request creation on this run. + schema: + type: string + format: uri-reference + Retry-After: + description: Suggested number of seconds before polling the creation status. + schema: + type: integer + minimum: 0 content: application/json: schema: - $ref: "#/components/schemas/PullRequestLink" + $ref: "#/components/schemas/PullRequestCreation" "400": description: Pull request creation does not apply to this run headers: @@ -2620,15 +2635,6 @@ paths: application/json: schema: $ref: "#/components/schemas/ErrorResponse" - "502": - description: GitHub rejected the pull request creation request - headers: - x-request-id: - $ref: "#/components/headers/XRequestId" - content: - application/json: - schema: - $ref: "#/components/schemas/ErrorResponse" "503": description: GitHub integration is unavailable on the server headers: @@ -2638,6 +2644,7 @@ paths: application/json: schema: $ref: "#/components/schemas/ErrorResponse" + put: operationId: linkRunPullRequest tags: [Runs] @@ -2723,6 +2730,31 @@ paths: schema: $ref: "#/components/schemas/ErrorResponse" + /api/v1/runs/{id}/pull_request/creation: + get: + operationId: getRunPullRequestCreation + tags: [Runs] + summary: Get Run Pull Request Creation + description: Returns the latest explicit pull request creation requested for this run. + parameters: + - $ref: "#/components/parameters/RunId" + responses: + "200": + description: Latest pull request creation state + content: + application/json: + schema: + $ref: "#/components/schemas/PullRequestCreation" + "404": + description: Run or pull request creation not found + headers: + x-request-id: + $ref: "#/components/headers/XRequestId" + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + /api/v1/runs/{id}/pull_request/merge: post: operationId: mergeRunPullRequest @@ -11541,6 +11573,10 @@ components: oneOf: - $ref: "#/components/schemas/PullRequestLink" - type: "null" + pull_request_creation: + oneOf: + - $ref: "#/components/schemas/PullRequestCreation" + - type: "null" superseded_by: type: ["string", "null"] retried_from: @@ -12347,6 +12383,53 @@ components: description: Optional model override for generating the pull request description. example: claude-sonnet-4-6 + PullRequestCreationId: + description: Stable identifier for one explicit pull request creation request. + type: string + example: 01KYYK70WTZT2E551P3H5P0059 + + PullRequestCreationStatus: + description: Durable state of a pull request creation request. + type: string + enum: + - pending + - succeeded + - failed + + PullRequestCreation: + description: Durable status for the latest explicit pull request creation requested for a run. + type: object + required: + - id + - status + - model + - force + - requested_at + - updated_at + properties: + id: + $ref: "#/components/schemas/PullRequestCreationId" + status: + $ref: "#/components/schemas/PullRequestCreationStatus" + model: + type: string + description: Resolved model identifier used to generate the pull request content. + force: + type: boolean + description: Whether creation was allowed for a run without a successful conclusion. + requested_at: + type: string + format: date-time + updated_at: + type: string + format: date-time + pull_request: + oneOf: + - $ref: "#/components/schemas/PullRequestLink" + - type: "null" + error: + type: ["string", "null"] + LinkRunPullRequestRequest: description: Request body for linking an existing GitHub pull request to a run. type: object 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..f02b71d539 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/pr_create.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/pr_create.rs @@ -70,13 +70,35 @@ fn pr_create_uses_server_endpoint_and_prints_url() { .json_body(serde_json::json!({ "force": false })); + then.status(202) + .header("Content-Type", "application/json") + .json_body(serde_json::json!({ + "id": "01KYYK70WTZT2E551P3H5P0059", + "status": "pending", + "model": "kimi-k3", + "force": false, + "requested_at": "2026-08-01T12:00:00Z", + "updated_at": "2026-08-01T12:00:00Z" + })); + }); + let status_mock = server.mock(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/pull_request/creation")); then.status(200) .header("Content-Type", "application/json") .json_body(serde_json::json!({ - "owner": "fabro-sh", - "repo": "fabro", - "number": 123, - "html_url": "https://github.com/fabro-sh/fabro/pull/123" + "id": "01KYYK70WTZT2E551P3H5P0059", + "status": "succeeded", + "model": "kimi-k3", + "force": false, + "requested_at": "2026-08-01T12:00:00Z", + "updated_at": "2026-08-01T12:00:15Z", + "pull_request": { + "owner": "fabro-sh", + "repo": "fabro", + "number": 123, + "html_url": "https://github.com/fabro-sh/fabro/pull/123" + } })); }); @@ -99,6 +121,7 @@ fn pr_create_uses_server_endpoint_and_prints_url() { resolve_mock.assert(); create_mock.assert(); + status_mock.assert(); } #[test] @@ -116,13 +139,35 @@ fn pr_create_passes_force_and_model_to_server() { "force": true, "model": "gpt-5.2" })); + then.status(202) + .header("Content-Type", "application/json") + .json_body(serde_json::json!({ + "id": "01KYYK70WTZT2E551P3H5P0059", + "status": "pending", + "model": "gpt-5.2", + "force": true, + "requested_at": "2026-08-01T12:00:00Z", + "updated_at": "2026-08-01T12:00:00Z" + })); + }); + let status_mock = server.mock(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/pull_request/creation")); then.status(200) .header("Content-Type", "application/json") .json_body(serde_json::json!({ - "owner": "fabro-sh", - "repo": "fabro", - "number": 123, - "html_url": "https://github.com/fabro-sh/fabro/pull/123" + "id": "01KYYK70WTZT2E551P3H5P0059", + "status": "succeeded", + "model": "gpt-5.2", + "force": true, + "requested_at": "2026-08-01T12:00:00Z", + "updated_at": "2026-08-01T12:00:15Z", + "pull_request": { + "owner": "fabro-sh", + "repo": "fabro", + "number": 123, + "html_url": "https://github.com/fabro-sh/fabro/pull/123" + } })); }); @@ -154,4 +199,5 @@ fn pr_create_passes_force_and_model_to_server() { resolve_mock.assert(); create_mock.assert(); + status_mock.assert(); } diff --git a/lib/apps/fabro-server/src/error.rs b/lib/apps/fabro-server/src/error.rs index 62831c9831..6979b8bc16 100644 --- a/lib/apps/fabro-server/src/error.rs +++ b/lib/apps/fabro-server/src/error.rs @@ -126,6 +126,10 @@ impl ApiError { self.status } + pub(crate) fn detail(&self) -> &str { + &self.detail + } + pub(crate) fn code(&self) -> Option<&str> { self.code.as_deref() } diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index f46bb7e0a2..5b345d9654 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -36,7 +36,7 @@ use crate::interp::process_env_var; use crate::server::{ self, AppState, AppStateConfig, ResolvedAppStateSettings, RouterOptions, build_app_state, build_router_with_options, reconcile_incomplete_runs_on_startup, shutdown_active_workers, - spawn_automation_scheduler, spawn_scheduler, + spawn_automation_scheduler, spawn_pull_request_creation_supervisor, spawn_scheduler, }; use crate::server_secrets::{ServerSecrets, process_env_snapshot}; use crate::startup::{migrate_startup_vault, resolve_startup, validate_startup_configuration}; @@ -826,6 +826,8 @@ where } spawn_scheduler(Arc::clone(&state)); spawn_automation_scheduler(Arc::clone(&state)); + let pull_request_creation_supervisor = + spawn_pull_request_creation_supervisor(Arc::clone(&state)); let router = build_router_with_options(Arc::clone(&state), &auth_mode, RouterOptions { web_enabled, #[cfg(debug_assertions)] @@ -997,6 +999,19 @@ where cleanup_handle.abort(); } + if shutdown.is_cancelled() { + if let Err(join_err) = pull_request_creation_supervisor.await { + warn!(error = %join_err, "Pull request creation supervisor task panicked"); + } + } else { + pull_request_creation_supervisor.abort(); + if let Err(join_err) = pull_request_creation_supervisor.await { + if !join_err.is_cancelled() { + warn!(error = %join_err, "Pull request creation supervisor task panicked"); + } + } + } + serve_result?; if let Some(manager) = webhook_manager { diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index bc5aeeeb3c..84810a68a2 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -186,6 +186,7 @@ pub(crate) use handler::graph::render_graph_bytes; pub(in crate::server) use handler::graph::{ RenderSubprocessError, render_dot_subprocess, render_graph_bytes_with_exe_override, }; +pub(crate) use handler::pull_requests::spawn_pull_request_creation_supervisor; #[cfg(test)] pub(in crate::server) use handler::system::validate_github_slug; use session_runtime::SessionRuntimeManager; @@ -1120,6 +1121,7 @@ pub struct AppState { pub(crate) worker_runtime: Arc, scheduler_notify: Notify, automation_scheduler_notify: Notify, + pull_request_scheduler_notify: Notify, global_event_tx: broadcast::Sender, /// Per-run coalescing registry for `GET /runs/{id}/files`. Concurrent /// callers for the same run share one materialization; different runs @@ -1207,6 +1209,16 @@ impl AppState { ) -> impl std::future::Future + '_ { self.automation_scheduler_notify.notified() } + + pub(crate) fn notify_pull_request_scheduler(&self) { + self.pull_request_scheduler_notify.notify_one(); + } + + pub(crate) fn pull_request_scheduler_notified( + &self, + ) -> impl std::future::Future + '_ { + self.pull_request_scheduler_notify.notified() + } } pub(crate) struct AskFabroReadiness { @@ -1621,6 +1633,7 @@ impl AppState { self.shutting_down.store(true, Ordering::Relaxed); self.scheduler_notify.notify_waiters(); self.automation_scheduler_notify.notify_waiters(); + self.pull_request_scheduler_notify.notify_waiters(); } pub(crate) fn shutdown_token(&self) -> CancellationToken { @@ -2559,6 +2572,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result Router> { "/runs/{id}/pull_request/merge", post(merge_run_pull_request), ) + .route( + "/runs/{id}/pull_request/creation", + get(get_run_pull_request_creation), + ) .route( "/runs/{id}/pull_request/close", post(close_run_pull_request), ) } +const PULL_REQUEST_CREATION_TIMEOUT: Duration = Duration::from_mins(10); +const PULL_REQUEST_CREATION_SCAN_INTERVAL: Duration = Duration::from_secs(5); +const MAX_CONCURRENT_PULL_REQUEST_CREATIONS: usize = 4; + #[expect( clippy::disallowed_types, reason = "Pull-request API validates public github.com URLs; these raw URLs are not credential-bearing log output." @@ -306,18 +322,23 @@ async fn create_run_pull_request( Err(err) => return err.into_response(), }; let run_state = cached.projection.as_ref(); - let inputs = match RunPrInputs::extract(run_state, body.force) { - Ok(inputs) => inputs, - Err(err) => return err.into_response(), - }; + if let Some(creation) = run_state + .pull_request_creation + .as_ref() + .filter(|creation| creation.is_pending()) + { + return accepted_pull_request_creation_response(&id, creation.clone()); + } + if let Err(err) = RunPrInputs::extract(run_state, body.force) { + return err.into_response(); + } let creds = match load_server_github_credentials(state.as_ref()).await { Ok(creds) => creds, Err(err) => return err.into_response(), }; - let github = match server_github_context(state.as_ref(), &creds) { - Ok(ctx) => ctx, - Err(err) => return err.into_response(), - }; + if let Err(err) = server_github_context(state.as_ref(), &creds) { + return err.into_response(); + } let model = if let Some(model) = body.model { model } else { @@ -328,8 +349,174 @@ async fn create_run_pull_request( .id .to_string() }; - let catalog = state.catalog(); + let creation_id = fabro_types::PullRequestCreationId::new(); + let event = workflow_event::Event::PullRequestCreationRequested { + creation_id, + model, + force: body.force, + }; + let appended = match workflow_event::append_event_if(&run_store, &id, &event, |projection| { + projection.pull_request.is_none() + && !projection + .pull_request_creation + .as_ref() + .is_some_and(fabro_types::PullRequestCreation::is_pending) + }) + .await + { + Ok(appended) => appended, + Err(err) => { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } + }; + + let cached = match state.cached_run(&id).await { + Ok(cached) => cached, + Err(err) => return err.into_response(), + }; + if !appended { + if let Some(creation) = cached + .projection + .pull_request_creation + .as_ref() + .filter(|creation| creation.is_pending()) + { + return accepted_pull_request_creation_response(&id, creation.clone()); + } + if let Some(record) = cached.projection.pull_request.as_ref() { + return ApiError::with_code( + StatusCode::CONFLICT, + format!("Pull request already exists at {}", record.html_url()), + "pull_request_exists", + ) + .into_response(); + } + return ApiError::new( + StatusCode::CONFLICT, + "Pull request creation state changed. Retry the request.", + ) + .into_response(); + } + + let Some(creation) = cached.projection.pull_request_creation.clone() else { + return ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "Pull request creation was accepted but its status is unavailable.", + ) + .into_response(); + }; + state.notify_pull_request_scheduler(); + accepted_pull_request_creation_response(&id, creation) +} +fn accepted_pull_request_creation_response( + run_id: &RunId, + creation: fabro_types::PullRequestCreation, +) -> Response { + let mut response = (StatusCode::ACCEPTED, Json(creation)).into_response(); + let location = format!("/api/v1/runs/{run_id}/pull_request/creation"); + if let Ok(location) = HeaderValue::from_str(&location) { + response.headers_mut().insert(header::LOCATION, location); + } + response + .headers_mut() + .insert(header::RETRY_AFTER, HeaderValue::from_static("1")); + response +} + +async fn get_run_pull_request_creation( + RequireRunScoped(id): RequireRunScoped, + State(state): State>, +) -> Response { + let cached = match state.cached_run(&id).await { + Ok(cached) => cached, + Err(err) => return err.into_response(), + }; + match cached.projection.pull_request_creation.clone() { + Some(creation) => Json(creation).into_response(), + None => ApiError::with_code( + StatusCode::NOT_FOUND, + "No explicit pull request creation was requested for this run.", + "no_pull_request_creation", + ) + .into_response(), + } +} + +async fn append_pull_request_creation_failure( + run_store: &fabro_store::RunDatabase, + run_id: &RunId, + creation_id: fabro_types::PullRequestCreationId, + error: String, +) -> anyhow::Result<()> { + let event = workflow_event::Event::PullRequestFailed { error }; + workflow_event::append_event_if(run_store, run_id, &event, |projection| { + projection + .pull_request_creation + .as_ref() + .is_some_and(|creation| creation.id == creation_id && creation.is_pending()) + }) + .await?; + Ok(()) +} + +pub(crate) async fn process_pull_request_creation( + state: Arc, + run_id: RunId, +) -> anyhow::Result<()> { + let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &run_id).await; + let run_store = state.stores.runs.open_run(&run_id).await?; + let run_state = run_store.state().await?; + let Some(creation) = run_state + .pull_request_creation + .as_ref() + .filter(|creation| creation.is_pending()) + .cloned() + else { + return Ok(()); + }; + if run_state.pull_request.is_some() { + return Ok(()); + } + + let inputs = match RunPrInputs::extract(&run_state, creation.force) { + Ok(inputs) => inputs, + Err(err) => { + return append_pull_request_creation_failure( + &run_store, + &run_id, + creation.id, + err.detail().to_string(), + ) + .await; + } + }; + let creds = match load_server_github_credentials(state.as_ref()).await { + Ok(creds) => creds, + Err(err) => { + return append_pull_request_creation_failure( + &run_store, + &run_id, + creation.id, + err.detail().to_string(), + ) + .await; + } + }; + let github = match server_github_context(state.as_ref(), &creds) { + Ok(github) => github, + Err(err) => { + return append_pull_request_creation_failure( + &run_store, + &run_id, + creation.id, + err.detail().to_string(), + ) + .await; + } + }; + let catalog = state.catalog(); let run_store_handle = run_store.clone().into(); let request = pull_request::OpenPullRequestRequest { github, @@ -339,18 +526,35 @@ async fn create_run_pull_request( expected_head_sha: inputs.final_git_sha, goal: inputs.goal, diff: inputs.diff, - model: &model, + model: &creation.model, draft: true, auto_merge: None, run_store: &run_store_handle, llm_source: state.llm_source.as_ref(), catalog, conclusion: Some(inputs.conclusion), - run_state: Some(run_state), + run_state: Some(&run_state), }; - let created_pull_request = match pull_request::open_pull_request(request).await { - Ok(created) => created, - Err(err) => return ApiError::new(StatusCode::BAD_GATEWAY, err).into_response(), + let shutdown = state.shutdown_token(); + let result = tokio::select! { + () = shutdown.cancelled() => return Ok(()), + result = time::timeout(PULL_REQUEST_CREATION_TIMEOUT, pull_request::open_pull_request(request)) => result, + }; + let created_pull_request = match result { + Ok(Ok(created)) => created, + Ok(Err(err)) => { + return append_pull_request_creation_failure(&run_store, &run_id, creation.id, err) + .await; + } + Err(_) => { + return append_pull_request_creation_failure( + &run_store, + &run_id, + creation.id, + "Pull request creation timed out after 10 minutes.".to_string(), + ) + .await; + } }; let event = workflow_event::Event::pull_request_created( @@ -361,11 +565,132 @@ async fn create_run_pull_request( &created_pull_request.title, true, ); - 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(); + workflow_event::append_event_if(&run_store, &run_id, &event, |projection| { + projection.pull_request.is_none() + && projection + .pull_request_creation + .as_ref() + .is_some_and(|current| current.id == creation.id && current.is_pending()) + }) + .await?; + Ok(()) +} + +async fn pending_pull_request_creation_run_ids(state: &AppState) -> anyhow::Result> { + let mut pending = state + .stores + .runs + .list_cached_runs(&ListRunsQuery::default(), chrono::Utc::now()) + .await? + .into_iter() + .filter_map(|cached| { + let creation = cached.projection.pull_request_creation.as_ref()?; + (cached.projection.pull_request.is_none() && creation.is_pending()) + .then_some((cached.run_id, creation.requested_at)) + }) + .collect::>(); + pending.sort_by_key(|(run_id, requested_at)| (*requested_at, *run_id)); + Ok(pending.into_iter().map(|(run_id, _)| run_id).collect()) +} + +pub(crate) fn spawn_pull_request_creation_supervisor(state: Arc) -> JoinHandle<()> { + tokio::spawn( + run_pull_request_creation_supervisor(state) + .instrument(tracing::info_span!("pull_request_creation_supervisor")), + ) +} + +async fn run_pull_request_creation_supervisor(state: Arc) { + let shutdown = state.shutdown_token(); + let mut workers = JoinSet::new(); + let mut active = HashSet::new(); + let mut task_run_ids = std::collections::HashMap::new(); + let mut scan_requested = true; + let mut scan_interval = time::interval(PULL_REQUEST_CREATION_SCAN_INTERVAL); + scan_interval.set_missed_tick_behavior(time::MissedTickBehavior::Delay); + + loop { + if scan_requested { + match pending_pull_request_creation_run_ids(state.as_ref()).await { + Ok(pending) => { + let available = + MAX_CONCURRENT_PULL_REQUEST_CREATIONS.saturating_sub(active.len()); + let ready = pending + .into_iter() + .filter(|run_id| !active.contains(run_id)) + .take(available) + .collect::>(); + for run_id in ready { + active.insert(run_id); + let task_state = Arc::clone(&state); + let handle = workers.spawn( + async move { + let result = + process_pull_request_creation(task_state, run_id).await; + (run_id, result) + } + .instrument( + tracing::info_span!("pull_request_creation", run_id = %run_id), + ), + ); + task_run_ids.insert(handle.id(), run_id); + } + } + Err(err) => { + tracing::warn!(error = %err, "Failed to scan queued pull request creations"); + } + } + scan_requested = false; + } + + if shutdown.is_cancelled() { + break; + } + + if workers.is_empty() { + tokio::select! { + () = shutdown.cancelled() => break, + () = state.pull_request_scheduler_notified() => scan_requested = true, + _ = scan_interval.tick() => scan_requested = true, + } + continue; + } + + tokio::select! { + () = shutdown.cancelled() => break, + () = state.pull_request_scheduler_notified() => scan_requested = true, + _ = scan_interval.tick() => scan_requested = true, + joined = workers.join_next_with_id() => { + match joined { + Some(Ok((task_id, (run_id, Ok(()))))) => { + task_run_ids.remove(&task_id); + active.remove(&run_id); + scan_requested = true; + } + Some(Ok((task_id, (run_id, Err(err))))) => { + task_run_ids.remove(&task_id); + active.remove(&run_id); + tracing::warn!(run_id = %run_id, error = %err, "Pull request creation worker failed"); + } + Some(Err(err)) => { + if let Some(run_id) = task_run_ids.remove(&err.id()) { + active.remove(&run_id); + tracing::warn!(run_id = %run_id, error = %err, "Pull request creation worker stopped unexpectedly"); + } else { + tracing::warn!(error = %err, "Pull request creation worker stopped unexpectedly"); + } + } + None => {} + } + } + } } - Json(created_pull_request.link).into_response() + while let Some(joined) = workers.join_next().await { + if let Err(err) = joined { + tracing::warn!(error = %err, "Pull request creation worker stopped during shutdown"); + } + } } async fn link_run_pull_request( @@ -373,6 +698,7 @@ async fn link_run_pull_request( State(state): State>, Json(body): Json, ) -> Response { + let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &id).await; let pull_request = match pull_request_record_from_link_request(&body) { Ok(record) => record, Err(err) => return err.into_response(), diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 1c04eb4a1a..54ce2fbe44 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -9671,6 +9671,17 @@ async fn create_run_pull_request_creates_and_persists_record() { .to_string(), ); }); + let find_mock = github.mock(|when, then| { + when.method("GET") + .path("/repos/acme/widgets/pulls") + .query_param("state", "open") + .query_param("base", "main") + .query_param("head", "acme:fabro/run/42") + .header("authorization", "Bearer ghu_test"); + then.status(200) + .header("content-type", "application/json") + .body("[]"); + }); let llm = MockServer::start_async().await; let response_mock = llm .mock_async(|when, then| { @@ -9730,12 +9741,50 @@ async fn create_run_pull_request_creates_and_persists_record() { ) .await .unwrap(); - let body = response_json!(response, StatusCode::OK).await; + assert_eq!( + response.headers().get(header::LOCATION).unwrap(), + &format!("/api/v1/runs/{run_id}/pull_request/creation") + ); + let body = response_json!(response, StatusCode::ACCEPTED).await; - assert_eq!(body["number"], 42); - assert_eq!(body["owner"], "acme"); - assert_eq!(body["repo"], "widgets"); - assert_eq!(body["html_url"], "https://github.com/acme/widgets/pull/42"); + assert_eq!(body["status"], "pending"); + assert_eq!(body["model"], "gpt-5.4"); + + // Starting the supervisor after the request simulates server recovery: + // the durable pending event is enough to resume the operation. + let supervisor = spawn_pull_request_creation_supervisor(Arc::clone(&state)); + + let creation_body = tokio::time::timeout(std::time::Duration::from_secs(3), async { + loop { + let response = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/pull_request/creation"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let body = response_json!(response, StatusCode::OK).await; + if body["status"] != "pending" { + break body; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("pull request creation should finish"); + + assert_eq!(creation_body["status"], "succeeded"); + assert_eq!(creation_body["pull_request"]["number"], 42); + assert_eq!(creation_body["pull_request"]["owner"], "acme"); + assert_eq!(creation_body["pull_request"]["repo"], "widgets"); + assert_eq!( + creation_body["pull_request"]["html_url"], + "https://github.com/acme/widgets/pull/42" + ); let state_response = app .oneshot( @@ -9754,7 +9803,153 @@ async fn create_run_pull_request_creates_and_persists_record() { response_mock.assert_async().await; branch_mock.assert(); + find_mock.assert(); create_mock.assert(); + state.shutdown_token().cancel(); + supervisor.await.unwrap(); +} + +#[tokio::test] +async fn create_run_pull_request_returns_the_active_durable_request() { + let github = MockServer::start(); + 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 configured_provider_ids = state.ready_llm_provider_ids().await; + let expected_default_model = state + .catalog() + .default_for_configured_ids(&configured_provider_ids) + .id + .to_string(); + let request_body = json!({ + "force": false, + "model": null + }) + .to_string(); + let first = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/pull_request"))) + .header("content-type", "application/json") + .body(Body::from(request_body.clone())) + .unwrap(), + ) + .await + .unwrap(); + let first_body = response_json!(first, StatusCode::ACCEPTED).await; + + let second = app + .oneshot( + Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/pull_request"))) + .header("content-type", "application/json") + .body(Body::from(request_body)) + .unwrap(), + ) + .await + .unwrap(); + let second_body = response_json!(second, StatusCode::ACCEPTED).await; + + assert_eq!(first_body["id"], second_body["id"]); + assert_eq!(first_body["model"], expected_default_model); + let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap(); + let events = run_store.list_events().await.unwrap(); + assert_eq!( + events + .iter() + .filter(|event| event.event.event_name() == "pull_request.creation_requested") + .count(), + 1 + ); +} + +#[tokio::test] +async fn create_run_pull_request_persists_generation_failure() { + let github = MockServer::start(); + let branch_mock = github.mock(|when, then| { + when.method("GET") + .path("/repos/acme/widgets/branches/fabro/run/42") + .header("authorization", "Bearer ghu_test"); + then.status(200) + .header("content-type", "application/json") + .body(json!({ "commit": { "sha": "final-sha" } }).to_string()); + }); + let find_mock = github.mock(|when, then| { + when.method("GET") + .path("/repos/acme/widgets/pulls") + .query_param("state", "open") + .query_param("base", "main") + .query_param("head", "acme:fabro/run/42") + .header("authorization", "Bearer ghu_test"); + then.status(200) + .header("content-type", "application/json") + .body("[]"); + }); + 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": "gpt-5.4" }).to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + response_json!(response, StatusCode::ACCEPTED).await; + let supervisor = spawn_pull_request_creation_supervisor(Arc::clone(&state)); + + let creation = tokio::time::timeout(std::time::Duration::from_secs(3), async { + loop { + let response = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/pull_request/creation"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let body = response_json!(response, StatusCode::OK).await; + if body["status"] != "pending" { + break body; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("pull request creation should finish"); + + assert_eq!(creation["status"], "failed"); + assert!( + creation["error"] + .as_str() + .is_some_and(|error| !error.is_empty()) + ); + assert!(creation["pull_request"].is_null()); + branch_mock.assert(); + find_mock.assert(); + state.shutdown_token().cancel(); + supervisor.await.unwrap(); } #[tokio::test] diff --git a/lib/components/fabro-github/src/lib.rs b/lib/components/fabro-github/src/lib.rs index 7e16f6ed13..d4d4baff8c 100644 --- a/lib/components/fabro-github/src/lib.rs +++ b/lib/components/fabro-github/src/lib.rs @@ -642,6 +642,108 @@ pub struct CreatedPullRequest { pub node_id: String, } +/// Existing open pull request found for an exact base, head branch, and head +/// commit. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ExistingPullRequest { + pub html_url: String, + pub number: u64, + pub node_id: String, + pub title: String, +} + +/// Find an open pull request that already carries the expected head commit. +/// +/// This supports recovery when GitHub created a pull request but the caller +/// stopped before it could persist the result locally. +pub async fn find_open_pull_request( + ctx: &GitHubContext<'_>, + owner: &str, + repo: &str, + base: &str, + head: &str, + expected_head_sha: &str, +) -> anyhow::Result> { + let client = ctx.http_client()?; + find_open_pull_request_with_client(&client, ctx, owner, repo, base, head, expected_head_sha) + .await +} + +#[allow( + clippy::too_many_arguments, + reason = "Pull request reconciliation needs explicit repo, branch, and commit coordinates." +)] +pub async fn find_open_pull_request_with_client( + client: &impl HttpClient, + ctx: &GitHubContext<'_>, + owner: &str, + repo: &str, + base: &str, + head: &str, + expected_head_sha: &str, +) -> anyhow::Result> { + #[derive(Deserialize)] + struct PullRequestHead { + sha: String, + } + + #[derive(Deserialize)] + struct PullRequestListItem { + html_url: String, + number: u64, + node_id: String, + title: String, + head: PullRequestHead, + } + + let token = ctx + .creds + .resolve_bearer_token( + client, + owner, + repo, + ctx.base_url, + serde_json::json!({ "contents": "write", "pull_requests": "write" }), + ) + .await?; + let mut url = DisplaySafeUrl::parse(&format!("{}/repos/{owner}/{repo}/pulls", ctx.base_url)) + .context("Failed to build pull request reconciliation URL")?; + url.query_pairs_mut() + .append_pair("state", "open") + .append_pair("base", base) + .append_pair("head", &format!("{owner}:{head}")); + let auth = format!("Bearer {token}"); + let resp = client + .request( + HttpMethod::Get, + &url.raw_string(), + &github_headers(&auth), + None, + ) + .await + .context("Failed to find an existing pull request")?; + if resp.status != 200 { + bail!( + "Unexpected status {} finding an existing pull request: {}", + resp.status, + resp.text() + ); + } + + let pull_requests = resp + .json::>() + .context("Failed to parse existing pull request response")?; + Ok(pull_requests + .into_iter() + .find(|pull_request| pull_request.head.sha == expected_head_sha) + .map(|pull_request| ExistingPullRequest { + html_url: pull_request.html_url, + number: pull_request.number, + node_id: pull_request.node_id, + title: pull_request.title, + })) +} + /// Create a pull request on GitHub. /// /// Signs a JWT, obtains a PR-scoped installation token, and POSTs to the @@ -1907,6 +2009,52 @@ mod tests { assert!(err.contains("repo"), "got: {err}"); } + #[tokio::test] + async fn find_open_pull_request_matches_the_expected_head_commit() { + let mock = MockHttpClient::new() + .on( + HttpMethod::Get, + "/repos/owner/repo/pulls?state=open&base=main&head=owner%3Afabro%2Frun%2F1", + 200, + r#"[ + { + "html_url": "https://github.com/owner/repo/pull/40", + "number": 40, + "node_id": "PR_wrong", + "title": "Old head", + "head": {"sha": "old-sha"} + }, + { + "html_url": "https://github.com/owner/repo/pull/42", + "number": 42, + "node_id": "PR_expected", + "title": "Expected head", + "head": {"sha": "final-sha"} + } + ]"#, + ) + .with_req_header("Authorization", "Bearer ghu_test"); + let creds = GitHubCredentials::Pat("ghu_test".to_string()); + let ctx = GitHubContext::new(&creds, "https://api.test"); + + let found = find_open_pull_request_with_client( + &mock, + &ctx, + "owner", + "repo", + "main", + "fabro/run/1", + "final-sha", + ) + .await + .unwrap() + .expect("matching pull request should be found"); + + assert_eq!(found.number, 42); + assert_eq!(found.node_id, "PR_expected"); + assert_eq!(found.title, "Expected head"); + } + #[tokio::test] async fn create_iat_auth_failed() { let mock = diff --git a/lib/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index d2042a4c6a..2e50d448c2 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -12,14 +12,14 @@ use fabro_types::{ ActivatedSkill, AgentControlState, AskFabro, BilledModelUsage, BilledTokenCounts, Checkpoint, CheckpointRecord, CommandTermination, Conclusion, EventBody, FailureCategory, FailureSignature, InterviewQuestionRecord, McpServerProjection, McpServerStatus, Outcome, PendingInterviewRecord, - PendingReason, PullRequestLink, RepositoryRef, Run, RunApproval, RunApprovalState, - RunBillingSummary, RunControlAction, RunDiff, RunEvent, RunId, RunLifecycle, RunLinks, - RunModel, RunOrigin, RunProjection, RunSandbox, RunSandboxFailure, RunSandboxInstance, - RunSandboxPlan, RunSandboxRuntime, RunSize, RunSpec, RunStatus, RunTimestamps, - SandboxProviderKind, StageCompletion, StageHandler, StageId, StageInferenceProjection, - StageModelUsage, StageOutcome, StageProjection, StageState, StartRecord, SubAgentProjection, - SubAgentStatus, TodoListKind, TodoListProjection, TodoProjection, WorkflowRef, first_event_seq, - timing, + PendingReason, PullRequestCreation, PullRequestCreationStatus, PullRequestLink, RepositoryRef, + Run, RunApproval, RunApprovalState, RunBillingSummary, RunControlAction, RunDiff, RunEvent, + RunId, RunLifecycle, RunLinks, RunModel, RunOrigin, RunProjection, RunSandbox, + RunSandboxFailure, RunSandboxInstance, RunSandboxPlan, RunSandboxRuntime, RunSize, RunSpec, + RunStatus, RunTimestamps, SandboxProviderKind, StageCompletion, StageHandler, StageId, + StageInferenceProjection, StageModelUsage, StageOutcome, StageProjection, StageState, + StartRecord, SubAgentProjection, SubAgentStatus, TodoListKind, TodoListProjection, + TodoProjection, WorkflowRef, first_event_seq, timing, }; use fabro_util::error::render_compact_with_causes; @@ -307,18 +307,57 @@ impl RunProjectionReducer for RunProjection { }, })); } + EventBody::PullRequestCreationRequested(props) => { + self.pull_request_creation = Some(PullRequestCreation { + id: props.creation_id, + status: PullRequestCreationStatus::Pending, + model: props.model.clone(), + force: props.force, + requested_at: ts, + updated_at: ts, + pull_request: None, + error: None, + }); + } EventBody::PullRequestCreated(props) => { - self.pull_request = Some(PullRequestLink { + let pull_request = PullRequestLink { owner: props.owner.clone(), repo: props.repo.clone(), number: props.pr_number, - }); + }; + self.pull_request = Some(pull_request.clone()); + if let Some(creation) = self.pull_request_creation.as_mut() { + if creation.is_pending() { + creation.status = PullRequestCreationStatus::Succeeded; + creation.updated_at = ts; + creation.pull_request = Some(pull_request); + creation.error = None; + } + } } EventBody::PullRequestLinked(props) => { self.pull_request = Some(props.pull_request.clone()); + if let Some(creation) = self.pull_request_creation.as_mut() { + if creation.is_pending() { + creation.status = PullRequestCreationStatus::Succeeded; + creation.updated_at = ts; + creation.pull_request = Some(props.pull_request.clone()); + creation.error = None; + } + } } EventBody::PullRequestUnlinked(_) => { self.pull_request = None; + self.pull_request_creation = None; + } + EventBody::PullRequestFailed(props) => { + if let Some(creation) = self.pull_request_creation.as_mut() { + if creation.is_pending() { + creation.status = PullRequestCreationStatus::Failed; + creation.updated_at = ts; + creation.error = Some(props.error.clone()); + } + } } EventBody::InterviewStarted(props) => { if props.question_id.is_empty() { @@ -1651,13 +1690,13 @@ mod tests { AgentBackend, AgentControlState, AttrValue, AutomationRef, BilledModelUsage, BilledTokenCounts, BlockedReason, Checkpoint, CheckpointRecord, CommandTermination, EventBody, FailureCategory, FailureDetail, FailureReason, Graph, McpServerStatus, Node, - Outcome, ParallelBranchId, PendingReason, PermissionLevel, PullRequestLink, QuestionType, - ReasoningEffort, RunApprovalState, RunBlobId, RunControlAction, RunDiff, RunEvent, RunSize, - RunSpec, RunStatus, Speed, StageContextWindowBreakdownItem, StageContextWindowCategory, - StageContextWindowCountMethod, StageContextWindowProjection, StageContextWindowStaleness, - StageContextWindowWarning, StageHandler, StageModelUsage, StageOutcome, StageState, - StageTiming, SubAgentStatus, SuccessReason, WorkflowSettings, first_event_seq, fixtures, - test_support, + Outcome, ParallelBranchId, PendingReason, PermissionLevel, PullRequestCreationStatus, + PullRequestLink, QuestionType, ReasoningEffort, RunApprovalState, RunBlobId, + RunControlAction, RunDiff, RunEvent, RunSize, RunSpec, RunStatus, Speed, + StageContextWindowBreakdownItem, StageContextWindowCategory, StageContextWindowCountMethod, + StageContextWindowProjection, StageContextWindowStaleness, StageContextWindowWarning, + StageHandler, StageModelUsage, StageOutcome, StageState, StageTiming, SubAgentStatus, + SuccessReason, WorkflowSettings, first_event_seq, fixtures, test_support, }; use serde_json::json; @@ -4599,6 +4638,77 @@ mod tests { assert_eq!(summary.pull_request, state.pull_request); } + #[test] + fn pull_request_creation_projects_failure_retry_and_success() { + use fabro_types::run_event::{ + PullRequestCreatedProps, PullRequestCreationRequestedProps, PullRequestFailedProps, + }; + + let mut state = running_projection(); + let first_id = "01KYYK70WTZT2E551P3H5P0059".parse().unwrap(); + state + .apply_event(&test_event( + 1, + EventBody::PullRequestCreationRequested(PullRequestCreationRequestedProps { + creation_id: first_id, + model: "kimi-k3".to_string(), + force: false, + }), + None, + )) + .unwrap(); + assert!(state.pull_request_creation.as_ref().unwrap().is_pending()); + + state + .apply_event(&test_event( + 2, + EventBody::PullRequestFailed(PullRequestFailedProps { + error: "provider unavailable".to_string(), + }), + None, + )) + .unwrap(); + let failed = state.pull_request_creation.as_ref().unwrap(); + assert_eq!(failed.status, PullRequestCreationStatus::Failed); + assert_eq!(failed.error.as_deref(), Some("provider unavailable")); + + let retry_id = "01KYYK70WTZT2E551P3H5P0060".parse().unwrap(); + state + .apply_event(&test_event( + 3, + EventBody::PullRequestCreationRequested(PullRequestCreationRequestedProps { + creation_id: retry_id, + model: "claude-sonnet-4-6".to_string(), + force: true, + }), + None, + )) + .unwrap(); + assert_eq!(state.pull_request_creation.as_ref().unwrap().id, retry_id); + + state + .apply_event(&test_event( + 4, + EventBody::PullRequestCreated(PullRequestCreatedProps { + pr_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(), + pr_number: 123, + owner: "fabro-sh".to_string(), + repo: "fabro".to_string(), + base_branch: "main".to_string(), + head_branch: "fabro/run/demo".to_string(), + head_sha: Some("final-sha".to_string()), + title: "Create asynchronously".to_string(), + draft: true, + }), + None, + )) + .unwrap(); + let succeeded = state.pull_request_creation.as_ref().unwrap(); + assert_eq!(succeeded.status, PullRequestCreationStatus::Succeeded); + assert_eq!(succeeded.pull_request.as_ref().unwrap().number, 123); + assert!(succeeded.error.is_none()); + } + #[test] fn pull_request_linked_replaces_and_unlinked_clears_projection() { use fabro_types::run_event::{ diff --git a/lib/components/fabro-workflow/src/event/convert.rs b/lib/components/fabro-workflow/src/event/convert.rs index 3a99f6b64e..3ac58c2d8b 100644 --- a/lib/components/fabro-workflow/src/event/convert.rs +++ b/lib/components/fabro-workflow/src/event/convert.rs @@ -1320,6 +1320,17 @@ fn event_body_from_event(event: &Event) -> EventBody { stderr: stderr.clone(), duration_ms: *duration_ms, }), + Event::PullRequestCreationRequested { + creation_id, + model, + force, + } => EventBody::PullRequestCreationRequested( + fabro_types::PullRequestCreationRequestedProps { + creation_id: *creation_id, + model: model.clone(), + force: *force, + }, + ), Event::PullRequestCreated { pr_url, pr_number, diff --git a/lib/components/fabro-workflow/src/event/events.rs b/lib/components/fabro-workflow/src/event/events.rs index 126b23d900..61a1ceeaa0 100644 --- a/lib/components/fabro-workflow/src/event/events.rs +++ b/lib/components/fabro-workflow/src/event/events.rs @@ -4,9 +4,9 @@ use ::fabro_types::{ AutomationRef, BilledTokenCounts, BlockedReason, CommandTermination, DiffSummary, FailureReason, ForkSourceRef, GitContext, PairId, PairMessageId, PairSystemMessageKind, PairTarget, ParallelBranchId, ParallelBranchResult, PendingReason, PermissionLevel, Principal, - PullRequestLink, ReviewTarget, RunBlobId, RunFailure, RunId, RunNoticeLevel, - RunPairEndedReason, RunPairFailedReason, RunProvenance, RunRunnableSource, RunTiming, - SandboxProviderKind, StageId, StageOutcome, StageTiming, SuccessReason, + PullRequestCreationId, PullRequestLink, ReviewTarget, RunBlobId, RunFailure, RunId, + RunNoticeLevel, RunPairEndedReason, RunPairFailedReason, RunProvenance, RunRunnableSource, + RunTiming, SandboxProviderKind, StageId, StageOutcome, StageTiming, SuccessReason, run_event as fabro_types, }; use fabro_agent::{AgentEvent, SandboxEvent}; @@ -726,6 +726,11 @@ pub enum Event { stderr: String, duration_ms: u64, }, + PullRequestCreationRequested { + creation_id: PullRequestCreationId, + model: String, + force: bool, + }, PullRequestCreated { pr_url: String, pr_number: u64, @@ -1534,6 +1539,13 @@ impl Event { } => { debug!(node_id, duration_ms, "Agent ACP timed out"); } + Self::PullRequestCreationRequested { + creation_id, + model, + force, + } => { + info!(creation_id = %creation_id, model, force, "Pull request creation requested"); + } Self::PullRequestCreated { pr_url, pr_number, diff --git a/lib/components/fabro-workflow/src/event/names.rs b/lib/components/fabro-workflow/src/event/names.rs index ee051c602a..12a26d1a90 100644 --- a/lib/components/fabro-workflow/src/event/names.rs +++ b/lib/components/fabro-workflow/src/event/names.rs @@ -151,6 +151,7 @@ pub fn event_name(event: &Event) -> &'static str { Event::AgentAcpCompleted { .. } => "agent.acp.completed", Event::AgentAcpCancelled { .. } => "agent.acp.cancelled", Event::AgentAcpTimedOut { .. } => "agent.acp.timed_out", + Event::PullRequestCreationRequested { .. } => "pull_request.creation_requested", Event::PullRequestCreated { .. } => "pull_request.created", Event::PullRequestLinked { .. } => "pull_request.linked", Event::PullRequestUnlinked { .. } => "pull_request.unlinked", diff --git a/lib/components/fabro-workflow/src/pipeline/pull_request.rs b/lib/components/fabro-workflow/src/pipeline/pull_request.rs index 65c4a43088..21d9164a5a 100644 --- a/lib/components/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/components/fabro-workflow/src/pipeline/pull_request.rs @@ -472,6 +472,47 @@ pub struct CreatedPullRequest { pub head_branch: String, } +fn recovered_pull_request( + existing: fabro_github::ExistingPullRequest, + owner: String, + repo: String, + base_branch: &str, + head_branch: &str, +) -> CreatedPullRequest { + CreatedPullRequest { + link: PullRequestLink { + owner, + repo, + number: existing.number, + }, + title: existing.title, + base_branch: base_branch.to_string(), + head_branch: head_branch.to_string(), + } +} + +async fn enable_auto_merge_if_requested( + github: &github_app::GitHubContext<'_>, + owner: &str, + repo: &str, + node_id: &str, + number: u64, + options: Option<&AutoMergeOptions>, +) { + let Some(options) = options else { + return; + }; + match github_app::enable_auto_merge(github, owner, repo, node_id, options.merge_strategy).await + { + Ok(()) => info!(pr_number = number, "Auto-merge enabled"), + Err(err) => warn!( + pr_number = number, + error = %err, + "Failed to enable auto-merge (repo may not have auto-merge enabled in settings)" + ), + } +} + /// How many times to read the remote branch head before giving up. /// /// `GET /repos/{owner}/{repo}/branches/{branch}` is replica-served, so shortly @@ -535,6 +576,36 @@ pub async fn open_pull_request( // branch would otherwise cost a full LLM call before failing. verify_remote_head(&req, &owner, &repo).await?; + if let Some(existing) = github_app::find_open_pull_request( + &req.github, + &owner, + &repo, + req.base_branch, + req.head_branch, + req.expected_head_sha, + ) + .await + .map_err(|err| format!("failed to reconcile an existing pull request: {err:#}"))? + { + info!(pr_url = %existing.html_url, pr_number = existing.number, "Existing pull request reconciled"); + enable_auto_merge_if_requested( + &req.github, + &owner, + &repo, + &existing.node_id, + existing.number, + req.auto_merge.as_ref(), + ) + .await; + return Ok(recovered_pull_request( + existing, + owner, + repo, + req.base_branch, + req.head_branch, + )); + } + let content = build_pr_content( req.diff, req.goal, @@ -550,7 +621,7 @@ pub async fn open_pull_request( 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, @@ -561,32 +632,58 @@ pub async fn open_pull_request( req.draft, ) .await - .map_err(|err| format!("{err:#}"))?; - - info!(pr_url = %created.html_url, created.number, "Pull request created"); - - if let Some(am_cfg) = req.auto_merge { - match github_app::enable_auto_merge( - &req.github, - &owner, - &repo, - &created.node_id, - am_cfg.merge_strategy, - ) - .await - { - Ok(()) => { - info!(pr_number = created.number, "Auto-merge enabled"); - } - Err(e) => { - tracing::warn!( - pr_number = created.number, - error = %e, - "Failed to enable auto-merge (repo may not have auto-merge enabled in settings)" - ); + { + Ok(created) => created, + Err(create_err) => { + match github_app::find_open_pull_request( + &req.github, + &owner, + &repo, + req.base_branch, + req.head_branch, + req.expected_head_sha, + ) + .await + { + Ok(Some(existing)) => { + info!(pr_url = %existing.html_url, pr_number = existing.number, "Pull request reconciled after create response failed"); + enable_auto_merge_if_requested( + &req.github, + &owner, + &repo, + &existing.node_id, + existing.number, + req.auto_merge.as_ref(), + ) + .await; + return Ok(recovered_pull_request( + existing, + owner, + repo, + req.base_branch, + req.head_branch, + )); + } + Ok(None) => return Err(format!("{create_err:#}")), + Err(reconcile_err) => { + return Err(format!( + "{create_err:#}; failed to reconcile the pull request after creation: {reconcile_err:#}" + )); + } } } - } + }; + + info!(pr_url = %created.html_url, created.number, "Pull request created"); + enable_auto_merge_if_requested( + &req.github, + &owner, + &repo, + &created.node_id, + created.number, + req.auto_merge.as_ref(), + ) + .await; let link = PullRequestLink { owner, @@ -1609,19 +1706,20 @@ 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, - branch_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, + branch_mock_id: usize, + reconcile_mock_id: usize, + github_mock_id: usize, + llm_source: Arc, + catalog: Arc, + creds: fabro_github::GitHubCredentials, + run_store: RunStoreHandle, } impl FallbackHarness { @@ -1632,6 +1730,9 @@ mod tests { httpmock::Mock::new(self.branch_mock_id, &self.github_server) .assert_async() .await; + httpmock::Mock::new(self.reconcile_mock_id, &self.github_server) + .assert_async() + .await; httpmock::Mock::new(self.github_mock_id, &self.github_server) .assert_async() .await; @@ -1690,6 +1791,19 @@ mod tests { })); }) .await; + let reconcile_mock = github_server + .mock_async(|when, then| { + when.method(GET) + .path("/repos/owner/repo/pulls") + .query_param("state", "open") + .query_param("base", "main") + .query_param("head", "owner:fabro/run/123") + .header("authorization", "Bearer test-token"); + then.status(200) + .header("content-type", "application/json") + .json_body(serde_json::json!([])); + }) + .await; let vault_dir = tempfile::tempdir().unwrap(); let mut vault = Vault::load(vault_dir.path().join("secrets.json")).unwrap(); @@ -1780,6 +1894,7 @@ mod tests { let openai_mock_id = openai_mock.id; let branch_mock_id = branch_mock.id; + let reconcile_mock_id = reconcile_mock.id; let github_mock_id = github_mock.id; FallbackHarness { @@ -1788,6 +1903,7 @@ mod tests { github_server, openai_mock_id, branch_mock_id, + reconcile_mock_id, github_mock_id, llm_source, catalog, diff --git a/lib/foundation/fabro-api/build.rs b/lib/foundation/fabro-api/build.rs index 8d3a682846..c09b8b0cb6 100644 --- a/lib/foundation/fabro-api/build.rs +++ b/lib/foundation/fabro-api/build.rs @@ -545,6 +545,21 @@ fn main() { ("EventEnvelope", "fabro_types::EventEnvelope", &[]), ("PullRequest", "fabro_types::PullRequest", &[]), ("PullRequestLink", "fabro_types::PullRequestLink", &[]), + ( + "PullRequestCreationId", + "fabro_types::PullRequestCreationId", + &[], + ), + ( + "PullRequestCreationStatus", + "fabro_types::PullRequestCreationStatus", + &[], + ), + ( + "PullRequestCreation", + "fabro_types::PullRequestCreation", + &[], + ), ("PullRequestDetails", "fabro_types::PullRequestDetails", &[]), ("PullRequestMeta", "fabro_types::PullRequestMeta", &[]), ( diff --git a/lib/foundation/fabro-api/src/lib.rs b/lib/foundation/fabro-api/src/lib.rs index 73a9e39beb..395868b61d 100644 --- a/lib/foundation/fabro-api/src/lib.rs +++ b/lib/foundation/fabro-api/src/lib.rs @@ -53,7 +53,8 @@ pub mod types { McpTransportView, Message, PairId, PairMessageId, PairMessageRecord, PairMessageRequest, PairRecord, PairStartRequest, PairStatus, PairTarget, PairTranscriptEntry, PairTranscriptResponse, ParallelBranchId, ParallelBranchResult, PendingInterviewRecord, - PermissionLevel, PreRunPushOutcome, Principal, PullRequest, PullRequestDetails, + PermissionLevel, PreRunPushOutcome, Principal, PullRequest, PullRequestCreation, + PullRequestCreationId, PullRequestCreationStatus, PullRequestDetails, PullRequestDetailsStatus, PullRequestDetailsUnavailableReason, PullRequestLink, PullRequestMeta, PullRequestResponse, QuestionType, ReasoningOutput, RepositoryRef, ReviewTarget, ReviewTargetKind, Role, Run, RunApproval, RunApprovalState, diff --git a/lib/foundation/fabro-api/tests/pull_request_round_trip.rs b/lib/foundation/fabro-api/tests/pull_request_round_trip.rs index 98ca72c92e..4cf7e38066 100644 --- a/lib/foundation/fabro-api/tests/pull_request_round_trip.rs +++ b/lib/foundation/fabro-api/tests/pull_request_round_trip.rs @@ -2,7 +2,10 @@ use std::any::{TypeId, type_name}; use fabro_api::types::MergeRunPullRequestRequest; use fabro_types::settings::run::MergeStrategy; -use fabro_types::{PullRequest, PullRequestLink, PullRequestResponse}; +use fabro_types::{ + PullRequest, PullRequestCreation, PullRequestCreationId, PullRequestCreationStatus, + PullRequestLink, PullRequestResponse, +}; use serde_json::json; #[test] @@ -28,6 +31,42 @@ fn pull_request_response_reuses_domain_types() { assert_same_type_as_pull_request_link(&response.data.link); } +#[test] +fn pull_request_creation_reuses_domain_type() { + let fixture = json!({ + "id": "01KYYK70WTZT2E551P3H5P0059", + "status": "pending", + "model": "kimi-k3", + "force": true, + "requested_at": "2026-08-01T12:00:00Z", + "updated_at": "2026-08-01T12:00:00Z" + }); + let creation: fabro_api::types::PullRequestCreation = + serde_json::from_value(fixture.clone()).expect("creation should deserialize"); + + assert_eq!( + TypeId::of::(), + TypeId::of::() + ); + assert_eq!( + TypeId::of::(), + TypeId::of::() + ); + assert_eq!( + TypeId::of::(), + TypeId::of::() + ); + assert_eq!(serde_json::to_value(creation).unwrap(), fixture); + assert_eq!( + serde_json::to_value(PullRequestCreationStatus::Succeeded).unwrap(), + "succeeded" + ); + assert_eq!( + serde_json::to_value(PullRequestCreationStatus::Failed).unwrap(), + "failed" + ); +} + #[test] fn merge_request_reuses_run_merge_strategy_type() { let request: MergeRunPullRequestRequest = serde_json::from_value(json!({ "method": "squash" })) diff --git a/lib/foundation/fabro-client/src/client.rs b/lib/foundation/fabro-client/src/client.rs index 24a7fd0c90..4b3537aa99 100644 --- a/lib/foundation/fabro-client/src/client.rs +++ b/lib/foundation/fabro-client/src/client.rs @@ -38,6 +38,7 @@ use crate::{AuthEntry, OAuthEntry, StoredSubject, sse}; const DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); const DEFAULT_HEALTH_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(250); +const PULL_REQUEST_CREATION_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); type TransportFuture = BoxFuture<'static, Result<(fabro_http::HttpClient, String)>>; @@ -1435,6 +1436,49 @@ impl Client { force: bool, model: Option, ) -> Result { + let mut creation = self + .request_run_pull_request_creation(run_id, force, model) + .await?; + let creation_id = creation.id; + loop { + match creation.status { + fabro_types::PullRequestCreationStatus::Pending => { + time::sleep(PULL_REQUEST_CREATION_POLL_INTERVAL).await; + creation = self.get_run_pull_request_creation(run_id).await?; + if creation.id != creation_id { + bail!( + "Pull request creation {creation_id} was superseded by {}", + creation.id + ); + } + } + fabro_types::PullRequestCreationStatus::Succeeded => { + return creation.pull_request.ok_or_else(|| { + anyhow!( + "Pull request creation {} succeeded without a pull request record", + creation.id + ) + }); + } + fabro_types::PullRequestCreationStatus::Failed => { + bail!( + "Pull request creation failed: {}", + creation + .error + .as_deref() + .unwrap_or("the server did not provide an error") + ); + } + } + } + } + + pub async fn request_run_pull_request_creation( + &self, + run_id: &RunId, + force: bool, + model: Option, + ) -> Result { let body = types::CreateRunPullRequestRequest { force, model }; let response = self .send_api(|client| async move { @@ -1447,7 +1491,24 @@ impl Client { }) .await .map_err(add_pr_upgrade_hint)?; - convert_type(response.into_inner()) + Ok(response.into_inner()) + } + + pub async fn get_run_pull_request_creation( + &self, + run_id: &RunId, + ) -> Result { + let response = self + .send_api(|client| async move { + client + .get_run_pull_request_creation() + .id(run_id.to_string()) + .send() + .await + }) + .await + .map_err(add_pr_upgrade_hint)?; + Ok(response.into_inner()) } pub async fn get_run_pull_request( diff --git a/lib/foundation/fabro-types/src/lib.rs b/lib/foundation/fabro-types/src/lib.rs index 309505c4ad..5ec509a9a1 100644 --- a/lib/foundation/fabro-types/src/lib.rs +++ b/lib/foundation/fabro-types/src/lib.rs @@ -100,7 +100,8 @@ pub use pair::{ pub use parallel::ParallelBranchResult; pub use principal::{AuthMethod, Principal, SystemActorKind, UserPrincipal}; pub use pull_request::{ - CheckRun, CheckRunStatus, PullRequest, PullRequestDetails, PullRequestDetailsStatus, + CheckRun, CheckRunStatus, PullRequest, PullRequestCreation, PullRequestCreationId, + PullRequestCreationStatus, PullRequestDetails, PullRequestDetailsStatus, PullRequestDetailsUnavailableReason, PullRequestGithubDetail, PullRequestLink, PullRequestMeta, PullRequestRef, PullRequestResponse, PullRequestTimestamps, PullRequestUser, }; diff --git a/lib/foundation/fabro-types/src/pull_request.rs b/lib/foundation/fabro-types/src/pull_request.rs index d2b91e8fe9..cdd922754f 100644 --- a/lib/foundation/fabro-types/src/pull_request.rs +++ b/lib/foundation/fabro-types/src/pull_request.rs @@ -1,7 +1,46 @@ +use chrono::{DateTime, Utc}; use serde::de::Error as DeError; use serde::ser::SerializeStruct; use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use crate::id::ulid_id; + +ulid_id!(PullRequestCreationId); + +/// Durable status for an explicitly requested pull request creation. +#[derive( + Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, strum::Display, strum::EnumString, +)] +#[serde(rename_all = "snake_case")] +#[strum(serialize_all = "snake_case")] +pub enum PullRequestCreationStatus { + Pending, + Succeeded, + Failed, +} + +/// Latest explicit pull request creation requested for a run. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct PullRequestCreation { + pub id: PullRequestCreationId, + pub status: PullRequestCreationStatus, + pub model: String, + pub force: bool, + pub requested_at: DateTime, + pub updated_at: DateTime, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pull_request: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +impl PullRequestCreation { + #[must_use] + pub fn is_pending(&self) -> bool { + self.status == PullRequestCreationStatus::Pending + } +} + /// Minimal GitHub pull request reference stored on a workflow run. #[derive(Debug, Clone, PartialEq, Eq)] pub struct PullRequestLink { diff --git a/lib/foundation/fabro-types/src/run_event/misc.rs b/lib/foundation/fabro-types/src/run_event/misc.rs index e1505c6d39..844df5dce1 100644 --- a/lib/foundation/fabro-types/src/run_event/misc.rs +++ b/lib/foundation/fabro-types/src/run_event/misc.rs @@ -3,7 +3,8 @@ use serde::{Deserialize, Serialize}; use super::ExecOutputTail; use crate::{ - CommandTermination, ParallelBranchResult, PullRequestLink, ReviewTarget, StageId, StageOutcome, + CommandTermination, ParallelBranchResult, PullRequestCreationId, PullRequestLink, ReviewTarget, + StageId, StageOutcome, }; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] @@ -261,6 +262,13 @@ pub struct AgentAcpTimedOutProps { pub duration_ms: u64, } +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct PullRequestCreationRequestedProps { + pub creation_id: PullRequestCreationId, + pub model: String, + pub force: bool, +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct PullRequestCreatedProps { pub pr_url: String, diff --git a/lib/foundation/fabro-types/src/run_event/mod.rs b/lib/foundation/fabro-types/src/run_event/mod.rs index 59892030fd..0b613920c2 100644 --- a/lib/foundation/fabro-types/src/run_event/mod.rs +++ b/lib/foundation/fabro-types/src/run_event/mod.rs @@ -352,6 +352,8 @@ pub enum EventBody { AgentAcpCancelled(AgentAcpCancelledProps), #[serde(rename = "agent.acp.timed_out")] AgentAcpTimedOut(AgentAcpTimedOutProps), + #[serde(rename = "pull_request.creation_requested")] + PullRequestCreationRequested(PullRequestCreationRequestedProps), #[serde(rename = "pull_request.created")] PullRequestCreated(PullRequestCreatedProps), #[serde(rename = "pull_request.linked")] @@ -567,6 +569,7 @@ impl EventBody { Self::AgentAcpCompleted(_) => "agent.acp.completed", Self::AgentAcpCancelled(_) => "agent.acp.cancelled", Self::AgentAcpTimedOut(_) => "agent.acp.timed_out", + Self::PullRequestCreationRequested(_) => "pull_request.creation_requested", Self::PullRequestCreated(_) => "pull_request.created", Self::PullRequestLinked(_) => "pull_request.linked", Self::PullRequestUnlinked(_) => "pull_request.unlinked", diff --git a/lib/foundation/fabro-types/src/run_projection.rs b/lib/foundation/fabro-types/src/run_projection.rs index 4c5cea665b..3a6be70d40 100644 --- a/lib/foundation/fabro-types/src/run_projection.rs +++ b/lib/foundation/fabro-types/src/run_projection.rs @@ -10,39 +10,41 @@ use crate::run_event::{AgentSessionActivatedProps, StagePromptProps}; use crate::{ AgentBackend, AgentMcpToolSummary, AgentSkillActivationSource, AgentSkillSummary, AgentToolSummary, BilledTokenCounts, Checkpoint, Conclusion, InterviewQuestionRecord, - InvalidTransition, LlmOutputKind, ModelRef, ParallelBranchId, PermissionLevel, PullRequestLink, - RunApproval, RunControlAction, RunDiff, RunId, RunSandbox, RunSpec, RunStatus, RunTiming, - StageCompletion, StageHandler, StageId, StageState, StageTiming, StartRecord, - TodoListProjection, timing, + InvalidTransition, LlmOutputKind, ModelRef, ParallelBranchId, PermissionLevel, + PullRequestCreation, PullRequestLink, RunApproval, RunControlAction, RunDiff, RunId, + RunSandbox, RunSpec, RunStatus, RunTiming, StageCompletion, StageHandler, StageId, StageState, + StageTiming, StartRecord, TodoListProjection, timing, }; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct RunProjection { #[serde(default, skip_serializing_if = "String::is_empty")] - pub title: String, + pub title: String, #[serde(default, skip_serializing_if = "Option::is_none")] - pub parent_id: Option, - pub spec: RunSpec, + pub parent_id: Option, + pub spec: RunSpec, #[serde(default, skip_serializing_if = "Option::is_none")] - pub web_url: Option, - pub start: Option, - pub status: RunStatus, + pub web_url: Option, + pub start: Option, + pub status: RunStatus, #[serde(default, skip_serializing_if = "Option::is_none")] - pub approval: Option, + pub approval: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub archived_at: Option>, - pub status_updated_at: DateTime, - pub last_event_at: DateTime, - pub pending_control: Option, - pub checkpoints: Vec, - pub conclusion: Option, - pub sandbox: Option, - pub pull_request: Option, - pub superseded_by: Option, + pub archived_at: Option>, + pub status_updated_at: DateTime, + pub last_event_at: DateTime, + pub pending_control: Option, + pub checkpoints: Vec, + pub conclusion: Option, + pub sandbox: Option, + pub pull_request: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub retried_from: Option, - pub pending_interviews: BTreeMap, - stages: HashMap, + pub pull_request_creation: Option, + pub superseded_by: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub retried_from: Option, + pub pending_interviews: BTreeMap, + stages: HashMap, } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -873,6 +875,7 @@ impl RunProjection { conclusion: None, sandbox: None, pull_request: None, + pull_request_creation: None, superseded_by: None, retried_from: None, pending_interviews: BTreeMap::new(), diff --git a/lib/packages/fabro-api-client/src/.openapi-generator/FILES b/lib/packages/fabro-api-client/src/.openapi-generator/FILES index f5d8fa8dae..fd5ae79ed6 100644 --- a/lib/packages/fabro-api-client/src/.openapi-generator/FILES +++ b/lib/packages/fabro-api-client/src/.openapi-generator/FILES @@ -307,6 +307,8 @@ models/provider.ts models/prune-run-entry.ts models/prune-runs-request.ts models/prune-runs-response.ts +models/pull-request-creation-status.ts +models/pull-request-creation.ts models/pull-request-details-status.ts models/pull-request-details-timestamps.ts models/pull-request-details-unavailable-reason.ts 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..36e9eb8d96 100644 --- a/lib/packages/fabro-api-client/src/api/runs-api.ts +++ b/lib/packages/fabro-api-client/src/api/runs-api.ts @@ -56,6 +56,8 @@ import type { PaginatedRunList } from '../models'; // @ts-ignore import type { PreflightResponse } from '../models'; // @ts-ignore +import type { PullRequestCreation } from '../models'; +// @ts-ignore import type { PullRequestLink } from '../models'; // @ts-ignore import type { PullRequestResponse } from '../models'; @@ -409,7 +411,7 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration) }; }, /** - * Creates a pull request for a completed run on GitHub and persists the record on the server. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -624,6 +626,46 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration) options: localVarRequestOptions, }; }, + /** + * Returns the latest explicit pull request creation requested for this run. + * @summary Get Run Pull Request Creation + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + getRunPullRequestCreation: async (id: string, options: RawAxiosRequestConfig = {}): Promise => { + // verify required parameter 'id' is not null or undefined + assertParamExists('getRunPullRequestCreation', 'id', id) + const localVarPath = `/api/v1/runs/{id}/pull_request/creation` + .replace(`{${"id"}}`, encodeURIComponent(String(id))); + // use dummy base URL string because the URL constructor only accepts absolute URLs. + const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL); + let baseOptions; + if (configuration) { + baseOptions = configuration.baseOptions; + } + + const localVarRequestOptions = { method: 'GET', ...baseOptions, ...options}; + const localVarHeaderParameter = {} as any; + const localVarQueryParameter = {} as any; + + // authentication SessionCookie required + + // authentication BearerAuth required + // http bearer authentication required + await setBearerAuthToObject(localVarHeaderParameter, configuration) + + localVarHeaderParameter['Accept'] = 'application/json'; + + setSearchParams(localVarUrlObj, localVarQueryParameter); + let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {}; + localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers}; + + return { + url: toPathString(localVarUrlObj), + options: localVarRequestOptions, + }; + }, /** * Returns checkpoint timeline entries from durable run-store checkpoints. Metadata branches are write-only archives and are not read by this endpoint. * @summary Get Run Timeline @@ -1646,14 +1688,14 @@ 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. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest * @param {*} [options] Override http request option. * @throws {RequiredError} */ - async createRunPullRequest(id: string, createRunPullRequestRequest: CreateRunPullRequestRequest, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + async createRunPullRequest(id: string, createRunPullRequestRequest: CreateRunPullRequestRequest, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { const localVarAxiosArgs = await localVarAxiosParamCreator.createRunPullRequest(id, createRunPullRequestRequest, options); const localVarOperationServerIndex = configuration?.serverIndex ?? 0; const localVarOperationServerBasePath = operationServerMap['RunsApi.createRunPullRequest']?.[localVarOperationServerIndex]?.url; @@ -1714,6 +1756,19 @@ export const RunsApiFp = function(configuration?: Configuration) { const localVarOperationServerBasePath = operationServerMap['RunsApi.getRunPullRequest']?.[localVarOperationServerIndex]?.url; return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, + /** + * Returns the latest explicit pull request creation requested for this run. + * @summary Get Run Pull Request Creation + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + async getRunPullRequestCreation(id: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + const localVarAxiosArgs = await localVarAxiosParamCreator.getRunPullRequestCreation(id, options); + const localVarOperationServerIndex = configuration?.serverIndex ?? 0; + const localVarOperationServerBasePath = operationServerMap['RunsApi.getRunPullRequestCreation']?.[localVarOperationServerIndex]?.url; + return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); + }, /** * Returns checkpoint timeline entries from durable run-store checkpoints. Metadata branches are write-only archives and are not read by this endpoint. * @summary Get Run Timeline @@ -2090,14 +2145,14 @@ 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. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest * @param {*} [options] Override http request option. * @throws {RequiredError} */ - createRunPullRequest(id: string, createRunPullRequestRequest: CreateRunPullRequestRequest, options?: RawAxiosRequestConfig): AxiosPromise { + createRunPullRequest(id: string, createRunPullRequestRequest: CreateRunPullRequestRequest, options?: RawAxiosRequestConfig): AxiosPromise { return localVarFp.createRunPullRequest(id, createRunPullRequestRequest, options).then((request) => request(axios, basePath)); }, /** @@ -2143,6 +2198,16 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath? getRunPullRequest(id: string, options?: RawAxiosRequestConfig): AxiosPromise { return localVarFp.getRunPullRequest(id, options).then((request) => request(axios, basePath)); }, + /** + * Returns the latest explicit pull request creation requested for this run. + * @summary Get Run Pull Request Creation + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + getRunPullRequestCreation(id: string, options?: RawAxiosRequestConfig): AxiosPromise { + return localVarFp.getRunPullRequestCreation(id, options).then((request) => request(axios, basePath)); + }, /** * Returns checkpoint timeline entries from durable run-store checkpoints. Metadata branches are write-only archives and are not read by this endpoint. * @summary Get Run Timeline @@ -2462,7 +2527,7 @@ export class RunsApi extends BaseAPI { } /** - * Creates a pull request for a completed run on GitHub and persists the record on the server. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -2520,6 +2585,17 @@ export class RunsApi extends BaseAPI { return RunsApiFp(this.configuration).getRunPullRequest(id, options).then((request) => request(this.axios, this.basePath)); } + /** + * Returns the latest explicit pull request creation requested for this run. + * @summary Get Run Pull Request Creation + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + public getRunPullRequestCreation(id: string, options?: RawAxiosRequestConfig) { + return RunsApiFp(this.configuration).getRunPullRequestCreation(id, options).then((request) => request(this.axios, this.basePath)); + } + /** * Returns checkpoint timeline entries from durable run-store checkpoints. Metadata branches are write-only archives and are not read by this endpoint. * @summary Get Run Timeline diff --git a/lib/packages/fabro-api-client/src/models/index.ts b/lib/packages/fabro-api-client/src/models/index.ts index d87d466cbf..3e6fc0919e 100644 --- a/lib/packages/fabro-api-client/src/models/index.ts +++ b/lib/packages/fabro-api-client/src/models/index.ts @@ -278,6 +278,8 @@ export * from './prune-run-entry'; export * from './prune-runs-request'; export * from './prune-runs-response'; export * from './pull-request'; +export * from './pull-request-creation'; +export * from './pull-request-creation-status'; export * from './pull-request-details'; export * from './pull-request-details-status'; export * from './pull-request-details-timestamps'; diff --git a/lib/packages/fabro-api-client/src/models/pull-request-creation-status.ts b/lib/packages/fabro-api-client/src/models/pull-request-creation-status.ts new file mode 100644 index 0000000000..f6e5a39416 --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/pull-request-creation-status.ts @@ -0,0 +1,27 @@ +/* tslint:disable */ +/* eslint-disable */ +/** + * Fabro Run API + * HTTP API for managing Fabro workflow run executions. + * + * The version of the OpenAPI document: 0.1.0 + * + * + * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + * https://openapi-generator.tech + * Do not edit the class manually. + */ + + + +/** + * Durable state of a pull request creation request. + */ + +export const PullRequestCreationStatus = { + PENDING: 'pending', + SUCCEEDED: 'succeeded', + FAILED: 'failed' +} as const; + +export type PullRequestCreationStatus = typeof PullRequestCreationStatus[keyof typeof PullRequestCreationStatus]; diff --git a/lib/packages/fabro-api-client/src/models/pull-request-creation.ts b/lib/packages/fabro-api-client/src/models/pull-request-creation.ts new file mode 100644 index 0000000000..fd1db4d0e1 --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/pull-request-creation.ts @@ -0,0 +1,44 @@ +/* tslint:disable */ +/* eslint-disable */ +/** + * Fabro Run API + * HTTP API for managing Fabro workflow run executions. + * + * The version of the OpenAPI document: 0.1.0 + * + * + * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + * https://openapi-generator.tech + * Do not edit the class manually. + */ + + +// May contain unused imports in some cases +// @ts-ignore +import type { PullRequestCreationStatus } from './pull-request-creation-status'; +// May contain unused imports in some cases +// @ts-ignore +import type { PullRequestLink } from './pull-request-link'; + +/** + * Durable status for the latest explicit pull request creation requested for a run. + */ +export interface PullRequestCreation { + /** + * Stable identifier for one explicit pull request creation request. + */ + 'id': string; + 'status': PullRequestCreationStatus; + /** + * Resolved model identifier used to generate the pull request content. + */ + 'model': string; + /** + * Whether creation was allowed for a run without a successful conclusion. + */ + 'force': boolean; + 'requested_at': string; + 'updated_at': string; + 'pull_request'?: PullRequestLink | null; + 'error'?: string | null; +} diff --git a/lib/packages/fabro-api-client/src/models/run-projection.ts b/lib/packages/fabro-api-client/src/models/run-projection.ts index e389baaf6b..bd527066f3 100644 --- a/lib/packages/fabro-api-client/src/models/run-projection.ts +++ b/lib/packages/fabro-api-client/src/models/run-projection.ts @@ -24,6 +24,9 @@ import type { Conclusion } from './conclusion'; import type { PendingInterviewRecord } from './pending-interview-record'; // May contain unused imports in some cases // @ts-ignore +import type { PullRequestCreation } from './pull-request-creation'; +// May contain unused imports in some cases +// @ts-ignore import type { PullRequestLink } from './pull-request-link'; // May contain unused imports in some cases // @ts-ignore @@ -74,6 +77,7 @@ export interface RunProjection { 'conclusion'?: Conclusion | null; 'sandbox'?: RunSandbox | null; 'pull_request'?: PullRequestLink | null; + 'pull_request_creation'?: PullRequestCreation | null; 'superseded_by'?: string | null; /** * Source run ID when this run was created by manual retry. From 5c6289df80d9a82dc500f5422402c9928b86b136 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 4 Aug 2026 14:51:29 -0400 Subject: [PATCH 2/2] Simplify async pull request creation Structural cleanup of the durable pull request creation feature, from a three-agent review (reuse, quality, efficiency) of the branch: - Move the supervisor out of handler/ into server/pull_request_supervisor.rs, collapse its double bookkeeping into one task-id map, and fold the five copy-pasted failure arms into attempt_pull_request_creation. - Tag pull_request.failed events with the creation id they resolve, so a publish-stage failure can never fail an unrelated explicit creation. The reducer gains PullRequestCreation::succeed/fail transition methods. - Scan pending creations through a narrow projection-cache accessor instead of materializing every run summary, raise the scan interval to 30s (notify covers the live path), and cap retries for runs whose worker cannot even record a failure. - Answer "creation already pending" POSTs before taking the per-run create lock, which a worker can hold for the whole creation. - Replace the hand-rolled per-run lock map with fabro_store::KeyedMutex. - Reuse cheap Arc'd projections (cached_run_projection) on the poll endpoint and in the worker instead of deep-cloning run summaries and diffs. - Merge ExistingPullRequest into fabro_github::CreatedPullRequest and extract one reconcile_existing_pull_request helper for both call sites. - Give the client poll loop a 15-minute deadline; document that Retry-After and the poll interval are the same constant. - Resolve a wedged pending creation (run already has a pull request) as a durable failure instead of skipping it forever. - Tests: shared wait_for_pull_request_creation helper, a pinned generation- failure assertion, and a new pipeline test proving reconciliation adopts an existing PR without an LLM call or create request. Verified: cargo build --workspace, cargo nextest run --workspace (7,767 passed), nightly clippy -D warnings, fmt --check, insta (no pending), bun typecheck in fabro-api-client. Co-Authored-By: Claude Fable 5 --- docs/internal/events.md | 6 + docs/public/api-reference/fabro-api.yaml | 5 +- .../src/commands/run/run_progress/mod.rs | 3 +- lib/apps/fabro-server/src/serve.rs | 13 +- lib/apps/fabro-server/src/server.rs | 72 +--- .../src/server/handler/pull_requests.rs | 370 +++--------------- .../src/server/pull_request_supervisor.rs | 265 +++++++++++++ lib/apps/fabro-server/src/server/tests.rs | 76 ++-- lib/components/fabro-github/src/lib.rs | 19 +- lib/components/fabro-store/src/run_state.rs | 65 +-- lib/components/fabro-store/src/slate/mod.rs | 7 + .../fabro-store/src/slate/projection_cache.rs | 21 + .../fabro-workflow/src/event/convert.rs | 5 +- .../fabro-workflow/src/event/events.rs | 5 +- .../fabro-workflow/src/pipeline/publish.rs | 6 +- .../src/pipeline/pull_request.rs | 185 +++++---- lib/foundation/fabro-client/src/client.rs | 14 + .../fabro-types/src/pull_request.rs | 16 + .../fabro-types/src/run_event/misc.rs | 6 +- .../fabro-api-client/src/api/runs-api.ts | 8 +- 20 files changed, 631 insertions(+), 536 deletions(-) create mode 100644 lib/apps/fabro-server/src/server/pull_request_supervisor.rs diff --git a/docs/internal/events.md b/docs/internal/events.md index d01c0991b6..859c1687ee 100644 --- a/docs/internal/events.md +++ b/docs/internal/events.md @@ -2199,6 +2199,7 @@ These legacy events may appear in older run logs. Current CLI backend runs do no "id": "...", "ts": "...", "run_id": "...", "event": "pull_request.failed", "properties": { + "creation_id": "01KYYK70WTZT2E551P3H5P0059", "error": "insufficient permissions" } } @@ -2206,8 +2207,13 @@ These legacy events may appear in older run logs. Current CLI backend runs do no | Property | Type | Description | |----------|------|-------------| +| `creation_id` | string (optional) | Explicit pull request creation this failure resolves. Absent for publish-stage failures. | | `error` | string | Error message | +When `creation_id` names the run's pending pull request creation, the run +projection marks that creation `failed`. A `pull_request.failed` event without +a `creation_id` (the workflow publish stage) does not change creation state. + ## Artifact events ### `artifact.captured` diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index b151695cc9..99f8deed11 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -2580,6 +2580,10 @@ paths: server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. + + If a creation is already pending for the run, the response returns + that creation unchanged; any different `model` or `force` values in + the new request are ignored. parameters: - $ref: "#/components/parameters/RunId" requestBody: @@ -2644,7 +2648,6 @@ paths: application/json: schema: $ref: "#/components/schemas/ErrorResponse" - put: operationId: linkRunPullRequest tags: [Runs] diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs index 9e7e850306..53d336d498 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs @@ -1422,7 +1422,8 @@ mod tests { draft: true, }); emit(&mut ui, Event::PullRequestFailed { - error: "auth token expired".into(), + creation_id: None, + error: "auth token expired".into(), }); insta::assert_snapshot!(rendered(&buffer), @r" diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index 5b345d9654..e22c61786b 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -997,19 +997,12 @@ where } } else { cleanup_handle.abort(); + pull_request_creation_supervisor.abort(); } - - if shutdown.is_cancelled() { - if let Err(join_err) = pull_request_creation_supervisor.await { + if let Err(join_err) = pull_request_creation_supervisor.await { + if !join_err.is_cancelled() { warn!(error = %join_err, "Pull request creation supervisor task panicked"); } - } else { - pull_request_creation_supervisor.abort(); - if let Err(join_err) = pull_request_creation_supervisor.await { - if !join_err.is_cancelled() { - warn!(error = %join_err, "Pull request creation supervisor task panicked"); - } - } } serve_result?; diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 84810a68a2..5445ad6197 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -86,7 +86,7 @@ use fabro_slack::{blocks as slack_blocks, connection as slack_connection}; use fabro_static::EnvVars; use fabro_store::{ ArtifactKey, ArtifactStore, CachedRunProjection, Database, EventEnvelope, EventPayload, - NodeArtifact, PendingInterviewRecord, RunSummaryStore, StageArtifactEntry, StageId, + KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore, StageArtifactEntry, StageId, }; #[cfg(test)] use fabro_types::BlockedReason; @@ -128,8 +128,7 @@ use tokio::process::Command; use tokio::runtime::Builder as TokioRuntimeBuilder; use tokio::sync::broadcast::error::RecvError; use tokio::sync::{ - Mutex as AsyncMutex, Notify, OwnedMutexGuard, RwLock as AsyncRwLock, Semaphore, broadcast, - mpsc, oneshot, + Mutex as AsyncMutex, Notify, RwLock as AsyncRwLock, Semaphore, broadcast, mpsc, oneshot, }; use tokio::task::spawn_blocking; use tokio::time::{sleep, timeout}; @@ -174,6 +173,7 @@ use crate::{ mod automation_scheduler; mod handler; +mod pull_request_supervisor; mod resource_sampler; mod session_runtime; @@ -186,9 +186,9 @@ pub(crate) use handler::graph::render_graph_bytes; pub(in crate::server) use handler::graph::{ RenderSubprocessError, render_dot_subprocess, render_graph_bytes_with_exe_override, }; -pub(crate) use handler::pull_requests::spawn_pull_request_creation_supervisor; #[cfg(test)] pub(in crate::server) use handler::system::validate_github_slug; +pub(crate) use pull_request_supervisor::spawn_pull_request_creation_supervisor; use session_runtime::SessionRuntimeManager; pub(crate) type EnvLookup = Arc Option + Send + Sync>; @@ -1127,7 +1127,7 @@ pub struct AppState { /// callers for the same run share one materialization; different runs /// proceed in parallel. See `crate::run_files` for semantics. pub(crate) files_in_flight: FilesInFlight, - pull_request_create_locks: PullRequestCreateLocks, + pull_request_create_locks: KeyedMutex, parent_link_lock: AsyncMutex<()>, pub(super) server_secrets: ServerSecrets, @@ -1160,8 +1160,6 @@ pub(crate) struct AppStores { pub(crate) variables: Arc, } -type PullRequestCreateLocks = Arc>>>>; - impl AppState { pub(crate) fn automation_store(&self) -> &AutomationStore { &self.stores.automations @@ -1255,50 +1253,6 @@ impl AskFabroReadiness { } } -struct PullRequestCreateGuard { - locks: PullRequestCreateLocks, - run_id: RunId, - mutex: Arc>, - guard: Option>, -} - -impl Drop for PullRequestCreateGuard { - fn drop(&mut self) { - self.guard.take(); - - let mut locks = self - .locks - .lock() - .expect("pull request create locks poisoned"); - if locks.get(&self.run_id).is_some_and(|mutex| { - Arc::ptr_eq(mutex, &self.mutex) && Arc::strong_count(&self.mutex) == 2 - }) { - locks.remove(&self.run_id); - } - } -} - -async fn lock_pull_request_create( - locks: &PullRequestCreateLocks, - run_id: &RunId, -) -> PullRequestCreateGuard { - let mutex = { - let mut locks = locks.lock().expect("pull request create locks poisoned"); - Arc::clone( - locks - .entry(*run_id) - .or_insert_with(|| Arc::new(AsyncMutex::new(()))), - ) - }; - let guard = mutex.clone().lock_owned().await; - PullRequestCreateGuard { - locks: Arc::clone(locks), - run_id: *run_id, - mutex, - guard: Some(guard), - } -} - pub(crate) struct AppStateConfig { pub(crate) resolved_settings: ResolvedAppStateSettings, pub(crate) registry_factory_override: Option>, @@ -1531,6 +1485,20 @@ impl AppState { .ok_or_else(|| ApiError::not_found("Run not found.")) } + /// Like [`Self::cached_run`], but returns only the shared projection — + /// no run summary clone or children count under the cache mutex. + pub(crate) async fn cached_run_projection( + &self, + run_id: &RunId, + ) -> Result, ApiError> { + self.stores + .runs + .get_cached_projection(run_id) + .await + .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))? + .ok_or_else(|| ApiError::not_found("Run not found.")) + } + pub(crate) fn session_runtimes(&self) -> &SessionRuntimeManager { &self.session_runtimes } @@ -2575,7 +2543,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result Router> { @@ -38,9 +33,10 @@ pub(super) fn routes() -> Router> { ) } -const PULL_REQUEST_CREATION_TIMEOUT: Duration = Duration::from_mins(10); -const PULL_REQUEST_CREATION_SCAN_INTERVAL: Duration = Duration::from_secs(5); -const MAX_CONCURRENT_PULL_REQUEST_CREATIONS: usize = 4; +/// Advertised via the 202 `Retry-After` header; the Rust client's poll +/// interval (`PULL_REQUEST_CREATION_POLL_INTERVAL` in `fabro-client`) matches +/// this value. +const PULL_REQUEST_CREATION_RETRY_AFTER: Duration = Duration::from_secs(1); #[expect( clippy::disallowed_types, @@ -81,7 +77,7 @@ fn pull_request_record_from_link_request( }) } -async fn load_server_github_credentials( +pub(in crate::server) async fn load_server_github_credentials( state: &AppState, ) -> Result { let settings = state.server_settings(); @@ -109,7 +105,7 @@ async fn load_server_github_credentials( } } -fn server_github_context<'a>( +pub(in crate::server) fn server_github_context<'a>( state: &'a AppState, creds: &'a fabro_github::GitHubCredentials, ) -> Result, ApiError> { @@ -135,6 +131,14 @@ fn github_pull_request_not_found_error(number: u64) -> ApiError { ) } +fn pull_request_exists_error(record: &PullRequestLink) -> ApiError { + ApiError::with_code( + StatusCode::CONFLICT, + format!("Pull request already exists at {}", record.html_url()), + "pull_request_exists", + ) +} + struct PullRequestGithubContext { record: PullRequestLink, owner: String, @@ -177,24 +181,23 @@ async fn load_pull_request_github_context( }) } -struct RunPrInputs<'a> { - goal: &'a str, - base_branch: &'a str, - run_branch: &'a str, - final_git_sha: &'a str, - diff: &'a str, - conclusion: &'a fabro_types::Conclusion, - normalized_origin: String, +pub(in crate::server) struct RunPrInputs<'a> { + pub(in crate::server) goal: &'a str, + pub(in crate::server) base_branch: &'a str, + pub(in crate::server) run_branch: &'a str, + pub(in crate::server) final_git_sha: &'a str, + pub(in crate::server) diff: &'a str, + pub(in crate::server) conclusion: &'a fabro_types::Conclusion, + pub(in crate::server) normalized_origin: String, } impl<'a> RunPrInputs<'a> { - fn extract(run_state: &'a fabro_store::RunProjection, force: bool) -> Result { + pub(in crate::server) fn extract( + run_state: &'a fabro_store::RunProjection, + force: bool, + ) -> Result { if let Some(record) = run_state.pull_request.as_ref() { - return Err(ApiError::with_code( - StatusCode::CONFLICT, - format!("Pull request already exists at {}", record.html_url()), - "pull_request_exists", - )); + return Err(pull_request_exists_error(record)); } let run_spec = &run_state.spec; let origin_url = run_spec.repo_origin_url().ok_or_else(|| { @@ -313,15 +316,16 @@ async fn create_run_pull_request( State(state): State>, Json(body): Json, ) -> Response { - let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &id).await; let Ok(run_store) = state.stores.runs.open_run(&id).await else { return ApiError::not_found("Run not found.").into_response(); }; - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let run_state = match state.cached_run_projection(&id).await { + Ok(run_state) => run_state, Err(err) => return err.into_response(), }; - let run_state = cached.projection.as_ref(); + // Answer before taking the per-run create lock: a running worker holds + // that lock for the whole creation, and an already-pending request only + // needs its current status. if let Some(creation) = run_state .pull_request_creation .as_ref() @@ -329,14 +333,10 @@ async fn create_run_pull_request( { return accepted_pull_request_creation_response(&id, creation.clone()); } - if let Err(err) = RunPrInputs::extract(run_state, body.force) { + if let Err(err) = RunPrInputs::extract(&run_state, body.force) { return err.into_response(); } - let creds = match load_server_github_credentials(state.as_ref()).await { - Ok(creds) => creds, - Err(err) => return err.into_response(), - }; - if let Err(err) = server_github_context(state.as_ref(), &creds) { + if let Err(err) = load_server_github_credentials(state.as_ref()).await { return err.into_response(); } let model = if let Some(model) = body.model { @@ -349,6 +349,7 @@ async fn create_run_pull_request( .id .to_string() }; + let _create_guard = state.pull_request_create_locks.lock(id).await; let creation_id = fabro_types::PullRequestCreationId::new(); let event = workflow_event::Event::PullRequestCreationRequested { creation_id, @@ -371,26 +372,20 @@ async fn create_run_pull_request( } }; - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let run_state = match state.cached_run_projection(&id).await { + Ok(run_state) => run_state, Err(err) => return err.into_response(), }; if !appended { - if let Some(creation) = cached - .projection + if let Some(creation) = run_state .pull_request_creation .as_ref() .filter(|creation| creation.is_pending()) { return accepted_pull_request_creation_response(&id, creation.clone()); } - if let Some(record) = cached.projection.pull_request.as_ref() { - return ApiError::with_code( - StatusCode::CONFLICT, - format!("Pull request already exists at {}", record.html_url()), - "pull_request_exists", - ) - .into_response(); + if let Some(record) = run_state.pull_request.as_ref() { + return pull_request_exists_error(record).into_response(); } return ApiError::new( StatusCode::CONFLICT, @@ -399,7 +394,7 @@ async fn create_run_pull_request( .into_response(); } - let Some(creation) = cached.projection.pull_request_creation.clone() else { + let Some(creation) = run_state.pull_request_creation.clone() else { return ApiError::new( StatusCode::INTERNAL_SERVER_ERROR, "Pull request creation was accepted but its status is unavailable.", @@ -416,12 +411,14 @@ fn accepted_pull_request_creation_response( ) -> Response { let mut response = (StatusCode::ACCEPTED, Json(creation)).into_response(); let location = format!("/api/v1/runs/{run_id}/pull_request/creation"); - if let Ok(location) = HeaderValue::from_str(&location) { - response.headers_mut().insert(header::LOCATION, location); - } - response - .headers_mut() - .insert(header::RETRY_AFTER, HeaderValue::from_static("1")); + response.headers_mut().insert( + header::LOCATION, + HeaderValue::try_from(location).expect("run ids are header-safe ASCII"), + ); + response.headers_mut().insert( + header::RETRY_AFTER, + HeaderValue::from(PULL_REQUEST_CREATION_RETRY_AFTER.as_secs()), + ); response } @@ -429,11 +426,11 @@ async fn get_run_pull_request_creation( RequireRunScoped(id): RequireRunScoped, State(state): State>, ) -> Response { - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let run_state = match state.cached_run_projection(&id).await { + Ok(run_state) => run_state, Err(err) => return err.into_response(), }; - match cached.projection.pull_request_creation.clone() { + match run_state.pull_request_creation.clone() { Some(creation) => Json(creation).into_response(), None => ApiError::with_code( StatusCode::NOT_FOUND, @@ -444,261 +441,12 @@ async fn get_run_pull_request_creation( } } -async fn append_pull_request_creation_failure( - run_store: &fabro_store::RunDatabase, - run_id: &RunId, - creation_id: fabro_types::PullRequestCreationId, - error: String, -) -> anyhow::Result<()> { - let event = workflow_event::Event::PullRequestFailed { error }; - workflow_event::append_event_if(run_store, run_id, &event, |projection| { - projection - .pull_request_creation - .as_ref() - .is_some_and(|creation| creation.id == creation_id && creation.is_pending()) - }) - .await?; - Ok(()) -} - -pub(crate) async fn process_pull_request_creation( - state: Arc, - run_id: RunId, -) -> anyhow::Result<()> { - let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &run_id).await; - let run_store = state.stores.runs.open_run(&run_id).await?; - let run_state = run_store.state().await?; - let Some(creation) = run_state - .pull_request_creation - .as_ref() - .filter(|creation| creation.is_pending()) - .cloned() - else { - return Ok(()); - }; - if run_state.pull_request.is_some() { - return Ok(()); - } - - let inputs = match RunPrInputs::extract(&run_state, creation.force) { - Ok(inputs) => inputs, - Err(err) => { - return append_pull_request_creation_failure( - &run_store, - &run_id, - creation.id, - err.detail().to_string(), - ) - .await; - } - }; - let creds = match load_server_github_credentials(state.as_ref()).await { - Ok(creds) => creds, - Err(err) => { - return append_pull_request_creation_failure( - &run_store, - &run_id, - creation.id, - err.detail().to_string(), - ) - .await; - } - }; - let github = match server_github_context(state.as_ref(), &creds) { - Ok(github) => github, - Err(err) => { - return append_pull_request_creation_failure( - &run_store, - &run_id, - creation.id, - err.detail().to_string(), - ) - .await; - } - }; - let catalog = state.catalog(); - let run_store_handle = run_store.clone().into(); - let request = pull_request::OpenPullRequestRequest { - github, - origin_url: &inputs.normalized_origin, - base_branch: inputs.base_branch, - head_branch: inputs.run_branch, - expected_head_sha: inputs.final_git_sha, - goal: inputs.goal, - diff: inputs.diff, - model: &creation.model, - draft: true, - auto_merge: None, - run_store: &run_store_handle, - llm_source: state.llm_source.as_ref(), - catalog, - conclusion: Some(inputs.conclusion), - run_state: Some(&run_state), - }; - let shutdown = state.shutdown_token(); - let result = tokio::select! { - () = shutdown.cancelled() => return Ok(()), - result = time::timeout(PULL_REQUEST_CREATION_TIMEOUT, pull_request::open_pull_request(request)) => result, - }; - let created_pull_request = match result { - Ok(Ok(created)) => created, - Ok(Err(err)) => { - return append_pull_request_creation_failure(&run_store, &run_id, creation.id, err) - .await; - } - Err(_) => { - return append_pull_request_creation_failure( - &run_store, - &run_id, - creation.id, - "Pull request creation timed out after 10 minutes.".to_string(), - ) - .await; - } - }; - - let event = workflow_event::Event::pull_request_created( - &created_pull_request.link, - &created_pull_request.base_branch, - &created_pull_request.head_branch, - inputs.final_git_sha, - &created_pull_request.title, - true, - ); - workflow_event::append_event_if(&run_store, &run_id, &event, |projection| { - projection.pull_request.is_none() - && projection - .pull_request_creation - .as_ref() - .is_some_and(|current| current.id == creation.id && current.is_pending()) - }) - .await?; - Ok(()) -} - -async fn pending_pull_request_creation_run_ids(state: &AppState) -> anyhow::Result> { - let mut pending = state - .stores - .runs - .list_cached_runs(&ListRunsQuery::default(), chrono::Utc::now()) - .await? - .into_iter() - .filter_map(|cached| { - let creation = cached.projection.pull_request_creation.as_ref()?; - (cached.projection.pull_request.is_none() && creation.is_pending()) - .then_some((cached.run_id, creation.requested_at)) - }) - .collect::>(); - pending.sort_by_key(|(run_id, requested_at)| (*requested_at, *run_id)); - Ok(pending.into_iter().map(|(run_id, _)| run_id).collect()) -} - -pub(crate) fn spawn_pull_request_creation_supervisor(state: Arc) -> JoinHandle<()> { - tokio::spawn( - run_pull_request_creation_supervisor(state) - .instrument(tracing::info_span!("pull_request_creation_supervisor")), - ) -} - -async fn run_pull_request_creation_supervisor(state: Arc) { - let shutdown = state.shutdown_token(); - let mut workers = JoinSet::new(); - let mut active = HashSet::new(); - let mut task_run_ids = std::collections::HashMap::new(); - let mut scan_requested = true; - let mut scan_interval = time::interval(PULL_REQUEST_CREATION_SCAN_INTERVAL); - scan_interval.set_missed_tick_behavior(time::MissedTickBehavior::Delay); - - loop { - if scan_requested { - match pending_pull_request_creation_run_ids(state.as_ref()).await { - Ok(pending) => { - let available = - MAX_CONCURRENT_PULL_REQUEST_CREATIONS.saturating_sub(active.len()); - let ready = pending - .into_iter() - .filter(|run_id| !active.contains(run_id)) - .take(available) - .collect::>(); - for run_id in ready { - active.insert(run_id); - let task_state = Arc::clone(&state); - let handle = workers.spawn( - async move { - let result = - process_pull_request_creation(task_state, run_id).await; - (run_id, result) - } - .instrument( - tracing::info_span!("pull_request_creation", run_id = %run_id), - ), - ); - task_run_ids.insert(handle.id(), run_id); - } - } - Err(err) => { - tracing::warn!(error = %err, "Failed to scan queued pull request creations"); - } - } - scan_requested = false; - } - - if shutdown.is_cancelled() { - break; - } - - if workers.is_empty() { - tokio::select! { - () = shutdown.cancelled() => break, - () = state.pull_request_scheduler_notified() => scan_requested = true, - _ = scan_interval.tick() => scan_requested = true, - } - continue; - } - - tokio::select! { - () = shutdown.cancelled() => break, - () = state.pull_request_scheduler_notified() => scan_requested = true, - _ = scan_interval.tick() => scan_requested = true, - joined = workers.join_next_with_id() => { - match joined { - Some(Ok((task_id, (run_id, Ok(()))))) => { - task_run_ids.remove(&task_id); - active.remove(&run_id); - scan_requested = true; - } - Some(Ok((task_id, (run_id, Err(err))))) => { - task_run_ids.remove(&task_id); - active.remove(&run_id); - tracing::warn!(run_id = %run_id, error = %err, "Pull request creation worker failed"); - } - Some(Err(err)) => { - if let Some(run_id) = task_run_ids.remove(&err.id()) { - active.remove(&run_id); - tracing::warn!(run_id = %run_id, error = %err, "Pull request creation worker stopped unexpectedly"); - } else { - tracing::warn!(error = %err, "Pull request creation worker stopped unexpectedly"); - } - } - None => {} - } - } - } - } - - while let Some(joined) = workers.join_next().await { - if let Err(err) = joined { - tracing::warn!(error = %err, "Pull request creation worker stopped during shutdown"); - } - } -} - async fn link_run_pull_request( RequireRunScoped(id): RequireRunScoped, State(state): State>, Json(body): Json, ) -> Response { - let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &id).await; + let _create_guard = state.pull_request_create_locks.lock(id).await; let pull_request = match pull_request_record_from_link_request(&body) { Ok(record) => record, Err(err) => return err.into_response(), @@ -720,15 +468,15 @@ async fn unlink_run_pull_request( RequireRunScoped(id): RequireRunScoped, State(state): State>, ) -> Response { - let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &id).await; + let _create_guard = state.pull_request_create_locks.lock(id).await; let Ok(run_store) = state.stores.runs.open_run(&id).await else { return ApiError::not_found("Run not found.").into_response(); }; - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let run_state = match state.cached_run_projection(&id).await { + Ok(run_state) => run_state, Err(err) => return err.into_response(), }; - let Some(pull_request) = cached.projection.pull_request.clone() else { + let Some(pull_request) = run_state.pull_request.clone() else { return ApiError::with_code( StatusCode::NOT_FOUND, format!("No pull request found in store. Create one first with: fabro pr create {id}"), diff --git a/lib/apps/fabro-server/src/server/pull_request_supervisor.rs b/lib/apps/fabro-server/src/server/pull_request_supervisor.rs new file mode 100644 index 0000000000..2bebd4c979 --- /dev/null +++ b/lib/apps/fabro-server/src/server/pull_request_supervisor.rs @@ -0,0 +1,265 @@ +//! Background processing for durably accepted pull request creations. +//! +//! `POST /runs/{id}/pull_request` records a `pull_request.creation_requested` +//! event and returns 202; this supervisor finds pending creations (including +//! after a server restart), runs them under a bounded worker pool, and +//! records a durable success or failure result. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use fabro_types::{PullRequestCreation, PullRequestCreationId, RunId}; +use tokio::task::{self, JoinHandle, JoinSet}; +use tokio::time; +use tracing::{Instrument as _, info_span, warn}; + +use super::handler::pull_requests::{ + RunPrInputs, load_server_github_credentials, server_github_context, +}; +use super::{AppState, pull_request, workflow_event}; + +const PULL_REQUEST_CREATION_TIMEOUT: Duration = Duration::from_mins(10); +const PULL_REQUEST_CREATION_SCAN_INTERVAL: Duration = Duration::from_secs(30); +const MAX_CONCURRENT_PULL_REQUEST_CREATIONS: usize = 4; +/// Stop retrying a run after this many worker attempts that could not even +/// record a durable failure (store errors). Without a cap, such a run would +/// re-run the whole attempt — including the LLM call — on every scan. +const MAX_WORKER_FAILURES_PER_RUN: u32 = 3; + +async fn append_pull_request_creation_failure( + run_store: &fabro_store::RunDatabase, + run_id: &RunId, + creation_id: PullRequestCreationId, + error: String, +) -> anyhow::Result<()> { + let event = workflow_event::Event::PullRequestFailed { + creation_id: Some(creation_id), + error, + }; + workflow_event::append_event_if(run_store, run_id, &event, |projection| { + is_pending_creation(projection, creation_id) + }) + .await?; + Ok(()) +} + +fn is_pending_creation( + projection: &fabro_store::RunProjection, + creation_id: PullRequestCreationId, +) -> bool { + projection + .pull_request_creation + .as_ref() + .is_some_and(|creation| creation.id == creation_id && creation.is_pending()) +} + +pub(in crate::server) async fn process_pull_request_creation( + state: Arc, + run_id: RunId, +) -> anyhow::Result<()> { + let _create_guard = state.pull_request_create_locks.lock(run_id).await; + let run_store = state.stores.runs.open_run(&run_id).await?; + let Some(run_state) = state.stores.runs.get_cached_projection(&run_id).await? else { + return Ok(()); + }; + let Some(creation) = run_state + .pull_request_creation + .as_ref() + .filter(|creation| creation.is_pending()) + .cloned() + else { + return Ok(()); + }; + + match attempt_pull_request_creation(&state, &run_store, &run_id, &run_state, &creation).await? { + Ok(()) => Ok(()), + Err(error) => { + append_pull_request_creation_failure(&run_store, &run_id, creation.id, error).await + } + } +} + +/// One end-to-end creation attempt. The inner `Err` is a durable creation +/// failure for the caller to record; the inner `Ok` covers success and +/// shutdown-interrupted attempts (which stay pending). The outer `Err` is an +/// infrastructure failure — nothing was recorded, so the supervisor may retry. +async fn attempt_pull_request_creation( + state: &AppState, + run_store: &fabro_store::RunDatabase, + run_id: &RunId, + run_state: &fabro_store::RunProjection, + creation: &PullRequestCreation, +) -> anyhow::Result> { + let inputs = match RunPrInputs::extract(run_state, creation.force) { + Ok(inputs) => inputs, + Err(err) => return Ok(Err(err.detail().to_string())), + }; + let creds = match load_server_github_credentials(state).await { + Ok(creds) => creds, + Err(err) => return Ok(Err(err.detail().to_string())), + }; + let github = match server_github_context(state, &creds) { + Ok(github) => github, + Err(err) => return Ok(Err(err.detail().to_string())), + }; + let catalog = state.catalog(); + let run_store_handle = run_store.clone().into(); + let request = pull_request::OpenPullRequestRequest { + github, + origin_url: &inputs.normalized_origin, + base_branch: inputs.base_branch, + head_branch: inputs.run_branch, + expected_head_sha: inputs.final_git_sha, + goal: inputs.goal, + diff: inputs.diff, + model: &creation.model, + draft: true, + auto_merge: None, + run_store: &run_store_handle, + llm_source: state.llm_source.as_ref(), + catalog, + conclusion: Some(inputs.conclusion), + run_state: Some(run_state), + }; + let shutdown = state.shutdown_token(); + let result = tokio::select! { + () = shutdown.cancelled() => return Ok(Ok(())), + result = time::timeout(PULL_REQUEST_CREATION_TIMEOUT, pull_request::open_pull_request(request)) => result, + }; + let created_pull_request = match result { + Ok(Ok(created)) => created, + Ok(Err(err)) => return Ok(Err(err)), + Err(_) => { + return Ok(Err(format!( + "Pull request creation timed out after {} minutes.", + PULL_REQUEST_CREATION_TIMEOUT.as_secs() / 60 + ))); + } + }; + + let event = workflow_event::Event::pull_request_created( + &created_pull_request.link, + &created_pull_request.base_branch, + &created_pull_request.head_branch, + inputs.final_git_sha, + &created_pull_request.title, + true, + ); + workflow_event::append_event_if(run_store, run_id, &event, |projection| { + projection.pull_request.is_none() && is_pending_creation(projection, creation.id) + }) + .await?; + Ok(Ok(())) +} + +pub(crate) fn spawn_pull_request_creation_supervisor(state: Arc) -> JoinHandle<()> { + tokio::spawn( + run_pull_request_creation_supervisor(state) + .instrument(info_span!("pull_request_creation_supervisor")), + ) +} + +async fn run_pull_request_creation_supervisor(state: Arc) { + let shutdown = state.shutdown_token(); + let mut workers = JoinSet::new(); + let mut active: HashMap = HashMap::new(); + let mut failures: HashMap = HashMap::new(); + let mut scan_requested = true; + let mut scan_interval = time::interval(PULL_REQUEST_CREATION_SCAN_INTERVAL); + scan_interval.set_missed_tick_behavior(time::MissedTickBehavior::Delay); + + loop { + if scan_requested { + match state + .stores + .runs + .pending_pull_request_creation_run_ids() + .await + { + Ok(pending) => { + let available = + MAX_CONCURRENT_PULL_REQUEST_CREATIONS.saturating_sub(active.len()); + let ready = pending + .into_iter() + .filter(|run_id| { + !active.values().any(|active_id| active_id == run_id) + && failures.get(run_id).copied().unwrap_or(0) + < MAX_WORKER_FAILURES_PER_RUN + }) + .take(available) + .collect::>(); + for run_id in ready { + let handle = workers.spawn( + process_pull_request_creation(Arc::clone(&state), run_id) + .instrument(info_span!("pull_request_creation", run_id = %run_id)), + ); + active.insert(handle.id(), run_id); + } + } + Err(err) => { + warn!(error = %err, "Failed to scan queued pull request creations"); + } + } + scan_requested = false; + } + + if shutdown.is_cancelled() { + break; + } + + if workers.is_empty() { + tokio::select! { + () = shutdown.cancelled() => break, + () = state.pull_request_scheduler_notified() => scan_requested = true, + _ = scan_interval.tick() => scan_requested = true, + } + continue; + } + + tokio::select! { + () = shutdown.cancelled() => break, + () = state.pull_request_scheduler_notified() => scan_requested = true, + _ = scan_interval.tick() => scan_requested = true, + joined = workers.join_next_with_id() => { + match joined { + Some(Ok((task_id, result))) => { + let run_id = active.remove(&task_id); + match (run_id, result) { + (Some(run_id), Ok(())) => { + failures.remove(&run_id); + scan_requested = true; + } + (Some(run_id), Err(err)) => { + // Deliberately no immediate rescan: the run's + // creation is still pending, and re-picking it + // now would retry the whole attempt in a tight + // loop. The next interval tick retries it. + *failures.entry(run_id).or_default() += 1; + warn!(run_id = %run_id, error = %err, "Pull request creation worker failed"); + } + (None, result) => { + warn!(?result, "Pull request creation worker finished without a tracked run id"); + } + } + } + Some(Err(err)) => { + if let Some(run_id) = active.remove(&err.id()) { + *failures.entry(run_id).or_default() += 1; + warn!(run_id = %run_id, error = %err, "Pull request creation worker stopped unexpectedly"); + } else { + warn!(error = %err, "Pull request creation worker stopped unexpectedly"); + } + } + None => {} + } + } + } + } + + while let Some(joined) = workers.join_next().await { + if let Err(err) = joined { + warn!(error = %err, "Pull request creation worker stopped during shutdown"); + } + } +} diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 54ce2fbe44..573c43e2b7 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -4098,6 +4098,30 @@ async fn wait_for_mock_hits(mock: &httpmock::Mock<'_>, expected: usize) { panic!("mock did not receive {expected} request(s)"); } +/// Poll `GET /runs/{id}/pull_request/creation` until the creation leaves +/// `pending`, returning the terminal creation body. +async fn wait_for_pull_request_creation(app: &Router, run_id: RunId) -> serde_json::Value { + for _ in 0..150 { + let response = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/pull_request/creation"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let body = response_json!(response, StatusCode::OK).await; + if body["status"] != "pending" { + return body; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + panic!("pull request creation for run {run_id} did not finish"); +} + async fn title_update_event_count(state: &AppState, run_id: RunId) -> usize { let run_store = state.stores.runs.open_run(&run_id).await.unwrap(); run_store @@ -9754,28 +9778,7 @@ async fn create_run_pull_request_creates_and_persists_record() { // the durable pending event is enough to resume the operation. let supervisor = spawn_pull_request_creation_supervisor(Arc::clone(&state)); - let creation_body = tokio::time::timeout(std::time::Duration::from_secs(3), async { - loop { - let response = app - .clone() - .oneshot( - Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}/pull_request/creation"))) - .body(Body::empty()) - .unwrap(), - ) - .await - .unwrap(); - let body = response_json!(response, StatusCode::OK).await; - if body["status"] != "pending" { - break body; - } - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } - }) - .await - .expect("pull request creation should finish"); + let creation_body = wait_for_pull_request_creation(&app, run_id).await; assert_eq!(creation_body["status"], "succeeded"); assert_eq!(creation_body["pull_request"]["number"], 42); @@ -9916,34 +9919,17 @@ async fn create_run_pull_request_persists_generation_failure() { response_json!(response, StatusCode::ACCEPTED).await; let supervisor = spawn_pull_request_creation_supervisor(Arc::clone(&state)); - let creation = tokio::time::timeout(std::time::Duration::from_secs(3), async { - loop { - let response = app - .clone() - .oneshot( - Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}/pull_request/creation"))) - .body(Body::empty()) - .unwrap(), - ) - .await - .unwrap(); - let body = response_json!(response, StatusCode::OK).await; - if body["status"] != "pending" { - break body; - } - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } - }) - .await - .expect("pull request creation should finish"); + let creation = wait_for_pull_request_creation(&app, run_id).await; assert_eq!(creation["status"], "failed"); + // The unconfigured LLM is what fails this fixture; pin the error to the + // generation step so the test cannot pass on an earlier validation error. assert!( creation["error"] .as_str() - .is_some_and(|error| !error.is_empty()) + .is_some_and(|error| error.contains("LLM generation failed")), + "unexpected error: {:?}", + creation["error"] ); assert!(creation["pull_request"].is_null()); branch_mock.assert(); diff --git a/lib/components/fabro-github/src/lib.rs b/lib/components/fabro-github/src/lib.rs index d4d4baff8c..fd4b98936a 100644 --- a/lib/components/fabro-github/src/lib.rs +++ b/lib/components/fabro-github/src/lib.rs @@ -635,17 +635,9 @@ pub async fn create_installation_access_token_for_pr( .await } -/// Result of a successful pull request creation. -pub struct CreatedPullRequest { - pub html_url: String, - pub number: u64, - pub node_id: String, -} - -/// Existing open pull request found for an exact base, head branch, and head -/// commit. +/// Pull request created on, or reconciled from, GitHub. #[derive(Debug, Clone, PartialEq, Eq)] -pub struct ExistingPullRequest { +pub struct CreatedPullRequest { pub html_url: String, pub number: u64, pub node_id: String, @@ -663,7 +655,7 @@ pub async fn find_open_pull_request( base: &str, head: &str, expected_head_sha: &str, -) -> anyhow::Result> { +) -> anyhow::Result> { let client = ctx.http_client()?; find_open_pull_request_with_client(&client, ctx, owner, repo, base, head, expected_head_sha) .await @@ -681,7 +673,7 @@ pub async fn find_open_pull_request_with_client( base: &str, head: &str, expected_head_sha: &str, -) -> anyhow::Result> { +) -> anyhow::Result> { #[derive(Deserialize)] struct PullRequestHead { sha: String, @@ -736,7 +728,7 @@ pub async fn find_open_pull_request_with_client( Ok(pull_requests .into_iter() .find(|pull_request| pull_request.head.sha == expected_head_sha) - .map(|pull_request| ExistingPullRequest { + .map(|pull_request| CreatedPullRequest { html_url: pull_request.html_url, number: pull_request.number, node_id: pull_request.node_id, @@ -849,6 +841,7 @@ pub async fn create_pull_request_with_client( html_url: pr.html_url, number: pr.number, node_id: pr.node_id, + title: title.to_string(), }) } diff --git a/lib/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index 2e50d448c2..d685f2bea4 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -326,37 +326,39 @@ impl RunProjectionReducer for RunProjection { number: props.pr_number, }; self.pull_request = Some(pull_request.clone()); - if let Some(creation) = self.pull_request_creation.as_mut() { - if creation.is_pending() { - creation.status = PullRequestCreationStatus::Succeeded; - creation.updated_at = ts; - creation.pull_request = Some(pull_request); - creation.error = None; - } + if let Some(creation) = self + .pull_request_creation + .as_mut() + .filter(|creation| creation.is_pending()) + { + creation.succeed(pull_request, ts); } } EventBody::PullRequestLinked(props) => { self.pull_request = Some(props.pull_request.clone()); - if let Some(creation) = self.pull_request_creation.as_mut() { - if creation.is_pending() { - creation.status = PullRequestCreationStatus::Succeeded; - creation.updated_at = ts; - creation.pull_request = Some(props.pull_request.clone()); - creation.error = None; - } + if let Some(creation) = self + .pull_request_creation + .as_mut() + .filter(|creation| creation.is_pending()) + { + creation.succeed(props.pull_request.clone(), ts); } } EventBody::PullRequestUnlinked(_) => { self.pull_request = None; + // Clear the creation record too: a lingering `Succeeded` + // record would point at a pull request that is no longer + // linked, and it would block a later explicit creation. self.pull_request_creation = None; } EventBody::PullRequestFailed(props) => { - if let Some(creation) = self.pull_request_creation.as_mut() { - if creation.is_pending() { - creation.status = PullRequestCreationStatus::Failed; - creation.updated_at = ts; - creation.error = Some(props.error.clone()); - } + // Only a failure that names the pending creation resolves it; + // publish-stage failures carry no creation id and must not + // fail an unrelated explicit creation. + if let Some(creation) = self.pull_request_creation.as_mut().filter(|creation| { + Some(creation.id) == props.creation_id && creation.is_pending() + }) { + creation.fail(props.error.clone(), ts); } } EventBody::InterviewStarted(props) => { @@ -4663,7 +4665,23 @@ mod tests { .apply_event(&test_event( 2, EventBody::PullRequestFailed(PullRequestFailedProps { - error: "provider unavailable".to_string(), + creation_id: None, + error: "publish stage failure".to_string(), + }), + None, + )) + .unwrap(); + assert!( + state.pull_request_creation.as_ref().unwrap().is_pending(), + "a failure without a creation id must not resolve the creation" + ); + + state + .apply_event(&test_event( + 3, + EventBody::PullRequestFailed(PullRequestFailedProps { + creation_id: Some(first_id), + error: "provider unavailable".to_string(), }), None, )) @@ -4675,7 +4693,7 @@ mod tests { let retry_id = "01KYYK70WTZT2E551P3H5P0060".parse().unwrap(); state .apply_event(&test_event( - 3, + 4, EventBody::PullRequestCreationRequested(PullRequestCreationRequestedProps { creation_id: retry_id, model: "claude-sonnet-4-6".to_string(), @@ -4688,7 +4706,7 @@ mod tests { state .apply_event(&test_event( - 4, + 5, EventBody::PullRequestCreated(PullRequestCreatedProps { pr_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(), pr_number: 123, @@ -4706,6 +4724,7 @@ mod tests { let succeeded = state.pull_request_creation.as_ref().unwrap(); assert_eq!(succeeded.status, PullRequestCreationStatus::Succeeded); assert_eq!(succeeded.pull_request.as_ref().unwrap().number, 123); + assert_eq!(succeeded.pull_request, state.pull_request); assert!(succeeded.error.is_none()); } diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 893b0d8f07..e0eaffb96f 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -364,6 +364,13 @@ impl Database { Ok(self.projection_cache.get_summary(run_id, now).await) } + /// Run ids whose latest explicit pull request creation is still pending, + /// oldest request first. + pub async fn pending_pull_request_creation_run_ids(&self) -> Result> { + self.warm_projection_cache().await?; + Ok(self.projection_cache.pending_pull_request_creations().await) + } + pub async fn put_session_run_index( &self, session_id: &SessionId, diff --git a/lib/components/fabro-store/src/slate/projection_cache.rs b/lib/components/fabro-store/src/slate/projection_cache.rs index 79ff0c03b8..6d53d2a6de 100644 --- a/lib/components/fabro-store/src/slate/projection_cache.rs +++ b/lib/components/fabro-store/src/slate/projection_cache.rs @@ -189,6 +189,27 @@ impl RunProjectionCache { .map(|entry| (Arc::clone(&entry.projection), entry.last_seq)) } + /// Run ids whose latest explicit pull request creation is still pending, + /// oldest request first. Clones only ids and timestamps, so callers can + /// poll on an interval without materializing run summaries. + pub(crate) async fn pending_pull_request_creations(&self) -> Vec { + let mut pending = self + .state + .lock() + .await + .entries + .values() + .filter_map(|entry| { + let creation = entry.projection.pull_request_creation.as_ref()?; + creation + .is_pending() + .then_some((creation.requested_at, entry.run_id)) + }) + .collect::>(); + pending.sort_unstable(); + pending.into_iter().map(|(_, run_id)| run_id).collect() + } + pub(crate) async fn get_summary(&self, run_id: &RunId, now: DateTime) -> Option { let mut entry = { let state = self.state.lock().await; diff --git a/lib/components/fabro-workflow/src/event/convert.rs b/lib/components/fabro-workflow/src/event/convert.rs index 3ac58c2d8b..5f81e18069 100644 --- a/lib/components/fabro-workflow/src/event/convert.rs +++ b/lib/components/fabro-workflow/src/event/convert.rs @@ -1362,9 +1362,10 @@ fn event_body_from_event(event: &Event) -> EventBody { pull_request: pull_request.clone(), }) } - Event::PullRequestFailed { error } => { + Event::PullRequestFailed { creation_id, error } => { EventBody::PullRequestFailed(fabro_types::PullRequestFailedProps { - error: error.clone(), + creation_id: *creation_id, + error: error.clone(), }) } } diff --git a/lib/components/fabro-workflow/src/event/events.rs b/lib/components/fabro-workflow/src/event/events.rs index 61a1ceeaa0..f277caac3b 100644 --- a/lib/components/fabro-workflow/src/event/events.rs +++ b/lib/components/fabro-workflow/src/event/events.rs @@ -751,7 +751,10 @@ pub enum Event { pull_request: PullRequestLink, }, PullRequestFailed { - error: String, + /// Set when the failure resolves an explicitly requested creation; + /// `None` for pull request failures in the workflow publish stage. + creation_id: Option, + error: String, }, } diff --git a/lib/components/fabro-workflow/src/pipeline/publish.rs b/lib/components/fabro-workflow/src/pipeline/publish.rs index 6adce32b93..30fe5b055a 100644 --- a/lib/components/fabro-workflow/src/pipeline/publish.rs +++ b/lib/components/fabro-workflow/src/pipeline/publish.rs @@ -116,7 +116,8 @@ impl Concluded { .await .map_err(|error| { self.services.emitter.emit(&Event::PullRequestFailed { - error: error.clone(), + creation_id: None, + error: error.clone(), }); Error::publish_with_source("failed to create pull request", anyhow::anyhow!(error)) })?; @@ -185,7 +186,8 @@ impl Concluded { fn pull_request_error(&self, message: &str) -> Error { self.services.emitter.emit(&Event::PullRequestFailed { - error: message.to_string(), + creation_id: None, + error: message.to_string(), }); Error::publish(message) } diff --git a/lib/components/fabro-workflow/src/pipeline/pull_request.rs b/lib/components/fabro-workflow/src/pipeline/pull_request.rs index 21d9164a5a..71de7edb4b 100644 --- a/lib/components/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/components/fabro-workflow/src/pipeline/pull_request.rs @@ -472,23 +472,47 @@ pub struct CreatedPullRequest { pub head_branch: String, } -fn recovered_pull_request( - existing: fabro_github::ExistingPullRequest, - owner: String, - repo: String, - base_branch: &str, - head_branch: &str, -) -> CreatedPullRequest { - CreatedPullRequest { +/// Adopt an open pull request that already exists for the head branch at the +/// expected commit, e.g. when GitHub created the pull request but the caller +/// stopped before persisting the result. +async fn reconcile_existing_pull_request( + req: &OpenPullRequestRequest<'_>, + owner: &str, + repo: &str, + context: &'static str, +) -> anyhow::Result> { + let Some(existing) = github_app::find_open_pull_request( + &req.github, + owner, + repo, + req.base_branch, + req.head_branch, + req.expected_head_sha, + ) + .await? + else { + return Ok(None); + }; + info!(pr_url = %existing.html_url, pr_number = existing.number, context, "Existing pull request reconciled"); + enable_auto_merge_if_requested( + &req.github, + owner, + repo, + &existing.node_id, + existing.number, + req.auto_merge.as_ref(), + ) + .await; + Ok(Some(CreatedPullRequest { link: PullRequestLink { - owner, - repo, + owner: owner.to_string(), + repo: repo.to_string(), number: existing.number, }, title: existing.title, - base_branch: base_branch.to_string(), - head_branch: head_branch.to_string(), - } + base_branch: req.base_branch.to_string(), + head_branch: req.head_branch.to_string(), + })) } async fn enable_auto_merge_if_requested( @@ -576,34 +600,11 @@ pub async fn open_pull_request( // branch would otherwise cost a full LLM call before failing. verify_remote_head(&req, &owner, &repo).await?; - if let Some(existing) = github_app::find_open_pull_request( - &req.github, - &owner, - &repo, - req.base_branch, - req.head_branch, - req.expected_head_sha, - ) - .await - .map_err(|err| format!("failed to reconcile an existing pull request: {err:#}"))? + if let Some(existing) = reconcile_existing_pull_request(&req, &owner, &repo, "before creation") + .await + .map_err(|err| format!("failed to reconcile an existing pull request: {err:#}"))? { - info!(pr_url = %existing.html_url, pr_number = existing.number, "Existing pull request reconciled"); - enable_auto_merge_if_requested( - &req.github, - &owner, - &repo, - &existing.node_id, - existing.number, - req.auto_merge.as_ref(), - ) - .await; - return Ok(recovered_pull_request( - existing, - owner, - repo, - req.base_branch, - req.head_branch, - )); + return Ok(existing); } let content = build_pr_content( @@ -635,35 +636,10 @@ pub async fn open_pull_request( { Ok(created) => created, Err(create_err) => { - match github_app::find_open_pull_request( - &req.github, - &owner, - &repo, - req.base_branch, - req.head_branch, - req.expected_head_sha, - ) - .await + match reconcile_existing_pull_request(&req, &owner, &repo, "after a failed create") + .await { - Ok(Some(existing)) => { - info!(pr_url = %existing.html_url, pr_number = existing.number, "Pull request reconciled after create response failed"); - enable_auto_merge_if_requested( - &req.github, - &owner, - &repo, - &existing.node_id, - existing.number, - req.auto_merge.as_ref(), - ) - .await; - return Ok(recovered_pull_request( - existing, - owner, - repo, - req.base_branch, - req.head_branch, - )); - } + Ok(Some(existing)) => return Ok(existing), Ok(None) => return Err(format!("{create_err:#}")), Err(reconcile_err) => { return Err(format!( @@ -1750,6 +1726,15 @@ mod tests { async fn setup_fallback_test_harness_with_branch_sha( openai_payload_text: &str, branch_sha: &str, + ) -> FallbackHarness { + setup_fallback_test_harness_with(openai_payload_text, branch_sha, serde_json::json!([])) + .await + } + + async fn setup_fallback_test_harness_with( + openai_payload_text: &str, + branch_sha: &str, + reconcile_response: serde_json::Value, ) -> FallbackHarness { let openai_server = MockServer::start_async().await; let openai_mock = openai_server @@ -1792,7 +1777,7 @@ mod tests { }) .await; let reconcile_mock = github_server - .mock_async(|when, then| { + .mock_async(move |when, then| { when.method(GET) .path("/repos/owner/repo/pulls") .query_param("state", "open") @@ -1801,7 +1786,7 @@ mod tests { .header("authorization", "Bearer test-token"); then.status(200) .header("content-type", "application/json") - .json_body(serde_json::json!([])); + .json_body(reconcile_response); }) .await; @@ -1912,6 +1897,66 @@ mod tests { } } + /// An open pull request already exists for the head branch at the + /// expected commit — for example after a crash between GitHub creating + /// the pull request and the caller persisting it. `open_pull_request` + /// adopts it without an LLM call and without a create request. + #[tokio::test] + async fn open_pull_request_adopts_an_existing_pull_request_without_creating() { + let payload = pr_content_json("Unused", "Unused."); + let harness = setup_fallback_test_harness_with( + &payload, + "final-sha", + serde_json::json!([{ + "html_url": "https://github.com/owner/repo/pull/7", + "number": 7, + "node_id": "PR_existing", + "title": "Reconciled title", + "head": {"sha": "final-sha"} + }]), + ) + .await; + + let github_base_url = harness.github_server.url(""); + let github = github_app::GitHubContext::new(&harness.creds, &github_base_url); + + let result = open_pull_request(OpenPullRequestRequest { + github, + origin_url: "https://github.com/owner/repo.git", + base_branch: "main", + head_branch: "fabro/run/123", + expected_head_sha: "final-sha", + goal: "Fix telemetry leak", + diff: "diff --git a/src/lib.rs b/src/lib.rs\n+fn x() {}\n", + model: "gpt-5.4", + draft: false, + auto_merge: None, + run_store: &harness.run_store, + llm_source: harness.llm_source.as_ref(), + catalog: harness.catalog.clone(), + conclusion: None, + run_state: None, + }) + .await + .expect("reconciliation should adopt the existing pull request"); + + assert_eq!(result.link.number, 7); + assert_eq!(result.title, "Reconciled title"); + // Adoption must not cost an LLM call or a create request. + assert_eq!( + httpmock::Mock::new(harness.openai_mock_id, &harness.openai_server) + .calls_async() + .await, + 0 + ); + assert_eq!( + httpmock::Mock::new(harness.github_mock_id, &harness.github_server) + .calls_async() + .await, + 0 + ); + } + /// LLM returns a usable body but an empty title; the content builder /// falls back to `pr_title_from_goal` (first line, decoration stripped) /// and PR creation succeeds with that title. diff --git a/lib/foundation/fabro-client/src/client.rs b/lib/foundation/fabro-client/src/client.rs index 4b3537aa99..4204272f7a 100644 --- a/lib/foundation/fabro-client/src/client.rs +++ b/lib/foundation/fabro-client/src/client.rs @@ -38,7 +38,13 @@ use crate::{AuthEntry, OAuthEntry, StoredSubject, sse}; const DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); const DEFAULT_HEALTH_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(250); +/// Matches the `Retry-After` the server sends on the 202 +/// (`PULL_REQUEST_CREATION_RETRY_AFTER` in `fabro-server`). const PULL_REQUEST_CREATION_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); +/// Overall polling deadline. The server abandons a creation attempt after 10 +/// minutes, but a creation can also sit pending behind the server's worker +/// pool (or a dead server), so the client needs its own bound. +const PULL_REQUEST_CREATION_POLL_DEADLINE: std::time::Duration = std::time::Duration::from_mins(15); type TransportFuture = BoxFuture<'static, Result<(fabro_http::HttpClient, String)>>; @@ -1440,9 +1446,17 @@ impl Client { .request_run_pull_request_creation(run_id, force, model) .await?; let creation_id = creation.id; + let deadline = std::time::Instant::now() + PULL_REQUEST_CREATION_POLL_DEADLINE; loop { match creation.status { fabro_types::PullRequestCreationStatus::Pending => { + if std::time::Instant::now() >= deadline { + bail!( + "Pull request creation {creation_id} is still pending after {} \ + minutes. Check its status with: fabro pr create {run_id}", + PULL_REQUEST_CREATION_POLL_DEADLINE.as_secs() / 60 + ); + } time::sleep(PULL_REQUEST_CREATION_POLL_INTERVAL).await; creation = self.get_run_pull_request_creation(run_id).await?; if creation.id != creation_id { diff --git a/lib/foundation/fabro-types/src/pull_request.rs b/lib/foundation/fabro-types/src/pull_request.rs index cdd922754f..4938083b5f 100644 --- a/lib/foundation/fabro-types/src/pull_request.rs +++ b/lib/foundation/fabro-types/src/pull_request.rs @@ -28,6 +28,9 @@ pub struct PullRequestCreation { pub force: bool, pub requested_at: DateTime, pub updated_at: DateTime, + /// Copy of the run's pull request link so that polling the creation + /// resource alone is enough to learn the outcome. Always equal to the + /// run's `pull_request` when the status is `Succeeded`. #[serde(default, skip_serializing_if = "Option::is_none")] pub pull_request: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -39,6 +42,19 @@ impl PullRequestCreation { pub fn is_pending(&self) -> bool { self.status == PullRequestCreationStatus::Pending } + + pub fn succeed(&mut self, pull_request: PullRequestLink, ts: DateTime) { + self.status = PullRequestCreationStatus::Succeeded; + self.updated_at = ts; + self.pull_request = Some(pull_request); + self.error = None; + } + + pub fn fail(&mut self, error: String, ts: DateTime) { + self.status = PullRequestCreationStatus::Failed; + self.updated_at = ts; + self.error = Some(error); + } } /// Minimal GitHub pull request reference stored on a workflow run. diff --git a/lib/foundation/fabro-types/src/run_event/misc.rs b/lib/foundation/fabro-types/src/run_event/misc.rs index 844df5dce1..cde08f65f3 100644 --- a/lib/foundation/fabro-types/src/run_event/misc.rs +++ b/lib/foundation/fabro-types/src/run_event/misc.rs @@ -295,5 +295,9 @@ pub struct PullRequestUnlinkedProps { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct PullRequestFailedProps { - pub error: String, + /// Set when the failure resolves an explicitly requested creation; absent + /// for pull request failures in the workflow publish stage. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub creation_id: Option, + pub error: String, } 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 36e9eb8d96..e3d936afe5 100644 --- a/lib/packages/fabro-api-client/src/api/runs-api.ts +++ b/lib/packages/fabro-api-client/src/api/runs-api.ts @@ -411,7 +411,7 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration) }; }, /** - * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. If a creation is already pending for the run, the response returns that creation unchanged; any different `model` or `force` values in the new request are ignored. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -1688,7 +1688,7 @@ export const RunsApiFp = function(configuration?: Configuration) { return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, /** - * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. If a creation is already pending for the run, the response returns that creation unchanged; any different `model` or `force` values in the new request are ignored. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -2145,7 +2145,7 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath? return localVarFp.createRun(runManifest, options).then((request) => request(axios, basePath)); }, /** - * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. If a creation is already pending for the run, the response returns that creation unchanged; any different `model` or `force` values in the new request are ignored. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -2527,7 +2527,7 @@ export class RunsApi extends BaseAPI { } /** - * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. If a creation is already pending for the run, the response returns that creation unchanged; any different `model` or `force` values in the new request are ignored. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest