"""Phase 7 report post-processing — the single reference point.

Contract 3.0 collects usage into team state, assembles the role-owned inputs
into final-report data.json once, then verifies and renders that published
record. Contract 2.0 keeps its historical in-place projection sequence for
read-only compatibility. The order is load-bearing: rendering before assembly
or usage collection publishes stale derived views, and validating before
rendering trips the report-views contract.

A non-zero exit no longer stops the sequence before `validate-run`. A schema
cap used to hide the contract scan: assembly refused, later steps never ran,
and the lead only saw those failures after a second finalize. Each step still
records its own exit; `ok` stays False when any step failed. `teardown-stages`
and `record-group-memory` are the exceptions — one reclaims worktrees, the
other hands this run's conclusion to the group's sibling tasks — so a failed
prefix skips both: a record that did not validate is not memory.

`token-usage` reads the lead session log (state outside this run). A refusal
there used to delete every later artifact; it no longer does, because the
sequence continues. `validators/validate-run.py` re-collects when the recorded
usage is all zeros against an `unavailable` session source (`_needs_token_autofix`)
and refuses the run rather than ship zeroed counts (`accuracy-failed`). A
legacy v1 report is caught earlier still, by its unsubstituted `{{...}}`
placeholders, which a v2 report never carries because its numeric cells are
`null` until this step fills them. Substituting the tokens on a later retry
then leaves the already-rendered html stale for `validators/validate-report-views.py`.

The translation sidecar is the `translate` step, directly before `render-views`,
which overlays it. It used to be a manual lead sequence outside this command (materialize the translator
prompt, dispatch it, resume finalize at `render-views`), written only in a doc
the lead lazy-reads; a lead that ran the whole sequence at once skipped it and
the run kept an English view under a `ko` report (2026-09-09, fontsninja-v3-site
dev-10628-3). `okstra_ctl.report_translation_dispatch` owns the step; an English
report or an existing sidecar makes it a no-op.

Every lead adapter drives Phase 7 through this module: the Codex adapter calls
it in-process (``codex_dispatch``), and a Claude-led run reaches the same code
through the ``okstra report-finalize`` CLI. Neither reimplements the sequence.
"""
from __future__ import annotations

import argparse
import json
from datetime import date, datetime, timezone
import subprocess
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Callable, Mapping, Sequence

from . import group_context
from .agent.activity import ActivityProjectionError, project_agent_activity
from .report_contract import apply_execution_roles
from .report_assembly import ReportAssemblyError, assemble_report
from .dispatch_state import (
    DispatchError,
    link_agent_dispatch_result,
    mutate_team_state,
)
from .final_report_paths import (
    final_report_data_path,
    final_report_markdown_path,
    sidecar_source_rel,
)
from .paths import task_dir, task_manifest_file
from .report_view_artifacts import html_view_path
from .release_gate import release_handoff_allowed
from .report_translation_dispatch import TranslateOutcome, translate_report
from .json_boundary import JsonBoundaryError, load_owned_object, write_owned_object_atomic
from .stage_integrate import IntegrateError
from .stage_targets import (
    StageTargetError,
    integrate_and_teardown_whole_task,
)
from .session import observe_lead_session
from .error_log_write import record_runtime_failure

# 포인터 값 타입만 쓴다. `okstra_project.phase_pointer` 는 okstra 안의
# 어떤 것도 import 하지 않는 leaf 라 순환이 없다 — 그 모듈 도크스트링.
from okstra_project.phase_pointer import (
    STATUS_BLOCKED,
    STATUS_PENDING,
    STATUS_READY,
    STATUS_TERMINAL,
    promote as promote_next_phase,
)


STEP_PROJECT_ACTIVITY = "project-activity"
STEP_TRANSLATE = "translate"
STEP_TOKEN_USAGE = "token-usage"
STEP_RENDER_VIEWS = "render-views"
STEP_SPAWN_FOLLOWUPS = "spawn-followups"
STEP_VALIDATE_RUN = "validate-run"
STEP_RECORD_GROUP_MEMORY = "record-group-memory"
STEP_TEARDOWN_STAGES = "teardown-stages"
STEP_PREFLIGHT = "preflight"

STEP_ORDER = (
    STEP_PROJECT_ACTIVITY,
    STEP_TRANSLATE,
    STEP_TOKEN_USAGE,
    STEP_RENDER_VIEWS,
    STEP_SPAWN_FOLLOWUPS,
    STEP_VALIDATE_RUN,
    # After validation: the group's sibling tasks read this run's conclusion
    # from `group-context.md`, and only a validated record is worth handing on.
    STEP_RECORD_GROUP_MEMORY,
    # Last, and only after the run validated: it removes the stage worktrees a
    # blocked verdict would send the user straight back to.
    STEP_TEARDOWN_STAGES,
)

V3_STEP_ORDER = (
    STEP_TOKEN_USAGE,
    STEP_PROJECT_ACTIVITY,
    STEP_PREFLIGHT,
    # Before the render that overlays its sidecar.
    STEP_TRANSLATE,
    STEP_RENDER_VIEWS,
    STEP_SPAWN_FOLLOWUPS,
    STEP_VALIDATE_RUN,
    STEP_RECORD_GROUP_MEMORY,
    STEP_TEARDOWN_STAGES,
)

# 앞 단계가 실패하면 건너뛰는 단계. 하나는 worktree 를 거두고, 하나는 검증 안 된
# 결론을 형제 task 에 넘기게 된다.
_SKIPPED_AFTER_FAILURE = frozenset({STEP_RECORD_GROUP_MEMORY, STEP_TEARDOWN_STAGES})


class FinalizeError(Exception):
    """Raised when the Phase 7 sequence cannot be assembled."""


def string_value(value: Any) -> str:
    if not isinstance(value, str):
        return ""
    return value.strip()


def require_string(payload: Mapping[str, Any], key: str) -> str:
    value = payload.get(key)
    if not isinstance(value, str) or not value.strip():
        raise FinalizeError(f"missing required string field: {key}")
    return value


def resolve_project_path(project_root: Path, value: str) -> Path:
    path = Path(value)
    return path if path.is_absolute() else project_root / path


def resolve_optional_path(project_root: Path, value: Any) -> Path | None:
    if not isinstance(value, str) or not value.strip():
        return None
    return resolve_project_path(project_root, value)


