heddle-cli 0.8.0

// SPDX-License-Identifier: Apache-2.0
use std::collections::HashSet;

use anyhow::{Context, Result, anyhow};
use objects::{
    object::{ChangeId, State, ThreadName},
    store::ObjectStore,
};
use oplog::{OpBatch, OpLogBackend, OpRecord};
use repo::{Repository, Thread, ThreadIntegrationPolicy, thread_flag};
use serde::Serialize;

use super::{
    action_line::print_next,
    advice::RecoveryAdvice,
    checkpoint::create_git_checkpoint,
    collapse::{CollapsePublishedRef, collapse_resolved_states},
    git_overlay_health::{
        RepositoryVerificationState, build_repository_verification_state, remote_drift_decision,
    },
    git_overlay_txn,
    merge::{build_thread_preview_report, merge_thread_into_current},
    next_action::{NextActionValidationContext, write_command_json},
    operator_core::{
        OperatorAction, OperatorCommandOutput, VerificationClaimPolicy,
        exit_if_blocked_operator_status,
    },
    operator_loop::primary_next_action,
    ready_cmd::worktree_dirty,
    snapshot::{SnapshotAgentOverrides, create_snapshot},
    thread_cmd::{
        current_thread, load_thread, refresh_thread, refresh_thread_freshness, thread_manager,
        thread_not_found_advice,
    },
    thread_landing::{land_local_command, switch_thread_command},
    worktree_safety::ensure_worktree_clean,
};
use crate::{
    bridge::GitBridge,
    cli::{
        Cli,
        cli_args::{LandArgs, SyncArgs},
        output_is_compact, should_output_json, style, worktree_status_options,
    },
    config::UserConfig,
};

#[derive(Serialize)]
struct SyncOutput {
    #[serde(flatten)]
    operator: OperatorCommandOutput,
    #[serde(skip_serializing)]
    #[serde(rename = "verification")]
    trust: RepositoryVerificationState,
    thread: String,
    current_state: Option<String>,
    chosen_path: String,
}

#[derive(Serialize)]
struct LandOutput {
    #[serde(flatten)]
    operator: OperatorCommandOutput,
    thread: String,
    captured: bool,
    checkpointed: bool,
    git_commit: Option<String>,
    synced: bool,
    integrated: bool,
    pushed: bool,
    pushed_remote: Option<String>,
    performed_steps: Vec<String>,
    skipped_steps: Vec<String>,
    merge_state: Option<String>,
    #[serde(skip_serializing)]
    #[serde(rename = "verification")]
    trust: RepositoryVerificationState,
    chosen_path: String,
}

pub async fn cmd_sync(cli: &Cli, args: SyncArgs) -> Result<()> {
    let repo = cli.open_repo()?;
    let mut thread = resolve_thread(
        &repo,
        args.thread.as_deref(),
        "sync",
        "heddle sync --thread <name>",
    )?;

    let stale_report = build_thread_preview_report(&repo, &mut thread, true)?;
    let stale_blockers = non_staleness_blockers(&stale_report.blockers);
    let operation = repo.operation_status()?;
    let remote_tracking = repo.git_remote_tracking_status()?;
    let import_hint = repo.git_overlay_import_hint()?;
    let mut output = if thread.freshness == repo::ThreadFreshness::Current {
        let recommended_action = primary_next_action(
            operation.as_ref(),
            remote_tracking.as_ref(),
            import_hint.as_ref(),
            Some(&land_local_command(&thread.id)),
        );
        let trust = build_repository_verification_state(&repo);
        SyncOutput {
            operator: OperatorCommandOutput {
                status: "current".to_string(),
                action: OperatorAction::Sync,
                message: format!("Thread '{}' is already current", thread.id),
                blockers: vec![],
                warnings: Vec::new(),
                next_action: Some(recommended_action.clone()),
                recommended_action: Some(recommended_action),
            },
            trust,
            thread: thread.id.clone(),
            current_state: thread.current_state.clone(),
            chosen_path: "no_op".to_string(),
        }
    } else if stale_report.conflict_count == 0 && !stale_blockers.is_empty() {
        // Genuine non-conflict blockers (e.g. failing verification) cannot be
        // auto-refreshed away. Surface the blocker without a refresh. (The
        // conflict case carries a "path conflict(s)" blocker too, but it is
        // routed to the refresh attempt below so the breadcrumb materializes.)
        let recommended_action = if stale_report.recommended_action.trim().is_empty()
            || stale_report.recommended_action.starts_with("heddle sync")
        {
            String::new()
        } else {
            primary_next_action(
                operation.as_ref(),
                remote_tracking.as_ref(),
                import_hint.as_ref(),
                Some(&stale_report.recommended_action),
            )
        };
        update_integration_policy(
            &repo,
            &thread.id,
            "blocked",
            stale_blockers
                .first()
                .cloned()
                .unwrap_or_else(|| "refresh requires manual resolution".to_string()),
        )?;
        let trust = build_repository_verification_state(&repo);
        SyncOutput {
            operator: OperatorCommandOutput {
                status: "blocked".to_string(),
                action: OperatorAction::Sync,
                message: format!("Thread '{}' needs manual sync", thread.id),
                blockers: stale_report.blockers.clone(),
                warnings: Vec::new(),
                next_action: non_empty_next_action(&recommended_action),
                recommended_action: non_empty_next_action(&recommended_action),
            },
            trust,
            thread: thread.id.clone(),
            current_state: thread.current_state.clone(),
            chosen_path: "blocked".to_string(),
        }
    } else {
        // Either a clean stale thread or one whose replay genuinely conflicts
        // (conflict_count > 0). Attempt the refresh in both cases.
        // `refresh_thread` persists the merge state + worktree conflict
        // markers in the thread's checkout *before* returning the conflict
        // advice, so the `resolve` breadcrumb below points at real state.
        // heddle#464 r2: the old conflict branch returned here early and
        // emitted `heddle resolve --list` with no merge in progress — a dead
        // breadcrumb that failed with `no_merge_in_progress`. A previewed
        // conflict that the 3-way merge resolves cleanly also completes here
        // and recommends `land`.
        match refresh_thread(&repo, &thread.id, cli) {
            Ok(refreshed) => {
                update_integration_policy(
                    &repo,
                    &refreshed.id,
                    "current",
                    "thread refreshed cleanly",
                )?;
                let recommended_action = primary_next_action(
                    operation.as_ref(),
                    remote_tracking.as_ref(),
                    import_hint.as_ref(),
                    Some(&land_local_command(&refreshed.id)),
                );
                let trust = build_repository_verification_state(&repo);
                SyncOutput {
                    operator: OperatorCommandOutput {
                        status: "refreshed".to_string(),
                        action: OperatorAction::Sync,
                        message: format!("Refreshed thread '{}'", refreshed.id),
                        blockers: vec![],
                        warnings: Vec::new(),
                        next_action: Some(recommended_action.clone()),
                        recommended_action: Some(recommended_action),
                    },
                    trust,
                    thread: refreshed.id.clone(),
                    current_state: refreshed.current_state.clone(),
                    chosen_path: "refresh".to_string(),
                }
            }
            Err(error) => {
                // refresh_thread materializes the conflict before returning;
                // only then is `resolve` a live breadcrumb. If no merge was
                // materialized the failure is genuine — propagate it.
                if !sync_conflict_merge_in_progress(&repo, &thread) {
                    return Err(error);
                }
                update_integration_policy(
                    &repo,
                    &thread.id,
                    "blocked",
                    "refresh produced conflicts requiring manual resolution",
                )?;
                let recommended_action = scoped_resolve_list_command(&thread);
                let trust = build_repository_verification_state(&repo);
                SyncOutput {
                    operator: OperatorCommandOutput {
                        status: "blocked".to_string(),
                        action: OperatorAction::Sync,
                        message: format!("Thread '{}' has merge conflicts to resolve", thread.id),
                        blockers: stale_report.blockers.clone(),
                        warnings: Vec::new(),
                        next_action: Some(recommended_action.clone()),
                        recommended_action: Some(recommended_action),
                    },
                    trust,
                    thread: thread.id.clone(),
                    current_state: thread.current_state.clone(),
                    chosen_path: "blocked".to_string(),
                }
            }
        }
    };
    output.operator.block_success_claim_if_verification_blocked(
        &output.trust,
        "sync",
        VerificationClaimPolicy::strict(),
    );

    write_sync_output(cli, &repo, &output)
}

