diff --git a/apps/staged/src-tauri/src/lib.rs b/apps/staged/src-tauri/src/lib.rs index 9f9305c04..ec2c14e27 100644 --- a/apps/staged/src-tauri/src/lib.rs +++ b/apps/staged/src-tauri/src/lib.rs @@ -2269,7 +2269,6 @@ pub fn run() { timeline::get_branch_timeline, timeline::refresh_branch_git_state, timeline::list_parent_branch_commits, - timeline::pull_branch_ff_only, timeline::reset_branch_to_remote, // Notes note_commands::create_note, @@ -2311,6 +2310,7 @@ pub fn run() { prs::refresh_all_pr_statuses, prs::has_unpushed_commits, prs::push_branch, + prs::pull_or_queue_branch, prs::rebase_branch, prs::squash_commits, prs::clear_branch_pr_status, diff --git a/apps/staged/src-tauri/src/prs.rs b/apps/staged/src-tauri/src/prs.rs index fd0563c3f..f747761b0 100644 --- a/apps/staged/src-tauri/src/prs.rs +++ b/apps/staged/src-tauri/src/prs.rs @@ -3,6 +3,7 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use crate::git; +use crate::session_commands::BranchSessionLaunchStatus; use crate::session_runner; use crate::store::{self, FailureStrategy, PipelineExecution, PipelineKind, PipelineStep, Store}; @@ -49,6 +50,34 @@ pub(crate) struct PrStatusEvent { // Pipeline session helper // ============================================================================= +/// Result of a start-or-queue branch pipeline command. +/// +/// Rebase and squash can be requested while the branch already has work in +/// flight, so the caller needs to know whether the returned session started +/// running or is waiting its turn on the branch queue. +#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct BranchPipelineResponse { + pub session_id: String, + pub session_status: BranchSessionLaunchStatus, +} + +impl BranchPipelineResponse { + fn running(session_id: String) -> Self { + Self { + session_id, + session_status: BranchSessionLaunchStatus::Running, + } + } + + fn queued(session_id: String) -> Self { + Self { + session_id, + session_status: BranchSessionLaunchStatus::Queued, + } + } +} + /// Resolved context for a branch, ready to start a pipeline session. struct BranchPipelineContext { branch: store::Branch, @@ -206,13 +235,73 @@ fn rebase_ref_for_target(branch: &store::Branch, target: Option<&str>) -> String } } -fn commit_pipeline_prompt(kind: &PipelineKind) -> &'static str { +const PUSH_PROMPT: &str = "Push the current branch to the remote with a normal push. If the push fails for a recoverable reason, diagnose and fix it, then retry with a normal push. Do not force push."; +const FORCE_PUSH_PROMPT: &str = "Force push the current branch to the remote"; +/// A pull never hands off to an agent (every step aborts), so this is purely the +/// session's label. +const PULL_PROMPT: &str = "Pull from origin"; + +/// Prompt for a pipeline session, which doubles as its timeline label. +/// +/// `push_force` only matters for [`PipelineKind::Push`]; the other kinds ignore +/// it. Queued and running paths share this so a session's label doesn't change +/// when it is drained. +fn pipeline_prompt(kind: &PipelineKind, push_force: bool) -> &'static str { match kind { PipelineKind::Rebase => "Rebase branch", PipelineKind::Squash => "Squash commits", + PipelineKind::Push if push_force => FORCE_PUSH_PROMPT, + PipelineKind::Push => PUSH_PROMPT, + PipelineKind::Pull => PULL_PROMPT, } } +/// The rebase target to persist on the pipeline, or `None` when the pipeline +/// targets the branch's configured base. +/// +/// Only "rebase onto origin" needs a persisted target; a base rebase re-derives +/// it from the branch on dequeue. Keeping this in one place means the queued and +/// running paths agree, which the same-kind dedupe check relies on. +fn persisted_rebase_target(target: Option<&str>, rebase_ref: &str) -> Option { + matches!(target, Some("origin")).then(|| rebase_ref.to_string()) +} + +/// Find a queued pipeline session on this branch that already performs exactly +/// the work being requested, so a second click doesn't stack a duplicate. +/// +/// `matches` has to compare every persisted field that changes what the pipeline +/// does, not just the kind: "rebase onto base" vs "rebase onto origin" and +/// "push" vs "force push" are different operations that share a [`PipelineKind`]. +fn find_queued_pipeline( + store: &Arc, + branch_id: &str, + matches: impl Fn(&PipelineExecution) -> bool, +) -> Result, String> { + let queued = store + .get_queued_sessions_for_branch(branch_id) + .map_err(|e| e.to_string())?; + + Ok(queued + .into_iter() + .find(|session| session.pipeline.as_ref().is_some_and(&matches)) + .map(|session| session.id)) +} + +/// Whether the branch already has work in flight, and a new request therefore +/// has to join the queue instead of starting now. +/// +/// Callers must hold the branch launch lock: the answer is only meaningful for +/// as long as no session can start or be enqueued underneath them. +fn branch_has_work_in_flight(store: &Arc, branch_id: &str) -> Result { + Ok(store + .has_running_session_for_branch(branch_id) + .map_err(|e| e.to_string())? + || !store + .get_queued_sessions_for_branch(branch_id) + .map_err(|e| e.to_string())? + .is_empty()) +} + const HTTPS_FALLBACK_CONFIG: &str = "url.https://github.com/.insteadOf=git@github.com:"; fn git_fetch_with_fallback(refspec: &str) -> String { @@ -237,12 +326,17 @@ fn git_push_with_fallback(args: &str) -> String { /// the branch's own name for "rebase onto origin" (used when local has /// diverged from `origin/{branch}`). Only the rebase variant consults this /// value; squash always operates against the base branch. +/// +/// Errors for [`PipelineKind::Push`] and [`PipelineKind::Pull`], which produce no +/// commit and belong to the git pipeline path — the only caller that reads the +/// kind from the database checks it first, so this guards against a misrouted +/// queued session rather than a reachable input. fn build_commit_pipeline_steps( kind: &PipelineKind, base_branch: &str, rebase_target: &str, -) -> Vec { - match kind { +) -> Result, String> { + let steps = match kind { PipelineKind::Rebase => { let target_note = if base_branch == rebase_target { String::new() @@ -329,53 +423,155 @@ Here is the context from the prior steps: .to_string(), }, ], - } + PipelineKind::Push | PipelineKind::Pull => { + return Err(format!("{kind:?} is not a commit pipeline")); + } + }; + + Ok(steps) } -#[allow(clippy::too_many_arguments)] -async fn start_running_commit_pipeline_for_branch( - ctx: BranchPipelineContext, +/// Build the single step that pushes the branch to its remote. +/// +/// Rebuilt from the persisted `push_force` flag on dequeue rather than replayed +/// from the queued pipeline, so a queued push always pushes the branch's current +/// name with the command the user asked for. +fn build_push_pipeline_steps(branch_name: &str, force: bool) -> Vec { + let push_command = if force { + git_push_with_fallback(&format!("-u origin {branch_name} --force-with-lease")) + } else { + git_push_with_fallback(&format!("-u origin {branch_name}")) + }; + + let on_failure = if force { + FailureStrategy::HandoffToAi { + prompt_template: + "The force push failed. Diagnose and fix the issue, then retry the force push.\n\n{step_outputs}" + .to_string(), + } + } else { + // For normal push, abort on non-fast-forward so the frontend can show + // the force-push dialog. The marker matches git's actual stderr output + // (e.g. "! [rejected] main -> main (non-fast-forward)"). + // + // If the push fails for a *different* reason (e.g. auth error, network + // timeout), the marker won't match and the pipeline falls through to an + // AI handoff for generic diagnosis — this is intentional. + FailureStrategy::Abort { + marker: Some("non-fast-forward".to_string()), + } + }; + + vec![PipelineStep::Command { + label: "Push to remote".to_string(), + command: push_command, + on_failure, + }] +} + +/// Build the steps for a queued fast-forward pull. +/// +/// Both steps abort rather than handing off to an agent: a failed `--ff-only` +/// merge means the branch diverged while the pull waited, and the fix is a user +/// decision (rebase onto origin, or reset to origin) rather than something an +/// agent should pick. `session_runner` turns that abort into a session error the +/// frontend toasts, since a drained pull has no one watching it. +/// +/// Rebuilt from the branch's current name on dequeue, like the push steps. +fn build_pull_pipeline_steps(branch_name: &str) -> Vec { + vec![ + PipelineStep::Command { + label: format!("Fetch origin/{branch_name}"), + command: git_fetch_with_fallback(branch_name), + on_failure: FailureStrategy::Abort { marker: None }, + }, + PipelineStep::Command { + label: format!("Fast-forward to origin/{branch_name}"), + command: format!("git merge --ff-only origin/{branch_name}"), + on_failure: FailureStrategy::Abort { marker: None }, + }, + ] +} + +/// The rows a run-now pipeline needs before the session runner can pick it up. +/// +/// Inserted under the branch launch lock and consumed by +/// [`launch_running_pipeline_session`] once the lock is released. +struct RunningPipelineSession { + session_id: String, + pipeline: PipelineExecution, + prompt: &'static str, +} + +/// Insert the session and pending-commit rows for a rebase/squash that runs now. +/// +/// Callers must hold the branch launch lock: these rows are what make the branch +/// look busy to a concurrent launch, so the busy check and this insert have to be +/// atomic. Synchronous for the same reason — the lock must not be held across an +/// await. +fn insert_running_commit_pipeline_session( + store: &Arc, + ctx: &BranchPipelineContext, kind: PipelineKind, - steps: Vec, + steps: &[PipelineStep], rebase_target: Option, - provider: Option, - store: Arc, - app_handle: &tauri::AppHandle, - registry: &Arc, -) -> Result { - let prompt = commit_pipeline_prompt(&kind); - let mut pipeline = PipelineExecution::from_steps(&steps).with_kind(kind); + provider: Option<&str>, +) -> Result { + let prompt = pipeline_prompt(&kind, false); + let mut pipeline = PipelineExecution::from_steps(steps).with_kind(kind); if let Some(target) = rebase_target { pipeline = pipeline.with_rebase_target(target); } let mut session = store::Session::new_running(prompt, &ctx.working_dir); - if let Some(ref p) = provider { + if let Some(p) = provider { session = session.with_provider(p); } session.pipeline = Some(pipeline.clone()); store.create_session(&session).map_err(|e| e.to_string())?; + let commit = store::Commit::new_pending(&ctx.branch.id).with_session(&session.id); + store.create_commit(&commit).map_err(|e| e.to_string())?; + + Ok(RunningPipelineSession { + session_id: session.id, + pipeline, + prompt, + }) +} + +/// Announce a run-now pipeline session and hand it to the session runner. +/// +/// Runs after the branch launch lock is released: the session row already exists, +/// so anything racing this already sees the branch as busy. +#[allow(clippy::too_many_arguments)] +fn launch_running_pipeline_session( + ctx: BranchPipelineContext, + running: RunningPipelineSession, + steps: Vec, + session_type: &str, + provider: Option, + store: Arc, + app_handle: &tauri::AppHandle, + registry: &Arc, +) -> Result { let branch_id = ctx.branch.id.clone(); let project_id = ctx.branch.project_id.clone(); - let commit = store::Commit::new_pending(&branch_id).with_session(&session.id); - store.create_commit(&commit).map_err(|e| e.to_string())?; - session_runner::emit_session_running( app_handle, - &session.id, + &running.session_id, &branch_id, &project_id, - "commit", + session_type, ); session_runner::start_pipeline_session( session_runner::PipelineConfig { - session_id: session.id.clone(), - prompt: prompt.to_string(), + session_id: running.session_id.clone(), + prompt: running.prompt.to_string(), steps, - pipeline, + pipeline: running.pipeline, working_dir: ctx.working_dir, pre_head_sha: None, provider, @@ -389,7 +585,75 @@ async fn start_running_commit_pipeline_for_branch( Arc::clone(registry), )?; - Ok(session.id) + Ok(running.session_id) +} + +/// Queue a rebase/squash pipeline when the branch has work in flight. +/// +/// Returns the queued session id — either a freshly created one, or an existing +/// queued pipeline that already covers this request — and `None` when the branch +/// is idle so the caller should start the pipeline immediately. +/// +/// The busy check, the dedupe scan, and the insert all run under the branch +/// launch lock, so two rapid clicks (or a click racing a session start) cannot +/// both observe an idle branch or both miss the same queued pipeline. This stays +/// synchronous on purpose: the lock must not be held across an await. +fn queue_commit_pipeline_if_branch_busy( + store: &Arc, + branch_id: &str, + kind: &PipelineKind, + provider: Option<&str>, + target: Option<&str>, +) -> Result, String> { + let launch_lock = crate::session_commands::branch_session_launch_lock_for(branch_id); + let _guard = launch_lock.lock().unwrap(); + queue_commit_pipeline_locked(store, branch_id, kind, provider, target) +} + +/// The body of [`queue_commit_pipeline_if_branch_busy`], for the run-now path, +/// which re-checks while already holding the branch launch lock. +fn queue_commit_pipeline_locked( + store: &Arc, + branch_id: &str, + kind: &PipelineKind, + provider: Option<&str>, + target: Option<&str>, +) -> Result, String> { + if !branch_has_work_in_flight(store, branch_id)? { + return Ok(None); + } + + let branch = store + .get_branch(branch_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("Branch not found: {branch_id}"))?; + let base_branch = base_branch_name(&branch); + let rebase_ref = rebase_ref_for_target(&branch, target); + let rebase_target = persisted_rebase_target(target, &rebase_ref); + + if let Some(existing) = find_queued_pipeline(store, branch_id, |pipeline| { + pipeline.kind.as_ref() == Some(kind) + && pipeline.rebase_target.as_deref() == rebase_target.as_deref() + })? { + return Ok(Some(existing)); + } + + let steps = build_commit_pipeline_steps(kind, base_branch, &rebase_ref)?; + let mut pipeline = PipelineExecution::from_steps(&steps).with_kind(kind.clone()); + if let Some(target) = rebase_target { + pipeline = pipeline.with_rebase_target(target); + } + let mut session = store::Session::new_queued(pipeline_prompt(kind, false)); + if let Some(p) = provider { + session = session.with_provider(p); + } + session.pipeline = Some(pipeline); + store.create_session(&session).map_err(|e| e.to_string())?; + + let commit = store::Commit::new_pending(branch_id).with_session(&session.id); + store.create_commit(&commit).map_err(|e| e.to_string())?; + + Ok(Some(session.id)) } #[allow(clippy::too_many_arguments)] @@ -401,60 +665,67 @@ pub(crate) async fn start_or_queue_commit_pipeline_for_branch( kind: PipelineKind, provider: Option, target: Option, -) -> Result { - let prompt = commit_pipeline_prompt(&kind); - - let branch_has_running_session = store - .has_running_session_for_branch(&branch_id) - .map_err(|e| e.to_string())?; - let branch_has_queued_session = !store - .get_queued_sessions_for_branch(&branch_id) - .map_err(|e| e.to_string())? - .is_empty(); - - if branch_has_running_session || branch_has_queued_session { - let branch = store - .get_branch(&branch_id) - .map_err(|e| e.to_string())? - .ok_or_else(|| format!("Branch not found: {branch_id}"))?; - let base_branch = base_branch_name(&branch); - let rebase_ref = rebase_ref_for_target(&branch, target.as_deref()); - let steps = build_commit_pipeline_steps(&kind, base_branch, &rebase_ref); - let mut pipeline = PipelineExecution::from_steps(&steps).with_kind(kind); - if matches!(target.as_deref(), Some("origin")) { - pipeline = pipeline.with_rebase_target(rebase_ref.clone()); - } - let mut session = store::Session::new_queued(prompt); - if let Some(ref p) = provider { - session = session.with_provider(p); - } - session.pipeline = Some(pipeline); - store.create_session(&session).map_err(|e| e.to_string())?; - - let commit = store::Commit::new_pending(&branch_id).with_session(&session.id); - store.create_commit(&commit).map_err(|e| e.to_string())?; - - return Ok(session.id); +) -> Result { + // Pre-flight check: a branch that is already busy queues without resolving a + // pipeline context, which for a remote branch can depend on a running + // workspace the queued work will only need later. + if let Some(session_id) = queue_commit_pipeline_if_branch_busy( + &store, + &branch_id, + &kind, + provider.as_deref(), + target.as_deref(), + )? { + return Ok(BranchPipelineResponse::queued(session_id)); } let ctx = resolve_branch_pipeline_context(&store, &branch_id)?; let base_branch = base_branch_name(&ctx.branch).to_string(); let rebase_ref = rebase_ref_for_target(&ctx.branch, target.as_deref()); - let steps = build_commit_pipeline_steps(&kind, &base_branch, &rebase_ref); - let persisted_rebase_target = - matches!(target.as_deref(), Some("origin")).then(|| rebase_ref.clone()); + let steps = build_commit_pipeline_steps(&kind, &base_branch, &rebase_ref)?; + let rebase_target = persisted_rebase_target(target.as_deref(), &rebase_ref); + + // Re-check and insert under the lock, mirroring + // `session_commands::start_or_queue_branch_session_for_store`: the pre-flight + // check released the lock to resolve the context above, so without a second + // look two near-simultaneous actions could both have seen an idle branch and + // both start running — exactly what git-pipeline exclusivity exists to stop. + let running = { + let launch_lock = crate::session_commands::branch_session_launch_lock_for(&branch_id); + let _guard = launch_lock.lock().unwrap(); + + if let Some(session_id) = queue_commit_pipeline_locked( + &store, + &branch_id, + &kind, + provider.as_deref(), + target.as_deref(), + )? { + return Ok(BranchPipelineResponse::queued(session_id)); + } + + insert_running_commit_pipeline_session( + &store, + &ctx, + kind, + &steps, + rebase_target, + provider.as_deref(), + )? + }; - start_running_commit_pipeline_for_branch( + let session_id = launch_running_pipeline_session( ctx, - kind, + running, steps, - persisted_rebase_target, + "commit", provider, store, &app_handle, ®istry, - ) - .await + )?; + + Ok(BranchPipelineResponse::running(session_id)) } pub(crate) async fn start_queued_commit_pipeline_for_branch( @@ -480,8 +751,8 @@ pub(crate) async fn start_queued_commit_pipeline_for_branch( let rebase_ref = queued_rebase_target .clone() .unwrap_or_else(|| base_branch.clone()); - let steps = build_commit_pipeline_steps(&kind, &base_branch, &rebase_ref); - let prompt = commit_pipeline_prompt(&kind); + let steps = build_commit_pipeline_steps(&kind, &base_branch, &rebase_ref)?; + let prompt = pipeline_prompt(&kind, false); let mut pipeline = PipelineExecution::from_steps(&steps).with_kind(kind); if let Some(target) = queued_rebase_target { pipeline = pipeline.with_rebase_target(target); @@ -506,37 +777,409 @@ pub(crate) async fn start_queued_commit_pipeline_for_branch( .update_session_pipeline(&session.id, &pipeline) .map_err(|e| e.to_string())?; - let branch_id = ctx.branch.id.clone(); - let project_id = ctx.branch.project_id.clone(); + let branch_id = ctx.branch.id.clone(); + let project_id = ctx.branch.project_id.clone(); + + session_runner::emit_session_running( + &app_handle, + &session.id, + &branch_id, + &project_id, + "commit", + ); + + session_runner::start_pipeline_session( + session_runner::PipelineConfig { + session_id: session.id.clone(), + prompt: prompt.to_string(), + steps, + pipeline, + working_dir: ctx.working_dir, + pre_head_sha: None, + provider: effective_provider, + workspace_name: ctx.workspace_name, + remote_working_dir: ctx.remote_working_dir, + branch_id: Some(branch_id), + project_id: Some(project_id), + }, + store, + app_handle, + Arc::clone(®istry), + )?; + + Ok(true) +} + +/// Insert the session row for a push that runs right now. +/// +/// Unlike [`start_pipeline_for_branch`], the session records its pipeline kind +/// and `branch_id`. A push creates no artifact, so without that link the branch +/// queue could not see it and a commit session could start mid-push. +/// +/// Like [`insert_running_commit_pipeline_session`], callers must hold the branch +/// launch lock: this row is what makes the branch look busy. +fn insert_running_push_pipeline_session( + store: &Arc, + ctx: &BranchPipelineContext, + force: bool, + steps: &[PipelineStep], + provider: Option<&str>, +) -> Result { + let prompt = pipeline_prompt(&PipelineKind::Push, force); + let pipeline = PipelineExecution::from_steps(steps) + .with_kind(PipelineKind::Push) + .with_push_force(force); + + let mut session = + store::Session::new_running(prompt, &ctx.working_dir).with_branch(&ctx.branch.id); + if let Some(p) = provider { + session = session.with_provider(p); + } + session.pipeline = Some(pipeline.clone()); + store.create_session(&session).map_err(|e| e.to_string())?; + + Ok(RunningPipelineSession { + session_id: session.id, + pipeline, + prompt, + }) +} + +/// Queue a push pipeline when the branch has work in flight. +/// +/// Returns the queued session id — either a freshly created one, or an existing +/// queued push that already covers this request — and `None` when the branch is +/// idle so the caller should push immediately. +/// +/// Mirrors [`queue_commit_pipeline_if_branch_busy`]: the busy check, the dedupe +/// scan, and the insert all run under the branch launch lock, and stay +/// synchronous because the lock must not be held across an await. A push and a +/// force push dedupe separately — they are different operations, so a queued +/// normal push must not swallow a force push request. +fn queue_push_pipeline_if_branch_busy( + store: &Arc, + branch_id: &str, + provider: Option<&str>, + force: bool, +) -> Result, String> { + let launch_lock = crate::session_commands::branch_session_launch_lock_for(branch_id); + let _guard = launch_lock.lock().unwrap(); + queue_push_pipeline_locked(store, branch_id, provider, force) +} + +/// The body of [`queue_push_pipeline_if_branch_busy`], for the run-now path, +/// which re-checks while already holding the branch launch lock. +fn queue_push_pipeline_locked( + store: &Arc, + branch_id: &str, + provider: Option<&str>, + force: bool, +) -> Result, String> { + if !branch_has_work_in_flight(store, branch_id)? { + return Ok(None); + } + + if let Some(existing) = find_queued_pipeline(store, branch_id, |pipeline| { + pipeline.kind.as_ref() == Some(&PipelineKind::Push) && pipeline.push_force == force + })? { + return Ok(Some(existing)); + } + + let branch = store + .get_branch(branch_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("Branch not found: {branch_id}"))?; + + let steps = build_push_pipeline_steps(&branch.branch_name, force); + let pipeline = PipelineExecution::from_steps(&steps) + .with_kind(PipelineKind::Push) + .with_push_force(force); + // No artifact row: a push produces no commit, and a pending-commit stub + // would render as a failed commit once the push finishes without a new sha. + // `branch_id` is what keeps this session on the branch queue instead. + let mut session = store::Session::new_queued(pipeline_prompt(&PipelineKind::Push, force)) + .with_branch(branch_id); + if let Some(p) = provider { + session = session.with_provider(p); + } + session.pipeline = Some(pipeline); + store.create_session(&session).map_err(|e| e.to_string())?; + + Ok(Some(session.id)) +} + +/// Start a queued push or pull pipeline that reached the front of the branch +/// queue. +/// +/// Steps are rebuilt from the branch's current name and the persisted +/// `push_force` flag rather than replayed from the queued pipeline, matching +/// [`start_queued_commit_pipeline_for_branch`] so a branch renamed while the work +/// waited still acts on the right ref. +pub(crate) async fn start_queued_git_pipeline_for_branch( + store: Arc, + registry: Arc, + app_handle: tauri::AppHandle, + branch_id: String, + session: store::Session, + provider: Option, +) -> Result { + let queued_pipeline = session + .pipeline + .as_ref() + .ok_or_else(|| format!("Queued session {} has no pipeline", session.id))?; + let kind = queued_pipeline + .kind + .clone() + .ok_or_else(|| format!("Queued session {} has no pipeline kind", session.id))?; + let force = queued_pipeline.push_force; + + let ctx = resolve_branch_pipeline_context(&store, &branch_id)?; + let (steps, session_type) = match kind { + PipelineKind::Push => ( + build_push_pipeline_steps(&ctx.branch.branch_name, force), + "push", + ), + PipelineKind::Pull => (build_pull_pipeline_steps(&ctx.branch.branch_name), "pull"), + PipelineKind::Rebase | PipelineKind::Squash => { + return Err(format!( + "Queued git pipeline session {} has non-git kind {kind:?}", + session.id + )); + } + }; + let prompt = pipeline_prompt(&kind, force); + let pipeline = PipelineExecution::from_steps(&steps) + .with_kind(kind) + .with_push_force(force); + let effective_provider = session.provider.clone().or(provider); + + let transitioned = store + .transition_queued_to_running(&session.id) + .map_err(|e| e.to_string())?; + if !transitioned { + return Ok(false); + } + + // No `mark_session_artifact_started` call: a git pipeline has no queued + // artifact stub whose timestamp needs restamping when the work actually + // starts. + store + .prepare_queued_session(&session.id, &ctx.working_dir.to_string_lossy(), prompt) + .map_err(|e| e.to_string())?; + store + .update_session_pipeline(&session.id, &pipeline) + .map_err(|e| e.to_string())?; + + let branch_id = ctx.branch.id.clone(); + let project_id = ctx.branch.project_id.clone(); + + session_runner::emit_session_running( + &app_handle, + &session.id, + &branch_id, + &project_id, + session_type, + ); + + session_runner::start_pipeline_session( + session_runner::PipelineConfig { + session_id: session.id.clone(), + prompt: prompt.to_string(), + steps, + pipeline, + working_dir: ctx.working_dir, + pre_head_sha: None, + provider: effective_provider, + workspace_name: ctx.workspace_name, + remote_working_dir: ctx.remote_working_dir, + branch_id: Some(branch_id), + project_id: Some(project_id), + }, + store, + app_handle, + Arc::clone(®istry), + )?; + + Ok(true) +} + +/// What the branch queue decided to do with a pull request. +enum PullDisposition { + /// Waiting its turn: either a freshly queued session, or the queued pull that + /// already covers this request. + Queued(String), + /// The branch was idle, so the pull runs now. The id is the session that marks + /// the branch busy for its duration. + RunningNow(String), +} + +/// Decide between pulling now and queueing behind in-flight branch work, and +/// record that decision. +/// +/// Mirrors [`queue_push_pipeline_if_branch_busy`]: the busy check, the dedupe +/// scan, and the insert all run under the branch launch lock, and stay +/// synchronous because the lock must not be held across an await. A pull has no +/// variants, so the dedupe keys on the kind alone. +/// +/// The run-now case still gets a session row — a running `PipelineKind::Pull` +/// linked to the branch with no artifact, exactly like a running push. It is what +/// makes the fetch-and-merge visible to `has_running_session_for_branch` and the +/// drain scan; without it the one mutating git operation that skips the pipeline +/// runner would look idle for its whole duration, and a commit session or another +/// git action could start against the same worktree mid-merge. No +/// `session-status-changed` event is emitted for it: the caller awaits the pull and +/// reports the outcome itself, so an event would double-report a failure and spin +/// the project tile for what is usually an instant operation. +/// +/// No provider is recorded either: every pull step aborts on failure, so the +/// pipeline never hands off to an agent. +fn claim_or_queue_pull_for_branch( + store: &Arc, + branch_id: &str, +) -> Result { + let launch_lock = crate::session_commands::branch_session_launch_lock_for(branch_id); + let _guard = launch_lock.lock().unwrap(); + + let busy = branch_has_work_in_flight(store, branch_id)?; + if busy { + if let Some(existing) = find_queued_pipeline(store, branch_id, |pipeline| { + pipeline.kind.as_ref() == Some(&PipelineKind::Pull) + })? { + return Ok(PullDisposition::Queued(existing)); + } + } + + let branch = store + .get_branch(branch_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("Branch not found: {branch_id}"))?; + + let steps = build_pull_pipeline_steps(&branch.branch_name); + let pipeline = PipelineExecution::from_steps(&steps).with_kind(PipelineKind::Pull); + // Like a push session, this one carries no artifact — `branch_id` is what + // keeps it on the branch queue. + let mut session = if busy { + store::Session::new_queued(PULL_PROMPT) + } else { + store::Session::new_running(PULL_PROMPT, &immediate_pull_working_dir(store, branch_id)) + } + .with_branch(branch_id); + session.pipeline = Some(pipeline); + store.create_session(&session).map_err(|e| e.to_string())?; + + Ok(if busy { + PullDisposition::Queued(session.id) + } else { + PullDisposition::RunningNow(session.id) + }) +} + +/// Best-effort working directory for the session row of an immediate pull. +/// +/// The pull resolves its own path (a remote branch fast-forwards through its +/// workspace shell), so this only decides what the session row displays — not +/// where anything runs, and not whether the pull can proceed. +fn immediate_pull_working_dir(store: &Arc, branch_id: &str) -> PathBuf { + store + .get_workdir_for_branch(branch_id) + .ok() + .flatten() + .map(|workdir| PathBuf::from(workdir.path)) + .unwrap_or_default() +} + +/// Release the session that marked the branch busy for an immediate pull. +/// +/// The marker never went through the session runner, so nothing else will end it. +/// The completion reason matches the pipeline path: the work ran to its conclusion +/// either way, only the outcome differs. +fn finish_immediate_pull_session(store: &Arc, session_id: &str, error: Option<&str>) { + let status = if error.is_some() { + store::SessionStatus::Error + } else { + store::SessionStatus::Completed + }; + + if let Err(e) = store.transition_from_running( + session_id, + status, + error, + Some(&store::CompletionReason::TurnComplete), + ) { + log::warn!("[prs] Failed to finish immediate pull session {session_id}: {e}"); + } +} + +/// Fast-forward the branch to origin now, or queue the pull behind in-flight +/// branch work. +/// +/// An idle branch pulls directly rather than through the pipeline runner, which is +/// why this returns an `Option` rather than the [`BranchPipelineResponse`] the +/// pipeline-only actions use: `None` means the pull already happened (and its +/// failure, if any, is this call's error), `Some(session_id)` means it is waiting +/// its turn on the branch queue. +pub(crate) async fn pull_or_queue_branch_for_branch( + store: Arc, + registry: Arc, + app_handle: tauri::AppHandle, + branch_id: String, +) -> Result, String> { + let session_id = match claim_or_queue_pull_for_branch(&store, &branch_id)? { + PullDisposition::Queued(session_id) => return Ok(Some(session_id)), + PullDisposition::RunningNow(session_id) => session_id, + }; + + let pull_store = Arc::clone(&store); + let pull_branch_id = branch_id.clone(); + // A join failure (the pull task panicking, or runtime shutdown) folds into the + // pull's own error path rather than returning early: the marker session below + // is what marks the branch busy, and nothing else ever finishes it. + let pulled = tauri::async_runtime::spawn_blocking(move || { + crate::timeline::pull_branch_ff_only_impl(&pull_store, &pull_branch_id) + }) + .await + .unwrap_or_else(|e| Err(format!("Pull task failed: {e}"))); - session_runner::emit_session_running( - &app_handle, - &session.id, - &branch_id, - &project_id, - "commit", + finish_immediate_pull_session( + &store, + &session_id, + pulled.as_ref().err().map(String::as_str), ); - session_runner::start_pipeline_session( - session_runner::PipelineConfig { - session_id: session.id.clone(), - prompt: prompt.to_string(), - steps, - pipeline, - working_dir: ctx.working_dir, - pre_head_sha: None, - provider: effective_provider, - workspace_name: ctx.workspace_name, - remote_working_dir: ctx.remote_working_dir, - branch_id: Some(branch_id), - project_id: Some(project_id), - }, - store, - app_handle, - Arc::clone(®istry), - )?; + // Anything the user requested while the pull held the branch is queued behind + // the marker session, and the marker bypassed the session runner that would + // normally drain it. Spawned rather than awaited so the pull's own result + // isn't held up by starting someone else's work. + tauri::async_runtime::spawn(async move { + if let Err(e) = crate::session_commands::drain_queued_sessions_for_branch( + store, + registry, + app_handle, + branch_id.clone(), + None, + ) + .await + { + log::warn!("[prs] Failed to drain queued sessions after pulling {branch_id}: {e}"); + } + }); - Ok(true) + pulled.map(|()| None) +} + +/// Pull origin's new commits into the branch. +/// +/// Returns the queued session id when the pull had to join the branch queue, and +/// `None` when it ran immediately. +#[tauri::command(rename_all = "camelCase")] +pub async fn pull_or_queue_branch( + store: tauri::State<'_, Mutex>>>, + registry: tauri::State<'_, Arc>, + app_handle: tauri::AppHandle, + branch_id: String, +) -> Result, String> { + let store = get_store(&store)?; + pull_or_queue_branch_for_branch(store, Arc::clone(®istry), app_handle, branch_id).await } fn create_pr_handoff_prompt( @@ -1189,71 +1832,60 @@ pub(crate) async fn has_unpushed_commits_impl( .map_err(|e| format!("has_unpushed_commits task failed: {e}"))? } -/// Push a branch to its remote by kicking off an agent session. -pub(crate) async fn start_push_branch_pipeline_for_branch( +/// Push a branch to its remote by kicking off an agent session, or queue the +/// push behind whatever the branch is already doing. +pub(crate) async fn start_or_queue_push_pipeline_for_branch( store: Arc, registry: Arc, app_handle: tauri::AppHandle, branch_id: String, provider: Option, force: Option, -) -> Result { - let ctx = resolve_branch_pipeline_context(&store, &branch_id)?; - +) -> Result { let force = force.unwrap_or(false); - let push_command = if force { - git_push_with_fallback(&format!( - "-u origin {} --force-with-lease", - ctx.branch.branch_name - )) - } else { - git_push_with_fallback(&format!("-u origin {}", ctx.branch.branch_name)) - }; + // Pre-flight check, then a re-check at insert time — see + // `start_or_queue_commit_pipeline_for_branch` for why both are needed. + if let Some(session_id) = + queue_push_pipeline_if_branch_busy(&store, &branch_id, provider.as_deref(), force)? + { + return Ok(BranchPipelineResponse::queued(session_id)); + } - let on_failure = if force { - FailureStrategy::HandoffToAi { - prompt_template: - "The force push failed. Diagnose and fix the issue, then retry the force push.\n\n{step_outputs}" - .to_string(), - } - } else { - // For normal push, abort on non-fast-forward so the frontend can show - // the force-push dialog. The marker matches git's actual stderr output - // (e.g. "! [rejected] main -> main (non-fast-forward)"). - // - // If the push fails for a *different* reason (e.g. auth error, network - // timeout), the marker won't match and the pipeline falls through to an - // AI handoff for generic diagnosis — this is intentional. - FailureStrategy::Abort { - marker: Some("non-fast-forward".to_string()), - } - }; + let ctx = resolve_branch_pipeline_context(&store, &branch_id)?; + let steps = build_push_pipeline_steps(&ctx.branch.branch_name, force); - let steps = vec![PipelineStep::Command { - label: "Push to remote".to_string(), - command: push_command, - on_failure, - }]; + let running = { + let launch_lock = crate::session_commands::branch_session_launch_lock_for(&branch_id); + let _guard = launch_lock.lock().unwrap(); - let prompt = if force { - "Force push the current branch to the remote".to_string() - } else { - "Push the current branch to the remote with a normal push. If the push fails for a recoverable reason, diagnose and fix it, then retry with a normal push. Do not force push.".to_string() + if let Some(session_id) = + queue_push_pipeline_locked(&store, &branch_id, provider.as_deref(), force)? + { + return Ok(BranchPipelineResponse::queued(session_id)); + } + + insert_running_push_pipeline_session(&store, &ctx, force, &steps, provider.as_deref())? }; - start_pipeline_for_branch( + let session_id = launch_running_pipeline_session( ctx, + running, steps, - &prompt, "push", provider, store, &app_handle, ®istry, - ) + )?; + + Ok(BranchPipelineResponse::running(session_id)) } +/// Push a branch to its remote. +/// +/// Queues behind in-flight branch work instead of failing, so the response +/// reports whether the push started or is waiting on the branch queue. #[tauri::command(rename_all = "camelCase")] pub async fn push_branch( store: tauri::State<'_, Mutex>>>, @@ -1262,9 +1894,9 @@ pub async fn push_branch( branch_id: String, provider: Option, force: Option, -) -> Result { +) -> Result { let store = get_store(&store)?; - start_push_branch_pipeline_for_branch( + start_or_queue_push_pipeline_for_branch( store, Arc::clone(®istry), app_handle, @@ -1281,6 +1913,9 @@ pub async fn push_branch( /// (the default behaviour used by the base-moved row and the `…` menu). /// When `target` is `"origin"`, rebases onto `origin/{branch_name}` so that /// the local branch incorporates remote-only commits (used by the diverged row). +/// +/// Queues behind in-flight branch work instead of failing, so the response +/// reports whether the rebase started or is waiting on the branch queue. #[tauri::command(rename_all = "camelCase")] pub async fn rebase_branch( store: tauri::State<'_, Mutex>>>, @@ -1289,7 +1924,7 @@ pub async fn rebase_branch( branch_id: String, provider: Option, target: Option, -) -> Result { +) -> Result { let store = get_store(&store)?; start_or_queue_commit_pipeline_for_branch( store, @@ -1310,6 +1945,9 @@ pub async fn rebase_branch( /// branch's own commits into staged changes. /// Hands off to AI to write a single conventional-commit message using the /// original commit history as context. +/// +/// Queues behind in-flight branch work instead of failing, so the response +/// reports whether the squash started or is waiting on the branch queue. #[tauri::command(rename_all = "camelCase")] pub async fn squash_commits( store: tauri::State<'_, Mutex>>>, @@ -1317,7 +1955,7 @@ pub async fn squash_commits( app_handle: tauri::AppHandle, branch_id: String, provider: Option, -) -> Result { +) -> Result { let store = get_store(&store)?; start_or_queue_commit_pipeline_for_branch( store, @@ -1356,6 +1994,571 @@ mod tests { } } + fn setup_branch_store() -> (Arc, store::Branch) { + let store = Arc::new(Store::in_memory().unwrap()); + let project = store::Project::new("test-owner/test-repo"); + store.create_project(&project).unwrap(); + let branch = store::Branch::new(&project.id, "feature", "main"); + store.create_branch(&branch).unwrap(); + (store, branch) + } + + /// Make the branch busy by linking a running note session to it, which is + /// what `has_running_session_for_branch` looks for. + fn start_running_note_session(store: &Arc, branch_id: &str) { + let session = store::Session::new_running("write a note", Path::new("/tmp/staged-test")); + store.create_session(&session).unwrap(); + let note = store::Note::new(branch_id, "note", "").with_session(&session.id); + store.create_note(¬e).unwrap(); + } + + fn queue_pipeline( + store: &Arc, + branch_id: &str, + kind: PipelineKind, + target: Option<&str>, + ) -> Option { + queue_commit_pipeline_if_branch_busy(store, branch_id, &kind, None, target).unwrap() + } + + #[test] + fn idle_branch_runs_pipeline_immediately_instead_of_queueing() { + let (store, branch) = setup_branch_store(); + + assert_eq!( + queue_pipeline(&store, &branch.id, PipelineKind::Rebase, None), + None + ); + assert!(store + .list_commits_for_branch(&branch.id) + .unwrap() + .is_empty()); + } + + #[test] + fn busy_branch_queues_pipeline_with_label_and_pending_commit() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let session_id = queue_pipeline(&store, &branch.id, PipelineKind::Rebase, None) + .expect("rebase should queue behind the running note session"); + + let session = store.get_session(&session_id).unwrap().unwrap(); + assert_eq!(session.status, store::SessionStatus::Queued); + // The timeline renders queued pipeline rows from this prompt, so it has + // to read as the git action rather than a bare "Pending commit". + assert_eq!(session.prompt, "Rebase branch"); + let pipeline = session.pipeline.unwrap(); + assert_eq!(pipeline.kind, Some(PipelineKind::Rebase)); + assert_eq!(pipeline.rebase_target, None); + + let pending: Vec<_> = store + .list_commits_for_branch(&branch.id) + .unwrap() + .into_iter() + .filter(|c| c.sha.is_none()) + .collect(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].session_id.as_deref(), Some(session_id.as_str())); + } + + #[test] + fn queued_squash_uses_squash_label() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let session_id = queue_pipeline(&store, &branch.id, PipelineKind::Squash, None).unwrap(); + + let session = store.get_session(&session_id).unwrap().unwrap(); + assert_eq!(session.prompt, "Squash commits"); + assert_eq!(session.pipeline.unwrap().kind, Some(PipelineKind::Squash)); + } + + #[test] + fn repeated_click_reuses_the_already_queued_pipeline() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let first = queue_pipeline(&store, &branch.id, PipelineKind::Rebase, None).unwrap(); + let second = queue_pipeline(&store, &branch.id, PipelineKind::Rebase, None).unwrap(); + + assert_eq!(first, second); + assert_eq!( + store + .get_queued_sessions_for_branch(&branch.id) + .unwrap() + .len(), + 1 + ); + } + + #[test] + fn different_pipeline_kinds_queue_separately() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let rebase = queue_pipeline(&store, &branch.id, PipelineKind::Rebase, None).unwrap(); + let squash = queue_pipeline(&store, &branch.id, PipelineKind::Squash, None).unwrap(); + + assert_ne!(rebase, squash); + assert_eq!( + store + .get_queued_sessions_for_branch(&branch.id) + .unwrap() + .len(), + 2 + ); + } + + #[test] + fn rebase_onto_origin_is_not_deduped_against_rebase_onto_base() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let onto_base = queue_pipeline(&store, &branch.id, PipelineKind::Rebase, None).unwrap(); + let onto_origin = + queue_pipeline(&store, &branch.id, PipelineKind::Rebase, Some("origin")).unwrap(); + + assert_ne!(onto_base, onto_origin); + let origin_session = store.get_session(&onto_origin).unwrap().unwrap(); + assert_eq!( + origin_session.pipeline.unwrap().rebase_target.as_deref(), + Some("feature") + ); + + // Clicking the diverged row's rebase again still dedupes. + assert_eq!( + queue_pipeline(&store, &branch.id, PipelineKind::Rebase, Some("origin")), + Some(onto_origin) + ); + } + + #[test] + fn queued_pipeline_alone_keeps_the_branch_busy() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + queue_pipeline(&store, &branch.id, PipelineKind::Rebase, None).unwrap(); + + // The note session finishes, but the still-queued rebase must keep a + // newly requested squash on the queue rather than running it now. + for session in store.get_running_sessions().unwrap() { + store + .update_session_status(&session.id, store::SessionStatus::Completed, None, None) + .unwrap(); + } + + assert!(queue_pipeline(&store, &branch.id, PipelineKind::Squash, None).is_some()); + } + + fn queue_push(store: &Arc, branch_id: &str, force: bool) -> Option { + queue_push_pipeline_if_branch_busy(store, branch_id, None, force).unwrap() + } + + #[test] + fn idle_branch_pushes_immediately_instead_of_queueing() { + let (store, branch) = setup_branch_store(); + + assert_eq!(queue_push(&store, &branch.id, false), None); + assert!(store + .get_queued_sessions_for_branch(&branch.id) + .unwrap() + .is_empty()); + } + + #[test] + fn busy_branch_queues_push_with_branch_link_and_no_artifact() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let session_id = queue_push(&store, &branch.id, false) + .expect("push should queue behind the running note session"); + + let session = store.get_session(&session_id).unwrap().unwrap(); + assert_eq!(session.status, store::SessionStatus::Queued); + assert_eq!(session.prompt, PUSH_PROMPT); + // The branch link is what keeps an artifact-less push on the queue. + assert_eq!(session.branch_id.as_deref(), Some(branch.id.as_str())); + let pipeline = session.pipeline.unwrap(); + assert_eq!(pipeline.kind, Some(PipelineKind::Push)); + assert!(!pipeline.push_force); + + // A pending commit would render as a failed commit once the push ends. + assert!(store + .list_commits_for_branch(&branch.id) + .unwrap() + .is_empty()); + } + + #[test] + fn repeated_push_click_reuses_the_already_queued_push() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let first = queue_push(&store, &branch.id, false).unwrap(); + let second = queue_push(&store, &branch.id, false).unwrap(); + + assert_eq!(first, second); + assert_eq!( + store + .get_queued_sessions_for_branch(&branch.id) + .unwrap() + .len(), + 1 + ); + } + + #[test] + fn force_push_is_not_deduped_against_a_queued_normal_push() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let push = queue_push(&store, &branch.id, false).unwrap(); + let force_push = queue_push(&store, &branch.id, true).unwrap(); + + assert_ne!(push, force_push); + let forced = store.get_session(&force_push).unwrap().unwrap(); + assert_eq!(forced.prompt, FORCE_PUSH_PROMPT); + assert!(forced.pipeline.unwrap().push_force); + assert_eq!( + store + .get_queued_sessions_for_branch(&branch.id) + .unwrap() + .len(), + 2 + ); + + // Clicking force push again still dedupes. + assert_eq!(queue_push(&store, &branch.id, true), Some(force_push)); + } + + #[test] + fn queued_push_alone_keeps_the_branch_busy() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + queue_push(&store, &branch.id, false).unwrap(); + + // The note session finishes, but the still-queued push must keep a newly + // requested squash on the queue rather than running it now. + for session in store.get_running_sessions().unwrap() { + store + .update_session_status(&session.id, store::SessionStatus::Completed, None, None) + .unwrap(); + } + + assert!(queue_pipeline(&store, &branch.id, PipelineKind::Squash, None).is_some()); + } + + #[test] + fn queued_rebase_keeps_a_later_push_on_the_queue() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + queue_pipeline(&store, &branch.id, PipelineKind::Rebase, None).unwrap(); + + for session in store.get_running_sessions().unwrap() { + store + .update_session_status(&session.id, store::SessionStatus::Completed, None, None) + .unwrap(); + } + + assert!(queue_push(&store, &branch.id, false).is_some()); + } + + fn pipeline_context(branch: &store::Branch) -> BranchPipelineContext { + BranchPipelineContext { + branch: branch.clone(), + working_dir: PathBuf::from("/tmp/staged-test"), + workspace_name: None, + remote_working_dir: None, + } + } + + #[test] + fn inserting_a_running_pipeline_is_what_makes_the_branch_busy() { + let (store, branch) = setup_branch_store(); + let steps = build_commit_pipeline_steps(&PipelineKind::Rebase, "main", "main").unwrap(); + + // The pre-flight check sees an idle branch and lets the rebase run now. + assert_eq!( + queue_pipeline(&store, &branch.id, PipelineKind::Rebase, None), + None + ); + insert_running_commit_pipeline_session( + &store, + &pipeline_context(&branch), + PipelineKind::Rebase, + &steps, + None, + None, + ) + .unwrap(); + + // The rows that insert wrote are what the re-check at insert time reads, so + // an action that raced it queues instead of also starting. + assert!(queue_commit_pipeline_locked( + &store, + &branch.id, + &PipelineKind::Squash, + None, + None + ) + .unwrap() + .is_some()); + assert!(queue_push(&store, &branch.id, false).is_some()); + } + + #[test] + fn inserting_a_running_push_is_what_makes_the_branch_busy() { + let (store, branch) = setup_branch_store(); + let steps = build_push_pipeline_steps(&branch.branch_name, false); + + insert_running_push_pipeline_session( + &store, + &pipeline_context(&branch), + false, + &steps, + None, + ) + .unwrap(); + + // A push has no artifact, so `branch_id` is what the re-check keys on. + assert!(queue_push_pipeline_locked(&store, &branch.id, None, true) + .unwrap() + .is_some()); + assert!(queue_commit_pipeline_locked( + &store, + &branch.id, + &PipelineKind::Rebase, + None, + None + ) + .unwrap() + .is_some()); + } + + #[test] + fn push_steps_are_rebuilt_from_the_current_branch_name_and_force_flag() { + let normal = build_push_pipeline_steps("feature", false); + let (label, command, on_failure) = command_at(&normal, 0); + assert_eq!(label, "Push to remote"); + assert!(command.contains("-u origin feature")); + assert!(!command.contains("--force-with-lease")); + assert!(matches!( + on_failure, + FailureStrategy::Abort { marker } if marker.as_deref() == Some("non-fast-forward") + )); + + // The drain path re-derives the ref, so a branch renamed while the push + // waited pushes the new name rather than the queued one. + let forced = build_push_pipeline_steps("feature-renamed", true); + let (_, forced_command, forced_on_failure) = command_at(&forced, 0); + assert!(forced_command.contains("-u origin feature-renamed --force-with-lease")); + assert!(matches!( + forced_on_failure, + FailureStrategy::HandoffToAi { .. } + )); + } + + #[test] + fn commit_pipeline_steps_reject_the_git_pipeline_kinds() { + let push_err = + build_commit_pipeline_steps(&PipelineKind::Push, "main", "main").unwrap_err(); + let pull_err = + build_commit_pipeline_steps(&PipelineKind::Pull, "main", "main").unwrap_err(); + + assert_eq!(push_err, "Push is not a commit pipeline"); + assert_eq!(pull_err, "Pull is not a commit pipeline"); + } + + /// The queued session id, or `None` when the branch was idle and the pull + /// claimed it to run now. + fn queue_pull(store: &Arc, branch_id: &str) -> Option { + match claim_or_queue_pull_for_branch(store, branch_id).unwrap() { + PullDisposition::Queued(session_id) => Some(session_id), + PullDisposition::RunningNow(_) => None, + } + } + + fn claim_pull(store: &Arc, branch_id: &str) -> String { + match claim_or_queue_pull_for_branch(store, branch_id).unwrap() { + PullDisposition::RunningNow(session_id) => session_id, + PullDisposition::Queued(session_id) => { + panic!("expected an immediate pull, got queued session {session_id}") + } + } + } + + #[test] + fn idle_branch_pulls_immediately_instead_of_queueing() { + let (store, branch) = setup_branch_store(); + + assert_eq!(queue_pull(&store, &branch.id), None); + assert!(store + .get_queued_sessions_for_branch(&branch.id) + .unwrap() + .is_empty()); + } + + #[test] + fn immediate_pull_claims_the_branch_with_a_running_marker_session() { + let (store, branch) = setup_branch_store(); + + let session_id = claim_pull(&store, &branch.id); + + let session = store.get_session(&session_id).unwrap().unwrap(); + assert_eq!(session.status, store::SessionStatus::Running); + // The branch link and pipeline kind are what make the pull visible to the + // queue; it creates no artifact, exactly like a push. + assert_eq!(session.branch_id.as_deref(), Some(branch.id.as_str())); + assert_eq!( + session.pipeline.unwrap().kind, + Some(store::PipelineKind::Pull) + ); + assert!(store + .list_commits_for_branch(&branch.id) + .unwrap() + .is_empty()); + assert!(store.has_running_session_for_branch(&branch.id).unwrap()); + } + + #[test] + fn a_pull_in_flight_keeps_other_branch_work_off_the_worktree() { + let (store, branch) = setup_branch_store(); + claim_pull(&store, &branch.id); + + // A commit pipeline or a second git action requested mid-pull has to wait + // rather than race the fast-forward in the same worktree. + assert!(queue_pipeline(&store, &branch.id, PipelineKind::Rebase, None).is_some()); + assert!(queue_push(&store, &branch.id, false).is_some()); + assert!(queue_pull(&store, &branch.id).is_some()); + } + + #[test] + fn finishing_an_immediate_pull_frees_the_branch_and_records_the_failure() { + let (store, branch) = setup_branch_store(); + + let ok = claim_pull(&store, &branch.id); + finish_immediate_pull_session(&store, &ok, None); + let completed = store.get_session(&ok).unwrap().unwrap(); + assert_eq!(completed.status, store::SessionStatus::Completed); + assert!(!store.has_running_session_for_branch(&branch.id).unwrap()); + + let failed_id = claim_pull(&store, &branch.id); + finish_immediate_pull_session( + &store, + &failed_id, + Some("Cannot pull with uncommitted changes"), + ); + let failed = store.get_session(&failed_id).unwrap().unwrap(); + assert_eq!(failed.status, store::SessionStatus::Error); + assert_eq!( + failed.error_message.as_deref(), + Some("Cannot pull with uncommitted changes") + ); + assert!(!store.has_running_session_for_branch(&branch.id).unwrap()); + } + + #[test] + fn busy_branch_queues_pull_with_branch_link_and_no_artifact() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let session_id = + queue_pull(&store, &branch.id).expect("pull should queue behind the running note"); + + let session = store.get_session(&session_id).unwrap().unwrap(); + assert_eq!(session.status, store::SessionStatus::Queued); + assert_eq!(session.prompt, PULL_PROMPT); + // Like a queued push, the branch link is what keeps an artifact-less + // pull on the queue. + assert_eq!(session.branch_id.as_deref(), Some(branch.id.as_str())); + assert_eq!( + session.pipeline.unwrap().kind, + Some(store::PipelineKind::Pull) + ); + assert!(store + .list_commits_for_branch(&branch.id) + .unwrap() + .is_empty()); + } + + #[test] + fn repeated_pull_click_reuses_the_already_queued_pull() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let first = queue_pull(&store, &branch.id).unwrap(); + let second = queue_pull(&store, &branch.id).unwrap(); + + assert_eq!(first, second); + assert_eq!( + store + .get_queued_sessions_for_branch(&branch.id) + .unwrap() + .len(), + 1 + ); + } + + #[test] + fn pull_and_push_queue_separately() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + + let pull = queue_pull(&store, &branch.id).unwrap(); + let push = queue_push(&store, &branch.id, false).unwrap(); + + assert_ne!(pull, push); + assert_eq!( + store + .get_queued_sessions_for_branch(&branch.id) + .unwrap() + .len(), + 2 + ); + } + + #[test] + fn queued_pull_alone_keeps_the_branch_busy() { + let (store, branch) = setup_branch_store(); + start_running_note_session(&store, &branch.id); + queue_pull(&store, &branch.id).unwrap(); + + // The note session finishes, but the still-queued pull must keep a newly + // requested push on the queue rather than running it now. + for session in store.get_running_sessions().unwrap() { + store + .update_session_status(&session.id, store::SessionStatus::Completed, None, None) + .unwrap(); + } + + assert!(queue_push(&store, &branch.id, false).is_some()); + } + + #[test] + fn pull_steps_fetch_then_fast_forward_and_never_hand_off() { + // The drain path re-derives the ref, so a branch renamed while the pull + // waited fast-forwards the new name rather than the queued one. + let steps = build_pull_pipeline_steps("feature-renamed"); + + let (fetch_label, fetch_command, fetch_on_failure) = command_at(&steps, 0); + assert_eq!(fetch_label, "Fetch origin/feature-renamed"); + assert!(fetch_command.contains("git fetch origin feature-renamed")); + assert!(matches!( + fetch_on_failure, + FailureStrategy::Abort { marker: None } + )); + + let (merge_label, merge_command, merge_on_failure) = command_at(&steps, 1); + assert_eq!(merge_label, "Fast-forward to origin/feature-renamed"); + assert_eq!(merge_command, "git merge --ff-only origin/feature-renamed"); + // A diverged branch is the user's call to make, not an agent's. + assert!(matches!( + merge_on_failure, + FailureStrategy::Abort { marker: None } + )); + assert_eq!(steps.len(), 2); + } + #[tokio::test] async fn collect_branch_refresh_results_tolerates_partial_task_failure() { let tasks = vec![ @@ -1447,7 +2650,7 @@ mod tests { #[test] fn rebase_pipeline_uses_signoff() { - let steps = build_commit_pipeline_steps(&PipelineKind::Rebase, "main", "main"); + let steps = build_commit_pipeline_steps(&PipelineKind::Rebase, "main", "main").unwrap(); let (label, command, _) = command_at(&steps, 0); assert_eq!(label, "Fetch latest base"); @@ -1463,7 +2666,8 @@ mod tests { #[test] fn rebase_pipeline_targets_origin_branch_when_target_differs() { - let steps = build_commit_pipeline_steps(&PipelineKind::Rebase, "main", "feature-branch"); + let steps = + build_commit_pipeline_steps(&PipelineKind::Rebase, "main", "feature-branch").unwrap(); let (label, command, _) = command_at(&steps, 0); assert_eq!(label, "Fetch origin/feature-branch"); @@ -1479,7 +2683,8 @@ mod tests { #[test] fn rebase_pipeline_prompt_mentions_base_when_target_differs() { - let steps = build_commit_pipeline_steps(&PipelineKind::Rebase, "main", "feature-branch"); + let steps = + build_commit_pipeline_steps(&PipelineKind::Rebase, "main", "feature-branch").unwrap(); let fetch_failure = match &steps[0] { PipelineStep::Command { @@ -1506,7 +2711,7 @@ mod tests { #[test] fn rebase_pipeline_prompt_omits_target_note_when_target_matches_base() { - let steps = build_commit_pipeline_steps(&PipelineKind::Rebase, "main", "main"); + let steps = build_commit_pipeline_steps(&PipelineKind::Rebase, "main", "main").unwrap(); for step in &steps { if let PipelineStep::Command { @@ -1522,7 +2727,7 @@ mod tests { #[test] fn squash_pipeline_prompt_requires_signoff() { - let steps = build_commit_pipeline_steps(&PipelineKind::Squash, "main", "main"); + let steps = build_commit_pipeline_steps(&PipelineKind::Squash, "main", "main").unwrap(); let (_, prompt) = ai_prompt_at(&steps, 3); assert!(prompt.contains("Use the user's global git identity")); diff --git a/apps/staged/src-tauri/src/session_commands.rs b/apps/staged/src-tauri/src/session_commands.rs index 7a38670f9..b108a15e6 100644 --- a/apps/staged/src-tauri/src/session_commands.rs +++ b/apps/staged/src-tauri/src/session_commands.rs @@ -1757,13 +1757,20 @@ enum BranchSessionScheduleKind { Note, Review, CommitPipeline, + /// A git command pipeline that mutates the branch without producing a + /// commit (push / force push / pull). Exclusive: it rewrites the remote from + /// the current worktree, or moves the worktree's HEAD, so nothing else may + /// touch the branch while it runs. + GitPipeline, } impl BranchSessionScheduleKind { fn is_exclusive(self) -> bool { matches!( self, - BranchSessionScheduleKind::Commit | BranchSessionScheduleKind::CommitPipeline + BranchSessionScheduleKind::Commit + | BranchSessionScheduleKind::CommitPipeline + | BranchSessionScheduleKind::GitPipeline ) } @@ -1776,7 +1783,9 @@ impl BranchSessionScheduleKind { BranchSessionScheduleKind::Commit => Some(BranchSessionType::Commit), BranchSessionScheduleKind::Note => Some(BranchSessionType::Note), BranchSessionScheduleKind::Review => Some(BranchSessionType::Review), - BranchSessionScheduleKind::CommitPipeline => None, + BranchSessionScheduleKind::CommitPipeline | BranchSessionScheduleKind::GitPipeline => { + None + } } } } @@ -1819,7 +1828,9 @@ fn review_session_schedule(review: &store::Review) -> BranchSessionSchedule { } } -fn commit_session_schedule(kind: BranchSessionScheduleKind) -> BranchSessionSchedule { +/// Schedule for the kinds that take the branch exclusively (commit sessions and +/// command pipelines): they always block the queue and carry no review. +fn exclusive_session_schedule(kind: BranchSessionScheduleKind) -> BranchSessionSchedule { BranchSessionSchedule { kind, review_id: None, @@ -1827,12 +1838,21 @@ fn commit_session_schedule(kind: BranchSessionScheduleKind) -> BranchSessionSche } } -fn is_commit_pipeline_session(session: &store::Session) -> bool { - session - .pipeline - .as_ref() - .and_then(|pipeline| pipeline.kind.as_ref()) - .is_some() +/// How a command-pipeline session is tied to the branch it belongs to. +enum PipelineBranchLink { + /// Rebase/squash: found through the pending-commit artifact they create. + Commit, + /// Push/pull: create no artifact, so they record `sessions.branch_id` instead. + Branch, +} + +fn pipeline_branch_link(session: &store::Session) -> Option { + match session.pipeline.as_ref()?.kind.as_ref()? { + store::PipelineKind::Rebase | store::PipelineKind::Squash => { + Some(PipelineBranchLink::Commit) + } + store::PipelineKind::Push | store::PipelineKind::Pull => Some(PipelineBranchLink::Branch), + } } fn resolve_branch_session_schedule( @@ -1841,21 +1861,37 @@ fn resolve_branch_session_schedule( session: &store::Session, require_artifact: bool, ) -> Result, String> { - if is_commit_pipeline_session(session) { - let commit = store - .get_commit_by_session(&session.id) - .map_err(|e| e.to_string())?; - return match commit { - Some(commit) if commit.branch_id == branch_id => Ok(Some(commit_session_schedule( - BranchSessionScheduleKind::CommitPipeline, - ))), - Some(_) => Ok(None), - None if require_artifact => Err(format!( - "Queued pipeline session {} has no linked commit", - session.id - )), - None => Ok(None), - }; + match pipeline_branch_link(session) { + Some(PipelineBranchLink::Commit) => { + let commit = store + .get_commit_by_session(&session.id) + .map_err(|e| e.to_string())?; + return match commit { + Some(commit) if commit.branch_id == branch_id => Ok(Some( + exclusive_session_schedule(BranchSessionScheduleKind::CommitPipeline), + )), + Some(_) => Ok(None), + None if require_artifact => Err(format!( + "Queued pipeline session {} has no linked commit", + session.id + )), + None => Ok(None), + }; + } + Some(PipelineBranchLink::Branch) => { + return match session.branch_id.as_deref() { + Some(linked) if linked == branch_id => Ok(Some(exclusive_session_schedule( + BranchSessionScheduleKind::GitPipeline, + ))), + Some(_) => Ok(None), + None if require_artifact => Err(format!( + "Queued git pipeline session {} has no linked branch", + session.id + )), + None => Ok(None), + }; + } + None => {} } if let Some(commit) = store @@ -1863,7 +1899,7 @@ fn resolve_branch_session_schedule( .map_err(|e| e.to_string())? { return Ok((commit.branch_id == branch_id) - .then(|| commit_session_schedule(BranchSessionScheduleKind::Commit))); + .then(|| exclusive_session_schedule(BranchSessionScheduleKind::Commit))); } if let Some(note) = store @@ -1912,7 +1948,7 @@ fn branch_session_launch_locks() -> &'static Mutex LOCKS.get_or_init(|| Mutex::new(HashMap::new())) } -fn branch_session_launch_lock_for(branch_id: &str) -> Arc> { +pub(crate) fn branch_session_launch_lock_for(branch_id: &str) -> Arc> { let mut locks = branch_session_launch_locks().lock().unwrap(); Arc::clone( locks @@ -2909,6 +2945,13 @@ async fn start_queued_session_for_branch( .await; } + if matches!(schedule.kind, BranchSessionScheduleKind::GitPipeline) { + return crate::prs::start_queued_git_pipeline_for_branch( + store, registry, app_handle, branch_id, session, provider, + ) + .await; + } + let session_type = schedule.kind.branch_session_type().ok_or_else(|| { format!( "Queued session {} cannot start as an agent session", @@ -5180,6 +5223,28 @@ mod tests { session } + /// A push or pull pipeline session, linked to the branch without any artifact + /// row — which is what makes it resolve as a `GitPipeline` schedule. + fn create_branch_git_pipeline_session( + store: &Arc, + branch_id: &str, + status: store::SessionStatus, + kind: store::PipelineKind, + ) -> store::Session { + let prompt = format!("{kind:?}").to_lowercase(); + let mut session = match status { + store::SessionStatus::Queued => store::Session::new_queued(&prompt), + store::SessionStatus::Running => { + store::Session::new_running(&prompt, Path::new("/tmp")) + } + other => panic!("unsupported scheduler test status: {}", other.as_str()), + } + .with_branch(branch_id); + session.pipeline = Some(store::PipelineExecution::from_steps(&[]).with_kind(kind)); + store.create_session(&session).unwrap(); + session + } + fn schedule(kind: BranchSessionScheduleKind) -> BranchSessionSchedule { BranchSessionSchedule { kind, @@ -5382,6 +5447,116 @@ mod tests { )); } + #[test] + fn running_git_pipeline_blocks_queued_note_review_and_commit() { + for kind in [store::PipelineKind::Push, store::PipelineKind::Pull] { + let (store, branch) = setup_branch_store(); + create_branch_git_pipeline_session( + &store, + &branch.id, + store::SessionStatus::Running, + kind, + ); + + let active = running_branch_session_kinds(&store, &branch.id).unwrap(); + + assert!(active.contains(&BranchSessionScheduleKind::GitPipeline)); + for blocked in [ + BranchSessionScheduleKind::Note, + BranchSessionScheduleKind::Review, + BranchSessionScheduleKind::Commit, + ] { + assert!(!can_start_with_active_branch_sessions(blocked, &active)); + } + } + } + + #[test] + fn running_push_pipeline_on_other_branch_does_not_block() { + let (store, branch) = setup_branch_store(); + let other = store::Branch::new(&branch.project_id, "other", "main"); + store.create_branch(&other).unwrap(); + create_branch_git_pipeline_session( + &store, + &other.id, + store::SessionStatus::Running, + store::PipelineKind::Push, + ); + + let active = running_branch_session_kinds(&store, &branch.id).unwrap(); + + assert!(active.is_empty()); + } + + #[test] + fn branch_start_decision_queues_all_user_modes_behind_queued_git_pipeline() { + for kind in [store::PipelineKind::Push, store::PipelineKind::Pull] { + let (store, branch) = setup_branch_store_with_workdir(); + create_branch_git_pipeline_session( + &store, + &branch.id, + store::SessionStatus::Queued, + kind, + ); + + for session_type in [ + BranchSessionType::Note, + BranchSessionType::Review, + BranchSessionType::Commit, + ] { + assert!( + should_queue_branch_session_start(&store, &branch.id, &session_type).unwrap() + ); + } + } + } + + #[test] + fn queued_push_acts_as_fifo_barrier() { + let mut active = HashSet::new(); + let queued = vec![ + ( + "note-1".to_string(), + schedule(BranchSessionScheduleKind::Note), + ), + ( + "push".to_string(), + schedule(BranchSessionScheduleKind::GitPipeline), + ), + ( + "note-2".to_string(), + schedule(BranchSessionScheduleKind::Note), + ), + ]; + + let drainable = drainable_session_ids_for_active_set(&queued, &mut active); + + assert_eq!(drainable, vec!["note-1".to_string()]); + } + + #[test] + fn drain_scan_starts_oldest_queued_push_alone() { + let mut active = HashSet::new(); + let queued = vec![ + ( + "push".to_string(), + schedule(BranchSessionScheduleKind::GitPipeline), + ), + ( + "note-1".to_string(), + schedule(BranchSessionScheduleKind::Note), + ), + ( + "commit".to_string(), + schedule(BranchSessionScheduleKind::Commit), + ), + ]; + + let drainable = drainable_session_ids_for_active_set(&queued, &mut active); + + assert_eq!(drainable, vec!["push".to_string()]); + } + #[test] fn queued_commit_acts_as_fifo_barrier() { let mut active = HashSet::new(); diff --git a/apps/staged/src-tauri/src/session_runner.rs b/apps/staged/src-tauri/src/session_runner.rs index 264705d57..c86c42055 100644 --- a/apps/staged/src-tauri/src/session_runner.rs +++ b/apps/staged/src-tauri/src/session_runner.rs @@ -1315,25 +1315,35 @@ pub fn start_pipeline_session( } } } - PipelineOutcome::Aborted { .. } => { + PipelineOutcome::Aborted { step_index } => { // Pipeline aborted (e.g. non-fast-forward). Mark as completed so - // the frontend can inspect the pipeline steps for the failure. + // the frontend can inspect the pipeline steps for the failure — + // except for a pull, which has no such affordance and, when it + // was drained off the branch queue, no one watching it. See + // `aborted_pipeline_error`. resolve_pipeline_artifacts_without_ai(&config, &store_for_status, false); + let error = aborted_pipeline_error(&config, &store_for_status, step_index); + // The pipeline ran to its conclusion either way; only the + // outcome differs, so the completion reason stays the same. let reason = CompletionReason::TurnComplete; + let status = if error.is_some() { + SessionStatus::Error + } else { + SessionStatus::Completed + }; registry.deregister(&session_id); let transitioned = store_for_status - .transition_from_running( - &session_id, - SessionStatus::Completed, - None, - Some(&reason), - ) + .transition_from_running(&session_id, status, error.as_deref(), Some(&reason)) .unwrap_or(false); emit_status( &app_handle, &session_id, - "completed", - None, + if error.is_some() { + "error" + } else { + "completed" + }, + error, Some(&reason), config.branch_id.clone(), config.project_id.clone(), @@ -1406,6 +1416,47 @@ fn finish_failed_pipeline_handoff_start( .unwrap_or(false) } +/// Error message for an aborted pipeline, or `None` when the abort is an expected +/// outcome the session should still complete with. +/// +/// Only a pull produces an error. A push rejected as non-fast-forward completes so +/// the frontend can read the steps and offer a force push, but a pull that can't +/// fast-forward is a dead end whose fix is a user decision (rebase onto origin, or +/// reset to origin) — and a pull drained off the branch queue runs headless, so +/// the error status is the only way the user hears about it. +/// +/// The failing step's output is read back from the persisted pipeline, which +/// `run_pipeline` writes before it returns `Aborted`. +fn aborted_pipeline_error( + config: &PipelineConfig, + store: &Store, + step_index: usize, +) -> Option { + if config.pipeline.kind.as_ref() != Some(&PipelineKind::Pull) { + return None; + } + + let step = store + .get_session(&config.session_id) + .ok() + .flatten() + .and_then(|session| session.pipeline) + .and_then(|pipeline| pipeline.steps.into_iter().nth(step_index)); + let Some(step) = step else { + return Some("Pull failed".to_string()); + }; + + let detail = step + .output + .as_deref() + .map(str::trim) + .filter(|output| !output.is_empty()); + Some(match detail { + Some(detail) => format!("{} failed:\n\n{detail}", step.label), + None => format!("{} failed", step.label), + }) +} + fn pre_head_for_pipeline_handoff(config: &PipelineConfig) -> Option { match config.pipeline.kind.as_ref() { Some(PipelineKind::Rebase) => config.pre_head_sha.clone(), @@ -1416,7 +1467,9 @@ fn pre_head_for_pipeline_handoff(config: &PipelineConfig) -> Option { None } }, - None => None, + // A push never rewrites local history and a pull never hands off, so + // neither has a pre-pipeline HEAD to compare against. + Some(PipelineKind::Push | PipelineKind::Pull) | None => None, } } @@ -1443,7 +1496,9 @@ fn resolve_pipeline_artifacts_without_ai(config: &PipelineConfig, store: &Store, ); } } - None => {} + // Push and pull pipelines create no artifact, so there is nothing to + // resolve. + Some(PipelineKind::Push | PipelineKind::Pull) | None => {} } } @@ -3631,6 +3686,94 @@ mod tests { let _ = std::fs::remove_dir_all(repo); } + /// A session whose pipeline recorded `step_index` as failed, mirroring what + /// `run_pipeline` persists before it returns `Aborted`. + fn store_with_aborted_pipeline( + kind: PipelineKind, + steps: &[PipelineStep], + step_index: usize, + output: Option<&str>, + ) -> (Store, PipelineConfig) { + let store = Store::in_memory().unwrap(); + let mut pipeline = PipelineExecution::from_steps(steps).with_kind(kind.clone()); + pipeline.steps[step_index].status = StepStatus::Failed; + pipeline.steps[step_index].output = output.map(str::to_string); + let mut session = + crate::store::Session::new_running("git pipeline", std::path::Path::new("/tmp")); + session.pipeline = Some(pipeline.clone()); + store.create_session(&session).unwrap(); + + let config = PipelineConfig { + session_id: session.id, + prompt: "git pipeline".to_string(), + steps: steps.to_vec(), + pipeline, + working_dir: PathBuf::from("/tmp"), + pre_head_sha: None, + provider: None, + workspace_name: None, + remote_working_dir: None, + branch_id: None, + project_id: None, + }; + (store, config) + } + + fn abort_step(label: &str) -> PipelineStep { + PipelineStep::Command { + label: label.to_string(), + command: "true".to_string(), + on_failure: FailureStrategy::Abort { marker: None }, + } + } + + #[test] + fn aborted_pull_reports_the_failing_step_and_its_output() { + let steps = vec![ + abort_step("Fetch origin/feature"), + abort_step("Fast-forward"), + ]; + let (store, config) = store_with_aborted_pipeline( + PipelineKind::Pull, + &steps, + 1, + Some(" fatal: Not possible to fast-forward, aborting.\n"), + ); + + let error = aborted_pipeline_error(&config, &store, 1).unwrap(); + + assert_eq!( + error, + "Fast-forward failed:\n\nfatal: Not possible to fast-forward, aborting." + ); + } + + #[test] + fn aborted_pull_without_step_output_still_names_the_step() { + let steps = vec![abort_step("Fetch origin/feature")]; + let (store, config) = store_with_aborted_pipeline(PipelineKind::Pull, &steps, 0, None); + + assert_eq!( + aborted_pipeline_error(&config, &store, 0).unwrap(), + "Fetch origin/feature failed" + ); + } + + #[test] + fn aborted_push_completes_without_an_error_message() { + // A push rejected as non-fast-forward is an expected outcome: the session + // completes so the frontend can offer a force push. + let steps = vec![abort_step("Push to remote")]; + let (store, config) = store_with_aborted_pipeline( + PipelineKind::Push, + &steps, + 0, + Some("! [rejected] feature -> feature (non-fast-forward)"), + ); + + assert_eq!(aborted_pipeline_error(&config, &store, 0), None); + } + #[test] fn rebase_pipeline_completion_deletes_noop_pending_commit() { let repo = make_git_repo("rebase-noop"); diff --git a/apps/staged/src-tauri/src/store/migration_tests.rs b/apps/staged/src-tauri/src/store/migration_tests.rs index bc55d2065..e0083da1e 100644 --- a/apps/staged/src-tauri/src/store/migration_tests.rs +++ b/apps/staged/src-tauri/src/store/migration_tests.rs @@ -145,7 +145,7 @@ fn test_store_bootstraps_fresh_database_with_baseline_migration() { ) .unwrap(); - assert_eq!(version, 21); + assert_eq!(version, 22); assert_eq!(app_version, super::APP_VERSION); assert!(table_exists(&conn, "projects")); assert!(table_exists(&conn, "project_notes")); @@ -160,6 +160,7 @@ fn test_store_bootstraps_fresh_database_with_baseline_migration() { )); assert!(column_exists(&conn, "sessions", "acp_config_selection")); assert!(column_exists(&conn, "sessions", "acp_title")); + assert!(column_exists(&conn, "sessions", "branch_id")); let trigger_count: i64 = conn .query_row( @@ -223,10 +224,11 @@ fn test_store_repairs_github_comment_tracking_user_version() { let version: i64 = conn .query_row("PRAGMA user_version", [], |row| row.get(0)) .unwrap(); - assert_eq!(version, 21); + assert_eq!(version, 22); assert!(column_exists(&conn, "sessions", "pipeline")); assert!(column_exists(&conn, "sessions", "acp_config_selection")); assert!(column_exists(&conn, "sessions", "acp_title")); + assert!(column_exists(&conn, "sessions", "branch_id")); assert!(column_exists( &conn, "session_messages", @@ -282,7 +284,7 @@ fn test_store_repairs_pipeline_user_version() { let version: i64 = conn .query_row("PRAGMA user_version", [], |row| row.get(0)) .unwrap(); - assert_eq!(version, 21); + assert_eq!(version, 22); assert!(column_exists(&conn, "comments", "github_comment_id")); assert!(column_exists(&conn, "comments", "github_comment_type")); assert!(column_exists(&conn, "comments", "github_comment_stale")); @@ -294,6 +296,7 @@ fn test_store_repairs_pipeline_user_version() { assert!(table_exists(&conn, "queued_session_messages")); assert!(column_exists(&conn, "sessions", "acp_config_selection")); assert!(column_exists(&conn, "sessions", "acp_title")); + assert!(column_exists(&conn, "sessions", "branch_id")); cleanup_db(&path); } diff --git a/apps/staged/src-tauri/src/store/migrations/0022-add-session-branch-id/up.sql b/apps/staged/src-tauri/src/store/migrations/0022-add-session-branch-id/up.sql new file mode 100644 index 000000000..8f67e2a61 --- /dev/null +++ b/apps/staged/src-tauri/src/store/migrations/0022-add-session-branch-id/up.sql @@ -0,0 +1,8 @@ +-- Link branch-scoped sessions that create no artifact directly to their branch. +-- +-- Queued/running sessions are normally found for a branch through their commit, +-- note, or review row. Push pipelines produce none of those, so they were +-- invisible to the branch queue. Existing rows keep resolving via artifacts; +-- only sessions with no artifact need this column populated. +ALTER TABLE sessions ADD COLUMN branch_id TEXT DEFAULT NULL REFERENCES branches(id) ON DELETE CASCADE; +CREATE INDEX idx_sessions_branch ON sessions(branch_id); diff --git a/apps/staged/src-tauri/src/store/models.rs b/apps/staged/src-tauri/src/store/models.rs index cb9dd0364..1325ae8fb 100644 --- a/apps/staged/src-tauri/src/store/models.rs +++ b/apps/staged/src-tauri/src/store/models.rs @@ -531,6 +531,14 @@ pub struct Session { /// Used as an interim display name while the session is running. #[serde(default)] pub acp_title: Option, + /// Branch this session belongs to, for sessions that create no artifact. + /// + /// Branch-scoped sessions are normally found through their commit, note, or + /// review row. Push pipelines have none of those, so they record the branch + /// here to stay visible to the branch queue. `None` for artifact-backed + /// sessions and for project-level sessions. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub branch_id: Option, } /// Persistent follow-up message waiting to be sent to an existing session. @@ -621,6 +629,7 @@ impl Session { pipeline: None, acp_config_selection: None, acp_title: None, + branch_id: None, } } @@ -644,6 +653,7 @@ impl Session { pipeline: None, acp_config_selection: None, acp_title: None, + branch_id: None, } } @@ -652,6 +662,12 @@ impl Session { self } + /// Link an artifact-less session to its branch so the branch queue sees it. + pub fn with_branch(mut self, branch_id: &str) -> Self { + self.branch_id = Some(branch_id.to_string()); + self + } + pub fn with_agent(mut self, agent_id: &str) -> Self { self.agent_id = Some(agent_id.to_string()); self @@ -1436,6 +1452,13 @@ pub struct PipelineExecution { /// rather than silently downgrading to a base rebase. #[serde(default, skip_serializing_if = "Option::is_none")] pub rebase_target: Option, + /// Whether the push variant force-pushes (`--force-with-lease`). + /// + /// Recorded so a queued push re-derives the same command on dequeue, and so + /// the queue can tell a pending push from a pending force push. Always + /// `false` for non-push pipelines. + #[serde(default, skip_serializing_if = "is_false")] + pub push_force: bool, pub steps: Vec, pub current_step: usize, /// Set when pipeline completes without needing AI. @@ -1472,6 +1495,7 @@ impl PipelineExecution { Self { kind: None, rebase_target: None, + push_force: false, steps: step_statuses, current_step: 0, completed_without_ai: false, @@ -1487,14 +1511,30 @@ impl PipelineExecution { self.rebase_target = Some(target); self } + + pub fn with_push_force(mut self, force: bool) -> Self { + self.push_force = force; + self + } +} + +/// `skip_serializing_if` predicate so non-push pipelines persist no push flag. +fn is_false(value: &bool) -> bool { + !*value } -/// Durable identity for commit-producing command pipelines. +/// Durable identity for command pipelines that the branch queue schedules. +/// +/// `Rebase` and `Squash` produce a commit and are linked to their branch through +/// a pending-commit artifact. `Push` and `Pull` produce no artifact and are +/// linked through `Session::branch_id` instead. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] pub enum PipelineKind { Rebase, Squash, + Push, + Pull, } #[cfg(test)] @@ -1534,6 +1574,36 @@ mod pipeline_tests { assert_eq!(execution.rebase_target, None); } + #[test] + fn pipeline_push_force_is_optional_for_legacy_pipeline_json() { + let execution: PipelineExecution = + serde_json::from_str(r#"{"steps":[],"currentStep":0,"completedWithoutAi":false}"#) + .unwrap(); + + assert!(!execution.push_force); + } + + #[test] + fn pipeline_push_force_round_trips_and_stays_out_of_non_push_json() { + let execution = PipelineExecution::from_steps(&[]) + .with_kind(PipelineKind::Push) + .with_push_force(true); + let json = serde_json::to_string(&execution).unwrap(); + + assert!(json.contains("\"kind\":\"push\"")); + assert!(json.contains("\"pushForce\":true")); + assert!( + serde_json::from_str::(&json) + .unwrap() + .push_force + ); + + let rebase = PipelineExecution::from_steps(&[]).with_kind(PipelineKind::Rebase); + assert!(!serde_json::to_string(&rebase) + .unwrap() + .contains("pushForce")); + } + #[test] fn pipeline_rebase_target_round_trips() { let execution: PipelineExecution = serde_json::from_str( diff --git a/apps/staged/src-tauri/src/store/sessions.rs b/apps/staged/src-tauri/src/store/sessions.rs index d6397631a..9581f7950 100644 --- a/apps/staged/src-tauri/src/store/sessions.rs +++ b/apps/staged/src-tauri/src/store/sessions.rs @@ -19,8 +19,8 @@ impl Store { let acp_config_selection_json = serialize_acp_config_selection(session.acp_config_selection.as_ref())?; conn.execute( - "INSERT INTO sessions (id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)", + "INSERT INTO sessions (id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title, branch_id) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)", params![ session.id, session.prompt, @@ -36,6 +36,7 @@ impl Store { pipeline_json, acp_config_selection_json, session.acp_title, + session.branch_id, ], )?; Ok(()) @@ -44,7 +45,7 @@ impl Store { pub fn get_session(&self, id: &str) -> Result, StoreError> { let conn = self.conn.lock().unwrap(); conn.query_row( - "SELECT id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title + "SELECT id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title, branch_id FROM sessions WHERE id = ?1", params![id], Self::row_to_session, @@ -223,7 +224,7 @@ impl Store { pub fn get_running_sessions(&self) -> Result, StoreError> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( - "SELECT id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title + "SELECT id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title, branch_id FROM sessions WHERE status = 'running'", )?; let sessions = stmt @@ -287,20 +288,22 @@ impl Store { /// Get all queued sessions for a branch, ordered by creation time (oldest first). /// - /// Sessions are linked to branches through artifacts (commits, notes, reviews). - /// This query joins across all three artifact tables to find every queued session - /// belonging to the given branch. + /// Sessions are linked to branches through artifacts (commits, notes, reviews), + /// or directly via `sessions.branch_id` for artifact-less work such as queued + /// push pipelines. This query covers both so a queued push takes its place in + /// the branch queue alongside artifact-backed sessions. pub fn get_queued_sessions_for_branch( &self, branch_id: &str, ) -> Result, StoreError> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( - "SELECT s.id, s.prompt, s.status, s.working_dir, s.provider, s.agent_id, s.error_message, s.completion_reason, s.created_at, s.updated_at, s.owner_pid, s.pipeline, s.acp_config_selection, s.acp_title + "SELECT s.id, s.prompt, s.status, s.working_dir, s.provider, s.agent_id, s.error_message, s.completion_reason, s.created_at, s.updated_at, s.owner_pid, s.pipeline, s.acp_config_selection, s.acp_title, s.branch_id FROM sessions s WHERE s.status = 'queued' AND ( - EXISTS (SELECT 1 FROM commits c WHERE c.session_id = s.id AND c.branch_id = ?1) + s.branch_id = ?1 + OR EXISTS (SELECT 1 FROM commits c WHERE c.session_id = s.id AND c.branch_id = ?1) OR EXISTS (SELECT 1 FROM notes n WHERE n.session_id = s.id AND n.branch_id = ?1) OR EXISTS (SELECT 1 FROM reviews r WHERE r.session_id = s.id AND r.branch_id = ?1) ) @@ -315,14 +318,17 @@ impl Store { /// Check whether a branch already has a running session. /// /// Auto-reviews (`is_auto = 1`) are excluded because they run in the - /// background and should never block user-initiated sessions. + /// background and should never block user-initiated sessions. Sessions + /// linked through `sessions.branch_id` (running push pipelines) count, so a + /// push in flight blocks new branch work the same way a commit does. pub fn has_running_session_for_branch(&self, branch_id: &str) -> Result { let conn = self.conn.lock().unwrap(); let count: i64 = conn.query_row( "SELECT COUNT(*) FROM sessions s WHERE s.status = 'running' AND ( - EXISTS (SELECT 1 FROM commits c WHERE c.session_id = s.id AND c.branch_id = ?1) + s.branch_id = ?1 + OR EXISTS (SELECT 1 FROM commits c WHERE c.session_id = s.id AND c.branch_id = ?1) OR EXISTS (SELECT 1 FROM notes n WHERE n.session_id = s.id AND n.branch_id = ?1) OR EXISTS (SELECT 1 FROM reviews r WHERE r.session_id = s.id AND r.branch_id = ?1 AND r.is_auto = 0) )", @@ -332,7 +338,8 @@ impl Store { Ok(count > 0) } - /// Resolve the branch that owns a session through its linked artifact. + /// Resolve the branch that owns a session through its linked artifact, or + /// through `sessions.branch_id` for artifact-less branch work (pushes). /// /// Project-note sessions do not belong to a branch and therefore return `None`. /// This assumes all branch-linked artifacts for a session point at the same @@ -345,6 +352,8 @@ impl Store { let conn = self.conn.lock().unwrap(); conn.query_row( "SELECT branch_id FROM ( + SELECT branch_id FROM sessions WHERE id = ?1 AND branch_id IS NOT NULL + UNION ALL SELECT branch_id FROM commits WHERE session_id = ?1 UNION ALL SELECT branch_id FROM notes WHERE session_id = ?1 @@ -382,6 +391,8 @@ impl Store { conn.query_row( "SELECT b.project_id FROM branches b INNER JOIN ( + SELECT branch_id FROM sessions WHERE id = ?1 AND branch_id IS NOT NULL + UNION ALL SELECT branch_id FROM commits WHERE session_id = ?1 UNION ALL SELECT branch_id FROM notes WHERE session_id = ?1 @@ -479,6 +490,7 @@ impl Store { pipeline, acp_config_selection, acp_title: row.get(13)?, + branch_id: row.get(14)?, }) } } diff --git a/apps/staged/src-tauri/src/store/tests.rs b/apps/staged/src-tauri/src/store/tests.rs index 0a973ccfe..0f4b36e11 100644 --- a/apps/staged/src-tauri/src/store/tests.rs +++ b/apps/staged/src-tauri/src/store/tests.rs @@ -947,6 +947,81 @@ fn test_queued_pipeline_commit_owns_branch_queue_position() { ); } +#[test] +fn test_queued_push_pipeline_is_found_through_session_branch_link() { + let store = Store::in_memory().unwrap(); + let project = Project::new("test-owner/test-repo"); + store.create_project(&project).unwrap(); + let branch = Branch::new(&project.id, "feature", "main"); + store.create_branch(&branch).unwrap(); + let other_branch = Branch::new(&project.id, "other", "main"); + store.create_branch(&other_branch).unwrap(); + + // No commit/note/review artifact: the branch link is the only way in. + let mut session = Session::new_queued("Push the current branch").with_branch(&branch.id); + session.pipeline = Some( + PipelineExecution::from_steps(&[]) + .with_kind(PipelineKind::Push) + .with_push_force(true), + ); + store.create_session(&session).unwrap(); + + let queued = store.get_queued_sessions_for_branch(&branch.id).unwrap(); + assert_eq!(queued.len(), 1); + assert_eq!(queued[0].id, session.id); + assert_eq!(queued[0].branch_id.as_deref(), Some(branch.id.as_str())); + assert_eq!( + queued[0].pipeline.as_ref().and_then(|p| p.kind.as_ref()), + Some(&PipelineKind::Push) + ); + assert!(queued[0].pipeline.as_ref().unwrap().push_force); + + assert!(store + .get_queued_sessions_for_branch(&other_branch.id) + .unwrap() + .is_empty()); +} + +#[test] +fn test_running_push_pipeline_marks_branch_busy_and_resolves_branch() { + let store = Store::in_memory().unwrap(); + let project = Project::new("test-owner/test-repo"); + store.create_project(&project).unwrap(); + let branch = Branch::new(&project.id, "feature", "main"); + store.create_branch(&branch).unwrap(); + + let mut session = + Session::new_running("Push the current branch", Path::new("/tmp")).with_branch(&branch.id); + session.pipeline = Some(PipelineExecution::from_steps(&[]).with_kind(PipelineKind::Push)); + store.create_session(&session).unwrap(); + + assert!(store.has_running_session_for_branch(&branch.id).unwrap()); + assert_eq!( + store + .get_branch_id_for_session(&session.id) + .unwrap() + .as_deref(), + Some(branch.id.as_str()) + ); + assert_eq!( + store + .get_project_id_for_session(&session.id) + .unwrap() + .as_deref(), + Some(project.id.as_str()) + ); + // The drain scan resolves schedules from `get_running_sessions`, so that + // query has to carry the branch link too. + let running = store.get_running_sessions().unwrap(); + assert_eq!( + running + .iter() + .find(|running| running.id == session.id) + .and_then(|running| running.branch_id.as_deref()), + Some(branch.id.as_str()) + ); +} + #[test] fn test_mark_session_artifact_started_restamps_empty_note_and_preserves_session_created_at() { let store = Store::in_memory().unwrap(); diff --git a/apps/staged/src-tauri/src/timeline.rs b/apps/staged/src-tauri/src/timeline.rs index bfa5aeb7a..88238f690 100644 --- a/apps/staged/src-tauri/src/timeline.rs +++ b/apps/staged/src-tauri/src/timeline.rs @@ -796,20 +796,11 @@ pub(crate) fn list_parent_branch_commits_impl( Ok(parse_parent_commit_lines(&lines)) } -#[tauri::command(rename_all = "camelCase")] -pub async fn pull_branch_ff_only( - store: tauri::State<'_, Mutex>>>, - branch_id: String, -) -> Result<(), String> { - let store = crate::get_store(&store)?; - - tauri::async_runtime::spawn_blocking(move || pull_branch_ff_only_impl(&store, &branch_id)) - .await - .map_err(|e| format!("Pull task failed: {e}"))? -} - -/// Synchronous body of [`pull_branch_ff_only`], shared verbatim with the -/// web-mode `dispatch()` arm. Callers run it inside `spawn_blocking`. +/// Fast-forward the branch to its upstream, right now. +/// +/// The immediate half of `prs::pull_or_queue_branch`, which owns the decision +/// between pulling now and queueing behind in-flight branch sessions. Callers run +/// this inside `spawn_blocking`. pub(crate) fn pull_branch_ff_only_impl(store: &Arc, branch_id: &str) -> Result<(), String> { let branch = store .get_branch(branch_id) diff --git a/apps/staged/src-tauri/src/web_server.rs b/apps/staged/src-tauri/src/web_server.rs index 283136046..d53454bbb 100644 --- a/apps/staged/src-tauri/src/web_server.rs +++ b/apps/staged/src-tauri/src/web_server.rs @@ -2322,15 +2322,18 @@ async fn dispatch(command: &str, args: Value, state: &WebAppState) -> Result { + "pull_or_queue_branch" => { let store = get_store(store_mutex)?; let branch_id: String = arg(&args, "branchId")?; - tauri::async_runtime::spawn_blocking(move || { - crate::timeline::pull_branch_ff_only_impl(&store, &branch_id) - }) - .await - .map_err(|e| format!("Pull task failed: {e}"))??; - Ok(Value::Null) + // Returns the queued session id, or null when the pull ran now. + let queued_session_id = crate::prs::pull_or_queue_branch_for_branch( + store, + Arc::clone(session_registry), + app_handle.clone(), + branch_id, + ) + .await?; + Ok(serde_json::to_value(queued_session_id).unwrap()) } "reset_branch_to_remote" => { let store = get_store(store_mutex)?; @@ -3376,7 +3379,7 @@ async fn dispatch(command: &str, args: Value, state: &WebAppState) -> Result = opt_arg(&args, "provider")?; let force: Option = opt_arg(&args, "force")?; - let session_id = crate::prs::start_push_branch_pipeline_for_branch( + let response = crate::prs::start_or_queue_push_pipeline_for_branch( store, Arc::clone(session_registry), app_handle.clone(), @@ -3385,29 +3388,32 @@ async fn dispatch(command: &str, args: Value, state: &WebAppState) -> Result { let store = get_store(store_mutex)?; let branch_id: String = arg(&args, "branchId")?; let provider: Option = opt_arg(&args, "provider")?; - let session_id = crate::prs::start_or_queue_commit_pipeline_for_branch( + // Forwarded so web mode honours "Rebase onto Origin" instead of + // silently downgrading it to a base rebase. + let target: Option = opt_arg(&args, "target")?; + let response = crate::prs::start_or_queue_commit_pipeline_for_branch( store, Arc::clone(session_registry), app_handle.clone(), branch_id, store::PipelineKind::Rebase, provider, - None, + target, ) .await?; - Ok(serde_json::to_value(session_id).unwrap()) + Ok(serde_json::to_value(response).unwrap()) } "squash_commits" => { let store = get_store(store_mutex)?; let branch_id: String = arg(&args, "branchId")?; let provider: Option = opt_arg(&args, "provider")?; - let session_id = crate::prs::start_or_queue_commit_pipeline_for_branch( + let response = crate::prs::start_or_queue_commit_pipeline_for_branch( store, Arc::clone(session_registry), app_handle.clone(), @@ -3417,7 +3423,7 @@ async fn dispatch(command: &str, args: Value, state: &WebAppState) -> Result { let store = get_store(store_mutex)?; diff --git a/apps/staged/src/lib/commands.test.ts b/apps/staged/src/lib/commands.test.ts index 9db0ed4dd..f6b2d63a5 100644 --- a/apps/staged/src/lib/commands.test.ts +++ b/apps/staged/src/lib/commands.test.ts @@ -259,6 +259,79 @@ describe('browser-native command wrappers', () => { }); }); + it('surfaces the queued-vs-running status returned by rebase and squash', async () => { + const invokeCommand = vi + .fn() + .mockResolvedValueOnce({ sessionId: 'session-1', sessionStatus: 'queued' }) + .mockResolvedValueOnce({ sessionId: 'session-2', sessionStatus: 'running' }); + vi.doMock('./transport', () => ({ + invokeCommand, + isTauri: true, + })); + + const { rebaseBranch, squashCommits } = await import('./commands'); + + await expect(rebaseBranch('branch-1', 'codex', 'origin')).resolves.toEqual({ + sessionId: 'session-1', + sessionStatus: 'queued', + }); + await expect(squashCommits('branch-1', 'codex')).resolves.toEqual({ + sessionId: 'session-2', + sessionStatus: 'running', + }); + + expect(invokeCommand.mock.calls).toEqual([ + ['rebase_branch', { branchId: 'branch-1', provider: 'codex', target: 'origin' }], + ['squash_commits', { branchId: 'branch-1', provider: 'codex' }], + ]); + }); + + it('surfaces the queued-vs-running status returned by push and force push', async () => { + const invokeCommand = vi + .fn() + .mockResolvedValueOnce({ sessionId: 'session-1', sessionStatus: 'running' }) + .mockResolvedValueOnce({ sessionId: 'session-2', sessionStatus: 'queued' }); + vi.doMock('./transport', () => ({ + invokeCommand, + isTauri: true, + })); + + const { pushBranch } = await import('./commands'); + + await expect(pushBranch('branch-1', 'codex', false)).resolves.toEqual({ + sessionId: 'session-1', + sessionStatus: 'running', + }); + await expect(pushBranch('branch-1', 'codex', true)).resolves.toEqual({ + sessionId: 'session-2', + sessionStatus: 'queued', + }); + + expect(invokeCommand.mock.calls).toEqual([ + ['push_branch', { branchId: 'branch-1', provider: 'codex', force: false }], + ['push_branch', { branchId: 'branch-1', provider: 'codex', force: true }], + ]); + }); + + it('distinguishes an immediate pull from a queued one', async () => { + const invokeCommand = vi.fn().mockResolvedValueOnce(null).mockResolvedValueOnce('session-1'); + vi.doMock('./transport', () => ({ + invokeCommand, + isTauri: true, + })); + + const { pullOrQueueBranch } = await import('./commands'); + + // null: the branch was idle, so the pull already fast-forwarded. + await expect(pullOrQueueBranch('branch-1')).resolves.toBeNull(); + await expect(pullOrQueueBranch('branch-1')).resolves.toBe('session-1'); + + expect(invokeCommand.mock.calls).toEqual([ + ['pull_or_queue_branch', { branchId: 'branch-1' }], + ['pull_or_queue_branch', { branchId: 'branch-1' }], + ]); + }); + it('forwards ACP config selection when resuming a session', async () => { const invokeCommand = vi.fn().mockResolvedValue(undefined); vi.doMock('./transport', () => ({ diff --git a/apps/staged/src/lib/commands.ts b/apps/staged/src/lib/commands.ts index 2663936ae..81ba745d4 100644 --- a/apps/staged/src/lib/commands.ts +++ b/apps/staged/src/lib/commands.ts @@ -21,6 +21,7 @@ import type { Branch, BranchTimeline, BranchRef, + BranchPipelineResponse, BranchSessionLaunchContext, BranchSessionType, BranchSessionResponse, @@ -546,8 +547,11 @@ export function invalidateProjectBranchTimelines(branchIds: string[]): void { window.dispatchEvent(new CustomEvent('timeline-invalidated', { detail: { branchIds } })); } -export function pullBranchFastForward(branchId: string): Promise { - return invokeCommand('pull_branch_ff_only', { branchId }); +/** Fast-forward a branch to origin, or queue the pull behind in-flight branch + * sessions. The backend owns that decision: it resolves to the queued session's + * ID when the pull has to wait, and to `null` when the pull already ran. */ +export function pullOrQueueBranch(branchId: string): Promise { + return invokeCommand('pull_or_queue_branch', { branchId }); } export function resetBranchToRemote(branchId: string): Promise { @@ -1254,8 +1258,13 @@ export function hasUnpushedCommits(branchId: string): Promise { /** Push a branch to its remote via an agent session. * The agent runs git push and can fix pre-push hook failures. - * Returns the session ID so the frontend can track progress. */ -export function pushBranch(branchId: string, provider?: string, force?: boolean): Promise { + * Queues behind in-flight branch sessions, so the response reports whether the + * returned session is pushing now or waiting on the branch queue. */ +export function pushBranch( + branchId: string, + provider?: string, + force?: boolean +): Promise { return invokeCommand('push_branch', { branchId, provider: provider ?? null, @@ -1281,12 +1290,13 @@ export interface GitHubCommentResult { /** Rebase a branch via a pipeline. * When target is 'base' (default), rebases onto origin/{base_branch}. * When target is 'origin', rebases onto origin/{branch_name}. - * Returns the session ID so the frontend can track progress. */ + * Queues behind in-flight branch sessions, so the response reports whether the + * returned session is running or waiting on the branch queue. */ export function rebaseBranch( branchId: string, provider?: string, target?: 'base' | 'origin' -): Promise { +): Promise { return invokeCommand('rebase_branch', { branchId, provider: provider ?? null, @@ -1296,8 +1306,12 @@ export function rebaseBranch( /** Squash all commits on a branch into a single commit via a pipeline. * Uses git reset --soft then hands off to AI to write the commit message. - * Returns the session ID so the frontend can track progress. */ -export function squashCommits(branchId: string, provider?: string): Promise { + * Queues behind in-flight branch sessions, so the response reports whether the + * returned session is running or waiting on the branch queue. */ +export function squashCommits( + branchId: string, + provider?: string +): Promise { return invokeCommand('squash_commits', { branchId, provider: provider ?? null, diff --git a/apps/staged/src/lib/features/branches/BranchCard.svelte b/apps/staged/src/lib/features/branches/BranchCard.svelte index eba9597af..d93a1b7c9 100644 --- a/apps/staged/src/lib/features/branches/BranchCard.svelte +++ b/apps/staged/src/lib/features/branches/BranchCard.svelte @@ -61,8 +61,12 @@ import * as AlertDialog from '$lib/components/ui/alert-dialog'; import { Button } from '$lib/components/ui/button'; import { + classifyPolledSession, + createPollFailureTracker, + createQueuedSessionCanceller, fileNameFromPath, formatBaseBranch, + isGitActionInFlight, isMaybeTextFile, isImageFile, } from './branchCardHelpers'; @@ -74,6 +78,7 @@ addPendingSession, getPendingSessionItems, prunePendingSessionItems, + queuedSessionMeta, } from './branchSessionLaunch.svelte'; import RemoteWorkspaceStatusBadge from './RemoteWorkspaceStatusBadge.svelte'; import RemoteWorkspaceStatusView from './RemoteWorkspaceStatusView.svelte'; @@ -83,6 +88,7 @@ import { timelineToHashtagItems, projectNotesToHashtagItems } from '../sessions/hashtagItems'; import { getPreferredAgent } from '../settings/preferences.svelte'; import { agentState, REMOTE_AGENTS } from '../agents/agent.svelte'; + import { pullStateStore } from '../../stores/pullState.svelte'; import { pushStateStore } from '../../stores/pushState.svelte'; import { onBranchGitStateUpdated, @@ -327,28 +333,68 @@ ); } let commandPipelinePending = $state(false); - let branchSessionBusy = $derived(timeline ? hasActiveSessions(timeline) : false); + + // Push and pull state is sourced from the global stores so it survives the + // BranchCard remount that happens when the user switches projects and back. + // Both are read up here, above the handlers that use them, because a push or + // pull is branch work `hasActiveSessions` cannot see: neither creates a + // timeline artifact. + let storePushState = $derived(pushStateStore.getPushState(branch.id)); + let storePullState = $derived(pullStateStore.getPullState(branch.id)); + /** A git action of our own is running or waiting on the branch queue. */ + let gitPipelineInFlight = $derived( + isGitActionInFlight({ + push: storePushState, + pull: storePullState, + immediatePull: pullingOrigin, + }) + ); + /** + * True when the branch has work in flight that a new action queues behind. + * + * Folding the git actions in matters for the gates that read this: without them + * the frontend reads a mid-push branch as idle and keeps disabling actions the + * backend would happily queue. The opposite skew — a timeline that still shows a + * session which has just finished — is unavoidable from here; it costs an + * immediate pull that fails with "Cannot pull with uncommitted changes" instead + * of a disabled button. + */ + let branchSessionBusy = $derived( + (timeline ? hasActiveSessions(timeline) : false) || gitPipelineInFlight + ); + + /** Timeline label the backend gives a queued rebase/squash pipeline session. */ + const branchCommandLabels = { rebase: 'Rebase branch', squash: 'Squash commits' } as const; async function startBranchCommandPipeline( kind: 'rebase' | 'squash', rebaseTarget?: 'base' | 'origin' ) { - if (commandPipelinePending || branchSessionBusy) return; + // Deliberately not gated on `branchSessionBusy`: the backend decides + // between running now and queueing behind in-flight branch work, and it + // dedupes repeat clicks of the same action. + if (commandPipelinePending) return; commandPipelinePending = true; const agents = isRemote ? REMOTE_AGENTS : agentState.providers; const provider = getPreferredAgent(agents) ?? undefined; const pendingKey = `pipeline-${kind}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; try { - let sessionId: string; - if (kind === 'rebase') { - sessionId = await commands.rebaseBranch(branch.id, provider, rebaseTarget); - } else { - sessionId = await commands.squashCommits(branch.id, provider); - } + const result = + kind === 'rebase' + ? await commands.rebaseBranch(branch.id, provider, rebaseTarget) + : await commands.squashCommits(branch.id, provider); // Add a pending session item so the session stub appears instantly - // instead of waiting for the full timeline refresh. - const title = kind === 'rebase' ? 'Rebasing…' : 'Squashing…'; - addPendingSession(branch.id, { key: pendingKey, type: 'pending-commit', title, sessionId }); + // instead of waiting for the full timeline refresh. Queued pipelines get + // the same label the persisted row will show, so the stub doesn't flash a + // progress title for work that hasn't started. + const queued = result.sessionStatus === 'queued'; + addPendingSession(branch.id, { + key: pendingKey, + type: queued ? 'queued-commit' : 'pending-commit', + title: queued ? branchCommandLabels[kind] : kind === 'rebase' ? 'Rebasing…' : 'Squashing…', + secondaryMeta: queued ? queuedSessionMeta(timeline) : undefined, + sessionId: result.sessionId, + }); await loadTimeline(); } catch (e) { notifyError(kind === 'rebase' ? 'Rebase failed' : 'Squash failed', e); @@ -604,13 +650,22 @@ let branchIdentityWarning = $derived(gitIdentityWarning(timeline?.gitState)); let gitUnsafeActionsDisabled = $derived(!!branchIdentityWarning); + /** + * Gate for the queueable git actions (Rebase/Squash, push, force-push). + * In-flight sessions are not a reason to disable: they all queue on the branch + * session queue and drain when the branch frees up. Only identity problems + * (detached HEAD, wrong branch) make them unsafe. + */ let branchCommandDisabledReason = $derived( - branchIdentityWarning ?? - (commandPipelinePending - ? 'Command in progress' - : branchSessionBusy - ? 'Session in progress' - : null) + branchIdentityWarning ?? (commandPipelinePending ? 'Command in progress' : null) + ); + /** + * Gate for reset to origin, which still executes immediately: it is validated + * against a point-in-time preview of what would be discarded, so a busy branch + * has to keep blocking it. + */ + let immediateGitActionDisabledReason = $derived( + branchCommandDisabledReason ?? (branchSessionBusy ? 'Session in progress' : null) ); // ========================================================================= @@ -994,11 +1049,31 @@ } } + // A queued pull outlives this component (it drains when the branch frees up), + // so its state lives in the global pullStateStore (read as `storePullState` + // above, alongside pushState). The immediate pull stays local: it is awaited + // right here. + /** Pull is waiting on the branch session queue rather than running. */ + let pullQueuedOrigin = $derived(storePullState?.state === 'queued'); + let pullSessionId = $derived(storePullState?.sessionId ?? null); + /** True for both the immediate pull and a drained one still running. */ + let pullInFlight = $derived(pullingOrigin || storePullState?.state === 'pulling'); + + /** + * Pull origin's new commits, or queue the pull behind in-flight branch work. + * + * The backend decides which — an idle branch fast-forwards immediately, with no + * session row for what is usually an instant operation — and it dedupes repeat + * clicks, so this is deliberately not gated on `branchSessionBusy`. + */ async function handlePullOrigin() { - if (pullingOrigin) return; + if (pullInFlight || pullQueuedOrigin) return; pullingOrigin = true; try { - await commands.pullBranchFastForward(branch.id); + const queuedSessionId = await commands.pullOrQueueBranch(branch.id); + if (queuedSessionId) { + pullStateStore.setPullQueued(branch.id, queuedSessionId); + } await loadTimeline(); } catch (e) { notifyError('Pull failed', e); @@ -1007,6 +1082,89 @@ } } + /** + * Drop a pull that is still waiting on the branch queue. + * + * The store entry is cleared here rather than by the completion listener: a + * session that never ran isn't in the session registry, so its cancellation + * event carries no pull session type to match on. It waits for the backend to + * confirm the cancellation — see `createQueuedSessionCanceller`. + */ + const runCancelQueuedPull = createQueuedSessionCanceller({ + cancel: (sessionId) => commands.cancelSession(sessionId), + clearState: () => pullStateStore.clearPullState(branch.id), + onError: (e) => notifyError('Could not cancel queued pull', e), + }); + + async function cancelQueuedPull() { + const sessionId = pullSessionId; + if (!pullQueuedOrigin || !sessionId) return; + if (await runCancelQueuedPull(sessionId)) { + commands.invalidateBranchTimeline(branch.id); + await loadTimeline(); + } + } + + /** + * Fallback polling for a pull on the branch queue, mirroring the push poller in + * BranchCardPrButton. + * + * A queued pull is drained headless, so `session-status-changed` is the only + * thing that moves it off "Pull queued" and reports a failure. If that event is + * missed the badge sticks until the user clicks Cancel and a failed pull loses + * its toast, so poll the session too. The effect tracks `storePullState`, so + * whichever of the two gets there first tears the other down. + * + * A deleted session comes back as `null` from `getSession`, so a rejection only + * ever means the backend was unreachable. Those are tolerated for a few + * consecutive attempts — dropping the badge on the first blip would strand a + * still-queued pull with no Cancel button and no signal to re-click. + */ + $effect(() => { + const sessionId = storePullState?.sessionId; + if (!sessionId) return; + + const failures = createPollFailureTracker(); + const interval = setInterval(async () => { + try { + const session = await commands.getSession(sessionId); + failures.recordSuccess(); + const disposition = classifyPolledSession(session); + if (disposition === 'waiting') return; + if (disposition === 'active') { + pullStateStore.markQueuedPullStarted(branch.id, sessionId); + return; + } + // 'gone' or 'finished': either way the badge has to go, and a drained pull + // that failed is the only outcome still worth telling the user about. + pullStateStore.clearPullState(branch.id); + if (session?.status === 'error') { + notifyError( + 'Pull failed', + session.errorMessage ?? 'The queued pull could not fast-forward this branch.' + ); + } + commands.invalidateBranchTimeline(branch.id); + await loadTimeline(); + } catch (e) { + if (!failures.recordFailure()) { + console.warn( + `[BranchCard] Could not poll pull session ${sessionId} for branch ${branch.id}, retrying:`, + e + ); + return; + } + console.error( + `[BranchCard] Lost track of pull session ${sessionId} for branch ${branch.id}:`, + e + ); + pullStateStore.clearPullState(branch.id); + } + }, 5_000); + + return () => clearInterval(interval); + }); + function formatCommitCount(count: number, noun = 'commit'): string { return `${count} ${noun}${count === 1 ? '' : 's'}`; } @@ -1049,31 +1207,56 @@ } } - // Push state is sourced from the global pushStateStore so it survives the - // BranchCard remount that happens when the user switches projects and back. - // The store is shared with BranchCardPrButton (single entry per branch.id), - // updated by the global sessionStatusListener on completion, and covered by - // a 5s polling fallback in BranchCardPrButton. - let storePushState = $derived(pushStateStore.getPushState(branch.id)); + // `storePushState` (read above) is shared with BranchCardPrButton (single entry + // per branch.id), updated by the global sessionStatusListener on completion, and + // covered by a 5s polling fallback in BranchCardPrButton. let pushingOrigin = $derived(storePushState?.state === 'pushing'); + /** Push is waiting on the branch session queue rather than running. */ + let pushQueuedOrigin = $derived(storePushState?.state === 'queued'); let pushSessionId = $derived(storePushState?.sessionId ?? null); let forcePushingOrigin = $derived(pushingOrigin); let forcePushSessionId = $derived(pushSessionId); async function handlePushOrigin() { - if (pushingOrigin || commandPipelinePending || branchSessionBusy) return; + // Deliberately not gated on `branchSessionBusy`: the backend decides + // between pushing now and queueing behind in-flight branch work, and it + // dedupes repeat clicks of the same action. + if (pushingOrigin || pushQueuedOrigin || commandPipelinePending) return; const agents = isRemote ? REMOTE_AGENTS : agentState.providers; const provider = getPreferredAgent(agents) ?? undefined; pushStateStore.setPushing(branch.id, '__pending__'); try { - const sessionId = await commands.pushBranch(branch.id, provider, false); - pushStateStore.setPushing(branch.id, sessionId); + const response = await commands.pushBranch(branch.id, provider, false); + pushStateStore.setPushLaunch(branch.id, response); } catch (e) { pushStateStore.setPushError(branch.id, e instanceof Error ? e.message : String(e)); notifyError('Push failed', e); } } + /** + * Drop a push that is still waiting on the branch queue. + * + * The store entry is cleared here rather than by the completion listener: a + * session that never ran isn't in the session registry, so its cancellation + * event carries no push session type to match on. It waits for the backend to + * confirm the cancellation — see `createQueuedSessionCanceller`. + */ + const runCancelQueuedPush = createQueuedSessionCanceller({ + cancel: (sessionId) => commands.cancelSession(sessionId), + clearState: () => pushStateStore.clearPushState(branch.id), + onError: (e) => notifyError('Could not cancel queued push', e), + }); + + async function cancelQueuedPush() { + const sessionId = pushSessionId; + if (!pushQueuedOrigin || !sessionId || sessionId === '__pending__') return; + if (await runCancelQueuedPush(sessionId)) { + commands.invalidateBranchTimeline(branch.id); + await loadTimeline(); + } + } + function openPushSession() { if (pushSessionId && pushSessionId !== '__pending__') { sessionMgr.openSessionId = pushSessionId; @@ -1087,7 +1270,9 @@ } async function confirmForcePush() { - if (forcePushingOrigin || commandPipelinePending || branchSessionBusy) { + // Like `handlePushOrigin`, not gated on `branchSessionBusy` — the backend + // queues the force push behind in-flight branch work. + if (forcePushingOrigin || pushQueuedOrigin || commandPipelinePending) { // Another operation is in progress — keep the dialog open so the user // understands why the action didn't proceed. return; @@ -1097,8 +1282,8 @@ const provider = getPreferredAgent(agents) ?? undefined; pushStateStore.setPushing(branch.id, '__pending__'); try { - const sessionId = await commands.pushBranch(branch.id, provider, true); - pushStateStore.setPushing(branch.id, sessionId); + const response = await commands.pushBranch(branch.id, provider, true); + pushStateStore.setPushLaunch(branch.id, response); } catch (e) { pushStateStore.setPushError(branch.id, e instanceof Error ? e.message : String(e)); notifyError('Force push failed', e); @@ -1727,10 +1912,7 @@ onNoteCreated={() => loadTimeline()} onRebaseBranch={() => startBranchCommandPipeline('rebase')} onSquashCommits={() => startBranchCommandPipeline('squash')} - newCommitDisabled={sessionMgr.isNewSessionDisabled || - commandPipelinePending || - branchSessionBusy || - gitUnsafeActionsDisabled} + rebaseSquashDisabled={!!branchCommandDisabledReason} {commitCount} /> @@ -1830,8 +2012,14 @@ onOpenForcePushSession={forcePushSessionId && forcePushSessionId !== '__pending__' ? openForcePushSession : undefined} + onCancelQueuedPush={cancelQueuedPush} + onCancelQueuedPull={cancelQueuedPull} {forcePushingOrigin} - rebaseBranchDisabledReason={branchCommandDisabledReason} + {pushQueuedOrigin} + {pullQueuedOrigin} + {branchSessionBusy} + {immediateGitActionDisabledReason} + queueableGitActionDisabledReason={branchCommandDisabledReason} onViewWorktreeDiff={isLocal ? () => openDiffDetail({ @@ -1846,7 +2034,7 @@ onDiscardWorktreeChanges={handleDiscardWorktreeChanges} onNewSessionReferring={(ref) => sessionMgr.openNewSessionReferring(ref)} newSessionDisabled={sessionMgr.isNewSessionDisabled || gitUnsafeActionsDisabled} - {pullingOrigin} + pullingOrigin={pullInFlight} {pushingOrigin} {resettingToOrigin} {discardingWorktreeChanges} diff --git a/apps/staged/src/lib/features/branches/BranchCardActionsBar.svelte b/apps/staged/src/lib/features/branches/BranchCardActionsBar.svelte index 297abbd5d..5a2d5518f 100644 --- a/apps/staged/src/lib/features/branches/BranchCardActionsBar.svelte +++ b/apps/staged/src/lib/features/branches/BranchCardActionsBar.svelte @@ -90,7 +90,9 @@ onNoteCreated?: () => void; onRebaseBranch?: () => void; onSquashCommits?: () => void; - newCommitDisabled?: boolean; + /** Rebase/Squash queue behind running sessions, so this covers only the + * cases where they can't run at all (detached HEAD, wrong branch). */ + rebaseSquashDisabled?: boolean; commitCount?: number; } @@ -106,7 +108,7 @@ onNoteCreated, onRebaseBranch, onSquashCommits, - newCommitDisabled = false, + rebaseSquashDisabled = false, commitCount = 0, }: Props = $props(); @@ -923,11 +925,11 @@ Rename Branch - onRebaseBranch?.()}> + onRebaseBranch?.()}> Rebase Branch {#if commitCount >= 2} - onSquashCommits?.()}> + onSquashCommits?.()}> Squash Commits {/if} diff --git a/apps/staged/src/lib/features/branches/BranchCardPrButton.svelte b/apps/staged/src/lib/features/branches/BranchCardPrButton.svelte index a08a3d2fe..c433e36ef 100644 --- a/apps/staged/src/lib/features/branches/BranchCardPrButton.svelte +++ b/apps/staged/src/lib/features/branches/BranchCardPrButton.svelte @@ -6,11 +6,13 @@ -->