def task_group(manifest: Mapping[str, Any]) -> str:
    value = string_value(manifest.get("taskGroup"))
    if value:
        return value
    task_key = require_string(manifest, "taskKey")
    colon_parts = task_key.split(":")
    if len(colon_parts) >= 3 and colon_parts[-2].strip():
        return colon_parts[-2].strip()
    slash_parts = task_key.split("/")
    if len(slash_parts) >= 2 and slash_parts[0].strip():
        return slash_parts[0].strip()
    raise FinalizeError(f"cannot infer task group from taskKey: {task_key}")


def task_id(manifest: Mapping[str, Any]) -> str:
    value = string_value(manifest.get("taskId"))
    if value:
        return value
    task_key = require_string(manifest, "taskKey")
    colon_parts = task_key.split(":")
    if len(colon_parts) >= 3 and colon_parts[-1].strip():
        return colon_parts[-1].strip()
    slash_parts = task_key.split("/")
    if len(slash_parts) >= 2 and slash_parts[-1].strip():
        return slash_parts[-1].strip()
    raise FinalizeError(f"cannot infer task id from taskKey: {task_key}")


def _category_seq(manifest: Mapping[str, Any], category: str) -> str:
    seqs = manifest.get("runSequencesByCategory")
    if not isinstance(seqs, Mapping):
        raise FinalizeError("run manifest has no runSequencesByCategory object")
    return require_string(seqs, category)


def run_seq(manifest: Mapping[str, Any]) -> str:
    """How many times this task type has run — what a lead event records."""
    return _category_seq(manifest, "manifests")


def report_seq(manifest: Mapping[str, Any]) -> str:
    """The `reports` sequence — the one the final-report filename carries.

    Categories advance independently (`paths.compute_run_paths`), so a run whose
    predecessor produced no report reuses the free report slot and `run_seq`
    outruns this one. This value reaches `render-report-views --seq`, which
    stamps it into the HTML `runMeta` naming the exported
    `user-response-<task-type>-<seq>.md`; `user_response.write_sidecar` derives
    that same name from the report path, so anything but `reports` makes the two
    answer paths write different files for one report.
    """
    return _category_seq(manifest, "reports")


def task_manifest_path(project_root: Path, manifest: Mapping[str, Any]) -> Path:
    """Resolve a run manifest's task manifest, falling back through the identity.

    Distinct from `paths.task_manifest_file`, which only names the file inside a
    known task root — this one decides *which* task root applies.
    """
    value = string_value(manifest.get("taskManifestPath"))
    if value:
        return resolve_project_path(project_root, value)
    task_root = string_value(manifest.get("taskRootPath"))
    if task_root:
        return task_manifest_file(resolve_project_path(project_root, task_root))
    return task_manifest_file(
        task_dir(project_root, task_group(manifest), task_id(manifest))
    )


def resolve_workspace_script(workspace_root: Path, script_name: str) -> Path:
    # The build lands helper scripts under `bin/`; `scripts/` is the repo layout.
    return _resolve_workspace_file(workspace_root, ("bin", "scripts"), script_name)


def resolve_workspace_validator(workspace_root: Path, validator_name: str) -> Path:
    return _resolve_workspace_file(workspace_root, ("validators",), validator_name)


def _resolve_workspace_file(
    workspace_root: Path,
    subdirs: tuple[str, ...],
    name: str,
) -> Path:
    candidates = [workspace_root / subdir / name for subdir in subdirs]
    candidates.append(workspace_root / name)
    for candidate in candidates:
        if candidate.is_file():
            return candidate.resolve()
    searched = ", ".join(str(candidate) for candidate in candidates)
    raise FinalizeError(f"{name} not found (searched: {searched})")


def tail(text: str, *, limit: int = 4000) -> str:
    if len(text) <= limit:
        return text
    return text[-limit:]


@dataclass(frozen=True)
class FinalizeContext:
    """Everything the Phase 7 sequence needs, resolved from a run manifest."""

    project_root: Path
    workspace_root: Path
    manifest_path: Path
    team_state_path: Path
    data_path: Path
    task_manifest_path: Path
    task_key: str
    task_type: str
    task_group: str
    task_id: str
    seq: str
    final_status_path: Path | None
    report_contract_version: str = "2.0"

    @property
    def markdown_path(self) -> Path:
        return final_report_markdown_path(self.data_path)

    @classmethod
    def from_manifest(
        cls,
        *,
        project_root: Path,
        workspace_root: Path,
        manifest_path: Path,
        team_state_path: Path,
        data_path: Path,
        manifest: Mapping[str, Any],
    ) -> "FinalizeContext":
        return cls(
            project_root=project_root,
            workspace_root=workspace_root,
            manifest_path=manifest_path,
            team_state_path=team_state_path,
            data_path=data_path,
            task_manifest_path=task_manifest_path(project_root, manifest),
            task_key=require_string(manifest, "taskKey"),
            task_type=require_string(manifest, "taskType"),
            task_group=task_group(manifest),
            task_id=task_id(manifest),
            seq=report_seq(manifest),
            # run 매니페스트는 이 경로를 `expectedStatusPath` 로 싣는다(render.py).
            # `finalStatusPath` 는 team-state 의 키라 여기서는 항상 비어 있었고,
            # 그래서 validate-run 이 `--final-status` 를 받지 못해 `.status` 파일이
            # 어느 run 에서도 생기지 않았다(실측 2026-09-06: 두 프로젝트 전체 1개).
            final_status_path=resolve_optional_path(
                project_root, manifest.get("expectedStatusPath")
            ),
            report_contract_version=str(
                manifest.get("reportContractVersion") or "2.0"
            ),
        )