pub async fn cmd_land(cli: &Cli, args: LandArgs) -> Result<()> {
    // Open at CWD only to discover the active thread, then re-open at
    // its metadata-recorded worktree. This makes `heddle land` work
    // from anywhere — operators don't need to `cd` into a lightweight
    // thread directory before landing. The capture/merge below run
    // against `repo`, so they all see the same checkout. See
    // `Repository::active_worktree_path`.
    let cwd_repo = cli.open_repo()?;
    let target_path = cwd_repo.active_worktree_path()?;
    let repo = if target_path == *cwd_repo.root() {
        cwd_repo
    } else {
        Repository::open(&target_path)?
    };
    let user_config = UserConfig::load_default().unwrap_or_default();
    let thread = resolve_thread(
        &repo,
        args.thread.as_deref(),
        "land",
        "heddle land --thread <name>",
    )?;
    let thread_repo = if thread.execution_path.as_os_str().is_empty() {
        None
    } else if thread.execution_path.exists() {
        Some(Repository::open(&thread.execution_path).with_context(|| {
            format!(
                "opening thread '{}' worktree at {}",
                thread.id,
                thread.execution_path.display()
            )
        })?)
    } else {
        let land_command = land_local_command(&thread.id);
        // `heddle start` would refuse here — the thread still holds an active
        // reservation, so it returns `active_reservation_advice` and the
        // operator is stuck. `heddle switch` rebuilds the dedicated worktree at
        // the recorded `execution_path` from the thread's current state (see
        // `cmd_thread_switch`), which is exactly the path this `land` reads, so
        // the rebuild clears the blocker and the follow-up `land` succeeds.
        let switch_command = switch_thread_command(&thread.id);
        return Err(anyhow!(RecoveryAdvice::safety_refusal(
            "thread_worktree_missing",
            format!("Thread '{}' worktree is missing", thread.id),
            format!(
                "Rebuild the thread's checkout with `{switch_command}` (it re-materializes the recorded worktree from the thread's current state), then retry `{land_command}`.",
            ),
            format!(
                "recorded execution path does not exist: {}",
                thread.execution_path.display()
            ),
            "land would need to inspect that checkout for unsaved work before merging",
            "repository state, refs, metadata, and worktree files were left unchanged",
            switch_command.clone(),
            vec![switch_command, land_command],
        )));
    };
    if args.push && args.no_push {
        return Err(anyhow!(RecoveryAdvice::land_push_option_conflict(
            &thread.id
        )));
    }
    if let Some(remote) = args.remote.as_deref()
        && !args.push
    {
        return Err(anyhow!(RecoveryAdvice::land_remote_requires_push(
            &thread.id, remote,
        )));
    }
    let should_push = args.push;
    let planned_push_remote = if should_push {
        match args
            .remote
            .clone()
            .or(super::remote::resolved_default_remote_name(&repo)?)
        {
            Some(remote) => Some(remote),
            None => {
                return Err(anyhow!(RecoveryAdvice::land_push_remote_missing(
                    &thread.id
                )));
            }
        }
    } else {
        None
    };
    let remote_synced = sync_remote_before_land_if_needed(&repo, &thread.id)?;
    git_overlay_txn::preflight_land_checkpoint(&repo, &thread.id)?;

    let mut captured = false;
    if let Some(thread_repo) = thread_repo.as_ref() {
        let status_options = worktree_status_options(Some(thread_repo.config()));
        if worktree_dirty(thread_repo, &status_options)? {
            let capture_message = args
                .message
                .clone()
                .or_else(|| Some(format!("Land {}", thread.id)));
            create_snapshot(
                thread_repo,
                &user_config,
                capture_message,
                None,
                SnapshotAgentOverrides {
                    provider: None,
                    model: None,
                    session: None,
                    segment: None,
                    policy: None,
                    no_policy: false,
                    no_agent: false,
                },
            )?;
            captured = true;
        }
    }

    let mut synced = remote_synced;
    let mut refreshed_thread = resolve_thread(
        &repo,
        Some(&thread.id),
        "land",
        "heddle land --thread <name>",
    )?;
    refresh_thread_freshness(&repo, &mut refreshed_thread)?;
    if refreshed_thread.freshness == repo::ThreadFreshness::Stale {
        let preview = build_thread_preview_report(&repo, &mut refreshed_thread, true)?;
        let stale_blockers = non_staleness_blockers(&preview.blockers);
        if preview.conflict_count == 0 && !stale_blockers.is_empty() {
            update_integration_policy(
                &repo,
                &refreshed_thread.id,
                "blocked",
                stale_blockers
                    .first()
                    .cloned()
                    .unwrap_or_else(|| "sync requires manual resolution".to_string()),
            )?;
            return write_land_output(
                cli,
                &repo,
                &LandOutput {
                    operator: OperatorCommandOutput {
                        status: "blocked".to_string(),
                        action: OperatorAction::Land,
                        message: format!(
                            "Thread '{}' must be synced manually",
                            refreshed_thread.id
                        ),
                        blockers: land_blockers_for_preview(&preview, &stale_blockers),
                        warnings: Vec::new(),
                        next_action: Some(format!(
                            "heddle sync {}",
                            thread_flag(&refreshed_thread.id)
                        )),
                        recommended_action: Some(format!(
                            "heddle sync {}",
                            thread_flag(&refreshed_thread.id)
                        )),
                    },
                    thread: refreshed_thread.id.clone(),
                    captured,
                    checkpointed: false,
                    git_commit: None,
                    synced: false,
                    integrated: false,
                    pushed: false,
                    pushed_remote: None,
                    merge_state: None,
                    trust: build_repository_verification_state(&repo),
                    chosen_path: "blocked".to_string(),
                    performed_steps: land_performed_steps(captured, false, false, false, false),
                    skipped_steps: land_skipped_steps(captured, false, false, false, false),
                },
            );
        }

        match refresh_thread(&repo, &refreshed_thread.id, cli) {
            Ok(refreshed) => {
                update_integration_policy(
                    &repo,
                    &refreshed.id,
                    "current",
                    "thread synced during land",
                )?;
                refreshed_thread = refreshed;
                synced = true;
            }
            Err(error) => {
                if !sync_conflict_merge_in_progress(&repo, &refreshed_thread) {
                    return Err(error);
                }
                update_integration_policy(
                    &repo,
                    &refreshed_thread.id,
                    "blocked",
                    "land sync produced conflicts requiring manual resolution",
                )?;
                let recommended_action = scoped_resolve_list_command(&refreshed_thread);
                return write_land_output(
                    cli,
                    &repo,
                    &LandOutput {
                        operator: OperatorCommandOutput {
                            status: "blocked".to_string(),
                            action: OperatorAction::Land,
                            message: format!(
                                "Thread '{}' has merge conflicts to resolve",
                                refreshed_thread.id
                            ),
                            blockers: land_blockers_for_preview(&preview, &stale_blockers),
                            warnings: Vec::new(),
                            next_action: Some(recommended_action.clone()),
                            recommended_action: Some(recommended_action),
                        },
                        thread: refreshed_thread.id.clone(),
                        captured,
                        checkpointed: false,
                        git_commit: None,
                        synced: false,
                        integrated: false,
                        pushed: false,
                        pushed_remote: None,
                        merge_state: None,
                        trust: build_repository_verification_state(&repo),
                        chosen_path: "blocked".to_string(),
                        performed_steps: land_performed_steps(
                            captured, synced, false, false, false,
                        ),
                        skipped_steps: land_skipped_steps(captured, synced, false, false, false),
                    },
                );
            }
        }
    }

    let mut merge_thread = resolve_thread(
        &repo,
        Some(&refreshed_thread.id),
        "land",
        "heddle land --thread <name>",
    )?;
    let preview = build_thread_preview_report(&repo, &mut merge_thread, true)?;
    let preview_warnings = land_warnings_for_preview(&preview);
    let integration_blockers = integration_blockers(&repo, &merge_thread, &preview);
    let manual_resolution_current = manual_resolution_current(&repo, &merge_thread);
    let squash_land = should_squash_land(&args, &user_config);
    if manual_resolution_current {
        let land_collapse_state = if squash_land
            && repo.capability() == repo::RepositoryCapability::GitOverlay
        {
            collapse_thread_for_land(&repo, &user_config, &merge_thread, args.message.as_deref())?
        } else {
            None
        };
        if land_collapse_state.is_some() {
            merge_thread = resolve_thread(
                &repo,
                Some(&merge_thread.id),
                "land",
                "heddle land --thread <name>",
            )?;
        }
        let merge_state = adopt_manual_resolution(&repo, &merge_thread.id)?;
        let mut checkpointed = false;
        let mut git_commit = None;
        update_integration_policy(
            &repo,
            &merge_thread.id,
            "auto_integrated",
            "accepted manually resolved integration state",
        )?;
        if repo.capability() == repo::RepositoryCapability::GitOverlay {
            let checkpoint_message = land_checkpoint_message(
                &repo,
                &merge_thread,
                args.message.as_deref(),
                land_collapse_state.is_some(),
            );
            let record = create_git_checkpoint(
                &repo,
                Some(&checkpoint_message),
                worktree_status_options(Some(repo.config())),
            )
            .map_err(|error| {
                anyhow!(RecoveryAdvice::land_checkpoint_partial_failure(
                    &merge_thread.id,
                    error,
                    land_performed_steps(captured, synced, true, false, false),
                ))
            })?;
            checkpointed = true;
            git_commit = Some(record.git_commit);
        }
        coalesce_land_integration_and_checkpoint(
            &repo,
            Some(&merge_state),
            git_commit.as_deref(),
            land_collapse_state.as_ref(),
        )
        .context(
            "land completed but failed to record manual integration and Git checkpoint as one undo batch",
        )?;
        let mut pushed = false;
        let mut pushed_remote = None;
        if should_push {
            let remote_name = push_after_land(
                cli,
                &repo,
                planned_push_remote.clone(),
                Some(merge_state.clone()),
            )
            .await
            .map_err(|error| {
                anyhow!(RecoveryAdvice::land_push_partial_failure(
                    &merge_thread.id,
                    error,
                    land_performed_steps(captured, synced, true, checkpointed, false),
                    git_commit.as_deref(),
                    planned_push_remote.as_deref(),
                ))
            })?;
            pushed = true;
            pushed_remote = Some(remote_name);
        }
        let resolved_manually = merge_thread
            .integration_policy_result
            .conflicts_resolved_manually;
        clear_manual_resolution_state(&repo, &merge_thread.id)?;
        let trust = git_overlay_txn::post_verify(&repo);
        let post_land_action = integrated_land_next_action(true, pushed, &trust);
        let mut operator = OperatorCommandOutput {
            status: "landed".to_string(),
            action: OperatorAction::Land,
            // Both genuine manual resolutions and fully-automatic conflict-free
            // integrations are adopted through this `manual_resolution_state`
            // branch (e.g. two threads forked from the same base that touch
            // disjoint files merge cleanly via a 3-way merge, yet still record a
            // resolution state to mark the thread land-ready). Only the former
            // should be reported as "manually resolved" — claiming a manual
            // resolution for an auto-clean merge is misleading.
            message: if resolved_manually {
                format!(
                    "Landed thread '{}' from a manually resolved integration state",
                    merge_thread.id
                )
            } else {
                format!(
                    "Landed thread '{}' via an automatic integration merge",
                    merge_thread.id
                )
            },
            blockers: Vec::new(),
            warnings: preview_warnings.clone(),
            next_action: post_land_action.clone(),
            recommended_action: post_land_action,
        };
        operator.block_success_claim_if_verification_blocked(
            &trust,
            "land",
            VerificationClaimPolicy::strict().allow_land_publish_followup(),
        );
        return write_land_output(
            cli,
            &repo,
            &LandOutput {
                operator,
                thread: merge_thread.id.clone(),
                captured,
                checkpointed,
                git_commit,
                synced,
                integrated: true,
                pushed,
                pushed_remote,
                merge_state: Some(merge_state),
                trust,
                performed_steps: land_performed_steps(captured, synced, true, checkpointed, pushed),
                skipped_steps: land_skipped_steps(captured, synced, true, checkpointed, pushed),
                chosen_path: if checkpointed {
                    if pushed {
                        "capture_sync_manual_resolution_checkpoint_push".to_string()
                    } else {
                        "capture_sync_manual_resolution_checkpoint".to_string()
                    }
                } else {
                    "capture_sync_manual_resolution".to_string()
                },
            },
        );
    }
    if preview.conflict_count > 0 || !integration_blockers.is_empty() {
        let reason = integration_blockers
            .first()
            .cloned()
            .unwrap_or_else(|| "integration requires manual review".to_string());
        let recovery_scope = recovery_scope_checkout(&merge_thread, repo.root());
        let policy_recovery_action = integration_blocker_recommended_action(
            &integration_blockers,
            recovery_scope.as_deref(),
        );
        if preview.conflict_count > 0
            && policy_recovery_action.is_none()
            && materialize_land_conflict_for_thread(&repo, &merge_thread)?
        {
            update_integration_policy(&repo, &merge_thread.id, "blocked", &reason)?;
            let recommended_action = scoped_resolve_list_command(&merge_thread);
            return write_land_output(
                cli,
                &repo,
                &LandOutput {
                    operator: OperatorCommandOutput {
                        status: "blocked".to_string(),
                        action: OperatorAction::Land,
                        message: format!(
                            "Thread '{}' has merge conflicts to resolve",
                            merge_thread.id
                        ),
                        blockers: land_blockers_for_preview(&preview, &integration_blockers),
                        warnings: preview_warnings.clone(),
                        next_action: Some(recommended_action.clone()),
                        recommended_action: Some(recommended_action),
                    },
                    thread: merge_thread.id.clone(),
                    captured,
                    checkpointed: false,
                    git_commit: None,
                    synced: false,
                    integrated: false,
                    pushed: false,
                    pushed_remote: None,
                    merge_state: None,
                    trust: build_repository_verification_state(&repo),
                    chosen_path: "blocked".to_string(),
                    performed_steps: land_performed_steps(captured, synced, false, false, false),
                    skipped_steps: land_skipped_steps(captured, synced, false, false, false),
                },
            );
        }
        // Never fall back to `preview.recommended_action` here: this is the
        // pre-merge bail, so a preview-originated `resolve` breadcrumb could
        // die with `no_merge_in_progress`, and the preview's own land
        // recommendation would self-loop this very command. When materializing
        // was not possible, drive the operator through the explicit sync path.
        let recommended_action = policy_recovery_action
            .unwrap_or_else(|| format!("heddle sync {}", thread_flag(&merge_thread.id)));
        update_integration_policy(&repo, &merge_thread.id, "blocked", &reason)?;
        return write_land_output(
            cli,
            &repo,
            &LandOutput {
                operator: OperatorCommandOutput {
                    status: "blocked".to_string(),
                    action: OperatorAction::Land,
                    message: format!("Thread '{}' is not eligible for auto-land", merge_thread.id),
                    blockers: land_blockers_for_preview(&preview, &integration_blockers),
                    warnings: preview_warnings.clone(),
                    next_action: Some(recommended_action.clone()),
                    recommended_action: Some(recommended_action),
                },
                thread: merge_thread.id.clone(),
                captured,
                checkpointed: false,
                git_commit: None,
                synced,
                integrated: false,
                pushed: false,
                pushed_remote: None,
                merge_state: None,
                trust: build_repository_verification_state(&repo),
                chosen_path: "blocked".to_string(),
                performed_steps: land_performed_steps(captured, synced, false, false, false),
                skipped_steps: land_skipped_steps(captured, synced, false, false, false),
            },
        );
    }

    let land_collapse_state =
        if squash_land && repo.capability() == repo::RepositoryCapability::GitOverlay {
            collapse_thread_for_land(&repo, &user_config, &merge_thread, args.message.as_deref())?
        } else {
            None
        };
    if land_collapse_state.is_some() {
        merge_thread = resolve_thread(
            &repo,
            Some(&merge_thread.id),
            "land",
            "heddle land --thread <name>",
        )?;
    }

    let merge_output = merge_thread_into_current(
        &repo,
        &merge_thread.id,
        None,
        false,
        false,
        false,
        false,
        false,
    )?;
    let integrated = merge_output.conflicts.is_empty() && merge_output.merge_state.is_some();
    let mut checkpointed = false;
    let mut git_commit = None;
    update_integration_policy(
        &repo,
        &merge_thread.id,
        if integrated {
            "auto_integrated"
        } else {
            "blocked"
        },
        if integrated {
            "clean integration path"
        } else {
            "merge produced conflicts"
        },
    )?;

    if integrated && repo.capability() == repo::RepositoryCapability::GitOverlay {
        let checkpoint_message = land_checkpoint_message(
            &repo,
            &merge_thread,
            args.message.as_deref(),
            land_collapse_state.is_some(),
        );
        let record = create_git_checkpoint(
            &repo,
            Some(&checkpoint_message),
            worktree_status_options(Some(repo.config())),
        )
        .map_err(|error| {
            anyhow!(RecoveryAdvice::land_checkpoint_partial_failure(
                &merge_thread.id,
                error,
                land_performed_steps(captured, synced, integrated, false, false),
            ))
        })?;
        checkpointed = true;
        git_commit = Some(record.git_commit);
    }
    coalesce_land_integration_and_checkpoint(
        &repo,
        merge_output.merge_state.as_deref(),
        git_commit.as_deref(),
        land_collapse_state.as_ref(),
    )
    .context("land completed but failed to record merge and Git checkpoint as one undo batch")?;

    let mut pushed = false;
    let mut pushed_remote = None;
    if integrated && should_push {
        let remote_name = push_after_land(
            cli,
            &repo,
            planned_push_remote.clone(),
            merge_output.merge_state.clone(),
        )
        .await
        .map_err(|error| {
            anyhow!(RecoveryAdvice::land_push_partial_failure(
                &merge_thread.id,
                error,
                land_performed_steps(captured, synced, integrated, checkpointed, false),
                git_commit.as_deref(),
                planned_push_remote.as_deref(),
            ))
        })?;
        pushed = true;
        pushed_remote = Some(remote_name);
    }

    if integrated {
        clear_manual_resolution_state(&repo, &merge_thread.id)?;
    }

    let trust = git_overlay_txn::post_verify(&repo);
    let integrated_next_action = integrated_land_next_action(integrated, pushed, &trust);
    let mut operator = OperatorCommandOutput {
        status: if integrated { "landed" } else { "blocked" }.to_string(),
        action: OperatorAction::Land,
        message: if integrated {
            format!("Landed thread '{}'", merge_thread.id)
        } else {
            format!("Thread '{}' could not be landed cleanly", merge_thread.id)
        },
        blockers: merge_output.operator.blockers.clone(),
        warnings: preview_warnings,
        next_action: if integrated {
            integrated_next_action.clone()
        } else {
            merge_output.operator.recommended_action.clone()
        },
        recommended_action: if integrated {
            integrated_next_action
        } else {
            merge_output.operator.recommended_action.clone()
        },
    };
    operator.block_success_claim_if_verification_blocked(
        &trust,
        "land",
        VerificationClaimPolicy::strict().allow_land_publish_followup(),
    );

    write_land_output(
        cli,
        &repo,
        &LandOutput {
            operator,
            thread: merge_thread.id.clone(),
            captured,
            checkpointed,
            git_commit,
            synced,
            integrated,
            pushed,
            pushed_remote,
            merge_state: merge_output.merge_state.clone(),
            trust,
            performed_steps: land_performed_steps(
                captured,
                synced,
                integrated,
                checkpointed,
                pushed,
            ),
            skipped_steps: land_skipped_steps(captured, synced, integrated, checkpointed, pushed),
            chosen_path: if integrated {
                if pushed {
                    "capture_sync_merge_checkpoint_push"
                } else if checkpointed {
                    "capture_sync_merge_checkpoint"
                } else {
                    "capture_sync_merge"
                }
                .to_string()
            } else {
                "blocked".to_string()
            },
        },
    )
}

