main.py

from src.call_graph_edges import load_call_edges
from plugins.entry_reasoning.plugin import (
    _entry_func_source_rel,
    _make_run_copy,
)
from src.file_utils import (
    collect_file_names,
    _has_source_code,
    _get_all_phase_files,
    _write_file_names,
    _json_file_is_valid,
    _is_under_submodules,
    _ensure_resume_mode_compatible,
    clear_test_file_exemptions,
)
from src.verification import _generate_all_bugs_validation_summary
from src.extract import run_extraction, EXT_TO_LANG
from src.generate_topdown_layers import generate_topdown_layers
from src.spec_generation_and_verification import run_spec_generation_and_verification
from src.incremental_reasoner import run_incremental_pipeline
from src.git import (
    frozen_worktree,
    _is_git_repo,
    _get_head_commit,
    _record_version,
)
from src.languages.codegraph import try_codegraph_init
from src.plugin import PLUGIN_OPTION_NAMES, get_unsupported_plugin_options, run_plugin_hook
from src.pipeline_setup import (
    _run_setup_extract,
    _run_generate_phases,
    _post_process_phases,
    _run_generate_domain_context,
)
from src.domain_knowledge import (
    collect_domain_knowledge_paths,
    stage_domain_knowledge_files,
)
from src.run_estimate import (
    preserve_history_before_clean,
    record_completed_run,
    write_history,
    write_preflight_estimate,
)
from src.specification import (
    SOFTWARE_PROFILE,
    SpecificationProfile,
    SpecificationProfileSession,
    bind_profile_session,
)
from config import settings
import os
import sys
import argparse
import json
import time
import shutil
import logging
import contextlib
import subprocess
from pathlib import Path


def _clean_previous_run(work_dir):
    """Remove the fm_agent working directory from the previous pipeline run."""
    if os.path.isdir(work_dir):
        shutil.rmtree(work_dir)


def _print_preflight_summary(estimate, output_path):
    scope = estimate["scope"]
    prediction = estimate["estimate"]
    uncounted_files = scope.get("function_count_uncounted_files") or []
    included = ", ".join(scope["included_directories"]) or "(none)"
    excluded = ", ".join(
        item["path"] for item in scope["excluded_directories"][:8]
    ) or "(none)"
    if len(scope["excluded_directories"]) > 8:
        excluded += f", … +{len(scope['excluded_directories']) - 8}"
    print("[Estimate] ESTIMATE — no LLM calls were made.")
    print(f"[Estimate] Included directories: {included}")
    print(f"[Estimate] Excluded directories: {excluded}")
    function_summary = f"~{scope['function_count']} function(s)"
    if uncounted_files:
        function_summary = (
            f"~{scope['function_count']} locally counted function(s), "
            f"{len(uncounted_files)} source file(s) require semantic extraction"
        )
    print(
        "[Estimate] "
        f"{scope['included_file_count']} included source file(s), "
        f"{scope['excluded_file_count']} excluded source file(s), "
        f"{function_summary}."
    )
    print(
        "[Estimate] Historical samples: "
        f"{prediction['based_on_runs']}; details: {output_path}"
    )
    ranges = (
        ("time", prediction.get("duration_seconds"), _format_estimate_duration),
        ("LLM calls", prediction.get("llm_calls"), lambda value: str(int(value))),
        ("tokens", prediction.get("tokens"), lambda value: f"{int(value):,}"),
        ("cost", prediction.get("cost_usd"), lambda value: f"${value:.3f}"),
    )
    for label, value, formatter in ranges:
        if isinstance(value, dict):
            rendered = f"{formatter(value['low'])} – {formatter(value['high'])}"
        else:
            rendered = "history unavailable"
        print(f"[Estimate] {label} (ESTIMATE): {rendered}")


def _format_estimate_duration(seconds):
    seconds = int(seconds)
    hours, remainder = divmod(seconds, 3600)
    minutes, seconds = divmod(remainder, 60)
    if hours:
        return f"{hours}h {minutes:02d}m"
    if minutes:
        return f"{minutes}m {seconds:02d}s"
    return f"{seconds}s"


def _existing_required_source_files(proj_dir, required_source_files):
    """Keep required files that still exist after a Stage 1 scope hook."""
    if required_source_files is None:
        return None
    return [
        source_file
        for source_file in required_source_files
        if os.path.isfile(os.path.join(proj_dir, source_file))
    ]