def build_commands(ctx: FinalizeContext) -> list[tuple[str, list[str]]]:
    """Assemble the ordered Phase 7 argv list. Order is contractual."""
    markdown_path = ctx.markdown_path
    commands = [
        (
            STEP_PROJECT_ACTIVITY,
            [
                "<in-process>",
                "agent-activity",
                "project",
                str(ctx.manifest_path),
                str(ctx.data_path),
            ],
        ),
        (
            STEP_TRANSLATE,
            ["<in-process>", "translate", str(ctx.data_path)],
        ),
        (
            STEP_TOKEN_USAGE,
            [
                sys.executable,
                str(resolve_workspace_script(ctx.workspace_root, "okstra-token-usage.py")),
                str(ctx.team_state_path),
                "--project-root",
                str(ctx.project_root),
                "--write",
                "--substitute-data",
                str(ctx.data_path),
            ],
        ),
        (
            STEP_RENDER_VIEWS,
            [
                sys.executable,
                str(
                    resolve_workspace_script(
                        ctx.workspace_root, "okstra-render-report-views.py"
                    )
                ),
                str(ctx.data_path),
                "--task-key",
                ctx.task_key,
                "--task-type",
                ctx.task_type,
                "--seq",
                ctx.seq,
                # 사이드카가 되짚는 기준은 태스크 루트다. 프로젝트 루트
                # 기준으로 넘기던 동안 HTML Export 가 만든 사이드카는
                # 방향 선택 게이트를 통과할 수 없었다.
                "--source-report",
                sidecar_source_rel(markdown_path),
                # 머리말 소요 시간의 출처. 보고서 seq 로 되짚으면 seq 가 갈린
                # run 의 team-state 를 읽는다 — 여기서는 이미 알고 있다.
                "--team-state",
                str(ctx.team_state_path),
            ],
        ),
        (
            STEP_SPAWN_FOLLOWUPS,
            [
                sys.executable,
                str(
                    resolve_workspace_script(
                        ctx.workspace_root, "okstra-spawn-followups.py"
                    )
                ),
                str(ctx.data_path),
                "--project-root",
                str(ctx.project_root),
                "--task-group",
                ctx.task_group,
                "--parent-task-key",
                ctx.task_key,
            ],
        ),
        (STEP_VALIDATE_RUN, _validate_run_command(ctx, ctx.data_path)),
        (
            STEP_RECORD_GROUP_MEMORY,
            ["<in-process>", "record-group-memory", str(ctx.data_path)],
        ),
        (
            STEP_TEARDOWN_STAGES,
            ["<in-process>", "teardown-stages", str(ctx.data_path)],
        ),
    ]
    if ctx.report_contract_version != "3.0":
        return commands
    by_name = dict(commands)
    by_name[STEP_PREFLIGHT] = [
        *_validate_run_command(ctx, ctx.data_path),
        "--section",
        "preflight",
    ]
    usage = by_name[STEP_TOKEN_USAGE]
    marker = usage.index("--substitute-data")
    by_name[STEP_TOKEN_USAGE] = usage[:marker]
    by_name[STEP_PROJECT_ACTIVITY] = [
        "<in-process>", "report-assembly", str(ctx.manifest_path), str(ctx.data_path)
    ]
    return [(name, by_name[name]) for name in V3_STEP_ORDER]


def _project_relative(ctx: FinalizeContext, path: Path) -> str:
    try:
        return path.resolve().relative_to(ctx.project_root.resolve()).as_posix()
    except ValueError:
        return str(path)


def _record_group_memory(
    ctx: FinalizeContext,
    command: Sequence[str],
) -> subprocess.CompletedProcess[str]:
    """이 run 의 결론을 그룹 문서의 okstra 영역에 쓴다.

    형제 task 의 다음 prepare 가 그 영역을 패킷 `## Task-Group Memory` 로 싣는다.
    그룹 문서가 없으면 만든다 — 메모리는 문서를 채웠는지에 달린 것이 아니다.
    레코드가 없으면 기억할 것이 없으므로 건너뛴다(실패가 아니다).
    """
    if not ctx.data_path.is_file():
        return subprocess.CompletedProcess(
            command, 0, json.dumps({"skipped": f"no final report record at {ctx.data_path}"}), ""
        )
    try:
        data = load_owned_object(ctx.data_path, artifact="final report record")
    except JsonBoundaryError as exc:
        return subprocess.CompletedProcess(
            command, 1, "", f"cannot read final-report data.json: {exc}")
    pointer, _ = _recorded_next_phase(ctx)
    next_phase = _next_phase_label(pointer)
    entry = group_context.memory_entry_from_record(
        data,
        task_id=ctx.task_id,
        task_type=ctx.task_type,
        seq=ctx.seq,
        record=_project_relative(ctx, ctx.data_path),
        next_phase=next_phase,
        today=date.today(),
    )
    try:
        written, queue = group_context.record_task_memory(
            ctx.project_root, ctx.task_group, entry, today=date.today()
        )
    except OSError as exc:
        return subprocess.CompletedProcess(
            command, 1, "", f"cannot write group memory: {exc}")
    following = group_context.next_in_group(queue)
    return subprocess.CompletedProcess(
        command, 0,
        json.dumps({
            "written": _project_relative(ctx, written),
            "taskId": entry.task_id,
            "nextInGroup": (
                {"taskId": following.task_id, "briefId": following.brief_id, "brief": following.brief}
                if following else None
            ),
        }),
        "",
    )


def _next_phase_label(pointer: Mapping[str, Any]) -> str:
    """`<phase> (<status>)`; a terminal pointer names no phase, so it reads `done (terminal)`."""
    phase = str(pointer.get("phase") or "").strip()
    status = str(pointer.get("status") or "").strip()
    if phase:
        return f"{phase} ({status})" if status else phase
    if status == STATUS_TERMINAL:
        return f"done ({STATUS_TERMINAL})"
    return ""


def _next_in_group(steps: Sequence[Mapping[str, Any]]) -> dict[str, str] | None:
    """`record-group-memory` 가 낸 그룹의 다음 task. 단계가 안 돌았거나 없으면 None."""
    for step in steps:
        if step.get("name") != STEP_RECORD_GROUP_MEMORY or step.get("exitCode") != 0:
            continue
        try:
            payload = json.loads(step.get("stdoutTail") or "")
        except json.JSONDecodeError:
            return None
        following = payload.get("nextInGroup") if isinstance(payload, Mapping) else None
        return dict(following) if isinstance(following, Mapping) else None
    return None


def _teardown_stage_worktrees(
    ctx: FinalizeContext,
    command: list[str],
) -> subprocess.CompletedProcess:
    """판정이 릴리스로 향할 때만 stage worktree 와 registry 키를 정리한다.

    whole-task 진입이 통합만 하고 정리를 남겨두므로(`stage_targets`), 정리는 판정이
    나온 뒤인 여기서 한다. `blocked` 이거나 릴리스를 막는 조건이 남은 판정에서는
    stage 작업물을 그대로 둬서 재작업이 바로 이어지게 한다. 이미 정리된 뒤 재실행돼도
    같은 결과를 낸다 — 병합은 `already_merged` 로, 없는 worktree 는 건너뛴다.
    """
    def _done(payload: Mapping[str, Any]) -> subprocess.CompletedProcess:
        return subprocess.CompletedProcess(command, 0, json.dumps(payload), "")

    if ctx.task_type != "final-verification":
        return _done({"skipped": "not a final-verification run"})
    try:
        data = load_owned_object(Path(ctx.data_path), artifact="final report record")
    except JsonBoundaryError as exc:
        return subprocess.CompletedProcess(
            command, 1, "", f"cannot read final-report data.json: {exc}")
    if data.get("verificationScope") != "whole-task":
        return _done({"skipped": "single-stage verification owns no teardown"})
    if not release_handoff_allowed(data):
        return _done({"skipped": "verdict does not clear the work for release"})
    try:
        result = integrate_and_teardown_whole_task(
            project_root=ctx.project_root,
            task_group=ctx.task_group,
            task_id=ctx.task_id,
        )
    except (IntegrateError, StageTargetError, OSError) as exc:
        return subprocess.CompletedProcess(command, 1, "", str(exc))
    return _done(result)