fn should_squash_land(args: &LandArgs, user_config: &UserConfig) -> bool {
    !args.no_squash && user_config.land.squash
}

fn sync_remote_before_land_if_needed(repo: &Repository, thread_id: &str) -> Result<bool> {
    let Some(remote) = repo.git_remote_tracking_status()? else {
        return Ok(false);
    };
    let remote_decision = remote_drift_decision(repo, &remote);
    if remote_decision.status != "remote_behind" {
        return Ok(false);
    }

    ensure_worktree_clean(repo, "land")?;
    let remote_name = super::remote::resolve_default_remote_name(repo, None)?;
    let mut bridge = GitBridge::new(repo);
    bridge.pull(&remote_name)?;

    let trust = git_overlay_txn::post_verify(repo);
    if !trust.verified {
        let primary_command = if trust.recommended_action.trim().is_empty() {
            "heddle status".to_string()
        } else {
            trust.recommended_action.clone()
        };
        return Err(anyhow!(RecoveryAdvice::safety_refusal(
            "land_remote_sync_blocked",
            format!(
                "Synced remote state before landing '{thread_id}', but repository verification is still blocked"
            ),
            format!("Run `{primary_command}`, then retry the land."),
            format!(
                "repository verification reports {}: {}",
                trust.status, trust.summary
            ),
            "land must not continue into integration while repository verification is blocked",
            "remote state was imported; thread refs and worktree changes from the land were left unchanged",
            primary_command.clone(),
            vec![primary_command],
        )));
    }

    Ok(true)
}