def _finalize_failed_entry_run(original_proj_dir, entry_run_dir):
    """Preserve partial entry results after failure and remove the run copy."""
    if not entry_run_dir or not os.path.isdir(entry_run_dir):
        return

    run_work_dir = os.path.join(entry_run_dir, "fm_agent")
    ready_marker = os.path.join(run_work_dir, ".entry_scope_ready")
    original_work_dir = os.path.join(original_proj_dir, "fm_agent")
    if os.path.isfile(ready_marker) and os.path.isdir(run_work_dir):
        os.remove(ready_marker)
        if os.path.isdir(original_work_dir):
            shutil.rmtree(original_work_dir)
        shutil.copytree(run_work_dir, original_work_dir, symlinks=True)
        print(f"[EntryPlugin] Copied partial fm_agent/ to {original_work_dir}.")
    shutil.rmtree(entry_run_dir, ignore_errors=True)


def _resolve_bug_validator_path(raw_path):
    """Resolve a custom bug-validator prompt path from the launch directory."""
    if not raw_path:
        return None

    path = os.path.abspath(os.path.expanduser(raw_path))
    if not os.path.isfile(path):
        raise ValueError(
            "--bug-validator must point to a file: "
            f"{raw_path}"
        )

    try:
        with open(path, "r"):
            pass
    except OSError as exc:
        raise ValueError(
            f"--bug-validator file is not readable: {exc}"
        ) from exc

    return path


def _start_dashboard(work_dir):
    """Start the live dashboard for this run and return its process handle."""
    dashboard_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dashboard.py")
    try:
        return subprocess.Popen([sys.executable, dashboard_path, work_dir])
    except Exception as exc:
        logging.warning("[Pipeline] dashboard failed to start: %s", exc)
        return None


def _stop_dashboard(process):
    if process is None:
        return
    try:
        process.terminate()
        process.wait(timeout=3)
    except Exception:
        try:
            process.kill()
        except Exception:
            pass


def _normalize_submodules(proj_dir, submodules):
    """Return validated project-relative submodule directories."""
    if not submodules:
        return []

    proj_dir = os.path.abspath(proj_dir)
    normalized = []
    seen = set()
    for raw in submodules:
        value = (raw or "").strip()
        if not value:
            continue
        candidate = value if os.path.isabs(value) else os.path.join(proj_dir, value)
        candidate = os.path.abspath(candidate)
        try:
            inside_project = os.path.commonpath([proj_dir, candidate]) == proj_dir
        except ValueError:
            inside_project = False
        if not inside_project or candidate == proj_dir:
            raise ValueError(
                f"--submodule must name subdirectories inside proj_dir, got: {raw}"
            )
        if not os.path.isdir(candidate):
            raise ValueError(f"--submodule path is not a directory: {raw}")

        rel = os.path.relpath(candidate, proj_dir).replace(os.sep, "/")
        if rel not in seen:
            normalized.append(rel)
            seen.add(rel)

    collapsed = []
    for rel in sorted(normalized, key=lambda path: (path.count("/"), path)):
        if not collapsed or not _is_under_submodules(rel, collapsed):
            collapsed.append(rel)
    return collapsed


def _active_plugin_options(args, effective_resume):
    """Return active plugin-sensitive options."""
    active = {
        name
        for name in PLUGIN_OPTION_NAMES
        if getattr(args, name, False)
    }
    if effective_resume:
        active.add("resume")
    if settings.runtime.domain_knowledge_paths.strip() or os.environ.get("FM_AGENT_DOMAIN_KNOWLEDGE"):
        active.add("domain_knowledge")
    return active


def _validate_plugin_options(parser, plugin_config, args, resume):
    """Fail before side effects when a plugin denies an active option."""
    unsupported = get_unsupported_plugin_options(
        plugin_config,
        _active_plugin_options(args, resume),
    )
    if unsupported:
        flags = ", ".join(f"--{name.replace('_', '-')}" for name in unsupported)
        parser.error(
            f"Plugin '{plugin_config.name}' does not support option(s): {flags}. "
            "Remove the option(s) or use a compatible plugin."
        )