def _validate_run_command(ctx: FinalizeContext, report_record_path: Path) -> list[str]:
    command = [
        sys.executable,
        str(resolve_workspace_validator(ctx.workspace_root, "validate-run.py")),
        "--team-state",
        str(ctx.team_state_path),
        "--report",
        str(report_record_path),
        "--run-manifest",
        str(ctx.manifest_path),
        "--task-manifest",
        str(ctx.task_manifest_path),
    ]
    if ctx.final_status_path is not None:
        command.extend(["--final-status", str(ctx.final_status_path)])
    return command


def step_payload(
    name: str,
    command: Sequence[str],
    result: subprocess.CompletedProcess[str],
) -> dict[str, Any]:
    payload = {
        "name": name,
        "command": list(command),
        "exitCode": result.returncode,
        "stdoutTail": tail(result.stdout),
        "stderrTail": tail(result.stderr),
    }
    if name == STEP_PROJECT_ACTIVITY and result.returncode != 0 and result.stdout:
        # 소유자 오류는 요약 문자열 길이에 잘리지 않은 전체 목록으로 전달한다.
        payload["issues"] = json.loads(result.stdout).get("issues", [])
    return payload


def run_finalize(
    ctx: FinalizeContext,
    *,
    before_step: Callable[[str], None] | None = None,
    only: Sequence[str] | None = None,
) -> dict[str, Any]:
    """계약 순서로 Phase 7 을 돌리고, 실패해도 `validate-run` 까지 모은다.

    `before_step` 은 각 단계 프로세스 직전에 호출된다. Codex 어댑터가
    `validate-run` 앞에 작성기 상태를 `completed` 로 표시하는 자리이다.

    `only` 는 고른 단계만 계약 순서로 남긴다.
    """
    steps: list[dict[str, Any]] = []
    try:
        if ctx.report_contract_version != "3.0":
            write_execution_roles(ctx)
        commands = build_commands(ctx)
    except FinalizeError as exc:
        return {"ok": False, "reason": str(exc), "steps": steps}

    if only:
        selected = set(only)
        if ctx.report_contract_version == "3.0":
            if STEP_TRANSLATE in selected:
                selected.add(STEP_PREFLIGHT)
            if selected - {STEP_TOKEN_USAGE}:
                # 부분 재개도 이전 조립본이 아닌 현재 소유자 입력을 소비한다.
                selected.add(STEP_PROJECT_ACTIVITY)
        contract_order = (
            V3_STEP_ORDER if ctx.report_contract_version == "3.0" else STEP_ORDER
        )
        unknown = sorted(selected - set(contract_order))
        if unknown:
            return {
                "ok": False,
                "reason": f"unknown finalize step(s): {', '.join(unknown)}",
                "steps": steps,
            }
        all_commands = commands
        commands = [(name, cmd) for name, cmd in commands if name in selected]
        commands = _with_view_the_validator_reads(
            ctx, commands, selected, all_commands
        )

    steps, first_failure = _execute_finalize_steps(ctx, commands, before_step)
    pointer, pointer_error = _recorded_next_phase(ctx)
    payload: dict[str, Any] = {
        "ok": not first_failure,
        "reason": first_failure,
        "steps": steps,
        "nextRecommendedPhase": pointer,
    }
    if pointer_error:
        payload["nextRecommendedPhaseError"] = pointer_error
    following = _next_in_group(steps)
    if following:
        payload["nextInGroup"] = following
    if first_failure:
        payload["recovery"] = _finalize_recovery(ctx, payload)
    # closeout 이 실제로 건넬 명령. 표를 리드의 기억에 맡기지 않는다.
    payload["nextCommand"] = closeout_command(payload)
    payload["reportPaths"] = _closeout_report_paths(ctx)
    return payload


def _execute_finalize_steps(
    ctx: FinalizeContext, commands: Sequence[tuple[str, list[str]]],
    before_step: Callable[[str], None] | None,
) -> tuple[list[dict[str, Any]], str]:
    """처리 순서를 지키며 각 실패를 기록하고, 전체 성공일 때 실행을 닫는다."""
    steps: list[dict[str, Any]] = []
    first_failure = ""
    validated = False
    blocked: dict[str, str] = {}
    for name, command in commands:
        # 실패한 시퀀스가 worktree 를 거두면 재작업 대상이 사라지고, 검증 안 된
        # 레코드를 기억하면 형제 task 가 그것을 결론으로 읽는다.
        if name in _SKIPPED_AFTER_FAILURE and first_failure and name not in blocked:
            continue
        if name in blocked:
            steps.append({"name": name, "status": "skipped", "exitCode": None,
                          "blockedBy": blocked[name]})
            continue
        if before_step is not None:
            before_step(name)
        result = _run_finalize_step(ctx, name, command)
        step = step_payload(name, command, result)
        if result.returncode != 0:
            step["errorLogAppend"] = record_runtime_failure(
                ctx.manifest_path, project_root=ctx.project_root,
                command=f"report-finalize {name}", exit_code=result.returncode,
                detail="\n".join(filter(None, [result.stderr, result.stdout])) or f"{name} exited {result.returncode}",
            )
        steps.append(step)
        if result.returncode != 0:
            if name == STEP_PROJECT_ACTIVITY:
                for consumer, _ in commands:
                    if consumer != STEP_TOKEN_USAGE or ctx.report_contract_version != "3.0":
                        blocked[consumer] = name
            elif name == STEP_PREFLIGHT:
                blocked[STEP_TRANSLATE] = name
                blocked[STEP_SPAWN_FOLLOWUPS] = name
            elif name == STEP_RENDER_VIEWS:
                blocked[STEP_VALIDATE_RUN] = name
        if result.returncode != 0 and not first_failure:
            first_failure = f"{name} failed with exit code {result.returncode}"
        if name == STEP_VALIDATE_RUN and result.returncode == 0:
            validated = True
    if validated and not first_failure:
        _record_run_end(ctx.team_state_path)
    return steps, first_failure