fn materialize_land_conflict_for_thread(repo: &Repository, thread: &Thread) -> Result<bool> {
    let Some(target_thread) = thread.target_thread.as_deref() else {
        return Ok(false);
    };

    if thread.execution_path.as_os_str().is_empty() {
        if repo.current_lane()?.as_deref() != Some(thread.thread.as_str()) {
            return Ok(false);
        }
        return materialize_land_conflict_in_repo(repo, target_thread);
    }

    if !thread.execution_path.exists() {
        return Ok(false);
    }
    let thread_repo = Repository::open(&thread.execution_path).with_context(|| {
        format!(
            "opening thread '{}' worktree at {} to materialize land conflict",
            thread.id,
            thread.execution_path.display()
        )
    })?;
    materialize_land_conflict_in_repo(&thread_repo, target_thread)
}

fn materialize_land_conflict_in_repo(repo: &Repository, target_thread: &str) -> Result<bool> {
    let output =
        merge_thread_into_current(repo, target_thread, None, false, false, false, false, false)?;
    Ok(!output.conflicts.is_empty() && repo.merge_state_manager().is_merge_in_progress())
}

fn collapse_thread_for_land(
    repo: &Repository,
    user_config: &UserConfig,
    thread: &Thread,
    message: Option<&str>,
) -> Result<Option<ChangeId>> {
    let sources = thread_source_states(repo, thread)?;
    if sources.len() <= 1 {
        return Ok(None);
    }
    let intent = message
        .filter(|message| !message.trim().is_empty())
        .map(ToOwned::to_owned)
        .unwrap_or_else(|| format!("Land {}", thread.id));
    let result = collapse_resolved_states(
        repo,
        user_config,
        &sources,
        intent,
        None,
        CollapsePublishedRef::Thread(ThreadName::new(&thread.id)),
    )?;
    Ok(Some(result.change_id))
}