def run_pipeline(
    proj_dir,
    resume=False,
    required_source_files=None,
    domain_knowledge_files=None,
    submodules=None,
    one_phase=False,
    extra_call_edges_path=None,
    only_spec=False,
    bug_validator_path=None,
    validate_bugs=True,
    plugin_config=None,
    initial_history=None,
    plugin_context=None,
    all_bugs=False,
    specification: SpecificationProfile | None = None,
):
    if not os.path.isdir(proj_dir):
        print(f"[Pipeline] ERROR: proj_dir does not exist or is not a directory: {proj_dir}")
        sys.exit(1)
    if not _has_source_code(proj_dir, submodules):
        scope = f" selected submodule(s): {', '.join(submodules)}" if submodules else f" {proj_dir}"
        print(f"[Pipeline] ERROR: No source code files found in{scope}. "
              f"Supported extensions: {', '.join(sorted(EXT_TO_LANG.keys()))}")
        sys.exit(1)

    work_dir = os.path.join(proj_dir, "fm_agent")
    input_dir = os.path.join(work_dir, "extracted_functions")
    output_dir = os.path.join(work_dir, "logic_verification_results")
    script_dir = os.path.dirname(os.path.abspath(__file__))
    extra_call_edges = load_call_edges(extra_call_edges_path)
    profile_session = SpecificationProfileSession(
        default_profile=specification or SOFTWARE_PROFILE,
        plugin_root=(plugin_config.root if plugin_config is not None else None),
        default_prompt_root=Path(script_dir),
    )

    # Clean files from the previous run — unless resuming, where we keep all
    # prior progress (phases.json, generated specs, verification results) and
    # only do the remaining work.
    preserved_history = list(initial_history or [])
    if resume:
        if os.path.isdir(work_dir):
            print(f"[Pipeline] RESUME: keeping existing {os.path.relpath(work_dir, proj_dir)}/ — only remaining work will run.")
        else:
            print("[Pipeline] RESUME requested but no previous fm_agent/ found — starting fresh.")
            resume = False
    else:
        if initial_history is None:
            preserved_history = preserve_history_before_clean(work_dir)
        _clean_previous_run(work_dir)
    os.makedirs(work_dir, exist_ok=True)
    if preserved_history:
        write_history(work_dir, preserved_history)
    estimate_path = os.path.join(work_dir, "estimate.json")
    estimate = write_preflight_estimate(
        proj_dir,
        work_dir,
        submodules=submodules,
    )
    _print_preflight_summary(estimate, estimate_path)
    if plugin_config is not None:
        plugin_context_path = os.path.join(work_dir, "plugin_context.json")
        effective_plugin_context = dict(plugin_context or {})
        if submodules:
            effective_plugin_context["submodules"] = list(submodules)
        with open(plugin_context_path, "w", encoding="utf-8") as file:
            json.dump(effective_plugin_context, file, indent=2)
        if plugin_config.configure_hook is not None:
            with bind_profile_session(profile_session):
                run_plugin_hook(
                    plugin_config.name,
                    "configure",
                    plugin_config.configure_function,
                    plugin_config.configure_hook,
                    proj_dir,
                )
    specification = profile_session.freeze_and_validate()
    if resume and not only_spec and specification.enable_reasoning:
        _ensure_resume_mode_compatible(output_dir, all_bugs)
    domain_knowledge_relpaths = stage_domain_knowledge_files(
        proj_dir, work_dir, domain_knowledge_files
    )
    if domain_knowledge_relpaths:
        print(
            "[Pipeline] User domain knowledge: "
            f"{len(domain_knowledge_relpaths)} markdown file(s)."
        )

    # Stage 1: generate phase.json (input: target code → phases.json)
    # Stage 2: generate domain context (input: phases.json → domain context files)
    phase_stage = plugin_config.get_stage("generate_phase_plan") if plugin_config else None
    context_stage = plugin_config.get_stage("generate_domain_context") if plugin_config else None
    extraction_stage = plugin_config.get_stage("extract_functions") if plugin_config else None
    file_list_stage = plugin_config.get_stage("collect_file_list") if plugin_config else None
    topdown_stage = plugin_config.get_stage("generate_topdown_layers") if plugin_config else None
    spec_stage = plugin_config.get_stage("generate_specs_and_verification") if plugin_config else None
    print("[Pipeline] Stage 1/6: Generating phase plan...")
    if phase_stage is not None and phase_stage.type == "pass":
        print(
            "[Pipeline] Stage 1/6: Plugin stage "
            "'generate_phase_plan' type=pass, skipping."
        )
    elif phase_stage is not None and phase_stage.type == "replace":
        run_plugin_hook(
            plugin_config.name,
            "generate_phase_plan",
            phase_stage.replace_function,
            phase_stage.replace_hook,
            proj_dir,
        )
    else:
        if phase_stage is not None and phase_stage.input_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "generate_phase_plan",
                phase_stage.input_function,
                phase_stage.input_hook,
                proj_dir,
            )
        _run_generate_phases(
            proj_dir, work_dir, script_dir, resume=resume,
            submodules=submodules,
            workflow_source=specification.prompts.phase_plan,
            specification=specification,
        )
        if phase_stage is not None and phase_stage.output_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "generate_phase_plan",
                phase_stage.output_function,
                phase_stage.output_hook,
                proj_dir,
            )

    surviving_required_source_files = _existing_required_source_files(
        proj_dir, required_source_files
    )
    phases_modified = _post_process_phases(
        proj_dir, work_dir,
        required_source_files=surviving_required_source_files,
        submodules=submodules,
        one_phase=one_phase,
        specification=specification,
    )

    print("[Pipeline] Stage 2/6: Generating domain context...")
    if context_stage is not None and context_stage.type == "pass":
        print(
            "[Pipeline] Stage 2/6: Plugin stage "
            "'generate_domain_context' type=pass, skipping."
        )
    elif context_stage is not None and context_stage.type == "replace":
        run_plugin_hook(
            plugin_config.name,
            "generate_domain_context",
            context_stage.replace_function,
            context_stage.replace_hook,
            proj_dir,
        )
    else:
        if context_stage is not None and context_stage.input_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "generate_domain_context",
                context_stage.input_function,
                context_stage.input_hook,
                proj_dir,
            )
        _run_generate_domain_context(
            proj_dir,
            work_dir,
            script_dir,
            resume=resume and not phases_modified,
            workflow_source=specification.prompts.domain_context,
        )
        if context_stage is not None and context_stage.output_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "generate_domain_context",
                context_stage.output_function,
                context_stage.output_hook,
                proj_dir,
            )

    # Build (or rebuild) the codegraph index if codegraph is installed. Both
    # run_extraction (Stage 3) and generate_topdown_layers (Stage 5) read from it.
    # force=not resume mirrors run_extraction below: a fresh run rebuilds so the
    # index matches the current tree, while a resume reuses the existing index
    # (same tree as the interrupted run — rebuilding would just be wasted work).
    try_codegraph_init(proj_dir, force=not resume)

    # Run function extraction using extract.py
    # force=False on resume preserves already-specced extracted files; on a fresh
    # run fm_agent/ was just wiped so it is equivalent to force=True.
    print("[Pipeline] Stage 3/6: Extracting functions from source files...")
    if extraction_stage is not None and extraction_stage.type == "pass":
        print(
            "[Pipeline] Stage 3/6: Plugin stage "
            "'extract_functions' type=pass, skipping."
        )
    elif extraction_stage is not None and extraction_stage.type == "replace":
        run_plugin_hook(
            plugin_config.name,
            "extract_functions",
            extraction_stage.replace_function,
            extraction_stage.replace_hook,
            proj_dir,
        )
    else:
        if extraction_stage is not None and extraction_stage.input_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "extract_functions",
                extraction_stage.input_function,
                extraction_stage.input_hook,
                proj_dir,
            )
        run_extraction(proj_dir, work_dir=work_dir, force=not resume, verbose=True, specification=specification)
        if extraction_stage is not None and extraction_stage.output_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "extract_functions",
                extraction_stage.output_function,
                extraction_stage.output_hook,
                proj_dir,
            )

    # Copy system_prompt.md to spec_prompts/system_prompt.md
    spec_prompts_dir = os.path.join(work_dir, "spec_prompts")
    os.makedirs(spec_prompts_dir, exist_ok=True)
    system_prompt_source = Path(specification.prompts.system)
    system_prompt_destination = Path(spec_prompts_dir) / "system_prompt.md"
    if system_prompt_source.resolve() != system_prompt_destination.resolve():
        shutil.copy2(system_prompt_source, system_prompt_destination)

    phases_path = os.path.join(work_dir, "phases.json")
    with open(phases_path, "r") as f:
        phases_data = json.load(f)

    print("[Pipeline] Stage 4/6: Collecting file list...")
    file_list_path = os.path.join(work_dir, "fm_agent_file_list.json")
    if file_list_stage is not None and file_list_stage.type == "pass":
        print(
            "[Pipeline] Stage 4/6: Plugin stage "
            "'collect_file_list' type=pass, skipping."
        )
        with open(file_list_path, "r", encoding="utf-8") as file:
            file_list = json.load(file)
    elif file_list_stage is not None and file_list_stage.type == "replace":
        run_plugin_hook(
            plugin_config.name,
            "collect_file_list",
            file_list_stage.replace_function,
            file_list_stage.replace_hook,
            proj_dir,
        )
        with open(file_list_path, "r", encoding="utf-8") as file:
            file_list = json.load(file)
    else:
        if file_list_stage is not None and file_list_stage.input_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "collect_file_list",
                file_list_stage.input_function,
                file_list_stage.input_hook,
                proj_dir,
            )
        file_list = collect_file_names(input_dir, file_list_path, specification)
        if submodules:
            file_list = _write_file_names(
                _get_all_phase_files(phases_data, input_dir, specification),
                file_list_path,
            )
        if file_list_stage is not None and file_list_stage.output_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "collect_file_list",
                file_list_stage.output_function,
                file_list_stage.output_hook,
                proj_dir,
            )
            with open(file_list_path, "r", encoding="utf-8") as file:
                file_list = json.load(file)

    if not file_list:
        print("[Pipeline] No functions found to verify. Skipping spec generation.")
        return

    # --- Stage 5: Generate topdown layers ---
    print("[Pipeline] Stage 5/6: Generating topdown layers...")
    if topdown_stage is not None and topdown_stage.type == "pass":
        print(
            "[Pipeline] Stage 5/6: Plugin stage "
            "'generate_topdown_layers' type=pass, skipping."
        )
    elif topdown_stage is not None and topdown_stage.type == "replace":
        run_plugin_hook(
            plugin_config.name,
            "generate_topdown_layers",
            topdown_stage.replace_function,
            topdown_stage.replace_hook,
            proj_dir,
        )
    else:
        if topdown_stage is not None and topdown_stage.input_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "generate_topdown_layers",
                topdown_stage.input_function,
                topdown_stage.input_hook,
                proj_dir,
            )
        generate_topdown_layers(work_dir, extra_call_edges=extra_call_edges, specification=specification)
        if topdown_stage is not None and topdown_stage.output_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "generate_topdown_layers",
                topdown_stage.output_function,
                topdown_stage.output_hook,
                proj_dir,
            )

    # --- Stage 6: Execute spec generation workflow (per phase, per layer) ---
    if only_spec or not specification.enable_reasoning:
        print("[Pipeline] Stage 6/6: Generating specs (reasoning & bug validation disabled)...")
    else:
        print("[Pipeline] Stage 6/6: Generating specs & verification...")
    if spec_stage is not None and spec_stage.type == "pass":
        print(
            "[Pipeline] Stage 6/6: Plugin stage "
            "'generate_specs_and_verification' type=pass, skipping."
        )
    elif spec_stage is not None and spec_stage.type == "replace":
        run_plugin_hook(
            plugin_config.name,
            "generate_specs_and_verification",
            spec_stage.replace_function,
            spec_stage.replace_hook,
            proj_dir,
        )
    else:
        if spec_stage is not None and spec_stage.input_hook is not None:
            run_plugin_hook(
                plugin_config.name,
                "generate_specs_and_verification",
                spec_stage.input_function,
                spec_stage.input_hook,
                proj_dir,
            )
        run_spec_generation_and_verification(
            proj_dir,
            work_dir,
            input_dir,
            output_dir,
            script_dir,
            spec_prompts_dir,
            phases_data,
            resume=resume,
            extra_call_edges=extra_call_edges,
            only_spec=only_spec,
            bug_validator_path=bug_validator_path,
            validate_bugs=validate_bugs,
            all_bugs=all_bugs,
            specification=specification,
        )

    # Print confirmed bug count only when reasoning and bug validation are
    # both enabled for the active profile and invocation.
    if not only_spec and validate_bugs and specification.enable_reasoning:
        if all_bugs:
            # A resumed run may find every function and candidate validation
            # already complete, so no watcher runs to refresh the persistent
            # summary. Rebuild it from the current terminal records here.
            _generate_all_bugs_validation_summary(work_dir)
        summary_path = os.path.join(work_dir, "bug_validation", "summary.json")
        if os.path.exists(summary_path):
            with open(summary_path, "r") as f:
                summary = json.load(f)
            confirmed = summary.get("total_confirmed", 0)
            print(f"[Pipeline] Confirmed bugs: {confirmed}")

    # Regenerate the static HTML report index from the completed artifacts.
    # Runs in every completed mode (including --only-spec, which yields an
    # empty-state page); a rendering problem must never fail the run.
    try:
        from report import generate_report
        index_path = generate_report(work_dir)
        print(f"[Pipeline] Report index written to {os.path.relpath(index_path, proj_dir)}")
    except Exception as exc:
        logging.warning("[Pipeline] report.html generation skipped: %s", exc)

    if only_spec or not specification.enable_reasoning:
        print("[Pipeline] Done (specs only; reasoning & bug validation skipped).")
    else:
        print("[Pipeline] Done.")

    if spec_stage is not None and spec_stage.output_hook is not None:
        run_plugin_hook(
            plugin_config.name,
            "generate_specs_and_verification",
            spec_stage.output_function,
            spec_stage.output_hook,
            proj_dir,
        )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        usage="python3 main.py <proj_dir> [--resume] [--incremental INTENT_FILE] "
              "[--domain-knowledge FILE ...] [--one-phase] [--isolate] "
              "[--all-bugs] "
              "[--submodule PATH [PATH ...]] [--entry-func PATH [PATH ...]] "
              "[--end-func PATH ...] [--extra-edge FILE] "
              "[--bug-validator FILE] [--only-spec] [--estimate] "
              "[--dashboard] "
              "[--list-plugin] [--plugin NAME]",
        description="Run the FM agent pipeline on a project directory.",
    )
    parser.add_argument("proj_dir", nargs="?", help="path to the project directory")
    parser.add_argument(
        "--resume",
        action="store_true",
        help="continue a previous run in <proj_dir>/fm_agent instead of wiping it: "
             "keeps phases.json, generated specs, and existing verification results; "
             "only does the remaining work. Repeat --all-bugs when resuming an "
             "all-bugs run.",
    )
    parser.add_argument(
        "--incremental",
        metavar="INTENT_FILE",
        help="Run in incremental mode. Value is the path to the intent file "
        "defining the goal of modification.",
    )
    parser.add_argument(
        "--isolate",
        action="store_true",
        help="Run the pipeline against an isolated git worktree snapshot of "
        "the project instead of the project directory itself.",
    )
    parser.add_argument(
        "--one-phase",
        action="store_true",
        help="Put all planned source files into a single analysis phase.",
    )
    parser.add_argument(
        "--all-bugs",
        action="store_true",
        help="continue reasoning after mismatches and report every candidate; "
        "full and incremental modes validate each candidate.",
    )
    parser.add_argument(
        "--only-spec",
        action="store_true",
        help="Only generate behavioral specs; skip the reasoning and bug "
        "validation stages.",
    )
    parser.add_argument(
        "--estimate",
        action="store_true",
        help="scan source scope and print a history-based run estimate without "
        "starting the pipeline or making LLM calls",
    )
    parser.add_argument(
        "--dashboard",
        action="store_true",
        help="automatically start the live dashboard for this run",
    )
    parser.add_argument(
        "--domain-knowledge",
        "--knowledge",
        metavar="FILE",
        action="append",
        nargs="+",
        default=[],
        help="additional Markdown domain-knowledge file(s) to copy into "
        "fm_agent/spec_prompts/domain_context/user_knowledge/ and provide to "
        "setup, spec generation, and validation agents. May be repeated. "
        "FM_AGENT_DOMAIN_KNOWLEDGE can also provide os.pathsep-separated files.",
    )
    parser.add_argument(
        "--submodule",
        metavar="PATH",
        nargs="+",
        default=None,
        help="Only process source code under one or more subdirectories of proj_dir.",
    )
    parser.add_argument(
        "--entry-func",
        metavar="PATH",
        nargs="+",
        default=None,
        help="one or more function paths of the entry points to start reasoning from.",
    )
    parser.add_argument(
        "--end-func",
        metavar="PATH",
        nargs="+",
        default=None,
        help="one or more function paths at which to stop (space-separated list); "
        "if omitted, the whole call graph reachable from --entry-func is analyzed.",
    )
    parser.add_argument(
        "--extra-edge",
        dest="extra_edge",
        metavar="FILE",
        default=None,
        help="optional JSON file, or directory of JSON files, containing "
        "supplemental caller->callee edges.",
    )
    parser.add_argument(
        "--bug-validator",
        metavar="FILE",
        default=None,
        help="use a custom Markdown prompt for bug validation instead of "
        "the built-in md/bug_validator.md",
    )
    parser.add_argument(
        "--list-plugin",
        action="store_true",
        help="list all valid plugins found under the plugins/ directory and exit.",
    )
    parser.add_argument(
        "--plugin",
        metavar="NAME",
        default=None,
        help="load and activate the named plugin from the plugins/ directory.",
    )
    args = parser.parse_args()

    if args.list_plugin:
        from pathlib import Path
        from src.plugin import load_plugins
        plugins_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "plugins")
        plugins = load_plugins(Path(plugins_dir))
        if not plugins:
            print("No valid plugins found.")
        else:
            print(f"{'Plugin':<30} {'Version':<12} Stages")
            print("-" * 65)
            for name, plugin in plugins.items():
                stage_names = ", ".join(plugin.stages.keys()) if plugin.stages else "(none)"
                print(f"{name:<30} {plugin.version:<12} {stage_names}")
        sys.exit(0)

    selected_plugin = args.plugin
    if args.entry_func is not None:
        if selected_plugin is not None:
            parser.error("--plugin cannot be combined with --entry-func.")
        selected_plugin = "entry_reasoning"
    elif selected_plugin == "entry_reasoning":
        parser.error("the entry_reasoning plugin requires --entry-func.")

    plugin_config = None
    if selected_plugin:
        if not args.proj_dir:
            parser.error("the following arguments are required when --plugin is used: proj_dir")
        from pathlib import Path
        from src.plugin import validate_plugin
        plugins_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "plugins")
        plugin_config = validate_plugin(Path(plugins_dir) / selected_plugin)
        if plugin_config is None:
            print(f"[Pipeline] ERROR: Plugin '{selected_plugin}' not found or invalid.")
            sys.exit(1)
        print(f"[Pipeline] Loaded plugin '{plugin_config.name}' v{plugin_config.version}")

    if not args.proj_dir:
        parser.error("the following arguments are required: proj_dir")

    resume = args.resume or os.environ.get("FM_AGENT_RESUME") == "1"
    if plugin_config is not None:
        _validate_plugin_options(parser, plugin_config, args, resume)
    proj_dir = os.path.abspath(args.proj_dir)
    is_entry = args.entry_func is not None
    extra_call_edges_path = args.extra_edge
    if extra_call_edges_path:
        extra_call_edges_path = os.path.abspath(extra_call_edges_path)
    plugin_context = {
        "extra_edge": extra_call_edges_path,
    }
    try:
        bug_validator_path = _resolve_bug_validator_path(args.bug_validator)
    except ValueError as exc:
        parser.error(str(exc))
    try:
        submodules = _normalize_submodules(proj_dir, args.submodule)
    except ValueError as exc:
        parser.error(str(exc))

    try:
        domain_knowledge_files = collect_domain_knowledge_paths(
            args.domain_knowledge,
            base_dir=proj_dir,
            fallback_base_dir=os.getcwd(),
        )
    except ValueError as exc:
        parser.error(str(exc))

    if submodules and args.entry_func is not None:
        parser.error("--submodule cannot be combined with --entry-func.")

    if args.only_spec and args.incremental:
        parser.error(
            "--only-spec cannot be combined with --incremental "
            "(incremental mode is inherently a reasoning/bug-validation flow)."
        )

    if args.estimate:
        incompatible = any(
            [
                resume,
                args.incremental,
                args.entry_func,
                args.end_func,
                args.isolate,
                args.only_spec,
                args.all_bugs,
                args.dashboard,
            ]
        )
        if incompatible:
            parser.error(
                "--estimate is a standalone preflight and cannot be combined "
                "with resume, incremental, entry/end functions, isolate, "
                "only-spec, all-bugs, or dashboard"
            )
        estimate_work_dir = os.path.join(proj_dir, "fm_agent_estimate")
        estimate = write_preflight_estimate(
            proj_dir,
            estimate_work_dir,
            submodules=submodules,
        )
        _print_preflight_summary(
            estimate,
            os.path.join(estimate_work_dir, "estimate.json"),
        )
        sys.exit(0)

    # ---- pre-flight environment check (shared by all pipeline modes) ----
    import config
    from src.env_check import run as env_check_run
    if not env_check_run(proj_dir, config):
        sys.exit(0)

    start_time = time.time()

    original_proj_dir = proj_dir
    entry_run_dir = None
    required_source_files = None
    validate_bugs = True
    if is_entry:
        entry_run_dir = original_proj_dir + ".fm-entry-run"
        _make_run_copy(original_proj_dir, entry_run_dir)
        required_source_files = list(dict.fromkeys(
            _entry_func_source_rel(entry) for entry in args.entry_func
        ))
        validate_bugs = False
        plugin_context.update({
            "original_proj_dir": original_proj_dir,
            "entry_run_dir": entry_run_dir,
            "entry_funcs": args.entry_func,
            "end_funcs": args.end_func or [],
            "all_bugs": args.all_bugs,
        })

    # Incremental mode diffs against the commit recorded by a previous run, and
    # --isolate snapshots the repo via a git worktree, so both require a git repo.
    # A non-git project can only run the full pipeline against the project directory
    # itself.
    if not is_entry and not _is_git_repo(proj_dir):
        parser.error(
            f"FM-Agent requires a git repository, but {proj_dir} is not."
        )

    # Resolve the intent path before snapshotting, since cwd-relative paths must
    # resolve against the real project, not the frozen worktree copy.
    intent_path = (
        os.path.abspath(args.incremental)
        if not is_entry and args.incremental
        else None
    )

    # In incremental mode the commit to diff against is the most recent one recorded
    # in version.log (the last line, since each run appends its commit). Read it from
    # the real project before snapshotting.
    old_commit = None
    if not is_entry and args.incremental:
        version_path = os.path.join(proj_dir, "fm_agent", "version.log")
        if os.path.exists(version_path):
            with open(version_path, "r") as f:
                commits = [line.strip() for line in f if line.strip()]
            old_commit = commits[-1] if commits else None

    # Capture the project's latest commit id before running. With --isolate the
    # pipeline runs against a throwaway worktree snapshot whose HEAD is a synthetic
    # snapshot commit, so the version to record must come from the real project.
    new_commit = _get_head_commit(proj_dir) if not is_entry else None

    # With --isolate, the pipeline runs against the snapshot's fm_agent/. Resuming
    # needs the previous run's fm_agent/ (phases.json, specs, verification results)
    # to be present in the snapshot, so copy the excluded workspace in for resume
    # too — not just incremental mode.
    run_ctx = (
        contextlib.nullcontext(entry_run_dir)
        if is_entry
        else frozen_worktree(
            proj_dir, copy_excluded=bool(args.incremental) or resume
        )
        if args.isolate
        else contextlib.nullcontext(proj_dir)
    )
    isolated_history = (
        preserve_history_before_clean(os.path.join(proj_dir, "fm_agent"))
        if args.isolate
        else None
    )
    with run_ctx as run_dir:
        dashboard_process = _start_dashboard(run_dir) if args.dashboard else None
        try:
            # Incremental mode requires a recorded commit to diff against; without a
            # version.log from a previous run, fall back to the full pipeline.
            if not is_entry and args.incremental and old_commit:
                run_incremental_pipeline(
                    run_dir,
                    intent_path,
                    old_commit,
                    domain_knowledge_files=domain_knowledge_files,
                    submodules=submodules,
                    one_phase=args.one_phase,
                    extra_call_edges_path=extra_call_edges_path,
                    bug_validator_path=bug_validator_path,
                    all_bugs=args.all_bugs,
                )
            else:
                run_pipeline(
                    run_dir,
                    resume=resume,
                    required_source_files=required_source_files,
                    domain_knowledge_files=domain_knowledge_files,
                    submodules=submodules,
                    one_phase=args.one_phase,
                    extra_call_edges_path=extra_call_edges_path,
                    only_spec=args.only_spec,
                    bug_validator_path=bug_validator_path,
                    validate_bugs=validate_bugs,
                    plugin_config=plugin_config,
                    initial_history=isolated_history,
                    plugin_context=plugin_context,
                    all_bugs=args.all_bugs,
                )
            # Record the commit that was processed. Written after the pipeline since
            # it recreates fm_agent/; with --isolate it lives in the snapshot and is
            # copied back to the real project below. Only recorded on success so a
            # partial run does not advance the version baseline.
            if not is_entry:
                _record_version(new_commit, os.path.join(run_dir, "fm_agent"))
                record_completed_run(
                    os.path.join(run_dir, "fm_agent"),
                    duration_seconds=time.time() - start_time,
                )
        finally:
            _stop_dashboard(dashboard_process)
            # With --isolate the pipeline ran against a throwaway snapshot, so its
            # fm_agent/ results live in the snapshot. Copy them back into the real
            # project so they are not lost when the snapshot is discarded — this runs
            # even when the pipeline crashes or is interrupted mid-run, so partial
            # progress survives and can be resumed with --resume.
            if is_entry:
                _finalize_failed_entry_run(original_proj_dir, entry_run_dir)
                clear_test_file_exemptions()
            elif args.isolate:
                src_fm = os.path.join(run_dir, "fm_agent")
                dst_fm = os.path.join(proj_dir, "fm_agent")
                if os.path.isdir(src_fm):
                    if os.path.isdir(dst_fm):
                        shutil.rmtree(dst_fm)
                    shutil.copytree(src_fm, dst_fm, symlinks=True)
                    print(f"[Pipeline] Copied results back to {dst_fm}")
    end_time = time.time()
    logging.info(f"Total time: {end_time - start_time:.2f} seconds")