def _finalize_recovery(ctx: FinalizeContext, result: Mapping[str, Any]) -> dict[str, Any]:
    """새 실행 대신 실패한 처리부터 재개할 명령과 원본 소유자 오류를 전달한다."""
    order = V3_STEP_ORDER if ctx.report_contract_version == "3.0" else STEP_ORDER
    failed = {step["name"] for step in result.get("steps") or [] if step.get("exitCode")}
    pointer = promote_next_phase(result.get("nextRecommendedPhase"))
    if (failed == {STEP_VALIDATE_RUN} and pointer["status"] == STATUS_BLOCKED
            and pointer["phase"] and pointer["phase"] != ctx.task_type):
        return {"mode": "phase-reentry", "phase": pointer["phase"], "instruction": pointer["rationale"]}
    command = [
        "okstra", "report-finalize", "--project-root", str(ctx.project_root),
        "--run-manifest", str(ctx.manifest_path), "--report", str(ctx.data_path),
        "--team-state", str(ctx.team_state_path),
    ]
    resume_steps = _recovery_step_names(result, order)
    if ctx.report_contract_version == "3.0" and STEP_PROJECT_ACTIVITY not in resume_steps:
        # 수정한 서사·판정 상태를 정본에 다시 조립해야 후속 검사가 새 입력을 읽는다.
        resume_steps = list(order[order.index(STEP_PROJECT_ACTIVITY):])
    for name in resume_steps:
        command.extend(["--only", name])
    issues = []
    for step in result.get("steps") or []:
        if step.get("name") != STEP_PROJECT_ACTIVITY or not step.get("exitCode"):
            continue
        issues.extend(step.get("issues") or [])
    return {
        "mode": "same-run", "resumeCommand": command, "issues": issues,
        "instruction": (
            "Repair the reported causes in this run using the owning input's correction command, "
            "then execute resumeCommand. Preserve approvals, model choices and completed evidence. "
            "If plan content changes, re-verify the affected items before finalizing. "
            "After a narrative correction, check existing translations with report-translate check-data "
            "and regenerate them if their source no longer matches. "
            "Ask only for an unresolved user decision or an actual external prerequisite."
        ),
    }


def _closeout_report_paths(ctx: FinalizeContext) -> dict[str, Any]:
    """closeout 이 인용할 task 한정 경로와, 그대로 답장에 붙일 링크.

    리드 계약은 이 답장의 모든 run 산출물 경로가 task 한정이어야 한다고 적는데,
    리드가 그 경로를 직접 조립하면 `runs/<task-type>/reports/...` 형태가 나온다 —
    같은 task-type 의 모든 task 에서 바이트까지 같아 어느 task 인지 못 가린다.
    조립할 값을 여기서 만들어 건넨다.

    `markdown` 은 같은 경로를 `[이름](경로)` 로 미리 렌더한 것이다. 호스트는
    답장을 마크다운으로 그리므로 괄호 안에 경로가 들어간 형태만 클릭되고,
    백틱으로 감싼 경로는 글자로만 남아 사용자가 직접 옮겨 적어야 한다. 리드가
    그 형태를 매번 다시 만들면 어긋날 자리가 생기므로 완성된 문자열을 준다.
    """
    def rel(path: Path) -> str:
        try:
            return path.resolve().relative_to(ctx.project_root.resolve()).as_posix()
        except ValueError:
            return str(path)

    paths = {
        "humanReport": rel(html_view_path(ctx.data_path)),
        "reportRecord": rel(ctx.data_path),
        "teamState": rel(ctx.team_state_path),
    }
    return {
        **paths,
        "renderFullCopy": f"okstra render-final-report {paths['reportRecord']}",
        "markdown": {
            key: f"[{Path(value).name}]({value})"
            for key, value in paths.items()
        },
    }


def _with_view_the_validator_reads(
    ctx: FinalizeContext,
    commands: list[tuple[str, list[str]]],
    selected: set[str],
    all_commands: list[tuple[str, list[str]]],
) -> list[tuple[str, list[str]]]:
    """`validate-run` 만 고른 호출에 빠진 열람본 렌더를 채워 넣는다.

    `validate-run` 은 HTML 열람본이 없으면 run 을 `contract-violated` 로 끝낸다
    (`okstra_ctl.blocking_checks`). 그런데 비영어 리포트의 Phase 7 은 번역
    워커를 사이에 두고 두 번에 나눠 도는 절차라, 번역이 막히면 렌더는 한 번도
    돌지 않은 채 검증만 도는 조합이 실제로 만들어진다 — 관측된 run 이 그렇게
    끝났다(2026-09-08 FontsNinja/app/jobs dev-10784 error-analysis: 번역
    디스패치가 호스트 승인에 막혀 열람본이 없었고, 그 부재 하나가 유일한 차단
    실패였다).

    열람본은 번역과 무관하게 만들어진다 — `okstra_ctl.report_views` 는 번역
    사이드카를 읽지 않고 리포트 본문 그대로 렌더한다. 그러니 없으면 만든다.
    번역이 나중에 도착하면 렌더를 다시 돌려 덮어쓰는 것이 그 절차의 마지막
    단계이고, 렌더는 그렇게 쓰도록 idempotent 하다.
    """
    if STEP_VALIDATE_RUN not in selected or STEP_RENDER_VIEWS in selected:
        return commands
    if html_view_path(ctx.data_path).is_file():
        return commands
    render = [
        (name, cmd) for name, cmd in all_commands if name == STEP_RENDER_VIEWS
    ]
    if not render:
        return commands
    contract_order = (
        V3_STEP_ORDER if ctx.report_contract_version == "3.0" else STEP_ORDER
    )
    merged = commands + render
    position = {name: index for index, name in enumerate(contract_order)}
    merged.sort(key=lambda row: position[row[0]])
    return merged