fn thread_source_states(repo: &Repository, thread: &Thread) -> Result<Vec<State>> {
    let Some(tip) = repo.refs().get_thread(&ThreadName::new(&thread.id))? else {
        return Ok(Vec::new());
    };
    let base = repo.resolve_state(&thread.base_state)?;
    let base_reachable = match base {
        Some(base) => reachable_state_set(repo, base)?,
        None => HashSet::new(),
    };
    let mut visited = HashSet::new();
    let mut ordered = Vec::new();
    collect_thread_sources(repo, tip, &base_reachable, &mut visited, &mut ordered)?;
    Ok(ordered)
}

fn collect_thread_sources(
    repo: &Repository,
    state_id: ChangeId,
    excluded: &HashSet<ChangeId>,
    visited: &mut HashSet<ChangeId>,
    ordered: &mut Vec<State>,
) -> Result<()> {
    if excluded.contains(&state_id) || !visited.insert(state_id) {
        return Ok(());
    }
    let Some(state) = repo.store().get_state(&state_id)? else {
        return Ok(());
    };
    for parent in &state.parents {
        collect_thread_sources(repo, *parent, excluded, visited, ordered)?;
    }
    ordered.push(state);
    Ok(())
}

fn reachable_state_set(repo: &Repository, root: ChangeId) -> Result<HashSet<ChangeId>> {
    let mut reachable = HashSet::new();
    let mut stack = vec![root];
    while let Some(state_id) = stack.pop() {
        if !reachable.insert(state_id) {
            continue;
        }
        if let Some(state) = repo.store().get_state(&state_id)? {
            stack.extend(state.parents.iter().copied());
        }
    }
    Ok(reachable)
}

async fn push_after_land(
    cli: &Cli,
    repo: &Repository,
    remote: Option<String>,
    state: Option<String>,
) -> Result<String> {
    if repo.capability() == repo::RepositoryCapability::GitOverlay && !repo.hosted_enabled() {
        let (remote_name, _, _, _, _, _) =
            super::remote::push_git_overlay_refs(repo, remote.as_deref(), false, false)?;
        Ok(remote_name)
    } else {
        let pushed_remote = remote
            .clone()
            .or(super::remote::resolved_default_remote_name(repo)?)
            .unwrap_or_else(|| "default".to_string());
        super::remote::cmd_push(cli, remote, None, state, false, false, None).await?;
        Ok(pushed_remote)
    }
}

fn land_performed_steps(
    captured: bool,
    synced: bool,
    integrated: bool,
    checkpointed: bool,
    pushed: bool,
) -> Vec<String> {
    [
        (captured, "capture"),
        (synced, "sync"),
        (integrated, "merge"),
        (checkpointed, "checkpoint"),
        (pushed, "push"),
    ]
    .into_iter()
    .filter(|&(done, _step)| done)
    .map(|(_done, step)| step.to_string())
    .collect()
}

fn land_skipped_steps(
    captured: bool,
    synced: bool,
    integrated: bool,
    checkpointed: bool,
    pushed: bool,
) -> Vec<String> {
    [
        (!captured, "capture(no changes)"),
        (!synced, "sync(current)"),
        (!integrated, "merge(blocked)"),
        (!checkpointed && integrated, "checkpoint(not needed)"),
        (!checkpointed && !integrated, "checkpoint(not reached)"),
        (!pushed && integrated, "push(not requested)"),
        (!pushed && !integrated, "push(not reached)"),
    ]
    .into_iter()
    .filter(|&(skipped, _step)| skipped)
    .map(|(_skipped, step)| step.to_string())
    .collect()
}

fn integrated_land_next_action(
    integrated: bool,
    pushed: bool,
    trust: &RepositoryVerificationState,
) -> Option<String> {
    if !integrated {
        return None;
    }
    if !pushed && trust.recommended_action == "heddle push" {
        Some(trust.recommended_action.clone())
    } else {
        Some("heddle thread cleanup --merged --dry-run".to_string())
    }
}

fn land_checkpoint_message(
    repo: &Repository,
    thread: &Thread,
    explicit: Option<&str>,
    prefer_land_subject: bool,
) -> String {
    if let Some(message) = explicit.filter(|message| !message.trim().is_empty()) {
        return message.to_string();
    }
    if prefer_land_subject {
        return format!("Land {}", thread.id);
    }
    if let Some(intent) = thread
        .current_state
        .as_deref()
        .and_then(|state| repo.resolve_state(state).ok().flatten())
        .and_then(|state_id| repo.store().get_state(&state_id).ok().flatten())
        .and_then(|state| state.intent)
        .filter(|intent| !intent.trim().is_empty())
    {
        return intent;
    }
    if let Some(task) = thread
        .task
        .as_deref()
        .filter(|task| !task.trim().is_empty())
    {
        return task.to_string();
    }
    format!("Land {}", thread.id)
}

fn resolve_thread(
    repo: &Repository,
    thread: Option<&str>,
    command: &'static str,
    primary_command: impl Into<String>,
) -> Result<Thread> {
    match thread {
        Some(thread) => load_thread(repo, thread),
        None => current_thread(repo)?.ok_or_else(|| {
            anyhow!(RecoveryAdvice::no_current_thread(
                command,
                Some("--thread"),
                primary_command,
            ))
        }),
    }
}

fn update_integration_policy(
    repo: &Repository,
    thread_id: &str,
    status: &str,
    reason: impl Into<String>,
) -> Result<()> {
    let manager = thread_manager(repo);
    let mut thread = manager.load(thread_id)?.ok_or_else(|| {
        anyhow!(thread_not_found_advice(
            thread_id,
            "update integration policy"
        ))
    })?;
    let prior_status = thread.integration_policy_result.status.clone();
    let reason = reason.into();
    let keep_previewed = status == "blocked" && prior_status.as_deref() == Some("previewed");
    let next_status = if keep_previewed { "previewed" } else { status };
    let next_reason = if keep_previewed {
        format!("auto-land blocked: {reason}")
    } else {
        reason
    };
    if status == "blocked" {
        thread.state = repo::ThreadState::Blocked;
    }
    thread.integration_policy_result = ThreadIntegrationPolicy {
        status: Some(next_status.to_string()),
        reason: Some(next_reason),
        manual_resolution_state: thread.integration_policy_result.manual_resolution_state,
        conflicts_resolved_manually: thread.integration_policy_result.conflicts_resolved_manually,
    };
    manager.save(&thread)?;
    Ok(())
}

fn clear_manual_resolution_state(repo: &Repository, thread_id: &str) -> Result<()> {
    let manager = thread_manager(repo);
    let mut thread = manager.load(thread_id)?.ok_or_else(|| {
        anyhow!(thread_not_found_advice(
            thread_id,
            "clear manual resolution"
        ))
    })?;
    thread.integration_policy_result.manual_resolution_state = None;
    thread.integration_policy_result.conflicts_resolved_manually = false;
    Ok(manager.save(&thread)?)
}

