### B4 PR WRAPPER INSPECTION ###
Sa. 30 Mai 2026 18:02:25 CEST

### BRANCH / STATUS ###
b4-pr-wrapper-mvp
146bda3 Add B4 remote transfer smoke payload
d3d5410 Refresh handoff state after PR945 (#946)
3b1a18f Add transfer closeout command (#945)
?? docs/reports/terminal/b4-pr-wrapper-inspection.txt

### agentic-kit pr help ###
                                                                                                   
 Usage: agentic-kit pr [OPTIONS] COMMAND [ARGS]...                                                 
                                                                                                   
 Evaluate deterministic PR closeout and readiness state.                                           
                                                                                                   
╭─ Options ───────────────────────────────────────────────────────────────────────────────────────╮
│ --help          Show this message and exit.                                                     │
╰─────────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ Commands ──────────────────────────────────────────────────────────────────────────────────────╮
│ closeout-check                                                                                  │
│ status          Print deterministic PR/CI status and fetch failed logs for red CI.              │
│ merge-if-green  Merge only when PR checks are green, refs match, and merge state is clean.      │
│ wait-ci         Wait for pull-request CI; guard merge preparation with --expected-head-sha.     │
╰─────────────────────────────────────────────────────────────────────────────────────────────────╯


### pr cli source ###
from __future__ import annotations

import json
from pathlib import Path

import typer

from agentic_project_kit.ci_readiness import (
    WAITING,
    gh_pr_snapshot_provider,
    render_pr_readiness,
    wait_for_pr_readiness,
)
from agentic_project_kit.next_turn_merge_if_green import main_verification_passed, merge_if_green, render_result
from agentic_project_kit.next_turn_pr_status import (
    attach_failed_run_logs,
    classify_pr_status,
    fetch_pr_payload,
    render_decision,
)
from agentic_project_kit.pr_closeout import BLOCKED, evaluate_pr_closeout, render_pr_closeout

pr_app = typer.Typer(help="Evaluate deterministic PR closeout and readiness state.")


@pr_app.command("closeout-check")
def closeout_check(json_file: Path) -> None:
    data = json.loads(json_file.read_text(encoding="utf-8"))
    result = evaluate_pr_closeout(data)
    typer.echo(render_pr_closeout(result))
    if result.outcome == BLOCKED:
        raise typer.Exit(code=1)


@pr_app.command("status")
def status(
    pr_number: int = typer.Argument(..., help="Pull request number to inspect."),
    json_output: bool = typer.Option(False, "--json", help="Print JSON instead of the text report."),
    no_failed_log_fetch: bool = typer.Option(
        False,
        "--no-failed-log-fetch",
        help="Do not fetch failed GitHub Actions logs for red checks.",
    ),
    failed_log_lines: int = typer.Option(120, min=0, help="Maximum failed-log excerpt lines."),
) -> None:
    """Print deterministic PR/CI status and fetch failed logs for red CI."""
    payload = fetch_pr_payload(str(pr_number))
    decision = classify_pr_status(payload, pr=str(pr_number))
    if decision.decision == "red" and not no_failed_log_fetch:
        decision = attach_failed_run_logs(decision, max_lines=failed_log_lines)
    if json_output:
        typer.echo(json.dumps(decision, default=lambda item: item.__dict__, indent=2, sort_keys=True))
    else:
        typer.echo(render_decision(decision))


@pr_app.command("merge-if-green")
def merge_if_green_command(
    pr_number: int = typer.Argument(..., help="Pull request number to merge only after green checks."),
    merge_method: str = typer.Option(
        "squash",
        help="GitHub merge method: squash, merge, or rebase.",
    ),
    delete_branch: bool = typer.Option(True, help="Delete the branch after a successful merge."),
    dry_run: bool = typer.Option(False, "--dry-run", help="Evaluate without merging."),
    no_verify_main: bool = typer.Option(False, "--no-verify-main", help="Do not verify main CI after merge."),
    main_branch: str = typer.Option("main", help="Expected base branch and post-merge verification branch."),
    expected_base_branch: str = typer.Option("", help="Expected PR base branch. Defaults to --main-branch."),
    expected_head_sha: str = typer.Option("", help="Expected PR head SHA. Refuses merge if the head moved."),
    main_ci_timeout_seconds: int = typer.Option(300, min=1, help="Post-merge main CI wait timeout."),
    main_ci_poll_seconds: int = typer.Option(10, min=1, help="Post-merge main CI polling interval."),
) -> None:
    """Merge only when PR checks are green, refs match, and merge state is clean."""
    result = merge_if_green(
        str(pr_number),
        merge_method=merge_method,
        delete_branch=delete_branch,
        dry_run=dry_run,
        verify_main=not no_verify_main,
        main_branch=main_branch,
        expected_base_branch=expected_base_branch,
        expected_head_sha=expected_head_sha,
        main_ci_timeout_seconds=main_ci_timeout_seconds,
        main_ci_poll_seconds=main_ci_poll_seconds,
    )
    typer.echo(render_result(result))
    if dry_run:
        return
    if result.decision != "merge" or not result.merged or not main_verification_passed(result):
        raise typer.Exit(code=1)


@pr_app.command("wait-ci")
def wait_ci(
    pr_number: int = typer.Argument(..., help="Pull request number to inspect."),
    expected_head_sha: str | None = typer.Option(
        None,
        "--expected-head-sha",
        help="Expected PR head SHA for --expected-head-sha. The command fails closed if the head moves.",
    ),
    timeout_seconds: int = typer.Option(2700, min=1, help="Maximum wait time."),
    interval_seconds: int = typer.Option(20, min=1, help="Polling interval."),
    expected_check: list[str] = typer.Option(
        [],
        "--expected-check",
        help="Check name that must be present before readiness can pass. Repeatable.",
    ),
) -> None:
    """Wait for pull-request CI; guard merge preparation with --expected-head-sha."""
    result = wait_for_pr_readiness(
        gh_pr_snapshot_provider(pr_number),
        expected_head_sha=expected_head_sha,
        timeout_seconds=timeout_seconds,
        interval_seconds=interval_seconds,
        expected_checks=tuple(expected_check),
    )
    typer.echo(render_pr_readiness(result))
    if not result.success:
        raise typer.Exit(code=2 if result.outcome == WAITING else 1)


def register_pr_closeout_alias(app: typer.Typer) -> None:
    @app.command("pr-closeout")
    def pr_closeout_alias(json_file: Path) -> None:
        data = json.loads(json_file.read_text(encoding="utf-8"))
        result = evaluate_pr_closeout(data)
        typer.echo(render_pr_closeout(result))
        if result.outcome == BLOCKED:
            raise typer.Exit(code=1)

### relevant pr tests ###
tests/test_boot_closeout_cli.py
tests/test_chat_switch_closeout.py
tests/test_control_file_preservation.py
tests/test_handoff_prompt.py
tests/test_interpreter_discovery_guard.py
tests/test_next_turn_evidence_repo_hygiene.py
tests/test_next_turn_merge_if_green.py
tests/test_next_turn_pr_status.py
tests/test_ns_interpreter_and_no_exec_guard.py
tests/test_ns_up_pr_completion.py
tests/test_patch_artifact_preflight.py
tests/test_patch_preflight_slice_gate_requirement.py
tests/test_post_merge_handoff_refresh.py
tests/test_pr_ci_readiness.py
tests/test_pr_cleanup.py
tests/test_pr_closeout.py
tests/test_pr_hygiene.py
tests/test_pre_gui_execution_hardening_plan.py
tests/test_protected_change_planner_ns_route.py
tests/test_protected_change_planner.py
tests/test_protected_control_file_preservation_gate.py
tests/test_release_preflight_phase.py
tests/test_release_prep_core.py
tests/test_remote_next_closeout.py
tests/test_rule_preservation.py
tests/test_terminal_remote_preflight.py
tests/test_transfer_closeout.py
tests/test_typed_work_orders_pre_gui_docs.py
tests/test_v0_3_30_gui_readiness_closeout_docs.py
tests/test_v031_pre_gui_execution_hardening_contract.py
tests/test_v031_status_handoff_closeout.py
tests/test_v032_status_handoff_closeout.py
tests/test_v036_logged_block_status_propagation.py
tests/test_v037_final_gui_preparation_closeout.py
tests/test_v040_gui_presenter_remote_mutation_safety.py
tests/test_v040_gui_presenter.py
tests/test_work_order_prepare.py

### transfer command source ###
from __future__ import annotations

import json
from pathlib import Path

import typer

from agentic_project_kit.transfer_closeout import closeout_transfer
from agentic_project_kit.transfer_local_runner import run_local_transfer
from agentic_project_kit.transfer_remote_next import run_remote_next_transfer
from agentic_project_kit.transfer_runner import (
    DEFAULT_INBOX,
    apply_transfer_order,
    inspect_transfer_order,
    load_transfer_order,
    transfer_result_as_json_data,
)
from agentic_project_kit.transfer_state import build_transfer_state

transfer_app = typer.Typer(help="Inspect and apply repo-backed text transfer orders.")


def _load_or_exit(path: Path):
    try:
        return load_transfer_order(path)
    except (FileNotFoundError, ValueError) as exc:
        typer.echo(str(exc))
        raise typer.Exit(code=1) from exc


def _emit_result(result, json_output: bool) -> None:
    if json_output:
        typer.echo(json.dumps(transfer_result_as_json_data(result), indent=2, sort_keys=True))
    else:
        typer.echo(f"transfer_id={result.transfer_id}")
        typer.echo(f"result_status={result.result_status}")
        typer.echo(f"returncode={result.returncode}")
        typer.echo(f"report_path={result.report_path}")
        typer.echo(f"message={result.message}")


@transfer_app.command("closeout")
def closeout(
    no_remove_transfer_dir: bool = typer.Option(
        False,
        "--no-remove-transfer-dir",
        help="Do not remove .agentic/transfer during closeout.",
    ),
    json_output: bool = typer.Option(True, "--json/--no-json", help="Print machine-readable JSON."),
) -> None:
    try:
        result = closeout_transfer(Path("."), remove_transfer_dir=not no_remove_transfer_dir)
    except RuntimeError as exc:
        typer.echo(str(exc))
        raise typer.Exit(code=1) from exc

    if json_output:
        typer.echo(json.dumps(result.as_json_data(), indent=2, sort_keys=True))
    else:
        typer.echo(f"result_status={result.result_status}")
        typer.echo(f"returncode={result.returncode}")
        typer.echo(f"removed_transfer_dir={result.removed_transfer_dir}")
        typer.echo(f"latest_command_run_path={result.latest_command_run_path}")
        typer.echo(f"blocked_dirty_paths={','.join(result.blocked_dirty_paths)}")
        typer.echo(f"next_action={result.next_action}")

    if result.returncode != 0:
        raise typer.Exit(code=result.returncode)


@transfer_app.command("remote-next")
def remote_next(
    branch: str = typer.Argument(..., help="Remote transfer branch to fetch, switch to, pull, and run."),
    json_output: bool = typer.Option(True, "--json/--no-json", help="Print machine-readable JSON."),
) -> None:
    try:
        result = run_remote_next_transfer(Path("."), branch)
    except (RuntimeError, ValueError, FileNotFoundError) as exc:
        typer.echo(str(exc))
        raise typer.Exit(code=1) from exc

    if json_output:
        typer.echo(json.dumps(result.as_json_data(), indent=2, sort_keys=True))
    else:
        typer.echo(f"branch={result.branch}")
        typer.echo(f"head={result.head}")
        typer.echo(f"result_status={result.local_run.result_status}")
        typer.echo(f"returncode={result.local_run.returncode}")
        typer.echo(f"next_action={result.local_run.next_action}")

    if result.local_run.returncode != 0:
        raise typer.Exit(code=result.local_run.returncode)


@transfer_app.command("run-local")
def run_local(
    path: Path = typer.Option(DEFAULT_INBOX, "--path", help="Transfer order path."),
    json_output: bool = typer.Option(True, "--json/--no-json", help="Print machine-readable JSON."),
) -> None:
    try:
        result = run_local_transfer(Path("."), path)
    except (FileNotFoundError, ValueError) as exc:
        typer.echo(str(exc))
        raise typer.Exit(code=1) from exc

    if json_output:
        typer.echo(json.dumps(result.as_json_data(), indent=2, sort_keys=True))
    else:
        typer.echo(f"transfer_id={result.transfer_id}")
        typer.echo(f"result_status={result.result_status}")
        typer.echo(f"returncode={result.returncode}")
        typer.echo(f"next_action={result.next_action}")

    if result.returncode != 0:
        raise typer.Exit(code=result.returncode)


@transfer_app.command("state")
def state(
    json_output: bool = typer.Option(True, "--json/--no-json", help="Print machine-readable JSON."),
) -> None:
    snapshot = build_transfer_state(Path("."))
    data = snapshot.as_json_data()
    if json_output:
        typer.echo(json.dumps(data, indent=2, sort_keys=True))
    else:
        typer.echo(f"primary_state={snapshot.primary_state}")
        typer.echo(f"next_action={snapshot.next_action}")


@transfer_app.command("status")
def status(
    path: Path = typer.Option(DEFAULT_INBOX, "--path", help="Transfer order path."),
    json_output: bool = typer.Option(False, "--json", help="Print machine-readable JSON."),
) -> None:
    order = _load_or_exit(path)
    result = inspect_transfer_order(order, Path("."))
    _emit_result(result, json_output)
    if result.returncode != 0:
        raise typer.Exit(code=result.returncode)


@transfer_app.command("inspect")
def inspect(
    path: Path = typer.Option(DEFAULT_INBOX, "--path", help="Transfer order path."),
    json_output: bool = typer.Option(False, "--json", help="Print machine-readable JSON."),
) -> None:
    order = _load_or_exit(path)
    result = inspect_transfer_order(order, Path("."))
    _emit_result(result, json_output)
    if result.returncode != 0:
        raise typer.Exit(code=result.returncode)


@transfer_app.command("apply")
def apply(
    path: Path = typer.Option(DEFAULT_INBOX, "--path", help="Transfer order path."),
    json_output: bool = typer.Option(False, "--json", help="Print machine-readable JSON."),
) -> None:
    order = _load_or_exit(path)
    result = apply_transfer_order(order, Path("."))
    _emit_result(result, json_output)
    if result.returncode != 0:
        raise typer.Exit(code=result.returncode)

### transfer state ###
{
  "branch": "b4-pr-wrapper-mvp",
  "capabilities": {
    "closeout_last_run": false,
    "diagnose": true,
    "refresh_rules": true,
    "run_next_command": false
  },
  "closeout": {
    "dirty_worktree": true,
    "pending_transfer_order": false
  },
  "created_at": "2026-05-30T16:02:25.816998+00:00",
  "head": "146bda3",
  "last_result": {
    "exists": true,
    "report_path": "docs/reports/command_runs/transfer-closeout-mvp.md"
  },
  "next_action": "Review or clean the worktree before running another transfer action.",
  "primary_state": "BLOCKED",
  "reasons": [
    "dirty_worktree"
  ],
  "repo": "vfi64/agentic-project-kit",
  "schema_version": 1
}