def _record_run_end(team_state_path: Path) -> None:
    """검증이 처음 통과한 시각을 `team-state.runEndedAt` 에 적는다. 이미 있으면 그대로.

    토큰 수집기(`okstra_token_usage.collect.resolve_run_window`)는 이 값을 창의
    끝으로 읽는데 어떤 코드도 적지 않았다. 그래서 완료된 run 을 다시 finalize
    하면 창의 끝이 지금이 되어, 같은 세션이 그 뒤에 돌린 run 이 이 run 의 리드
    사용량에 들어왔다(관측 2026-09-03, dev-10626 error-analysis r04: 하루 뒤
    재수집에 오늘 run 이 포함, 리드 3h → 11h). run 이 끝났다고 판정하는 자리는
    검증 통과이므로 여기서 한 번만 적는다. 실패해도 finalize 는 이미 끝난
    시퀀스라 결과를 바꾸지 않고 경고만 낸다.
    """
    ended = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    def stamp(state: dict[str, Any]) -> bool:
        if state.get("runEndedAt"):
            return False
        state["runEndedAt"] = ended
        return True

    try:
        mutate_team_state(team_state_path, stamp)
    except (DispatchError, OSError) as exc:
        print(
            f"report-finalize: could not record runEndedAt on {team_state_path} "
            f"({exc}); the next usage collection will end at the current time",
            file=sys.stderr,
        )


def _recorded_next_phase(ctx: FinalizeContext) -> tuple[dict[str, str], str]:
    """`validate-run` 이 방금 태스크 매니페스트에 쓴 다음 phase 포인터.

    리드의 closeout 표는 이 포인터의 `status` / `rationale` 로 분기하는데,
    포인터는 리드가 쓰는 필드가 아니고(`prompts/lead/report-writer.md`
    "Nobody writes this pointer by hand"), 이 시퀀스의 결과에도 실리지
    않았다. 값 없이 표만 받은 리드에게 남는 행동은 상태를 늘어놓고 턴을
    끝내는 것뿐이다.

    여기서 다시 계산하지 않는다 — 투영은 `validate-run` 의
    `update_workflow_metadata` 가 하고, 이 함수는 그것이 파일에 남긴 값을
    읽어 옮기기만 한다. `--only` 로 `validate-run` 을 건너뛴 호출에서는
    이전 run 이 남긴 값이 그대로 읽히므로, 이번 시퀀스가 그 단계를 돌았는지는
    `steps` 로 판단한다.

    읽기가 실패하면 값을 지어내지 않고 사유를 함께 돌려준다.
    """
    try:
        manifest = load_owned_object(
            ctx.task_manifest_path, artifact="task-manifest"
        )
    except (JsonBoundaryError, OSError) as exc:
        return promote_next_phase(None), f"{ctx.task_manifest_path}: {exc}"
    workflow = manifest.get("workflow")
    if not isinstance(workflow, Mapping):
        return promote_next_phase(None), ""
    return promote_next_phase(workflow.get("nextRecommendedPhase")), ""


def _run_finalize_step(
    ctx: FinalizeContext,
    name: str,
    command: Sequence[str],
) -> subprocess.CompletedProcess[str]:
    """한 Phase 7 단계를 실행하고 그 단계의 종료 코드만 돌려준다."""
    if name == STEP_PROJECT_ACTIVITY:
        return _run_project_activity(ctx, command)
    if name == STEP_TRANSLATE:
        return _run_translate(ctx, command)
    if name == STEP_TEARDOWN_STAGES:
        return _teardown_stage_worktrees(ctx, command)
    if name == STEP_RECORD_GROUP_MEMORY:
        return _record_group_memory(ctx, command)
    if name == STEP_VALIDATE_RUN:
        try:
            _link_lead_result_for_validation(ctx)
        except (DispatchError, OSError, json.JSONDecodeError) as exc:
            return subprocess.CompletedProcess(
                command, 1, "", f"lead result linkage failed: {exc}"
            )
    return subprocess.run(
        command,
        cwd=ctx.project_root,
        text=True,
        capture_output=True,
    )


def _run_translate(
    ctx: FinalizeContext,
    command: Sequence[str],
) -> subprocess.CompletedProcess[str]:
    """비영어 리포트의 번역 사이드카를 만든다. 영어 리포트는 건너뛴다.

    실패는 다른 단계처럼 기록만 하고 시퀀스는 계속 돈다 — `render-views` 는
    사이드카 없이 영어 열람본을 내고, `validate-run` 은 권고를 남기며, 결과의
    `--only translate --only render-views …` 재개 힌트가 그 둘을 다시 돌린다.
    """
    try:
        outcome = translate_report(
            project_root=ctx.project_root,
            workspace_root=ctx.workspace_root,
            manifest_path=ctx.manifest_path,
            manifest=_load_manifest(ctx.manifest_path),
            data_path=ctx.data_path,
        )
    except (FinalizeError, OSError, JsonBoundaryError) as exc:
        outcome = TranslateOutcome(1, "", f"translate failed: {exc}")
    return subprocess.CompletedProcess(
        command, outcome.returncode, outcome.stdout, outcome.stderr
    )


def _run_project_activity(
    ctx: FinalizeContext,
    command: Sequence[str],
) -> subprocess.CompletedProcess[str]:
    """계약 3.0 은 조립, 그 외는 활동 투영만 한다."""
    try:
        if ctx.report_contract_version == "3.0":
            assembled = assemble_report(ctx.project_root, ctx.manifest_path)
            _point_task_manifest_at_report(ctx)
            count = len(assembled.get("agentActivity") or [])
        else:
            rows = project_agent_activity(
                ctx.project_root,
                ctx.manifest_path,
                ctx.data_path,
            )
            count = len(rows)
    except ReportAssemblyError as exc:
        return subprocess.CompletedProcess(
            command, 1, json.dumps({"issues": [asdict(issue) for issue in exc.issues]}), str(exc)
        )
    except ActivityProjectionError as exc:
        return subprocess.CompletedProcess(command, 1, "", str(exc))
    return subprocess.CompletedProcess(
        command, 0, json.dumps({"count": count}), ""
    )


def write_execution_roles(ctx: FinalizeContext) -> None:
    """Write the exact manifest role set onto the final-report data.json."""
    if not ctx.data_path.is_file() or not ctx.manifest_path.is_file():
        return
    manifest = _load_manifest(ctx.manifest_path)
    data = _load_manifest(ctx.data_path)
    updated = apply_execution_roles(dict(data), dict(manifest))
    if updated == data:
        return
    write_owned_object_atomic(ctx.data_path, updated, artifact="final report record")