fn coalesce_land_integration_and_checkpoint(
    repo: &Repository,
    merge_state: Option<&str>,
    git_commit: Option<&str>,
    collapse_state: Option<&ChangeId>,
) -> Result<()> {
    let Some(merge_state) = merge_state else {
        return Ok(());
    };
    let Some(git_commit) = git_commit else {
        return Ok(());
    };

    let integration_batch = find_recent_land_integration_batch(repo, merge_state)?;
    let checkpoint_batch = find_recent_land_git_checkpoint_batch(repo, git_commit)?;
    repo.oplog()
        .coalesce_batches(integration_batch.id, checkpoint_batch.id)?;
    if let Some(collapse_state) = collapse_state {
        let collapse_batch = find_recent_land_collapse_batch(repo, collapse_state)?;
        repo.oplog()
            .coalesce_batches(integration_batch.id, collapse_batch.id)?;
    }
    Ok(())
}

fn find_recent_land_collapse_batch(
    repo: &Repository,
    collapse_state: &ChangeId,
) -> Result<OpBatch> {
    repo.oplog()
        .recent_batches_scoped(12, Some(&repo.op_scope()))?
        .into_iter()
        .find(|batch| {
            batch.entries.iter().any(|entry| {
                matches!(
                    &entry.operation,
                    OpRecord::Collapse { result, .. } if result == collapse_state
                )
            })
        })
        .ok_or_else(|| anyhow!("land squash succeeded but its oplog batch was not found"))
}

fn find_recent_land_integration_batch(repo: &Repository, merge_state: &str) -> Result<OpBatch> {
    repo.oplog()
        .recent_batches_scoped(12, Some(&repo.op_scope()))?
        .into_iter()
        .find(|batch| {
            batch
                .entries
                .iter()
                .any(|entry| op_targets_merge_state(&entry.operation, merge_state))
        })
        .ok_or_else(|| anyhow!("land merge succeeded but its oplog batch was not found"))
}

fn find_recent_land_git_checkpoint_batch(repo: &Repository, git_commit: &str) -> Result<OpBatch> {
    repo.oplog()
        .recent_batches_scoped(12, Some(&repo.op_scope()))?
        .into_iter()
        .find(|batch| {
            batch.entries.iter().any(|entry| {
                matches!(
                    &entry.operation,
                    OpRecord::GitCheckpoint { new_git_oid, .. } if new_git_oid == git_commit
                )
            })
        })
        .ok_or_else(|| anyhow!("land Git checkpoint succeeded but its oplog batch was not found"))
}

fn op_targets_merge_state(op: &OpRecord, merge_state: &str) -> bool {
    match op {
        OpRecord::Snapshot { new_state, .. } => change_id_matches_display(new_state, merge_state),
        OpRecord::Checkpoint { state, .. } => change_id_matches_display(state, merge_state),
        OpRecord::Goto { target, .. } => change_id_matches_display(target, merge_state),
        OpRecord::FastForward { post_target_id, .. } => {
            change_id_matches_display(post_target_id, merge_state)
        }
        // These records don't advance HEAD/thread to a merge target the land
        // flow tracks. Enumerated explicitly (no wildcard) so a new
        // state-advancing variant must be considered as a possible merge
        // target here (heddle#354 r9).
        OpRecord::ThreadCreate { .. }
        | OpRecord::ThreadDelete { .. }
        | OpRecord::ThreadUpdate { .. }
        | OpRecord::Fork { .. }
        | OpRecord::Collapse { .. }
        | OpRecord::MarkerCreate { .. }
        | OpRecord::MarkerDelete { .. }
        | OpRecord::TransactionAbort { .. }
        | OpRecord::EphemeralThreadCollapse { .. }
        | OpRecord::ConflictResolved { .. }
        | OpRecord::TransactionCommit { .. }
        | OpRecord::Redact { .. }
        | OpRecord::Purge { .. }
        | OpRecord::GitCheckpoint { .. }
        | OpRecord::RemoteThreadUpdate { .. }
        | OpRecord::RemoteThreadDelete { .. }
        | OpRecord::UndoRecoveryUpdate { .. }
        | OpRecord::StateVisibilitySet { .. }
        | OpRecord::StateVisibilityPromote { .. } => false,
    }
}

fn change_id_matches_display(id: &ChangeId, display: &str) -> bool {
    id.short() == display || id.to_string_full() == display
}

fn adopt_manual_resolution(repo: &Repository, thread_id: &str) -> Result<String> {
    let manager = thread_manager(repo);
    let mut thread = manager.load(thread_id)?.ok_or_else(|| {
        anyhow!(thread_not_found_advice(
            thread_id,
            "adopt manual resolution"
        ))
    })?;
    let target = repo
        .refs()
        .get_thread(&ThreadName::new(&thread.thread))?
        .ok_or_else(|| {
            anyhow!(
                "Thread '{}' has no current state to integrate",
                thread.thread
            )
        })?;
    super::ff_record::record_ff_advance(repo, &thread.thread, &target)?;
    thread.state = repo::ThreadState::Merged;
    thread.merged_state = Some(target.short());
    thread.current_state = Some(target.short());
    thread.updated_at = chrono::Utc::now();
    thread.freshness = repo::ThreadFreshness::Current;
    manager.save(&thread)?;
    Ok(target.short())
}

const AUTO_LAND_CONFIDENCE_THRESHOLD: f32 = 0.75;
const AUTO_LAND_CONFIDENCE_RECOVERY_ACTION: &str =
    "heddle commit -m \"...\" --confidence <confidence>";

pub(crate) fn integration_blockers(
    repo: &Repository,
    thread: &Thread,
    preview: &super::merge::ThreadPreviewReport,
) -> Vec<String> {
    let manual_resolution_current = manual_resolution_current(repo, thread);
    let mut blockers = if manual_resolution_current {
        Vec::new()
    } else {
        non_staleness_blockers(&preview.blockers)
    };
    blockers.extend(auto_land_policy_blockers(repo, thread));
    blockers
}

pub(crate) fn auto_land_policy_blockers(repo: &Repository, thread: &Thread) -> Vec<String> {
    let mut blockers = Vec::new();
    let agent_authored = thread_is_agent_authored(repo, thread);
    if agent_authored
        && let Some(confidence) = thread.confidence_summary.value
        && confidence < AUTO_LAND_CONFIDENCE_THRESHOLD
    {
        blockers.push(format!(
            "confidence {:.2} is below the auto-land threshold of 0.75",
            confidence
        ));
    }
    if matches!(thread.verification_summary.tests_passed, Some(false)) {
        blockers.push("verification summary reports failing tests".to_string());
    }
    blockers
}

pub(crate) fn integration_blocker_recommended_action(
    blockers: &[String],
    scope_to_checkout: Option<&std::path::Path>,
) -> Option<String> {
    blockers
        .iter()
        .any(|blocker| {
            blocker.starts_with("confidence ")
                || blocker == "verification summary reports failing tests"
        })
        .then(|| auto_land_confidence_recovery_action(scope_to_checkout))
}

/// The `confidence`/`verification` policy blocker is cleared by re-capturing the
/// thread's state with a fresh confidence. That capture must land in the
/// *thread's* checkout, not whatever checkout `ready`/`land` was invoked from —
/// running an unscoped `heddle commit` from the parent of an isolated
/// agent-authored thread commits the parent and never updates the blocked
/// thread. When the thread's checkout differs from the current one, scope the
/// recovery with the global `--repo` flag so the capture targets the thread.
/// (heddle#464.)
fn auto_land_confidence_recovery_action(scope_to_checkout: Option<&std::path::Path>) -> String {
    match scope_to_checkout {
        Some(path) => format!(
            "heddle --repo {} {}",
            crate::cli::render::shell_quote(&path.display().to_string()),
            AUTO_LAND_CONFIDENCE_RECOVERY_ACTION
                .strip_prefix("heddle ")
                .expect("recovery action is a heddle command"),
        ),
        None => AUTO_LAND_CONFIDENCE_RECOVERY_ACTION.to_string(),
    }
}