def _point_task_manifest_at_report(ctx: FinalizeContext) -> None:
    """게시한 data.json 을 latestReport 로 올린다.

    render-only prepare 는 기존 포인터를 보존한다. 조립이 새 리포트를 썼는데도
    task-manifest 가 예전 seq 를 가리키면 검사가 통과해도 상태가 004 에 남는다.
    """
    path = ctx.task_manifest_path
    if not path.is_file() or not ctx.data_path.is_file():
        return
    try:
        payload = load_owned_object(path, artifact="task manifest")
    except (OSError, JsonBoundaryError):
        return
    if not isinstance(payload, Mapping):
        return
    relative = _project_relative(ctx, ctx.data_path)
    if payload.get("latestReportRecordPath") == relative:
        return
    updated = dict(payload)
    updated["latestReportRecordPath"] = relative
    write_owned_object_atomic(path, updated, artifact="task manifest")


def _link_lead_result_for_validation(ctx: FinalizeContext) -> None:
    """Link the lead-authored synthesis artifact before Phase 7 validation."""
    manifest = _load_manifest(ctx.manifest_path)
    contract = manifest.get("agentContract")
    if not isinstance(contract, Mapping) or contract.get("schemaVersion") != 1:
        return
    team_state = _load_manifest(ctx.team_state_path)
    lead_dispatches = [
        row for row in (team_state.get("agentDispatches") or [])
        if isinstance(row, Mapping)
        and (
            row.get("audience") == "lead"
            or row.get("workerId") == "lead"
            or str(row.get("assignmentRef") or "") in {"lead", "lead/lead"}
        )
    ]
    if len(lead_dispatches) != 1:
        return
    convergence_value = string_value(manifest.get("convergenceStatePath"))
    convergence_path = (
        resolve_project_path(ctx.project_root, convergence_value)
        if convergence_value else None
    )
    if convergence_path is not None and convergence_path.is_file():
        result_path = convergence_path
    elif ctx.task_type == "release-handoff" and ctx.data_path.is_file():
        result_path = ctx.data_path
    else:
        return
    link_agent_dispatch_result(
        project_root=ctx.project_root,
        run_manifest_path=ctx.manifest_path,
        dispatch_id=str(lead_dispatches[0]["dispatchId"]),
        result_path=result_path,
    )


def _load_manifest(path: Path) -> Mapping[str, Any]:
    try:
        payload = load_owned_object(path, artifact="run manifest")
    except JsonBoundaryError as exc:
        raise FinalizeError(f"run manifest is not valid: {path} ({exc})") from exc
    if not isinstance(payload, Mapping):
        raise FinalizeError(f"run manifest is not a JSON object: {path}")
    return payload


def _context_from_args(args: argparse.Namespace) -> FinalizeContext:
    project_root = Path(args.project_root).resolve()
    manifest_path = resolve_project_path(project_root, args.run_manifest)
    manifest = _load_manifest(manifest_path)
    if args.team_state:
        team_state_path = resolve_project_path(project_root, args.team_state)
    else:
        team_state_path = resolve_project_path(
            project_root, require_string(manifest, "teamStatePath")
        )
    report_path = resolve_project_path(project_root, args.report)
    return FinalizeContext.from_manifest(
        project_root=project_root,
        workspace_root=Path(args.workspace_root).resolve(),
        manifest_path=manifest_path,
        team_state_path=team_state_path,
        data_path=final_report_data_path(report_path),
        manifest=manifest,
    )


_CLI_EPILOG = r"""Usage:
  okstra report-finalize --project-root <dir> --run-manifest <path> \
    --report <final-report-<task-type>-<seq>.md> [--team-state <path>]

Runs the Phase 7 steps in their contractual order against one final-report:

  1. project-activity  project the run's activity events into the data.json
  2. translate         for a non-English reportLanguage, materialize and
                       dispatch the translator worker and require its
                       *.i18n.<lang>.json sidecar; a no-op for English or
                       when the sidecar already exists
  3. token-usage       substitute real token/cost numbers into the data.json
  4. render-views      write the schema-v2 task-specific *.html sibling with
                       the translation overlaid; v1 keeps the
                       legacy conditional interactive view
  5. spawn-followups   turn section 4 rows into task stubs
  6. validate-run      validate the finished run artifacts
  7. record-group-memory / 8. teardown-stages  after a clean validation

Every step is idempotent, so re-running after a fixed failure is safe. The
sequence stops at the first non-zero exit and reports which step failed, except
token-usage: its input is the lead session log, so a failure there does not
take the html, follow-ups and validation after it down with it. Whenever a step
exits non-zero the command prints the --only flags that resume the sequence
from the earliest failing step — follow those rather than rerunning the failing
step alone, which would leave the html rendered before the tokens landed.

This is the same code path the Codex lead adapter runs automatically after its
report-writer completes, so a Claude-led and a Codex-led run finalize
identically.

  --workspace-root is owned by this command.
"""


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="okstra report-finalize",
        epilog=_CLI_EPILOG,
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description="Run the Phase 7 report post-processing sequence.",
    )
    parser.add_argument("--project-root", required=True)
    parser.add_argument("--run-manifest", required=True)
    parser.add_argument(
        "--report",
        required=True,
        help="final-report record (.data.json); a leftover markdown path still locates the sibling record",
    )
    # `okstra report-finalize` 래퍼(src/lib/python-command.mts)가 넣고, 사용자가
    # 직접 주면 거절한다. usage 에 필수로 찍히면 그 거절과 모순되므로 감춘다.
    parser.add_argument("--workspace-root", required=True, help=argparse.SUPPRESS)
    parser.add_argument(
        "--team-state",
        default="",
        help="defaults to the run manifest's teamStatePath",
    )
    parser.add_argument(
        "--only",
        action="append",
        default=[],
        choices=list(dict.fromkeys((*STEP_ORDER, *V3_STEP_ORDER))),
        help=(
            "run only these steps (repeatable, contractual order preserved). "
            "Use `--only validate-run` to retry the step that usually fails "
            "without repeating the preceding idempotent steps."
        ),
    )
    return parser


def _step_summary_lines(
    result: Mapping[str, Any], order: Sequence[str] = STEP_ORDER,
) -> list[str]:
    """One human-readable line per step, so the outcome is legible without
    parsing the JSON payload."""
    lines = []
    for step in result.get("steps") or []:
        code = step.get("exitCode")
        if step.get("status") == "skipped":
            lines.append(f"  [skip] {step.get('name')} (blocked by {step.get('blockedBy')})")
            continue
        mark = "ok  " if code == 0 else "FAIL"
        lines.append(f"  [{mark}] {step.get('name')} (exit {code})")
    for name in order:
        if not any(s.get("name") == name for s in (result.get("steps") or [])):
            lines.append(f"  [skip] {name}")
    return lines


def _recovery_step_names(
    result: Mapping[str, Any], order: Sequence[str] = STEP_ORDER,
) -> list[str]:
    """The steps a retry has to re-run: every step from the earliest failure on.

    Naming only the failed steps would prescribe half a recovery. `token-usage`
    can fail while `render-views` still writes html from unsubstituted data;
    substituting the tokens on a retry leaves that view stale
    (`validators/validate-report-views.py` checks `source-sha256` against the
    md body). Resuming from the earliest failure redoes the tail while skipping
    the prefix that succeeded — the saving `--only` exists for.
    """
    failed = {
        string_value(step.get("name"))
        for step in (result.get("steps") or [])
        if step.get("exitCode") not in (None, 0)
    }
    for index, name in enumerate(order):
        if name in failed:
            return list(order[index:])
    return []


def closeout_command(result: Mapping[str, Any]) -> dict[str, str]:
    """이 run 이 사용자에게 건네야 할 명령.

    `prompts/launch.template.md` "User closeout (BLOCKING)" 의 표를 코드로
    내린 것이다. 그 표는 문서에만 있었고, 표가 분기하는 포인터는 리드에게
    도착하지도 않았다 — 규칙이 있고 집행이 없는 자리였다. 여기서 판정하면
    리드가 표를 기억해 따라갈 필요가 없고, 리드가 침묵해도 명령은
    `report-finalize` 출력에 남는다.

    산문을 해석하지 않는다. `blocked` / `pending` 의 근거 문장은 이미 무엇을
    실행할지 말하고 있으므로(`okstra_ctl.next_phase` 의 승인 차단 근거,
    `validators/validate-run.py` 의 검증 실패 근거), 그 문장을 그대로 인용할
    자리로 넘기고 명령 칸은 비운다. 근거에서 명령을 다시 뽑아내는 것은
    기록된 값의 재구현이다.

    돌려주는 것은 `command` 와 `note` 두 칸이다. `command` 가 비면 근거
    문장이 그 자리를 대신한다.
    """
    recovery = result.get("recovery")
    if isinstance(recovery, Mapping) and recovery.get("mode") == "same-run":
        return {"command": "", "note": str(recovery["instruction"])}
    failed_validate = any(
        step.get("name") == STEP_VALIDATE_RUN and step.get("exitCode") != 0
        for step in (result.get("steps") or [])
        if isinstance(step, Mapping)
    )
    if failed_validate:
        recovery = promote_next_phase(result.get("nextRecommendedPhase"))
        if recovery["status"] == STATUS_BLOCKED and recovery["phase"]:
            return {
                "command": f"/okstra-run → {recovery['phase']}",
                "note": recovery["rationale"],
            }
        return {
            "command": "/okstra-run",
            "note": "validate-run failed — name the blocking cause in one line, "
                    "then re-run this same phase",
        }
    pointer = result.get("nextRecommendedPhase")
    if not isinstance(pointer, Mapping):
        return {"command": "/okstra-inspect status", "note": "no pointer recorded"}
    status = str(pointer.get("status") or "").strip()
    phase = str(pointer.get("phase") or "").strip()
    rationale = str(pointer.get("rationale") or "").strip()
    if status == STATUS_READY and phase:
        return {"command": f"/okstra-run → {phase}", "note": ""}
    if status == STATUS_TERMINAL:
        following = result.get("nextInGroup")
        if isinstance(following, Mapping) and following.get("briefId"):
            return {
                "command": "/okstra-run",
                "note": "this task is finished; the task-group's next task in start "
                        f"order is `{following['briefId']}` — start it from the brief "
                        f"`{following.get('brief') or ''}`, and name any follow-up tasks "
                        "this run registered",
            }
        return {
            "command": "",
            "note": "the lifecycle ends here — say the task is finished, name any "
                    "follow-up tasks this run registered, and do not send the user "
                    "to /okstra-inspect",
        }
    if status in (STATUS_BLOCKED, STATUS_PENDING) and rationale:
        return {
            "command": "",
            "note": "quote the rationale above and issue the command it names",
        }
    return {"command": "/okstra-inspect status", "note": ""}


def _next_phase_summary_lines(result: Mapping[str, Any]) -> list[str]:
    """다음 phase 포인터와 그것이 정하는 명령을 사람이 읽는 줄로."""
    pointer = result.get("nextRecommendedPhase")
    if not isinstance(pointer, Mapping):
        return []
    lines = [
        f"next phase status: {pointer.get('status') or '-'}",
        f"next phase: {pointer.get('phase') or '-'}",
    ]
    rationale = str(pointer.get("rationale") or "").strip()
    if rationale:
        lines.append(f"next phase rationale: {rationale}")
    error = str(result.get("nextRecommendedPhaseError") or "").strip()
    if error:
        lines.append(f"next phase pointer unreadable: {error}")
    closeout = closeout_command(result)
    lines.append(f"next command: {closeout['command'] or '-'}")
    if closeout["note"]:
        lines.append(f"next command note: {closeout['note']}")
    return lines


def main(argv: Sequence[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    try:
        ctx = _context_from_args(args)
    except FinalizeError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    # Phase 7 is the last boundary the lead crosses, and the last chance to add
    # the generations resume and compaction split it into. It runs ahead of the
    # sequence because the token-usage step below reads `leadSessionIds`.
    observe_lead_session(ctx.project_root, ctx.team_state_path)
    result = run_finalize(ctx, only=args.only or None)
    order = V3_STEP_ORDER if ctx.report_contract_version == "3.0" else STEP_ORDER
    print(json.dumps(result, indent=2, ensure_ascii=False))
    print("finalize steps:", file=sys.stderr)
    for line in _step_summary_lines(result, order):
        print(line, file=sys.stderr)
    # closeout 이 분기하는 값. JSON 을 파싱하지 않고 훑는 리드에게도 보이도록
    # 요약 줄로 한 번 더 낸다.
    for line in _next_phase_summary_lines(result):
        print(line, file=sys.stderr)
    if not result["ok"]:
        print(f"error: {result['reason']}", file=sys.stderr)
        recovery = _recovery_step_names(result, order)
        if recovery:
            flags = " ".join(f"--only {name}" for name in recovery)
            print(
                f"resume the sequence from the earliest failure with `{flags}`",
                file=sys.stderr,
            )
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())