/// Returns the thread's recorded checkout iff it is a real, distinct path from
/// `current_checkout` — i.e. when a recovery breadcrumb that mutates thread
/// state must be re-scoped away from the current checkout. Canonicalizes both
/// sides (falling back to the raw path) so a symlinked worktree doesn't read as
/// "different" and over-scope the in-thread case.
pub(crate) fn recovery_scope_checkout(
    thread: &Thread,
    current_checkout: &std::path::Path,
) -> Option<std::path::PathBuf> {
    let execution_path = &thread.execution_path;
    if execution_path.as_os_str().is_empty() {
        return None;
    }
    let canonical =
        |path: &std::path::Path| path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
    (canonical(execution_path) != canonical(current_checkout)).then(|| execution_path.clone())
}

fn land_blockers_for_preview(
    preview: &super::merge::ThreadPreviewReport,
    blockers: &[String],
) -> Vec<String> {
    let mut out = blockers.to_vec();
    if preview.conflict_count > 0 {
        out.push(format!(
            "{} path conflict(s) need manual resolution",
            preview.conflict_count
        ));
        out.extend(
            preview
                .conflicts
                .iter()
                .map(|path| format!("conflict: {path}")),
        );
    }
    out.sort();
    out.dedup();
    out
}

fn land_warnings_for_preview(preview: &super::merge::ThreadPreviewReport) -> Vec<String> {
    let mut warnings = preview
        .blockers
        .iter()
        .filter(|blocker| is_heavy_impact_advisory(blocker))
        .cloned()
        .collect::<Vec<_>>();
    if warnings.is_empty() && !preview.heavy_impact_paths.is_empty() {
        warnings.push(format!(
            "Heavy-impact change: {} — review broader impact before merging",
            preview.heavy_impact_paths.join(", ")
        ));
    }
    warnings.sort();
    warnings.dedup();
    warnings
}

fn is_heavy_impact_advisory(blocker: &str) -> bool {
    blocker.to_lowercase().contains("heavy-impact change")
}

fn manual_resolution_current(repo: &Repository, thread: &Thread) -> bool {
    let thread_tip = repo
        .refs()
        .get_thread(&ThreadName::new(&thread.thread))
        .ok()
        .flatten()
        .map(|id| id.short());
    thread
        .integration_policy_result
        .manual_resolution_state
        .as_deref()
        .zip(thread_tip.as_deref())
        .is_some_and(|(resolved, current)| resolved == current)
        && thread.freshness == repo::ThreadFreshness::Current
}

fn thread_is_agent_authored(repo: &Repository, thread: &Thread) -> bool {
    let current_state = thread
        .current_state
        .as_deref()
        .and_then(|state| repo.resolve_state(state).ok().flatten())
        .or_else(|| {
            repo.refs()
                .get_thread(&ThreadName::new(&thread.thread))
                .ok()
                .flatten()
        });
    current_state
        .and_then(|id| repo.store().get_state(&id).ok().flatten())
        .map(|state| state.attribution.agent.is_some())
        .unwrap_or(false)
}

pub(crate) fn non_staleness_blockers(blockers: &[String]) -> Vec<String> {
    blockers
        .iter()
        .filter(|blocker| {
            !blocker.contains(" is stale against ") && !is_heavy_impact_advisory(blocker)
        })
        .cloned()
        .collect()
}

impl super::compact::CompactProjection for SyncOutput {
    fn compact(&self) -> super::compact::CompactOutput {
        <OperatorCommandOutput as super::compact::CompactProjection>::compact(&self.operator)
    }
}

impl super::compact::CompactProjection for LandOutput {
    fn compact(&self) -> super::compact::CompactOutput {
        <OperatorCommandOutput as super::compact::CompactProjection>::compact(&self.operator)
    }
}

fn write_sync_output(cli: &Cli, repo: &Repository, output: &SyncOutput) -> Result<()> {
    if should_output_json(cli, None) {
        write_command_json(
            output,
            output_is_compact(cli),
            NextActionValidationContext::new(&["sync"], repo.capability()),
        )?;
    } else {
        println!("{}", serde_json::to_string_pretty(output)?);
    }
    Ok(())
}

fn non_empty_next_action(action: &str) -> Option<String> {
    (!action.trim().is_empty()).then(|| action.to_string())
}

/// True when `refresh_thread` materialized a conflicted merge for `thread`
/// (so `heddle resolve` has state to read). The merge state lives in the
/// thread's own checkout, which may differ from the repo `sync` ran against.
fn sync_conflict_merge_in_progress(repo: &Repository, thread: &Thread) -> bool {
    if thread.execution_path.as_os_str().is_empty() {
        repo.merge_state_manager().is_merge_in_progress()
    } else if thread.execution_path.exists() {
        Repository::open(&thread.execution_path)
            .map(|worktree| worktree.merge_state_manager().is_merge_in_progress())
            .unwrap_or(false)
    } else {
        false
    }
}

/// `heddle resolve --list` scoped to wherever the conflict was materialized.
/// When the thread has its own checkout the breadcrumb must carry `--repo` so
/// it reads the merge state in that checkout rather than the repo `sync` ran
/// against (where no merge is in progress).
fn scoped_resolve_list_command(thread: &Thread) -> String {
    if thread.execution_path.as_os_str().is_empty() {
        super::command_catalog::heddle_action(["resolve", "--list"])
    } else {
        super::command_catalog::heddle_action(vec![
            "--repo".to_string(),
            thread.execution_path.display().to_string(),
            "resolve".to_string(),
            "--list".to_string(),
        ])
    }
}

fn write_land_output(cli: &Cli, repo: &Repository, output: &LandOutput) -> Result<()> {
    if should_output_json(cli, None) {
        write_command_json(
            output,
            output_is_compact(cli),
            NextActionValidationContext::new(&["land"], repo.capability()),
        )?;
    } else {
        let marker = match output.operator.status.as_str() {
            "landed" => style::ok_marker(),
            "blocked" => style::warn_marker(),
            _ => style::working_marker(),
        };
        println!("{marker} {}", output.operator.message);
        println!("  {}", style::field("thread", &style::bold(&output.thread)));
        if output.integrated {
            println!("  {}", style::field("landed", "on parent"));
            let push_status = if output.pushed {
                output
                    .pushed_remote
                    .as_deref()
                    .map(|remote| format!("pushed to {remote}"))
                    .unwrap_or_else(|| "pushed".to_string())
            } else {
                "not pushed".to_string()
            };
            println!("  {}", style::field("push", &push_status));
        } else {
            if !output.performed_steps.is_empty() {
                println!(
                    "  {}",
                    style::field(
                        "completed",
                        &output
                            .performed_steps
                            .iter()
                            .map(|step| land_text_step(step))
                            .collect::<Vec<_>>()
                            .join(", ")
                    )
                );
            }
            if !output.skipped_steps.is_empty() {
                println!(
                    "  {}",
                    style::field(
                        "up to date",
                        &output
                            .skipped_steps
                            .iter()
                            .map(|step| land_text_step(step))
                            .collect::<Vec<_>>()
                            .join(", ")
                    )
                );
            }
        }
        if output.captured {
            println!("  {}", style::field("captured", "yes"));
        }
        if output.synced {
            println!("  {}", style::field("refreshed", "yes"));
        }
        if output.checkpointed {
            println!("  {}", style::field("saved", "local Git commit recorded"));
        }
        for blocker in &output.operator.blockers {
            println!("  blocker: {}", style::warn(blocker));
        }
        for warning in &output.operator.warnings {
            println!("  warning: {}", style::warn(warning));
        }
        println!(
            "Workspace: {}",
            if output.trust.verified {
                style::accent("verified")
            } else {
                style::warn(&output.trust.status)
            }
        );
        if let Some(next) = output
            .operator
            .recommended_action
            .as_ref()
            .or(output.operator.next_action.as_ref())
        {
            print_next(next);
        }
    }
    exit_if_blocked_operator_status(&output.operator.status);
    Ok(())
}

fn land_text_step(step: &str) -> String {
    match step {
        "capture" => "saved".to_string(),
        "sync" => "refreshed".to_string(),
        "merge" => "merged".to_string(),
        "checkpoint" => "committed".to_string(),
        "push" => "pushed".to_string(),
        "capture(no changes)" => "no unsaved changes".to_string(),
        "sync(current)" => "already refreshed".to_string(),
        "merge(blocked)" => "merge blocked".to_string(),
        "checkpoint(not needed)" => "no Git commit needed".to_string(),
        "checkpoint(not reached)" => "Git commit not reached".to_string(),
        "push(not requested)" => "push not requested".to_string(),
        "push(not reached)" => "push not reached".to_string(),
        other => other.to_string(),
    }
}

#[cfg(test)]
mod tests {
    use std::path::{Path, PathBuf};

    use super::*;
    use crate::cli::commands::command_catalog::validate_recommended_action;

    fn thread_with_execution_path(execution_path: PathBuf) -> Thread {
        Thread {
            id: "agent-thread".to_string(),
            thread: "agent-thread".to_string(),
            target_thread: None,
            parent_thread: None,
            mode: repo::ThreadMode::Solid,
            state: repo::ThreadState::Active,
            base_state: "base".to_string(),
            base_root: "root".to_string(),
            current_state: Some("base".to_string()),
            merged_state: None,
            task: None,
            execution_path,
            materialized_path: None,
            changed_paths: vec![],
            impact_categories: vec![],
            heavy_impact_paths: vec![],
            promotion_suggested: false,
            freshness: repo::ThreadFreshness::Current,
            verification_summary: repo::ThreadVerificationSummary::default(),
            confidence_summary: repo::ThreadConfidenceSummary::default(),
            integration_policy_result: ThreadIntegrationPolicy::default(),
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
            ephemeral: None,
            auto: false,
            shared_target_dir: None,
        }
    }

    // heddle#464 bug 2: the confidence/verification policy-blocker recovery used
    // to be a bare `heddle commit ... --confidence`, which commits the CURRENT
    // checkout. Run from the parent of an isolated thread, that never updates the
    // blocked thread's state. Scope it to the thread's checkout via `--repo`.
    #[test]
    fn confidence_blocker_recovery_scopes_to_thread_checkout() {
        let blockers = vec!["confidence 0.40 is below the auto-land threshold of 0.75".to_string()];
        let action = integration_blocker_recommended_action(
            &blockers,
            Some(Path::new("/work/threads/agent-thread")),
        )
        .expect("a confidence blocker must yield a recovery action");
        assert_eq!(
            action,
            "heddle --repo /work/threads/agent-thread commit -m \"...\" --confidence <confidence>"
        );
        validate_recommended_action(&action)
            .unwrap_or_else(|e| panic!("scoped recovery must validate: {e}"));
    }

    #[test]
    fn verification_blocker_recovery_scopes_to_thread_checkout() {
        let blockers = vec!["verification summary reports failing tests".to_string()];
        let action = integration_blocker_recommended_action(
            &blockers,
            Some(Path::new("/work/threads/agent-thread")),
        )
        .expect("a verification blocker must yield a recovery action");
        assert_eq!(
            action,
            "heddle --repo /work/threads/agent-thread commit -m \"...\" --confidence <confidence>"
        );
        validate_recommended_action(&action)
            .unwrap_or_else(|e| panic!("scoped recovery must validate: {e}"));
    }

    // The in-thread case (recovery run from inside the thread's own checkout)
    // must stay unscoped — a `--repo` pointing back at the current checkout is
    // noise, and the bare command already targets the right state.
    #[test]
    fn confidence_blocker_recovery_stays_unscoped_in_thread() {
        let blockers = vec!["confidence 0.40 is below the auto-land threshold of 0.75".to_string()];
        let action = integration_blocker_recommended_action(&blockers, None)
            .expect("a confidence blocker must yield a recovery action");
        assert_eq!(action, AUTO_LAND_CONFIDENCE_RECOVERY_ACTION);
        validate_recommended_action(&action)
            .unwrap_or_else(|e| panic!("unscoped recovery must validate: {e}"));
    }

    #[test]
    fn non_policy_blockers_yield_no_recovery_action() {
        let blockers = vec!["3 path conflict(s) need manual resolution".to_string()];
        assert!(integration_blocker_recommended_action(&blockers, None).is_none());
    }

    #[test]
    fn heavy_impact_review_is_advisory_for_land() {
        let blockers = vec![
            "Thread 'agent-thread' is stale against 'main'".to_string(),
            "Heavy-impact change: crates/wire/src/lib.rs — review broader impact before merging"
                .to_string(),
            "confidence 0.40 is below the auto-land threshold of 0.75".to_string(),
        ];

        assert_eq!(
            non_staleness_blockers(&blockers),
            vec!["confidence 0.40 is below the auto-land threshold of 0.75".to_string()]
        );
    }

    #[test]
    fn land_warnings_surface_heavy_impact_review() {
        let preview = crate::cli::commands::merge::ThreadPreviewReport {
            thread: "agent-thread".to_string(),
            thread_mode: "solid".to_string(),
            thread_state: "active".to_string(),
            freshness: "current".to_string(),
            task: None,
            changed_paths: vec!["crates/wire/src/lib.rs".to_string()],
            changed_path_count: 1,
            impact_categories: vec![],
            heavy_impact_paths: vec!["crates/wire/src/lib.rs".to_string()],
            merge_relation: "would_merge".to_string(),
            conflicts: vec![],
            conflict_count: 0,
            blockers: vec![
                "Heavy-impact change: crates/wire/src/lib.rs — review broader impact before merging"
                    .to_string(),
            ],
            recommended_action: "heddle land --thread agent-thread --no-push".to_string(),
            recommended_action_template: None,
            thread_health: "review".to_string(),
        };

        assert_eq!(
            land_warnings_for_preview(&preview),
            vec![
                "Heavy-impact change: crates/wire/src/lib.rs — review broader impact before merging"
                    .to_string()
            ]
        );
    }

    // `recovery_scope_checkout` is the gate that decides whether to scope: an
    // isolated thread (execution_path differs from the current checkout) scopes;
    // the in-thread case (paths equal) and a worktree-less thread (empty path)
    // do not.
    #[test]
    fn recovery_scope_checkout_distinguishes_isolated_from_in_thread() {
        let isolated = thread_with_execution_path(PathBuf::from("/work/threads/agent-thread"));
        assert_eq!(
            recovery_scope_checkout(&isolated, Path::new("/work/parent")),
            Some(PathBuf::from("/work/threads/agent-thread")),
        );

        let in_thread = thread_with_execution_path(PathBuf::from("/work/threads/agent-thread"));
        assert_eq!(
            recovery_scope_checkout(&in_thread, Path::new("/work/threads/agent-thread")),
            None,
        );

        let no_worktree = thread_with_execution_path(PathBuf::new());
        assert_eq!(
            recovery_scope_checkout(&no_worktree, Path::new("/work/parent")),
            None,
        );
    }
}