"""prepare_task_bundle — the single python entrypoint that materializes a
complete okstra task bundle on disk.

This function replaces the ~50-step wiring that previously lived in bash
`okstra.sh`. It is called by:
  - `okstra.sh` (thin wrapper: argv → prepare_task_bundle → optional exec claude)
  - `okstra-run` skill (collects inputs via AskUserQuestion → calls this)
  - okstra-ctl rerun (passes recorded invocation argv through)

The function is read-modify-write on disk inside per-task mutex; it does not
mutate the calling process environment or rely on inherited env values for
any per-run identity. The only env vars honored are user-knob defaults
(`OKSTRA_DEFAULT_*_MODEL`, `OKSTRA_HOME`) — these are intentional config, not
state passing, and are read once at the start.
"""
from __future__ import annotations

import hashlib
from argparse import Action, ArgumentError, ArgumentParser, Namespace
import json
import os
import re
import shutil
import subprocess as _subprocess
import sys
import tempfile
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Mapping, Sequence

from okstra_project import project_json_path, upsert_project_json
from okstra_project.state import slugify
from . import fix_cycles
from .analysis_packet import build_analysis_packet
from .stage_ledger import (
    build_stage_ledger,
    render_stage_ledger,
    stage_ledger_notice,
)
from .analysis_inputs import (
    ANALYSIS_TASK_TYPES,
    AnalysisInputError,
    load_candidate_map,
    parse_evidence_paths,
    resolve_analysis_head,
    resolve_analysis_target,
    resolve_evidence_inputs,
)
from .stage_fix_carry import derive_stage_fix_carry
from .clarification_items import (
    APPROVAL_BLOCKS,
    attached_user_responses_section,
    clarification_response_with_sidecars,
    progress_blocking_ids,
    scan_approval_gate,
)
from .error_report import prior_run_error_digest
from .incremental_scope import ReverifyScopeError, parse_user_reverify_scope
from .implementation_direction import (
    DirectionSelectionError,
    SelectedDirection,
    lexical_absolute_path,
    resolve_selected_direction,
    validate_selected_direction_plan,
    validate_task_artifact_path,
    write_selected_direction_snapshot,
)
from .qa_commands import format_errors as _format_qa_errors, validate_qa_commands
from .material import (
    build_analysis_material,
    related_tasks_bullets,
    related_tasks_inline,
    resolve_related_tasks,
)
from .final_report_schema import load_schema_version
from .final_report_paths import (
    final_report_data_path as _final_report_data_path,
    final_report_markdown_path as _final_report_markdown_path,
    require_approved_plan_record as _require_approved_plan_record_path,
)
from .lead_events import LeadEvent, append_lead_event
from .assignment_environment import load_assignment_context
from .assignment_resolver import (
    AssignmentContext,
    AssignmentPlan,
    AssignmentResolutionError,
    ResolvedAssignment,
    TerminalBackend,
    resolve_assignments,
    resolve_context_lead_provider,
    resolve_dispatch_assignment,
    resolve_model_selection,
)
from .domain.host import (
    CurrentSessionModelAttestation,
    HostCapabilityMismatch,
    HostNotRegistered,
    HostSessionContext,
    ProviderUnavailable,
)
from .domain.provider import ServedModelAttestation
from .execution_identity import (
    ExecutionManifest,
    ParticipantAssignment,
    RoleExecution,
    execution_label,
    model_spec_digest,
)
from .execution_manifest import persist_static_execution_identity
from .legacy_model_selection import (
    CanonicalModelSelection,
    ModelSelectionInputError,
    deserialize_host_session_context,
    normalize_model_selection_inputs,
    serialize_host_session_context,
)
from .model_defaults import ModelDefaultScopes
from .worker_prompt_policy import (
    ANALYSIS_DUTY_BY_TASK_TYPE,
    critic_assignment_ref,
)
from .models import (
    UnknownProviderError,
    default_model,
    lead_launch_argv,
    lead_launch_spec,
    provider_default_model,
    provider_ids,
    provider_spec,
    provider_supports_role,
)
from .schema_excerpt import build_schema_excerpt
from .registry.host_registry import default_host_registry
from .registry.provider_registry import default_provider_registry
from .role_requirements import RoleProfile, RoleProfileError, load_role_profile
from .agent_invocation import (
    AgentInstruction,
    AgentInstructionSource,
    AgentInvocationRequest,
    AgentModelAssignment,
    AgentInvocationError,
    agent_model_assignment_from_payload,
    digest_duty_catalog,
    invocation_execution_identity_from_manifest,
    prepare_agent_invocation,
    verify_agent_invocation,
)
from .report_contract import CURRENT_REPORT_SCHEMA_VERSION
from .json_boundary import (
    JsonBoundaryError,
    load_owned_object,
    write_owned_object_atomic,
)
from .path_resolve import relative_to_project_root, resolve_user_file
from .render import (
    apply_lead_prompt_defaults,
    inject_lead_prompt_computed_tokens,
    migrate_legacy_run_artifacts,
    render_active_run_context,
    render_latest_task_discovery,
    render_reference_expectations,
    render_run_manifest,
    render_task_catalog_discovery,
    render_task_index,
    render_task_manifest,
    render_team_state,
    render_template_with_ctx,
    render_timeline,
)
from okstra_project.dirs import okstra_home

from .dispatch_state import (
    BACKEND_CMUX_PANE,
    detect_terminal_backend,
    generate_claude_session_id,
)
from .run_context import (
    compute_and_write_run_context,
    refresh_run_context_snapshot,
    write_run_inputs,
)
from .seeding import (
    SettingsLinkError,
    cleanup_obsolete_generated_docs,
    ensure_project_settings_symlink,
    installed_version,
    verify_installation,
)
from .session import (
    resolve_inproc_lead_session_id,
    write_claude_resume_command_file,
)
from .pr_template import PrTemplateError, resolve_pr_template_path
from .workers import (
    normalize_workers,
    resolve_optional_workers,
    resolve_profile_workers,
    validate_workers_against_profile,
)
from .workflow import compute_workflow_state, load_phase_forbidden
from .locks import worktree_provision_mutex
from . import stage_targets as _stage_targets
from .plan_run_root import plan_run_root_from_approved_plan
from . import implementation_stage as _implementation_stage
from .work_categories import resolve_work_category
from .worktree import (
    WorktreeProvision,
    provision_task_worktree,
)
from .brief_frontmatter import (
    has_reporter_confirmation_contract,
    read_brief_frontmatter,
)
from .scope_provenance import brief_end_state_id_sequence

# Frontmatter approval-flag matcher.
#
# Final-report 의 YAML frontmatter 안에서 `approved: true` / `approved: false`
# 한 줄만 식별한다. schema-v1 에서는 이 줄이 정본이고, schema-v2 에서는
# 정본 `frontmatter.approved` 의 표시다. 본문(body) 의 다른 `approved:` 등장과
# 충돌하지 않도록 호출자는 frontmatter 블록을 먼저 추출
# (`_extract_frontmatter_block`) 한 뒤 이 패턴을 적용한다.
APPROVED_FRONTMATTER_PATTERN = re.compile(
    r"^approved:[ \t]+(true|false)[ \t]*$",
    re.IGNORECASE | re.MULTILINE,
)

PLAN_BODY_GATE_PATTERN = re.compile(
    r"Gate result[^A-Za-z\n]+(?P<value>[a-z][a-z\-]+)",
    re.IGNORECASE,
)
BLOCKING_PLAN_BODY_GATES = {"blocked-by-disagreement", "aborted-non-result"}

# Frontmatter implementation-option matcher.
#
# `approved:` 바로 아래 줄의 `implementation-option:` 한 줄을 식별한다.
# report-writer 가 빈 값으로 항상 emit 하므로 라인은 존재하되 값은 비어 있을
# 수 있다. `--implementation-option <name>` CLI 가 schema-v1 에서 이 줄의
# 값만 치환한다. schema-v2 는 정본 `frontmatter.implementationOption` 을 쓰고
# 열람본을 다시 렌더한다. 값이 비면 implementation 은 plan 의
# `Recommended Option` 으로 폴백한다.
IMPLEMENTATION_OPTION_FRONTMATTER_PATTERN = re.compile(
    r"^implementation-option:[ \t]*(.*)$",
    re.MULTILINE,
)

# validators/validate-run.py:_FRONTMATTER_BLOCK_RE 의 미러 — 선행 BOM/빈 줄을
# 허용해 두 모듈이 같은 리포트의 frontmatter 게이트를 동일하게 판정하게 한다.
_FRONTMATTER_BLOCK_PATTERN = re.compile(r"\A\ufeff?\s*---\n(.*?)\n---\n", re.DOTALL)


def _extract_frontmatter_block(body: str) -> str | None:
    """final-report 의 leading `---` 펜스 사이에 든 YAML 블록 텍스트를 반환.

    펜스가 없거나 닫히지 않으면 None. report-writer 의 표준 산출물은 항상
    `---\\n...frontmatter...\\n---\\n` 로 시작한다.
    """
    m = _FRONTMATTER_BLOCK_PATTERN.match(body)
    return m.group(1) if m else None


def _load_final_report_data_if_present(report_path: Path) -> tuple[Path, dict] | None:
    data_path = _final_report_data_path(report_path)
    if not data_path.is_file():
        return None
    try:
        data = load_owned_object(data_path, artifact="approved plan final report")
    except JsonBoundaryError as exc:
        raise PrepareError(
            f"approved plan data.json is invalid JSON: {data_path}: {exc}"
        ) from exc
    if not isinstance(data, dict):
        raise PrepareError(f"approved plan data.json must be a JSON object: {data_path}")
    return data_path, data


def _data_json_gate_result(data: dict) -> str:
    planning = data.get("implementationPlanning")
    if not isinstance(planning, dict):
        return ""
    verification = planning.get("planBodyVerification")
    if not isinstance(verification, dict):
        return ""
    return str(verification.get("gateResult") or "").strip().lower()


def _blocking_gate_survives_user_decision(data: dict, gate: str) -> bool:
    """사용자가 진행 처분을 골라도 이 게이트 값이 승인을 막는가.

    `aborted-non-result` 는 투표가 없어 사용자 판단의 대상이 아니다.
    `blocked-by-disagreement` 는 승인 행이 있고 그 행이 전부 진행 처분이면
    DISAGREE 를 증거로 남긴 채 막지 않는다. 승인 행이 없으면 판단 기록이
    없으므로 막는다.
    """
    if gate != "blocked-by-disagreement":
        return True
    rows = data.get("clarificationItems")
    if not isinstance(rows, list):
        return True
    has_approval_row = any(
        isinstance(row, dict)
        and str(row.get("blocks") or "").strip().lower() in APPROVAL_BLOCKS
        for row in rows
    )
    return (not has_approval_row) or bool(
        progress_blocking_ids(rows, APPROVAL_BLOCKS, report_data=data)
    )


def _record_approved_flag(path: Path) -> bool | None:
    """정본 `frontmatter.approved`. 정본이 없으면(schema-v1) None."""
    loaded = _load_final_report_data_if_present(path)
    if loaded is None:
        return None
    data_path, data = loaded
    frontmatter = data.get("frontmatter")
    if not isinstance(frontmatter, dict) or "approved" not in frontmatter:
        raise PrepareError(
            f"approved plan data.json has no frontmatter.approved field: {data_path}"
        )
    data_approved = frontmatter.get("approved")
    if not isinstance(data_approved, bool):
        raise PrepareError(
            f"approved plan data.json frontmatter.approved is not boolean: {data_path}"
        )
    return data_approved


def _unapproved_plan_message(path: str, displayed: str) -> str:
    return (
        f"approved plan is not yet approved (frontmatter `approved: {displayed}`): "
        f"{path}\n"
        "  re-run okstra with `--approve`, or confirm approval in the "
        "in-session wizard.\n"
        "  resolve any `Blocks=approval` rows in `## 1. Clarification Items` first."
    )


def _reject_blocking_plan_body_gate(path: Path, body: str, *, action: str) -> None:
    loaded = _load_final_report_data_if_present(path)
    if loaded is not None:
        data_path, data = loaded
        data_gate = _data_json_gate_result(data)
        if (
            data_gate in BLOCKING_PLAN_BODY_GATES
            and _blocking_gate_survives_user_decision(data, data_gate)
        ):
            raise PrepareError(
                f"{action} rejected because approved plan data.json Gate result is "
                f"`{data_gate}`: {data_path}\n"
                "  resolve the plan-body verification disagreement and regenerate "
                "the implementation-planning report before approval."
            )
        return
    gate_match = PLAN_BODY_GATE_PATTERN.search(body)
    if not gate_match:
        return
    gate_value = gate_match.group("value").strip().lower()
    if gate_value in BLOCKING_PLAN_BODY_GATES:
        raise PrepareError(
            f"{action} rejected because approved plan Gate result is "
            f"`{gate_value}`: {path}\n"
            "  resolve the plan-body verification disagreement and regenerate "
            "the implementation-planning report before approval."
        )


def _validate_approved_plan_conformance(path: Path) -> None:
    """승인 계획의 stage conformance 선언 형식을 승인 경계에서 판정한다.

    같은 판정이 지금까지는 구현 런의 마지막 `validate-run` 에서만 나왔다. 그때는
    워커 배치와 수렴이 이미 끝난 뒤라 런 하나를 통째로 버려야 했다 — 실측된
    실패에서 승인 계획의 stage 2~7 이 기계 형식이 아니라 산문이었다.

    선언이 없는 stage 는 건드리지 않는다(`conformance.malformed_conformance_stages`
    참조): XOR 부재는 계획 단계의 check S11 이 이미 막고, 구현 진입 게이트도
    validate-run 이 계속 요구한다. 여기서 다시 하면 같은 규칙이 두 곳이 된다.
    """
    from .conformance import malformed_conformance_stages

    loaded = _load_final_report_data_if_present(path)
    if loaded is None:
        return
    data_path, data = loaded
    bad = malformed_conformance_stages(data)
    if not bad:
        return
    # 한 stage 씩 고치고 다시 막히는 왕복을 피하려면 전부 나열해야 한다 —
    # 실패한 런에서 검증기는 stage 2 만 지목했지만 실제로는 2~7 전부였다.
    stages = ", ".join(str(number) for number in bad)
    raise PrepareError(
        f"approved plan data.json has malformed conformanceTests for stage(s) "
        f"{stages}: {data_path}\n"
        "  each declaring stage must read "
        "`<task_root>/qa/scripts/stage-<N>.<ext> "
        "(requires=[db|io|http|external,...])` after the "
        "`Conformance tests: stage-<N> — ` prefix "
        "(prompts/profiles/implementation-planning.md), or carry "
        "`Conformance exemption: <reason>` instead.\n"
        "  re-run implementation-planning so the declaration is regenerated in "
        "that form, then approve it."
    )


def _commit_data_json_and_rerender(
    record_path: Path, data: dict, *, action: str
) -> None:
    """정본을 쓰고 전문 열람본을 그 정본에서 다시 렌더한다."""
    try:
        from .render_final_report import (
            FinalReportRenderError,
            find_default_template_for_data,
            render,
        )

        rendered = render(
            data,
            template_path=find_default_template_for_data(data),
        )
    except FinalReportRenderError as exc:
        raise PrepareError(
            f"{action} could not re-render approved plan from data.json: {exc}"
        ) from exc

    md_path = _final_report_markdown_path(record_path)
    data_tmp = record_path.with_suffix(record_path.suffix + f".tmp.{os.getpid()}")
    md_tmp = md_path.with_suffix(md_path.suffix + f".tmp.{os.getpid()}")
    data_tmp.write_text(
        json.dumps(data, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    md_tmp.write_text(rendered, encoding="utf-8")
    # 두 파일을 같이 atomic 하게 commit 할 수 없으므로, 첫 replace 직전의
    # data.json 바이트를 보관했다가 markdown replace 가 실패하면 data.json 을
    # 되돌린다. 남는 불일치 창은 두 replace 사이의 hard process kill 뿐이다.
    data_prev = record_path.read_bytes()
    data_tmp.replace(record_path)
    try:
        md_tmp.replace(md_path)
    except OSError:
        record_path.write_bytes(data_prev)
        raise


def _set_data_json_approved_true_if_present(path: Path) -> bool:
    loaded = _load_final_report_data_if_present(path)
    if loaded is None:
        return False
    data_path, data = loaded
    data_gate = _data_json_gate_result(data)
    if (
        data_gate in BLOCKING_PLAN_BODY_GATES
        and _blocking_gate_survives_user_decision(data, data_gate)
    ):
        raise PrepareError(
            f"--approve rejected because approved plan data.json Gate result is "
            f"`{data_gate}`: {data_path}"
        )
    frontmatter = data.setdefault("frontmatter", {})
    if not isinstance(frontmatter, dict):
        raise PrepareError(
            f"approved plan data.json frontmatter must be an object: {data_path}"
        )
    frontmatter["approved"] = True
    _commit_data_json_and_rerender(data_path, data, action="--approve")
    return True


# PrepareError 는 stage_targets(저수준)에 정의돼 final-verification 과 container
# 가 whole-task 통합 실패를 동일 타입으로 받는다. run.py 는 재노출만 한다.
PrepareError = _stage_targets.PrepareError


def _normalize_lead_runtime(value: str) -> str:
    runtime = (value or "claude-code").strip()
    registry = default_host_registry()
    try:
        return registry.resolve(runtime).descriptor.id
    except HostNotRegistered as exc:
        allowed = ", ".join(registry.ids())
        raise PrepareError(
            f"unsupported lead runtime: {runtime} (allowed: {allowed})"
        ) from exc


@dataclass
class PrepareInputs:
    workspace_root: Path
    project_root: Path
    project_id: str
    task_group: str
    task_id: str
    task_type: str
    brief_path: Path  # absolute, already resolved
    analysis_target: str = ""
    evidence_inputs_raw: str = ""
    directive: str = ""
    workers_override: str = ""
    role_counts_raw: tuple[str, ...] = ()
    role_models_raw: tuple[str, ...] = ()
    host_session_context: HostSessionContext = field(
        default_factory=lambda: HostSessionContext(
            host_id="",
            entry_mode="spawn-process",
            available_functions=frozenset(),
            interaction_surface="unavailable",
            current_model=CurrentSessionModelAttestation.unknown(""),
        )
    )
    lead_provider: str = ""
    lead_model: str = ""
    claude_model: str = ""
    codex_model: str = ""
    antigravity_model: str = ""
    worker_models_raw: str = ""
    report_writer_provider: str = ""
    report_writer_model: str = ""
    lead_runtime: str = "claude-code"
    lead_runtime_request: str = ""
    runtime_resolution_json: str = ""
    executor: str = ""
    critic: str = ""
    related_tasks_raw: str = ""
    work_category: str = ""
    base_ref: str = ""
    approved_plan_path: str = ""
    # implementation 전용: `--qa-waiver "<stageKey>:<reason>"` 사용자 확인형 우회.
    # prepare-time 에 task-level conformance 매니페스트 entry.waiver 를 채운다.
    qa_waiver: str = ""
    stage: str = "auto"
    # release-handoff 전용: PR 로 내보낼 stage 묶음 (csv, 예: "2,3"). 빈 값 =
    # whole-task 모드. `--stage`(impl/fv 의 Stage Map 실행/검증 선택)와는
    # 별개 채널이다.
    stages: str = ""
    clarification_response_path: str = ""  # absolute or empty
    # implementation-planning 신규 실행 전용: 검증된 방향 선택 최종 보고서.
    selected_direction_path: str = ""
    # implementation-planning 전용: 사용자가 고른 이번 재실행의 재검증 범위.
    # "" / "auto" = 리드의 `okstra incremental-scope` 판정에 맡김, "full" =
    # 전체 재검증 강제, "<stage csv>" = 그 stage 들을 impacted 로 지정.
    reverify_scope: str = ""
    # release-handoff 전용: PR 본문 템플릿 1회성 override. 빈 문자열이면
    # project.json → global config → 스킬 디폴트 순으로 해석된다.
    pr_template_path: str = ""
    render_only: bool = False
    approve_plan_ack: bool = False
    # implementation 전용: 유저가 implementation-planning final-report 에서 고른
    # Option Candidate 이름. `--implementation-option <name>` 으로 전달되어
    # approved-plan frontmatter 의 `implementation-option:` 라인을 채운다. 빈
    # 문자열이면 implementation 은 plan 의 `Recommended Option` 으로 폴백한다.
    implementation_option: str = ""
    # Phase 6 plan-body verification opt-out. Default True (round runs after
    # report-writer draft). Flipped to False by CLI `--no-plan-verification`.
    # Only meaningful for `--task-type implementation-planning`; the manifest
    # records the value for other phases too to keep the schema stable.
    plan_verification_enabled: bool = True
    # "" | "yes" | "no" — done(release-handoff) task 재진입을 fix-cycle 로
    # 기록할지의 사용자 의사. "yes" 면 새 cycle 을 열고 이번 run 을 부착한다.
    fix_cycle: str = ""


@dataclass
class PrepareOutputs:
    ctx: dict
    prompt_text: str
    extras: dict = field(default_factory=dict)


def _default(name: str, fallback: str) -> str:
    return os.environ.get(name, "") or fallback


def _approved_plan_record(path: str) -> Path:
    try:
        return _require_approved_plan_record_path(Path(path))
    except ValueError as exc:
        raise PrepareError(str(exc)) from exc


def _validate_approved_plan(path: str) -> None:
    p = _approved_plan_record(path)
    record_approved = _record_approved_flag(p)
    if record_approved is not True:
        raise PrepareError(_unapproved_plan_message(str(p), "false"))
    _reject_blocking_plan_body_gate(p, "", action="approved plan validation")
    _validate_approved_plan_conformance(p)
    # frontmatter approved == true 상태. §1 Clarification Items 의
    # Blocks=approval 행이 아직 진행 처분 없이 열려 있으면 승인을 무효화한다.
    scan = scan_approval_gate(p)
    if scan.unreadable_reason:
        raise PrepareError(
            f"approved plan §1 approval gate could not be read: {path}\n"
            f"  {scan.unreadable_reason}.\n"
            "  the gate refuses to soft-pass — re-render the report with "
            "scripts/okstra-render-final-report.py so §1 matches the schema, then retry."
        )
    blockers = scan.blockers
    if blockers:
        lines = [
            f"approved plan frontmatter has `approved: true` but §1 has {len(blockers)} "
            f"unresolved `Blocks=approval` row(s); resolve them or mark them obsolete first:",
        ]
        for b in blockers:
            lines.append(f"  - {b.row_id} (Status={b.raw_status})")
        lines.append(f"  file: {path}")
        raise PrepareError("\n".join(lines))


_STAGE_VALIDATOR_PATH = (
    Path(__file__).resolve().parents[2] / "validators"
    / "validate-implementation-plan-stages.py"
)


def _validate_stage_structure(plan_path: str) -> None:
    """Run validators/validate-implementation-plan-stages.py against the approved plan."""
    r = _subprocess.run(
        [sys.executable, str(_STAGE_VALIDATOR_PATH), "--plan", plan_path],
        capture_output=True, text=True,
    )
    if r.returncode != 0:
        raise PrepareError(
            f"approved plan failed stage validation:\n{r.stderr.strip()}"
        )


RUN_STEP_BUDGET = _stage_targets.RUN_STEP_BUDGET


def _parse_stage_map_into_ctx(plan_path: str) -> list:
    """Parse the approved plan into context records for execution consumers."""
    from .stage_map import StageMapError, parse_stage_map_file, stage_map_records

    try:
        return stage_map_records(parse_stage_map_file(Path(plan_path)))
    except StageMapError as exc:
        raise PrepareError(
            "approved-plan 의 Stage Map 을 신뢰할 수 없어 거부합니다 "
            f"({plan_path}): {exc.reason}. plan 의 Stage Map 을 점검하세요."
        ) from exc


def _apply_cli_approval(path: str) -> None:
    """`--approve` 가 지정된 경우 정본 `frontmatter.approved` 를 true 로 토글.

    정본을 쓰고 열람본을 다시 렌더한다. 이미 true 면 아무 것도 쓰지 않는다.
    schema-v1 계획은 정본이 없어 `--approved-plan` 으로 받을 수 없다.
    """
    p = _approved_plan_record(path)
    _reject_blocking_plan_body_gate(p, "", action="--approve")
    if _record_approved_flag(p) is True:
        return
    if not _set_data_json_approved_true_if_present(p):
        raise PrepareError(
            f"--approve found a report record but could not update it: {p}"
        )


def _apply_cli_implementation_option(path: str, option_name: str) -> None:
    """`--implementation-option <name>` 을 정본에 쓰고 열람본을 다시 렌더한다."""
    p = _approved_plan_record(path)
    loaded = _load_final_report_data_if_present(p)
    if loaded is None:
        raise PrepareError(f"approved plan data.json disappeared: {p}")
    data_path, data = loaded
    frontmatter = data.get("frontmatter")
    if not isinstance(frontmatter, dict) or "implementationOption" not in frontmatter:
        raise PrepareError(
            f"--implementation-option was given but the approved-plan "
            f"data.json has no frontmatter.implementationOption field: {data_path}"
        )
    frontmatter["implementationOption"] = option_name
    _commit_data_json_and_rerender(data_path, data, action="--implementation-option")


def _ensure_task_directories(ctx: dict) -> None:
    for key in (
        "TASK_ROOT", "INSTRUCTION_SET_PATH", "RUNS_DIR", "HISTORY_DIR",
        "RUN_DIR", "RUN_MANIFESTS_DIR", "RUN_STATE_DIR", "RUN_PROMPTS_DIR",
        "RUN_REPORTS_DIR", "RUN_STATUS_DIR", "RUN_SESSIONS_DIR",
        "RUN_LOGS_DIR", "WORKER_RESULTS_PATH", "RUN_CARRY_PATH", "OKSTRA_DISCOVERY_DIR",
    ):
        Path(ctx[key]).mkdir(parents=True, exist_ok=True)


def _role_execution_refs_from_ctx(ctx: dict) -> list[str]:
    try:
        payload = json.loads(ctx.get("EXECUTION_IDENTITY_JSON") or "{}")
    except (TypeError, ValueError):
        return []
    return [
        str(row.get("roleExecutionRef"))
        for row in payload.get("roleExecutions") or []
        if isinstance(row, dict) and row.get("roleExecutionRef")
    ]


def _record_start(
    *,
    workspace_root: Path,
    ctx: dict,
    initial_status: str,
    canonical_argv: list[str],
    cwd: str,
    brief_sha256: str,
) -> None:
    """record_start hook 호출. okstra-central.sh 의 bash wrapper 와 같은 동작
    이지만 python 직접 호출이라 환경 변수 의존 없음.
    """
    from datetime import datetime, timezone
    from okstra_ctl import record_start
    from .locks import central_lock

    home = okstra_home()
    home.mkdir(mode=0o700, parents=True, exist_ok=True)
    os.chmod(home, 0o700)
    # bootstrap (okstra_central_bootstrap 와 동등) — 디렉터리·jsonl 파일 보장.
    for sub in ("archive", ".locks", "batches", "projects"):
        (home / sub).mkdir(exist_ok=True)
    for f in ("active.jsonl", "recent.jsonl"):
        (home / f).touch(exist_ok=True)
    state_file = home / "state.json"
    if not state_file.exists():
        write_owned_object_atomic(
            state_file,
            {
                "schemaVersion": "1",
                "createdAt": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
                "backfilledAt": None,
            },
            artifact="global state",
        )
        os.chmod(state_file, 0o600)
    # 중앙 인덱스 락은 reconcile.py 와 공유하는 central_lock 단일 기준점을 쓴다.
    # 직접 flock 을 손으로 깔면 락 파일/모드가 두 경로에서 갈라질 수 있다.
    with central_lock(home):
        record_start(
            home,
            project_id=ctx["PROJECT_ID"],
            project_root=ctx["PROJECT_ROOT"],
            task_group=ctx["TASK_GROUP"],
            task_id=ctx["TASK_ID"],
            task_type=ctx.get("TASK_TYPE", ""),
            run_seq=int(ctx["RUN_MANIFESTS_SEQ"]),
            when=ctx["RUN_TIMESTAMP_ISO"],
            workers=[w for w in ctx.get("RECOMMENDED_ANALYSERS", "").split(",") if w],
            lead_model=ctx.get("LEAD_MODEL", ""),
            run_dir_rel=ctx.get("RUN_DIR_RELATIVE_PATH", ""),
            final_report_record_rel=ctx.get("FINAL_REPORT_RECORD_RELATIVE_PATH", ""),
            final_status_rel=ctx.get("FINAL_STATUS_RELATIVE_PATH", ""),
            argv=canonical_argv,
            cwd=cwd,
            env_overrides={},
            okstra_version=ctx.get("OKSTRA_VERSION", ""),
            initial_status=initial_status,
            brief_sha256=brief_sha256,
            execution_manifest_path=ctx.get("RUN_MANIFEST_RELATIVE_PATH", ""),
            role_execution_refs=_role_execution_refs_from_ctx(ctx),
        )


def _remove_leaked_reservation(ctx: dict, run_seq: int) -> None:
    """record_start 가 실패해 'reserving' row 를 'running' 으로 승격하지 못했을 때,
    rerun 이 미리 박아 둔 예약 row 를 중앙 락 아래에서 제거해 누수를 막는다.
    cleanup 자체가 실패해도 best-effort — prepare 결과를 무르지 않는다."""
    from okstra_ctl import remove_reservation
    from .locks import central_lock

    try:
        home = okstra_home()
        with central_lock(home):
            remove_reservation(
                home,
                project_id=ctx["PROJECT_ID"],
                task_group=ctx["TASK_GROUP"],
                task_id=ctx["TASK_ID"],
                task_type=ctx.get("TASK_TYPE", ""),
                run_seq=run_seq,
            )
    except Exception as exc:  # noqa: BLE001 — cleanup 실패는 비치명적
        print(
            f"okstra-central: reservation cleanup failed after record_start error ({exc})",
            file=sys.stderr,
        )


def _brief_sha256(path: Path) -> str:
    import hashlib

    try:
        with Path(path).open("rb") as f:
            return hashlib.sha256(f.read()).hexdigest()
    except OSError:
        return ""


def _record_artifact_runtime_render_only_event(inp: PrepareInputs, ctx: dict) -> None:
    descriptor = default_host_registry().resolve(inp.lead_runtime).descriptor
    if descriptor.session_accounting != "artifact-only" or not inp.render_only:
        return
    append_lead_event(
        Path(ctx["LEAD_EVENTS_PATH"]),
        LeadEvent(
            event_type="bundle-prepared",
            lead_runtime=inp.lead_runtime,
            task_key=ctx["TASK_KEY"],
            task_type=ctx["TASK_TYPE"],
            run_seq=ctx["RUN_MANIFESTS_SEQ"],
            timestamp=ctx["RUN_TIMESTAMP_ISO"],
            details={
                "dispatchMode": "render-only",
                "workerDispatch": "not-started",
                "runManifestPath": ctx.get("RUN_MANIFEST_RELATIVE_PATH", ""),
                "teamStatePath": ctx.get("TEAM_STATE_RELATIVE_PATH", ""),
                "promptSnapshotPath": ctx.get("RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""),
            },
        ),
    )


def _canonical_argv(inp: PrepareInputs, ctx: dict) -> list[str]:
    """rerun 충실 재현을 위한 canonical argv 재구성."""
    has_role_selection = bool(inp.role_counts_raw or inp.role_models_raw)
    legacy_fallbacks = {} if has_role_selection else ctx
    lead_fallbacks = (
        {} if inp.host_session_context.entry_mode == "current-session"
        else legacy_fallbacks
    )
    workers = inp.workers_override or (
        legacy_fallbacks.get("RECOMMENDED_ANALYSERS", "")
    )
    pairs = [
        ("--task-type", inp.task_type),
        ("--project-id", inp.project_id),
        ("--task-group", inp.task_group),
        ("--task-id", inp.task_id),
        ("--task-brief", str(inp.brief_path)),
        ("--analysis-target", inp.analysis_target),
        ("--evidence-inputs", inp.evidence_inputs_raw),
        ("--directive", inp.directive),
        ("--approved-plan", inp.approved_plan_path),
        ("--implementation-option", inp.implementation_option),
        ("--clarification-response", inp.clarification_response_path),
        ("--selected-direction", inp.selected_direction_path),
        ("--workers", workers),
        (
            "--lead-provider",
            inp.lead_provider or lead_fallbacks.get("LEAD_PROVIDER", ""),
        ),
        (
            "--lead-model",
            inp.lead_model or lead_fallbacks.get("LEAD_MODEL", ""),
        ),
        (
            "--claude-model",
            inp.claude_model or legacy_fallbacks.get("CLAUDE_WORKER_MODEL", ""),
        ),
        (
            "--codex-model",
            inp.codex_model or legacy_fallbacks.get("CODEX_WORKER_MODEL", ""),
        ),
        (
            "--antigravity-model",
            inp.antigravity_model
            or legacy_fallbacks.get("ANTIGRAVITY_WORKER_MODEL", ""),
        ),
        ("--worker-model", inp.worker_models_raw),
        (
            "--report-writer-provider",
            inp.report_writer_provider
            or legacy_fallbacks.get("REPORT_WRITER_PROVIDER", ""),
        ),
        (
            "--report-writer-model",
            inp.report_writer_model
            or legacy_fallbacks.get("REPORT_WRITER_MODEL", ""),
        ),
        ("--lead-runtime", inp.lead_runtime if inp.lead_runtime != "claude-code" else ""),
        (
            "--executor",
            inp.executor or legacy_fallbacks.get("EXECUTOR_PROVIDER", ""),
        ),
        ("--critic", inp.critic or legacy_fallbacks.get("CRITIC_CHOICE", "")),
        ("--related-tasks", inp.related_tasks_raw),
        ("--work-category", inp.work_category),
    ]
    argv: list[str] = []
    for flag, val in pairs:
        if val:
            argv.extend([flag, val])
    for role_count in inp.role_counts_raw:
        argv.extend(["--role-count", role_count])
    for role_model in inp.role_models_raw:
        argv.extend(["--role-model", role_model])
    if inp.host_session_context.host_id:
        argv.extend([
            "--host-session-context-json",
            serialize_host_session_context(inp.host_session_context),
        ])
    if inp.render_only:
        argv.append("--render-only")
    if not inp.plan_verification_enabled:
        argv.append("--no-plan-verification")
    argv.append("--yes")
    return argv


_INCLUDE_DIRECTIVE = re.compile(r"\{\{INCLUDE:([^}]+?)\}\}")
_HTML_COMMENT = re.compile(r"<!--.*?-->", re.DOTALL)


def _expand_profile_includes(profile_path: Path, _depth: int = 0) -> str:
    """Resolve `{{INCLUDE:<name>}}` directives in a profile file.

    Includes are resolved relative to the profile's directory. A maximum
    recursion depth of 4 prevents accidental cycles; the included file
    contents replace the directive line in-place. Missing include targets
    raise PrepareError so a bad reference fails fast instead of silently
    leaving a `{{INCLUDE:...}}` token in the rendered profile.

    The top-level result is stripped of HTML comments: the shared fragments
    carry maintainer-only `<!-- ... -->` notes (edit-here-once guidance, dedup
    rationale) that would otherwise ride into the rendered profile the lead
    reads every run without changing a single instruction. Nested-include
    bodies are left intact and get stripped once by this depth-0 pass.
    """
    if _depth > 4:
        raise PrepareError(
            f"profile include recursion exceeded depth 4 while resolving {profile_path}"
        )
    text = profile_path.read_text(encoding="utf-8")

    def _sub(match: "re.Match[str]") -> str:
        target_name = match.group(1).strip()
        target = profile_path.parent / target_name
        if not target.is_file():
            raise PrepareError(
                f"profile include target missing: {target} (referenced from {profile_path})"
            )
        return _expand_profile_includes(target, _depth + 1).rstrip("\n")

    expanded = _INCLUDE_DIRECTIVE.sub(_sub, text)
    if _depth == 0:
        expanded = _HTML_COMMENT.sub("", expanded)
        expanded = re.sub(r"\n{3,}", "\n\n", expanded)
    return expanded


# ---------------------------------------------------------------------------
# prepare_task_bundle 의 단계(phase) 헬퍼.
#
# 왜 분리하는가: prepare_task_bundle 은 input 검증 → 에셋 해소 → 프로젝트 등록
# → roster/model/executor 해소 → run-context/worktree → stage 예약 →
# instruction-set/manifest 렌더 → central record 까지 8개 독립 책임을 순차로
# 수행한다. 각 책임을 명시적 입력/반환을 갖는 헬퍼로 빼내면 (a) 개별 단위
# 테스트가 가능해지고 (b) prepare_task_bundle 본문이 "무엇을 어떤 순서로
# 엮는가" 만 드러내는 thin orchestrator 로 남는다. ctx dict 조립은 의도적으로
# orchestrator 에 남겨 데이터 흐름을 한눈에 보이게 한다.
# ---------------------------------------------------------------------------

_INSTALL_HINT = (
    " This file ships with the okstra package; its absence usually means a stale "
    "or partial install. Run 'okstra ensure-installed' (or 'okstra install' again) "
    "to repair the runtime. If the problem persists, run 'okstra doctor' for a "
    "fuller diagnostic."
)


@dataclass
class _ResolvedAssets:
    """런타임 에셋 경로 묶음. 모든 파일은 패키지와 함께 배포되며, 누락은
    user-content 문제가 아니라 설치 불완전을 뜻한다 (_INSTALL_HINT)."""

    profile_file: Path
    prompt_template: Path
    task_index_template: Path
    final_report_template: Path
    lead_contract: Path
    run_validator: Path
    brief_validator: Path
    # None when this task-type's host orchestration carries no gates; see
    # prompts/host-orchestration/README.md. Absence is a normal state, not a
    # broken install, so it is exempt from the _INSTALL_HINT contract above.
    host_rules_file: Path | None


def _resolve_runtime_assets(workspace_root: Path, inp: PrepareInputs) -> _ResolvedAssets:
    """task-type 별 profile + 공통 템플릿/스킬/validator 경로를 해소하고 존재를 확인한다."""
    profile_file = workspace_root / "prompts" / "profiles" / f"{inp.task_type}.md"
    if not profile_file.is_file():
        raise PrepareError(
            f"analysis profile file not found for task-type {inp.task_type}: "
            f"{profile_file}.{_INSTALL_HINT}"
        )
    prompt_template = workspace_root / "prompts" / "launch.template.md"
    if not prompt_template.is_file():
        raise PrepareError(
            f"okstra prompt template not found: {prompt_template}.{_INSTALL_HINT}"
        )
    task_index_template = (
        workspace_root / "templates" / "project-docs" / "task-index.template.md"
    )
    final_report_template = (
        workspace_root / "templates" / "reports" / "final-report-v2.template.md"
    )
    lead_contract = workspace_root / "prompts" / "lead" / "okstra-lead-contract.md"
    run_validator = workspace_root / "validators" / "validate-run.py"
    brief_validator = workspace_root / "validators" / "validate-brief.py"
    for required in (
        task_index_template,
        final_report_template,
        run_validator,
        brief_validator,
        lead_contract,
    ):
        if not required.is_file():
            raise PrepareError(
                f"required okstra template or lead contract missing: {required}.{_INSTALL_HINT}"
            )
    host_rules_file = (
        workspace_root / "prompts" / "host-orchestration" / f"{inp.task_type}.md"
    )
    return _ResolvedAssets(
        profile_file=profile_file,
        prompt_template=prompt_template,
        task_index_template=task_index_template,
        final_report_template=final_report_template,
        lead_contract=lead_contract,
        run_validator=run_validator,
        brief_validator=brief_validator,
        host_rules_file=host_rules_file if host_rules_file.is_file() else None,
    )


def _validate_task_brief_preflight(
    project_root: Path,
    brief_path: Path,
    validator_path: Path,
) -> None:
    """Validate canonical briefs before any prepare-time side effects."""
    frontmatter = read_brief_frontmatter(brief_path)
    if not has_reporter_confirmation_contract(frontmatter):
        return

    proc = _subprocess.run(
        [
            sys.executable,
            str(validator_path),
            str(brief_path),
            "--briefs-root",
            str(project_root / ".okstra" / "briefs"),
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    if proc.returncode != 0:
        detail = " ".join(
            line.strip()
            for output in (proc.stdout, proc.stderr)
            for line in output.splitlines()
            if line.strip()
        )
        raise PrepareError(f"task brief failed validation: {detail}")
    if frontmatter["reporter-confirmations"] == "pending":
        raise PrepareError(
            "task brief reporter-confirmations is pending; rerun okstra-brief-gen "
            "Step 6.5 before starting error-analysis"
        )


def _validate_prepare_inputs(project_root: Path, inp: PrepareInputs) -> list:
    """Validate pure prepare inputs and return a final-verification stage map."""
    if not project_root.is_dir():
        raise PrepareError(f"project root not found: {project_root}")
    _validate_analysis_prepare_inputs(project_root, inp)
    if inp.stages and inp.task_type != "release-handoff":
        raise PrepareError(
            "--stages is only meaningful with --task-type release-handoff; "
            f"got {inp.task_type}"
        )
    if inp.task_type == "release-handoff":
        # brief 는 entry phase 의 입력물 — release-handoff 는 검증 보고서를
        # 인용하는 input 문서를 prepare 가 직접 생성해 brief 자리에 채운다
        # (_materialize_release_handoff_input, 이 검증 직후 실행).
        pass
    elif not inp.brief_path.is_file():
        raise PrepareError(f"task brief not found: {inp.brief_path}")
    elif inp.task_type == "implementation-option-selection" and not brief_end_state_id_sequence(
        inp.brief_path
    ):
        raise PrepareError(
            "regenerate the brief with stable end-state IDs before starting "
            "implementation-option-selection"
        )
    ctx_stage_map: list = []
    # implementation 과 final-verification 은 둘 다 승인된 plan 의 Stage Map 을
    # 입력으로 받는다(전자는 실행 scope, 후자는 검증 scope).
    if inp.task_type in ("implementation", "final-verification"):
        if not inp.approved_plan_path:
            raise PrepareError(
                f"task-type {inp.task_type} requires "
                "--approved-plan <path-to-final-report.md>"
            )
        if inp.task_type == "final-verification":
            # final-verification 에서 --approve / --implementation-option 은
            # 의미가 없다 (승인은 implementation 진입 시 이미 끝났다).
            if inp.approve_plan_ack:
                raise PrepareError(
                    "--approve is only meaningful with --task-type implementation "
                    "and --approved-plan <path>"
                )
            if inp.implementation_option:
                raise PrepareError(
                    "--implementation-option is only meaningful with --task-type "
                    "implementation and --approved-plan <path>"
                )
            ctx_stage_map = _parse_stage_map_into_ctx(inp.approved_plan_path)
    else:
        if inp.approve_plan_ack:
            # implementation 외 task-type 에서 `--approve` 는 의미가 없다. 사용자에게
            # 정확한 시점을 알려주기 위해 조용히 무시하지 않고 즉시 거부한다.
            raise PrepareError(
                "--approve is only meaningful with --task-type implementation "
                "and --approved-plan <path>"
            )
        if inp.implementation_option:
            # implementation 외 task-type 에서 `--implementation-option` 도 의미가
            # 없다 (`--approve` 와 동일). 조용히 무시하지 않고 즉시 거부한다.
            raise PrepareError(
                "--implementation-option is only meaningful with --task-type "
                "implementation and --approved-plan <path>"
            )
        # wizard 는 모든 flag 를 빈 값 포함 항상 전달하므로(`--stage ""`),
        # 빈 문자열은 "stage 미지정" 으로 받아들인다.
        if inp.stage not in ("", "auto"):
            raise PrepareError(
                "--stage is only meaningful with --task-type implementation or "
                f"final-verification; got {inp.task_type}"
            )
    if inp.clarification_response_path and not Path(inp.clarification_response_path).is_file():
        raise PrepareError(
            f"clarification response file not found: {inp.clarification_response_path}"
        )
    _validate_planning_entry_inputs(project_root, inp)
    _validate_reverify_scope(inp)
    return ctx_stage_map


_PLANNING_REPORT_RE = re.compile(
    r"^final-report-implementation-planning-\d{3,}\.(?:md|data\.json)$"
)


def _resolve_planning_input_path(
    path_value: str, project_root: Path
) -> Path | None:
    """Resolve cwd-first planning inputs without following symbolic links."""
    raw_path = Path(path_value).expanduser()
    cwd_candidate = lexical_absolute_path(raw_path)
    if cwd_candidate.is_file():
        return cwd_candidate
    if not raw_path.is_absolute():
        project_candidate = lexical_absolute_path(project_root / raw_path)
        if project_candidate.is_file():
            return project_candidate
    return None


def _is_existing_planning_report(
    path_value: str, project_root: Path, inp: PrepareInputs
) -> bool:
    from .paths import task_dir

    try:
        report = lexical_absolute_path(Path(path_value))
        task_root = lexical_absolute_path(
            task_dir(project_root, inp.task_group, inp.task_id)
        )
        relative = report.relative_to(task_root)
    except ValueError:
        return False
    has_planning_layout = (
        _PLANNING_REPORT_RE.fullmatch(report.name) is not None
        and relative.parts[:3] == (
            "runs",
            "implementation-planning",
            "reports",
        )
        and len(relative.parts) == 4
    )
    if not has_planning_layout:
        return False
    try:
        validate_task_artifact_path(report, task_root, "planning report")
    except DirectionSelectionError:
        return False
    return True


def _validate_planning_entry_inputs(
    project_root: Path, inp: PrepareInputs
) -> None:
    if inp.task_type != "implementation-planning":
        if inp.selected_direction_path:
            raise PrepareError(
                "--selected-direction is only meaningful with --task-type "
                f"implementation-planning; got {inp.task_type}"
            )
        return
    if inp.selected_direction_path and inp.clarification_response_path:
        raise PrepareError(
            "implementation-planning accepts either --selected-direction for a "
            "new plan or --clarification-response for a rerun, not both"
        )
    if inp.selected_direction_path:
        return
    if inp.clarification_response_path:
        if not _is_existing_planning_report(
            inp.clarification_response_path, project_root, inp
        ):
            raise PrepareError(
                "implementation-planning rerun --clarification-response must point "
                "to an existing implementation-planning report for the same task"
            )
        return
    raise PrepareError(
        "implementation-planning requires --selected-direction for a new plan or "
        "--clarification-response to an existing implementation-planning report"
    )


def _resolve_planning_direction(inp: PrepareInputs) -> SelectedDirection | None:
    if inp.task_type != "implementation-planning" or not inp.selected_direction_path:
        return None
    from .paths import task_dir

    try:
        report = lexical_absolute_path(Path(inp.selected_direction_path))
        expected_task_root = lexical_absolute_path(
            task_dir(Path(inp.project_root), inp.task_group, inp.task_id)
        )
        report.relative_to(expected_task_root)
    except ValueError as exc:
        raise PrepareError(
            "selected direction report must belong to the same task"
        ) from exc
    try:
        validate_task_artifact_path(
            report, expected_task_root, "selected direction report"
        )
        return resolve_selected_direction(
            report,
            expected_task_key=(
                f"{inp.project_id}:{inp.task_group}:{inp.task_id}"
            ),
        )
    except DirectionSelectionError as exc:
        raise PrepareError(str(exc)) from exc


def _validate_reverify_scope(inp: PrepareInputs) -> None:
    """A pinned re-verification scope is only actionable on a planning re-run.

    Every other phase renders the tokens too (the template always reads them),
    but nothing consumes them there — so a value outside the one phase that
    acts on it is a caller mistake, not a preference to honour silently.
    """
    if not (inp.reverify_scope or "").strip():
        return
    if inp.task_type != "implementation-planning":
        raise PrepareError(
            "--reverify-scope is only meaningful with --task-type "
            f"implementation-planning; got {inp.task_type}"
        )
    if not inp.clarification_response_path:
        raise PrepareError(
            "--reverify-scope needs --clarification-response: there is no prior "
            "report to narrow re-verification against"
        )
    try:
        parse_user_reverify_scope(inp.reverify_scope)
    except ReverifyScopeError as exc:
        raise PrepareError(str(exc)) from exc


def _implementation_plan_contract(
    inp: PrepareInputs,
) -> tuple[str, Path, dict | None]:
    """Classify an implementation plan from its report record."""
    plan_path = _approved_plan_record(inp.approved_plan_path)
    loaded = _load_final_report_data_if_present(plan_path)
    if loaded is None:
        raise PrepareError(f"approved plan record could not be read: {plan_path}")
    _, data = loaded
    planning = data.get("implementationPlanning")
    if not isinstance(planning, dict):
        raise PrepareError(
            "approved plan sibling data.json must contain an "
            "implementationPlanning object"
        )
    contract = planning.get("planningContract")
    selected_payload = any(
        field in planning
        for field in (
            "selectedDirectionRef",
            "directionRealization",
            "directionInvalidation",
            "outcome",
        )
    )
    if contract is None:
        if selected_payload:
            raise PrepareError(
                "selected-direction markers require "
                "implementationPlanning.planningContract=`selected-direction`"
            )
        return "legacy", plan_path, data
    if contract != "selected-direction":
        raise PrepareError(
            "approved plan sibling data.json has unsupported planningContract "
            f"{contract!r}"
        )
    return "selected-direction", plan_path, data


def _validate_selected_implementation_plan(
    inp: PrepareInputs,
    plan_path: Path,
    data: dict,
) -> None:
    """Apply the Task 8 selected-direction semantics at implementation entry."""
    planning = data.get("implementationPlanning") or {}
    if planning.get("outcome") != "plan-ready":
        raise PrepareError(
            "selected-direction plan is not implementation-ready: "
            f"outcome={planning.get('outcome')!r}; direction-invalidated plans "
            "must re-enter implementation-option-selection"
        )
    summary = planning.get("coverageSummary") or {}
    if summary.get("coverageVerdict") != "exact":
        raise PrepareError(
            "selected-direction plan requires exact coverage before implementation"
        )

    from .paths import task_dir

    expected_task_root = lexical_absolute_path(
        task_dir(Path(inp.project_root), inp.task_group, inp.task_id)
    )
    try:
        validate_task_artifact_path(plan_path, expected_task_root, "approved plan")
        data_path = validate_task_artifact_path(
            _final_report_data_path(plan_path),
            expected_task_root,
            "approved plan data.json",
        )
        if data_path != _final_report_data_path(plan_path):
            raise DirectionSelectionError("approved plan data.json must be a sibling")
        snapshot_path = validate_task_artifact_path(
            expected_task_root / "instruction-set" / "selected-direction.json",
            expected_task_root,
            "selected direction snapshot",
        )
        brief_path = validate_task_artifact_path(
            expected_task_root / "instruction-set" / "task-brief.md",
            expected_task_root,
            "selected direction task brief",
        )
    except DirectionSelectionError as exc:
        raise PrepareError(f"selected-direction plan input is invalid: {exc}") from exc

    expected_task_key = f"{inp.project_id}:{inp.task_group}:{inp.task_id}"
    actual_task_key = (data.get("header") or {}).get("taskKey")
    if actual_task_key != expected_task_key:
        raise PrepareError(
            "selected-direction plan header.taskKey must match the implementation "
            f"task: expected {expected_task_key!r}, got {actual_task_key!r}"
        )
    failures = validate_selected_direction_plan(data, brief_path, snapshot_path)
    if failures:
        raise PrepareError(
            "selected-direction plan failed implementation entry validation:\n  - "
            + "\n  - ".join(failures)
        )


def _prepare_implementation_approved_plan(inp: PrepareInputs) -> list:
    """Apply approved-plan inputs only after canonical brief preflight succeeds."""
    contract, plan_path, data = _implementation_plan_contract(inp)
    if contract == "selected-direction":
        assert data is not None
        _validate_selected_implementation_plan(inp, plan_path, data)
        if inp.implementation_option:
            raise PrepareError(
                "--implementation-option is not accepted for a selected-direction plan"
            )
    if inp.approve_plan_ack or inp.implementation_option:
        with worktree_provision_mutex(
            okstra_home(), inp.project_id,
            slugify(inp.task_group), slugify(inp.task_id),
        ):
            if inp.approve_plan_ack:
                _apply_cli_approval(inp.approved_plan_path)
            if inp.implementation_option:
                _apply_cli_implementation_option(
                    inp.approved_plan_path, inp.implementation_option
                )
    contract, plan_path, data = _implementation_plan_contract(inp)
    _validate_approved_plan(inp.approved_plan_path)
    if contract == "selected-direction":
        assert data is not None
        _validate_selected_implementation_plan(inp, plan_path, data)
    _validate_stage_structure(inp.approved_plan_path)
    return _parse_stage_map_into_ctx(inp.approved_plan_path)


def _validate_analysis_prepare_inputs(project_root: Path, inp: PrepareInputs) -> None:
    """Reject invalid analysis inputs before task worktree or run sequence creation."""
    is_analysis = inp.task_type in ANALYSIS_TASK_TYPES
    if not is_analysis:
        if inp.analysis_target:
            raise PrepareError(
                f"--analysis-target is only accepted for analysis task types; got {inp.task_type}"
            )
        if inp.evidence_inputs_raw:
            raise PrepareError(
                f"--evidence-inputs is only accepted for analysis task types; got {inp.task_type}"
            )
        return
    if inp.task_type == "project-analysis":
        if inp.analysis_target:
            raise PrepareError("project-analysis does not accept an analysis target")
        if inp.evidence_inputs_raw:
            raise PrepareError("project-analysis does not accept evidence inputs")
        return
    if inp.task_type == "feature-analysis" and not inp.analysis_target.strip():
        raise PrepareError("feature-analysis requires --analysis-target")
    if inp.task_type != "feature-analysis" and inp.analysis_target:
        raise PrepareError(f"{inp.task_type} does not accept an analysis target")
    paths = parse_evidence_paths(inp.evidence_inputs_raw, project_root)
    try:
        candidates = load_candidate_map(project_root, paths)
        evidence = resolve_evidence_inputs(
            project_root, inp.task_type, paths, "0" * 40,
        )
        if inp.task_type == "feature-analysis":
            resolve_analysis_target(inp.analysis_target, evidence, candidates)
    except AnalysisInputError as exc:
        raise PrepareError(str(exc)) from exc


def _collect_handoff_source_report_rows(
    rows: list, nums: list,
) -> list:
    """선택 stage 들의 마지막 verified 행에서 인용 표 행을 만든다 (last-wins)."""
    last_verified: dict = {}
    for r in rows:
        if r.get("status") == "verified" and isinstance(r.get("stage"), int):
            last_verified[r["stage"]] = r
    return [
        f"| {n} | {last_verified[n].get('report_path', '')} "
        f"| `{last_verified[n].get('verdict', '')}` |"
        for n in nums
    ]


def _materialize_release_handoff_input(
    workspace_root: Path, project_root: Path, inp: "PrepareInputs",
) -> dict:
    """release-handoff 의 입력 문서를 검증 보고서 인용으로 자동 생성한다.

    brief 는 entry phase 의 입력물이라 release-handoff 에는 없다 — 이 run 의
    `--task-brief` 자리는 여기서 생성된 input 문서가 채운다. stage 자격은
    okstra_ctl.handoff 의 SSOT 판정을 prepare 에서도 그대로 강제한다.
    반환: ctx 에 올릴 {"HANDOFF_MODE": ..., "HANDOFF_STAGES": ...}."""
    from .consumers import read_consumers
    from .handoff import (HandoffError, _require_eligible,
                          latest_whole_task_fv_release_ready)
    from .paths import task_dir
    from .render import render_template_with_ctx
    from .run_context import _now_task_date

    # Path("") 는 "." 로 정규화된다 — 빈 brief 센티널은 둘 다로 들어올 수 있다.
    if str(inp.brief_path).strip() not in ("", "."):
        raise PrepareError(
            "--task-brief is not accepted for --task-type release-handoff — "
            "the input document is generated from the cited final-verification "
            "reports (pick stages via --stages, or leave it empty for "
            "whole-task)"
        )
    if not inp.approved_plan_path:
        raise PrepareError(
            "release-handoff requires --approved-plan "
            "<implementation-planning final-report> — the Stage Map and the "
            "consumers ledger drive stage eligibility"
        )
    plan = Path(inp.approved_plan_path)
    if not plan.is_file():
        raise PrepareError(f"approved plan not found: {plan}")
    stage_map = _parse_stage_map_into_ctx(str(plan))
    rows = read_consumers(plan_run_root_from_approved_plan(plan))

    if inp.stages:
        try:
            nums = sorted({int(x) for x in inp.stages.split(",") if x.strip()})
        except ValueError:
            raise PrepareError(
                f"--stages must be a comma-separated int list, got {inp.stages!r}"
            )
        if not nums:
            raise PrepareError("--stages must select at least one stage")
        try:
            _require_eligible(stage_map, rows, nums)
        except HandoffError as exc:
            raise PrepareError(str(exc)) from exc
        mode = "stage-group"
        stages_csv = ",".join(str(n) for n in nums)
        report_rows = _collect_handoff_source_report_rows(rows, nums)
    else:
        report = latest_whole_task_fv_release_ready(
            project_root, inp.project_id, inp.task_group, inp.task_id)
        if not report:
            raise PrepareError(
                "whole-task release-handoff requires an accepted whole-task "
                "final-verification report — run final-verification first, "
                "or pass --stages <csv> for a stage-group PR"
            )
        mode = "whole-task"
        stages_csv = ""
        report_relative = relative_to_project_root(Path(report), project_root)
        report_rows = [f"| (all) | {report_relative or report} | `accepted` |"]

    template = (workspace_root / "templates" / "reports"
                / "release-handoff-input.template.md")
    if not template.is_file():
        raise PrepareError(
            f"release-handoff input template missing: {template}.{_INSTALL_HINT}"
        )
    out = (task_dir(project_root, inp.task_group, inp.task_id)
           / "release-handoff-input.md")
    out.parent.mkdir(parents=True, exist_ok=True)
    render_template_with_ctx(str(template), str(out), {
        "TASK_KEY": f"{inp.project_id}:{inp.task_group}:{inp.task_id}",
        "TASK_ID": inp.task_id,
        "TASK_GROUP": inp.task_group,
        "PROJECT_ID": inp.project_id,
        "TASK_TYPE": inp.task_type,
        "TASK_DATE": _now_task_date(),
        "HANDOFF_MODE": mode,
        "HANDOFF_STAGES": stages_csv or "(all)",
        "HANDOFF_SOURCE_REPORTS": "\n".join(report_rows),
    })
    inp.brief_path = out
    return {"HANDOFF_MODE": mode, "HANDOFF_STAGES": stages_csv}


QA_COMMAND_EXECUTING_TASK_TYPES = ("implementation", "final-verification")


def validate_project_qa_commands(task_type: str, project_root: Path) -> None:
    """`qaCommands` 를 실행하는 phase 진입에서 변경성 토큰 선언을 막는다.

    `implementation` 은 verifier 의 QA gate baseline 으로, `final-verification` 은
    프로파일이 정의한 Tier 2 재실행 집합으로 같은 선언을 읽는다. 실행 직전에 리드가
    스스로 걸러내게 두면 그 자기검사가 유일한 방어선이 되므로 진입에서 막는다.
    나머지 task-type 은 이 선언을 읽지 않아 잘못된 값이 있어도 동작에 닿지 않는다.
    """
    if task_type not in QA_COMMAND_EXECUTING_TASK_TYPES:
        return
    project_json = project_json_path(project_root)
    if not project_json.is_file():
        return
    try:
        project_meta = load_owned_object(project_json, artifact="project config")
    except (OSError, JsonBoundaryError) as exc:
        raise PrepareError(
            f"project.json read failed at {project_json}: {exc}"
        ) from exc
    qa_errors = validate_qa_commands(project_meta.get("qaCommands"))
    if qa_errors:
        raise PrepareError(_format_qa_errors(qa_errors))


def _apply_qa_waiver_if_requested(inp: "PrepareInputs", project_root: Path) -> None:
    """`--qa-waiver` 가 있으면 task-level 매니페스트 entry 의 waiver 를 채운다.

    이 read-modify-write 는 같은 task-key 의 동시 implementation run 이 거치는
    `_clear_stale_stage_waiver` 와 동일한 conformance-manifest.json 을 건드린다.
    `_clear_stale_stage_waiver` 는 prepare_task_bundle 의 worktree_provision_mutex
    안에서 실행되므로, 이 함수도 같은 per-task-key 락을 잡아 두 writer 를
    직렬화한다. 락이 없으면 후발 write 가 선발의 waiver 변경을 덮어써(lost update)
    verifier 가 Tier 3 conformance 를 건너뛰고 stage 를 가린다.
    """
    if not inp.qa_waiver:
        return
    from .conformance import apply_qa_waiver, parse_qa_waiver_arg
    from .paths import task_dir
    parsed = parse_qa_waiver_arg(inp.qa_waiver)
    if parsed is None:
        raise PrepareError(
            f'--qa-waiver must be "<stageKey>:<reason>", got {inp.qa_waiver!r}'
        )
    stage_key, reason = parsed
    manifest_path = task_dir(project_root, inp.task_group, inp.task_id) / "qa" / "conformance-manifest.json"
    with worktree_provision_mutex(
        okstra_home(), inp.project_id, slugify(inp.task_group), slugify(inp.task_id),
    ):
        if not manifest_path.is_file():
            raise PrepareError(f"--qa-waiver: conformance manifest not found at {manifest_path}")
        manifest = load_owned_object(manifest_path, artifact="conformance manifest")
        when = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        if not apply_qa_waiver(manifest, stage_key, reason, at=when):
            raise PrepareError(f"--qa-waiver: stageKey {stage_key!r} not in manifest {manifest_path}")
        write_owned_object_atomic(
            manifest_path, manifest, artifact="conformance manifest"
        )


def _clear_stale_stage_waiver(inp: "PrepareInputs", project_root: Path, stage: int) -> None:
    """A fresh `implementation` run of stage N must not inherit a waiver left on
    its conformance entry by an earlier run (e.g. an all-gate run that pre-waived
    future stages, or an abandoned attempt). A stale waiver makes the verifier
    skip Tier 3 conformance and silently mask this stage, so clear it — UNLESS
    the user re-waived this exact stage for this run via `--qa-waiver` (already
    applied upstream in `_apply_qa_waiver_if_requested`)."""
    from .conformance import clear_qa_waiver, parse_qa_waiver_arg
    from .paths import task_dir
    manifest_path = (
        task_dir(project_root, inp.task_group, inp.task_id)
        / "qa" / "conformance-manifest.json"
    )
    if not manifest_path.is_file():
        return
    manifest = load_owned_object(manifest_path, artifact="conformance manifest")
    entries = manifest.get("entries") if isinstance(manifest, dict) else None
    if not isinstance(entries, list):
        return
    # The manifest stageKey is `<task-id>-stage-<N>` authored by planning; match
    # on the `-stage-<N>` suffix so we do not assume the task-id's exact form.
    suffix = f"-stage-{stage}"
    stage_key = next(
        (e["stageKey"] for e in entries
         if isinstance(e, dict) and isinstance(e.get("stageKey"), str)
         and e["stageKey"].endswith(suffix)),
        None,
    )
    if stage_key is None:
        return
    if inp.qa_waiver:
        parsed = parse_qa_waiver_arg(inp.qa_waiver)
        if parsed is not None and parsed[0] == stage_key:
            return  # user intentionally waived this stage for this run
    if clear_qa_waiver(manifest, stage_key):
        write_owned_object_atomic(
            manifest_path, manifest, artifact="conformance manifest"
        )


def _register_and_check_project(project_root: Path, inp: PrepareInputs) -> None:
    """project.json self-registration + (implementation 한정) qaCommands gate 검증."""
    from okstra_project import ResolverError

    try:
        upsert_project_json(project_root, inp.project_id)
    except ResolverError as exc:
        # Surface the project_root in the prefix so the user can tell which
        # registration failed when multiple projects are in play. The full
        # underlying ResolverError text (which carries remediation guidance)
        # is preserved by the `: {exc}` suffix and the `raise ... from exc`.
        raise PrepareError(f"project.json upsert failed for {project_root}: {exc}") from exc

    validate_project_qa_commands(inp.task_type, project_root)
    # waiver 는 stage 단위 Tier 3 면제라 implementation 진입에서만 적용한다.
    if inp.task_type == "implementation":
        _apply_qa_waiver_if_requested(inp, project_root)


def _resolve_roster(inp: PrepareInputs, profile_file: Path) -> tuple[list[str], str]:
    """profile 의 `Required workers:` 블록을 권위로 roster 를 해소한다.

    release-handoff 은 의도적으로 single-lead (worker dispatch / teammate /
    convergence 없음). profile 에 `- Required workers:` 블록이 없으므로 어떤
    override 가 와도 빈 roster 를 강제해 profile 계약과 일관성을 유지한다."""
    if inp.task_type == "release-handoff":
        return [], ""
    profile_workers = resolve_profile_workers(profile_file)
    optional_workers = resolve_optional_workers(profile_file)
    profile_workers_csv = ",".join(profile_workers)
    workers = normalize_workers(inp.workers_override or profile_workers_csv)
    if inp.workers_override.strip():
        validate_workers_against_profile(workers, profile_workers, optional_workers)
    try:
        role_profile = load_role_profile(profile_file)
    except RoleProfileError as exc:
        raise PrepareError(str(exc)) from exc
    requires_report_writer = any(
        row.role == "report-writer" and row.min_count == 1
        for row in role_profile.roles
    )
    if requires_report_writer and "report-writer" not in workers:
        workers.append("report-writer")
    if not workers:
        raise PrepareError(f"no workers resolved for profile: {inp.task_type}")
    _validate_option_selection_roster(inp.task_type, workers)
    return workers, ",".join(workers)


def _profile_requires_executor(profile_file: Path) -> bool:
    try:
        profile = load_role_profile(profile_file)
    except RoleProfileError as exc:
        raise PrepareError(str(exc)) from exc
    return any(
        requirement.role == "implementer" and requirement.min_count > 0
        for requirement in profile.roles
    )


def _normalize_prepare_model_selection(
    inp: PrepareInputs,
    profile_file: Path,
) -> tuple[CanonicalModelSelection, RoleProfile]:
    try:
        _validate_critic_choice(inp.critic, inp.task_type)
        profile = load_role_profile(profile_file)
        profile_workers = resolve_profile_workers(profile_file)
        has_worker_models = bool(
            inp.worker_models_raw.strip()
            or inp.claude_model.strip()
            or inp.codex_model.strip()
            or inp.antigravity_model.strip()
        )
        workers = inp.workers_override
        if not workers and has_worker_models:
            workers = ",".join(profile_workers)
        selection = normalize_model_selection_inputs(
            profile=profile,
            role_counts_raw=inp.role_counts_raw,
            role_models_raw=inp.role_models_raw,
            workers=workers,
            worker_model=inp.worker_models_raw,
            critic=inp.critic,
            executor=inp.executor,
            lead_provider=inp.lead_provider,
            lead_model=inp.lead_model,
            report_writer_provider=inp.report_writer_provider,
            report_writer_model=inp.report_writer_model,
            provider_models={
                "claude": inp.claude_model,
                "codex": inp.codex_model,
                "antigravity": inp.antigravity_model,
            },
        )
    except (ModelSelectionInputError, RoleProfileError) as exc:
        raise PrepareError(str(exc)) from exc
    return selection, profile


def _effective_host_session_context(
    inp: PrepareInputs,
    lead_runtime: str,
) -> HostSessionContext:
    context = inp.host_session_context
    if context.host_id:
        return context
    descriptor = default_host_registry().resolve(lead_runtime).descriptor
    current_model = context.current_model
    if not current_model.provider_id:
        current_model = CurrentSessionModelAttestation.unknown(
            descriptor.native_provider_id
        )
    return HostSessionContext(
        host_id=descriptor.id,
        entry_mode=context.entry_mode,
        available_functions=context.available_functions,
        interaction_surface=context.interaction_surface,
        current_model=current_model,
    )


def _resolve_prepare_assignments(
    *,
    inp: PrepareInputs,
    profile: RoleProfile,
    selection: CanonicalModelSelection,
    workers: list[str],
    lead_runtime: str,
    terminal_backend: TerminalBackend,
    requires_executor: bool,
) -> tuple[AssignmentContext, AssignmentPlan]:
    scopes = _model_default_scopes(Path(inp.project_root))
    provider_ids_for_snapshot = _selection_provider_ids(
        inp,
        profile,
        selection,
        workers,
        scopes,
        requires_executor,
    )
    context = load_assignment_context(
        host_runtime=lead_runtime,
        terminal_backend=terminal_backend,
        execution_provider_ids=provider_ids_for_snapshot,
    )
    host = _effective_host_session_context(inp, lead_runtime)
    role_models = _materialize_selection_models(selection, context, host)
    role_counts = _resolver_role_counts(profile, selection.role_counts)
    try:
        plan = resolve_assignments(
            profile=profile,
            role_counts=role_counts,
            role_models=role_models,
            scopes=scopes,
            pool=context.pool,
            host=host,
            environment=context.environment,
        )
    except AssignmentResolutionError as exc:
        raise PrepareError(_assignment_resolution_message(exc)) from exc
    except (ValueError, RuntimeError) as exc:
        raise PrepareError(str(exc)) from exc
    return context, plan


def _build_static_execution_manifest(
    plan: AssignmentPlan,
    context: AssignmentContext,
    lead: ResolvedAssignment,
    translator: ResolvedAssignment,
) -> ExecutionManifest:
    participants: list[ParticipantAssignment] = []
    roles: list[RoleExecution] = []
    # The translator is resolved outside the role plan, like the lead, and both
    # are named in `invocationAssignments`. Without a role execution here,
    # `agent-prompt materialize --audience translator` has no canonical
    # identity to bind to and refuses — so a `reportLanguage: ko` run could
    # never write its translation sidecar and rendered the English source.
    assignments = (
        lead,
        *(row for row in plan.assignments if row.role != "leader"),
        translator,
    )
    for index, assignment in enumerate(assignments, start=1):
        participant_ref = f"participant-{index:03d}"
        participants.append(_participant_assignment(assignment, participant_ref))
        roles.append(
            _static_role_execution(
                assignment,
                participant_ref,
                context,
            )
        )
    return ExecutionManifest(
        entry_mode=lead.entry_mode,
        participant_assignments=tuple(participants),
        role_executions=tuple(roles),
    )


def _participant_assignment(
    assignment: ResolvedAssignment,
    participant_ref: str,
) -> ParticipantAssignment:
    binding = assignment.binding
    return ParticipantAssignment(
        participant_ref=participant_ref,
        provider=assignment.provider_id,
        model_ref=assignment.model_ref,
        model_id=assignment.model_id,
        runner=binding.runner if binding is not None else "current-session",
        host_runtime=assignment.host_runtime,
        host_model_value=binding.host_model_value if binding is not None else None,
        session_ref=None,
        window_ref=None,
        worker_write_capability=(
            binding.worker_write_capability if binding is not None else None
        ),
        entry_mode=assignment.entry_mode,
        status="prepared",
    )


def _static_role_execution(
    assignment: ResolvedAssignment,
    participant_ref: str,
    context: AssignmentContext,
) -> RoleExecution:
    binding = assignment.binding
    digest = None
    if assignment.model_ref is not None and binding is not None:
        digest = model_spec_digest(context.pool.resolve(assignment.model_ref), binding)
    return RoleExecution(
        role_execution_ref=(
            f"role-exec-{assignment.role}-{assignment.ordinal:03d}"
        ),
        participant_ref=participant_ref,
        source_role_execution_ref=None,
        role=assignment.role,
        provider=assignment.provider_id,
        model_ref=assignment.model_ref,
        model_id=assignment.model_id,
        ordinal=assignment.ordinal,
        execution_label=execution_label(
            assignment.role,
            assignment.provider_id,
            assignment.model_id,
            assignment.ordinal,
        ),
        model_spec_digest=digest,
        binding=binding,
        served_model_attestation=ServedModelAttestation.unknown(),
        status="prepared",
    )


def _assignment_resolution_message(exc: AssignmentResolutionError) -> str:
    lines = [str(exc)]
    for candidate in exc.candidates:
        if candidate.available:
            continue
        lines.append(
            f"- {candidate.role}#{candidate.ordinal} {candidate.model_ref}: "
            f"{candidate.reason}"
        )
    for constraint in exc.unsatisfied_constraints:
        lines.append(
            f"- {constraint.role} {constraint.constraint}: "
            f"required={constraint.required}, available={constraint.available}"
        )
    return "\n".join(lines)


def _selection_provider_ids(
    inp: PrepareInputs,
    profile: RoleProfile,
    selection: CanonicalModelSelection,
    workers: list[str],
    scopes: ModelDefaultScopes,
    requires_executor: bool,
) -> tuple[str, ...]:
    default_roles = _roles_requiring_defaults(
        inp,
        profile,
        selection,
    )
    has_canonical_role_selection = bool(
        inp.role_counts_raw or inp.role_models_raw
    )
    if has_canonical_role_selection:
        selected = _canonical_selection_provider_ids(
            inp,
            profile,
            selection,
            workers,
            scopes,
            default_roles,
            requires_executor,
        )
    else:
        selected = list(
            _selected_execution_provider_ids(inp, workers, requires_executor)
        )
    selected.extend(
        provider
        for providers in selection.provider_constraints.values()
        for provider in providers
        if provider
    )
    selected.extend(_providers_from_model_refs(selection.role_models.values()))
    selected.extend(
        _providers_from_model_refs(
            _selected_scoped_default_groups(default_roles, scopes)
        )
    )
    current_provider = inp.host_session_context.current_model.provider_id
    if current_provider:
        selected.append(current_provider)
    return tuple(dict.fromkeys(selected))


def _selected_scoped_default_groups(
    roles: Sequence[str],
    scopes: ModelDefaultScopes,
) -> tuple[tuple[str, ...], ...]:
    groups: list[tuple[str, ...]] = []
    for role in roles:
        if role in scopes.project:
            groups.append(scopes.project[role])
        elif role in scopes.global_:
            groups.append(scopes.global_[role])
    return tuple(groups)


def _roles_requiring_defaults(
    inp: PrepareInputs,
    profile: RoleProfile,
    selection: CanonicalModelSelection,
) -> tuple[str, ...]:
    roles: list[str] = []
    if (
        inp.host_session_context.entry_mode != "current-session"
        and not _explicit_selection_count(selection, "leader")
    ):
        roles.append("leader")
    for requirement in profile.roles:
        if requirement.dynamic:
            continue
        desired = selection.role_counts.get(
            requirement.role,
            requirement.recommended_count,
        )
        explicit = _explicit_selection_count(selection, requirement.role)
        if explicit < desired:
            roles.append(requirement.role)
    return tuple(roles)


def _explicit_selection_count(
    selection: CanonicalModelSelection,
    role: str,
) -> int:
    return max(
        len(selection.role_models.get(role, ())),
        len(selection.provider_constraints.get(role, ())),
    )


def _canonical_selection_provider_ids(
    inp: PrepareInputs,
    profile: RoleProfile,
    selection: CanonicalModelSelection,
    workers: list[str],
    scopes: ModelDefaultScopes,
    default_roles: Sequence[str],
    requires_executor: bool,
) -> list[str]:
    selected = [inp.lead_provider]
    report_writer_uses_bundled_default = (
        "report-writer" in default_roles
        and "report-writer" not in scopes.project
        and "report-writer" not in scopes.global_
    )
    if report_writer_uses_bundled_default:
        selected.append("claude")
    implementer_uses_bundled_default = (
        requires_executor
        and "implementer" in default_roles
        and "implementer" not in scopes.project
        and "implementer" not in scopes.global_
    )
    if implementer_uses_bundled_default:
        # `--executor` names the provider that implements, so it belongs in the
        # roster the same way the bundled default does. Reading only the default
        # left `--executor <provider>` with no assignment of its own: the run
        # rendered without that provider, and asking for it with `--workers`
        # was refused as not being in the roster.
        selected.append(
            (inp.executor or "").strip().lower()
            or _default("OKSTRA_DEFAULT_EXECUTOR", "claude")
        )
    if _needs_profile_worker_candidates(profile, selection, scopes):
        selected.extend(
            worker for worker in workers if worker != "report-writer"
        )
    return selected


def _needs_profile_worker_candidates(
    profile: RoleProfile,
    selection: CanonicalModelSelection,
    scopes: ModelDefaultScopes,
) -> bool:
    for requirement in profile.roles:
        if requirement.dynamic or requirement.role == "report-writer":
            continue
        desired = selection.role_counts.get(
            requirement.role,
            requirement.recommended_count,
        )
        explicit = _explicit_selection_count(selection, requirement.role)
        has_scoped_default = (
            requirement.role in scopes.project
            or requirement.role in scopes.global_
        )
        if explicit < desired and not has_scoped_default:
            return True
    return False


def _providers_from_model_refs(
    model_groups: Sequence[Sequence[str]],
) -> tuple[str, ...]:
    return tuple(
        model_ref.split("/", 1)[0]
        for models in model_groups
        for model_ref in models
        if "/" in model_ref
    )


def _materialize_selection_models(
    selection: CanonicalModelSelection,
    context: AssignmentContext,
    host: HostSessionContext,
) -> dict[str, tuple[str, ...]]:
    role_models = dict(selection.role_models)
    for role, providers in selection.provider_constraints.items():
        model_values = selection.legacy_model_values.get(role, {})
        resolved: list[str] = []
        for raw_provider in providers:
            provider = raw_provider or _fallback_legacy_provider(role, context, host)
            raw_model = model_values.get(provider, model_values.get("", ""))
            try:
                model = resolve_model_selection(
                    context.pool,
                    provider,
                    role,
                    raw_model,
                )
            except (ValueError, RuntimeError) as exc:
                raise PrepareError(str(exc)) from exc
            resolved.append(str(model.model_ref))
        role_models[role] = tuple(resolved)
    return role_models


def _fallback_legacy_provider(
    role: str,
    context: AssignmentContext,
    host: HostSessionContext,
) -> str:
    if role == "leader":
        return (
            host.current_model.provider_id
            or context.environment.host_descriptor.native_provider_id
        )
    return "claude"


def _resolver_role_counts(
    profile: RoleProfile,
    counts: Mapping[str, int],
) -> dict[str, int]:
    adjustable = {
        requirement.role
        for requirement in profile.roles
        if (
            not requirement.dynamic
            and requirement.min_count < requirement.max_count
        )
    }
    return {role: count for role, count in counts.items() if role in adjustable}


def _model_default_scopes(project_root: Path) -> ModelDefaultScopes:
    return ModelDefaultScopes(
        project=_read_model_defaults(project_json_path(project_root)),
        global_=_read_model_defaults(okstra_home() / "config.json"),
        bundled={},
    )


def _read_model_defaults(path: Path) -> dict[str, tuple[str, ...]]:
    if not path.is_file():
        return {}
    try:
        payload = load_owned_object(path, artifact="model defaults config")
    except (OSError, JsonBoundaryError) as exc:
        raise PrepareError(f"model defaults file is invalid: {path}: {exc}") from exc
    raw_defaults = payload.get("modelDefaults", {}) if isinstance(payload, dict) else {}
    if not isinstance(raw_defaults, dict):
        raise PrepareError(f"modelDefaults must be an object: {path}")
    defaults: dict[str, tuple[str, ...]] = {}
    for role, models in raw_defaults.items():
        if (
            not isinstance(role, str)
            or not isinstance(models, list)
            or not models
            or not all(isinstance(model, str) and model for model in models)
        ):
            raise PrepareError(f"modelDefaults entries must be non-empty string arrays: {path}")
        defaults[role] = tuple(models)
    return defaults


def _validate_option_selection_roster(
    task_type: str, workers: Sequence[str]
) -> None:
    """Require independent analysis from three workers before option selection."""
    if task_type != "implementation-option-selection":
        return
    analyser_count = sum(worker != "report-writer" for worker in workers)
    if analyser_count < 3:
        raise PrepareError(
            "implementation-option-selection requires at least 3 analyser workers"
        )


def _resolve_pr_template(inp: PrepareInputs) -> tuple[str, str]:
    """release-handoff 전용 PR 본문 템플릿 경로 + 출처를 해소한다 (그 외엔 빈 값)."""
    if inp.task_type != "release-handoff":
        return "", ""
    try:
        resolved_tpl = resolve_pr_template_path(Path(inp.project_root), inp.pr_template_path)
    except PrTemplateError as exc:
        raise PrepareError(f"PR template resolution failed: {exc}") from exc
    return str(resolved_tpl.path), resolved_tpl.source


@dataclass
class RoleAssignment:
    role: str
    provider: str
    model_display: str
    model_id: str
    model_execution_value: str
    runner: str
    host_runtime: str
    host_model_value: str | None
    worker_id: str = ""

    def to_model_payload(self) -> dict[str, object]:
        # `model` carries the catalog model id, not the display name: the run
        # manifest's role executions record `modelId`, and the two are compared
        # to bind an assignment to its execution. A display name that differs
        # from its id (every antigravity and kimi model) made that comparison
        # fail for the whole provider.
        return {
            "provider": self.provider,
            "model": self.model_id,
            "modelExecutionValue": self.model_execution_value,
            "runner": self.runner,
            "hostRuntime": self.host_runtime,
            "hostModelValue": self.host_model_value,
        }

    def to_payload(self) -> dict[str, object]:
        payload = {
            "role": self.role,
            **self.to_model_payload(),
        }
        if self.worker_id:
            payload["workerId"] = self.worker_id
        return payload


@dataclass
class _ModelBindings:
    """이 run 의 lead/worker 모델 + critic + executor 바인딩.

    `*_execution` 은 dispatch 에 쓰이는 CLI 신원 철자다. 카탈로그 값과 달리
    provider CLI 조회를 거치므로 실패할 수 있고, 그래서 이 묶음에서 함께
    해소한다 — prepare 가 run seq 를 할당하거나 worktree 를 만들기 전에.
    """

    lead: ResolvedAssignment
    cw: "_LegacyModelProjection"
    co: "_LegacyModelProjection"
    ge: "_LegacyModelProjection"
    rw: ResolvedAssignment
    critic_choice: str
    critic_model_execution: str
    executor_display_name: str
    codex_worker_execution: str
    antigravity_worker_execution: str
    executor_assignment: RoleAssignment | None
    lead_assignment: RoleAssignment
    worker_assignments: tuple[RoleAssignment, ...]
    invocation_assignments: dict[str, dict[str, object]]
    translator: ResolvedAssignment


@dataclass(frozen=True)
class _LegacyModelProjection:
    """Display/catalog values retained for the fixed v1 manifest fields."""

    model_id: str
    display: str
    execution: str


def _legacy_claude_model_override(provider: str, env_name: str) -> str:
    """Read a legacy Claude model override only for a Claude assignment."""
    if provider.strip().lower() != "claude":
        return ""
    return _default(env_name, "")


def recommended_role_models(
    *,
    lead_provider: str = "",
    report_writer_provider: str = "",
) -> dict[str, str]:
    """역할 → 추천 모델 display 값 (env override 반영). prepare 의 모델 해소와
    wizard 의 안내 표기가 공유하는 단일 기준점.

    lead 는 호스트가 provider 를 정한다 — codex 호스트의 in-session lead 는
    codex 이고 다른 provider 요청은 거부된다(`resolve_lead_provider`). 그래서
    `lead_provider` 를 받아 그 provider 의 기본값으로 해소한다. 받지 않으면
    role 별 레거시 기본값(claude 계열)으로 떨어지는데, 그 값을 codex 호스트
    화면에 그대로 쓰면 안내는 `opus` 인데 prepare 는 `gpt-5.6-sol` 을 배정한다.
    """
    lead_provider_id = lead_provider or "claude"
    report_writer_provider_id = report_writer_provider or "claude"
    lead_default = (
        _legacy_claude_model_override(
            lead_provider_id,
            "OKSTRA_DEFAULT_LEAD_MODEL",
        )
        or provider_default_model(lead_provider_id, "lead")
    )
    report_writer_default = (
        _legacy_claude_model_override(
            report_writer_provider_id,
            "OKSTRA_DEFAULT_REPORT_WRITER_MODEL",
        )
        or provider_default_model(report_writer_provider_id, "report-writer")
    )
    recommendations = {
        "lead": lead_default,
        "claude": _default("OKSTRA_DEFAULT_CLAUDE_MODEL", default_model("claude")),
        "codex": _default("OKSTRA_DEFAULT_CODEX_MODEL", default_model("codex")),
        "antigravity": _default("OKSTRA_DEFAULT_ANTIGRAVITY_MODEL", default_model("antigravity")),
        "report-writer": report_writer_default,
    }
    for provider in provider_ids("analyser"):
        recommendations.setdefault(
            provider, provider_default_model(provider, "analyser")
        )
    return recommendations


def _parse_worker_model_overrides(
    raw_value: str,
    context: AssignmentContext,
) -> dict[str, str]:
    overrides: dict[str, str] = {}
    for item in (raw_value or "").split(","):
        item = item.strip()
        if not item:
            continue
        provider, separator, model = item.partition("=")
        provider = provider.strip().lower()
        model = model.strip()
        if not separator or not provider or not model:
            raise PrepareError(
                "--worker-model must use provider=model entries separated by commas"
            )
        if provider in overrides:
            raise PrepareError(f"duplicate --worker-model provider: {provider}")
        context.pool.resolve_provider(provider)
        overrides[provider] = model
    return overrides


def _model_for_role(
    *,
    context: AssignmentContext,
    provider: str,
    raw_value: str,
    role: str,
    default_override: str = "",
    duty_id: str = "",
) -> ResolvedAssignment:
    try:
        return resolve_dispatch_assignment(
            context=context,
            host_runtime=context.environment.host_descriptor.id,
            role=role,
            duty_id=duty_id or role,
            provider=provider,
            model=raw_value or default_override,
        )
    except (ValueError, RuntimeError) as exc:
        raise PrepareError(str(exc)) from exc


def _resolve_worker_models(
    inp: PrepareInputs,
    workers: list[str],
    context: AssignmentContext,
) -> dict:
    """Resolve provider models while preserving legacy fixed-provider projections."""
    overrides = _parse_worker_model_overrides(inp.worker_models_raw, context)
    legacy = {
        "claude": inp.claude_model,
        "codex": inp.codex_model,
        "antigravity": inp.antigravity_model,
    }
    legacy_default_env = {
        "claude": "OKSTRA_DEFAULT_CLAUDE_MODEL",
        "codex": "OKSTRA_DEFAULT_CODEX_MODEL",
        "antigravity": "OKSTRA_DEFAULT_ANTIGRAVITY_MODEL",
    }

    def worker_model_input(provider: str) -> str:
        selected = overrides.get(provider, legacy.get(provider, ""))
        env_name = legacy_default_env.get(provider)
        return selected or (_default(env_name, "") if env_name else "")

    selected_worker_providers = tuple(dict.fromkeys(
        worker for worker in workers if worker != "report-writer"
    ))
    provider_models = {}
    for provider in selected_worker_providers:
        provider_models[provider] = _model_for_role(
            context=context,
            provider=provider,
            raw_value=worker_model_input(provider),
            role="analyser",
        )
    try:
        lead_assignment = resolve_context_lead_provider(
            context=context,
            requested_provider=inp.lead_provider,
        )
    except (
        HostCapabilityMismatch,
        HostNotRegistered,
        ProviderUnavailable,
        UnknownProviderError,
    ) as exc:
        raise PrepareError(str(exc)) from exc
    lead_provider = lead_assignment.provider
    report_writer_provider = inp.report_writer_provider.strip().lower() or "claude"
    if not context.pool.resolve_provider(report_writer_provider).supports_role(
        "report-writer"
    ):
        raise PrepareError(
            f"provider {report_writer_provider!r} does not support the report-writer role"
        )

    def legacy_projection(provider: str) -> _LegacyModelProjection:
        assigned = provider_models.get(provider)
        if assigned is not None:
            return _LegacyModelProjection(
                model_id=assigned.model_id,
                display=assigned.display,
                execution=assigned.execution,
            )
        try:
            selected = resolve_model_selection(
                context.pool,
                provider,
                "analyser",
                worker_model_input(provider),
            )
        except (ValueError, RuntimeError) as exc:
            raise PrepareError(str(exc)) from exc
        return _LegacyModelProjection(
            model_id=selected.model_ref.model_id,
            display=selected.display_name,
            execution=selected.execution_value,
        )

    return {
        "lead_provider": lead_provider,
        "report_writer_provider": report_writer_provider,
        "lead": _model_for_role(
            context=context,
            provider=lead_provider,
            raw_value=inp.lead_model,
            role="leader",
            default_override=_legacy_claude_model_override(
                lead_provider,
                "OKSTRA_DEFAULT_LEAD_MODEL",
            ),
            duty_id="lead",
        ),
        "cw": legacy_projection("claude"),
        "co": legacy_projection("codex"),
        "ge": legacy_projection("antigravity"),
        "rw": _model_for_role(
            context=context,
            provider=report_writer_provider,
            raw_value=inp.report_writer_model,
            role="report-writer",
            default_override=_legacy_claude_model_override(
                report_writer_provider,
                "OKSTRA_DEFAULT_REPORT_WRITER_MODEL",
            ),
        ),
        "providers": provider_models,
        "worker_model_inputs": {
            provider: worker_model_input(provider)
            for provider in dict.fromkeys((
                *selected_worker_providers,
                *overrides,
                *legacy,
            ))
        },
    }


def _resolve_model_bindings(
    inp: PrepareInputs,
    workers: list[str],
    context: AssignmentContext,
    requires_executor: bool,
) -> _ModelBindings:
    """worker 모델 + critic 선택 + executor 바인딩을 한 묶음으로 해소·검증한다."""
    m = _resolve_worker_models(inp, workers, context)
    critic_choice = _validate_critic_choice(inp.critic, inp.task_type)
    critic_model_execution = ""
    critic_meta = None
    if critic_choice in provider_ids("critic"):
        critic_meta = _model_for_role(
            context=context,
            provider=critic_choice,
            raw_value=m["worker_model_inputs"].get(critic_choice, ""),
            role="critic",
            duty_id="scope-critic",
        )
        critic_model_execution = critic_meta.execution
    executor_assignment = None
    display_name = ""
    if requires_executor:
        executor_assignment, display_name = _resolve_executor_assignment(
            inp,
            workers,
            context,
            m["worker_model_inputs"],
        )
    lead_assignment = _role_assignment(
        resolved=m["lead"],
        role="lead",
    )
    worker_assignments = _build_worker_assignments(workers, m)
    critic_assignment = (
        _role_assignment(
            resolved=critic_meta,
            role="critic",
        )
        if critic_meta is not None
        else None
    )
    translator_meta = _model_for_role(
        context=context,
        provider=m["report_writer_provider"],
        raw_value=m["rw"].model_id,
        role="translator",
        duty_id="translator",
    )
    translator_assignment = _role_assignment(
        resolved=translator_meta,
        role="translator",
    )
    _reject_split_worker_models(executor_assignment, worker_assignments)
    invocation_assignments = _build_invocation_assignments(
        lead_assignment,
        worker_assignments,
        critic_assignment,
        translator_assignment,
        inp.task_type,
    )
    worker_by_provider = {
        row.provider: row for row in worker_assignments
    }
    return _ModelBindings(
        lead=m["lead"], cw=m["cw"], co=m["co"], ge=m["ge"], rw=m["rw"],
        critic_choice=critic_choice,
        critic_model_execution=critic_model_execution,
        executor_display_name=display_name,
        codex_worker_execution=(
            worker_by_provider["codex"].model_execution_value
            if "codex" in worker_by_provider
            else m["co"].execution
        ),
        antigravity_worker_execution=(
            worker_by_provider["antigravity"].model_execution_value
            if "antigravity" in worker_by_provider
            else m["ge"].execution
        ),
        executor_assignment=executor_assignment,
        lead_assignment=lead_assignment,
        worker_assignments=worker_assignments,
        invocation_assignments=invocation_assignments,
        translator=translator_meta,
    )


_CRITIC_REQUIRED_TASK_TYPES = frozenset({
    "requirements-discovery",
    "error-analysis",
    "implementation-planning",
    "final-verification",
})


def _validate_critic_choice(raw_value: str, task_type: str = "") -> str:
    critic_choice = (raw_value or "").strip().lower()
    allowed_critics = ["", *provider_ids("critic")]
    if critic_choice == "off":
        if task_type in _CRITIC_REQUIRED_TASK_TYPES:
            raise PrepareError(
                "--critic off is not allowed; critic is required "
                f"for {task_type}"
            )
        return "off"
    if critic_choice not in allowed_critics:
        raise PrepareError(
            f"--critic must be one of: {', '.join(provider_ids('critic'))} "
            f"(got: {critic_choice!r})"
        )
    return critic_choice


def _uses_canonical_assignment_projection(inp: PrepareInputs) -> bool:
    return bool(
        inp.role_counts_raw
        or inp.role_models_raw
        or inp.host_session_context.entry_mode == "current-session"
    )


def _project_assignment_plan(
    inp: PrepareInputs,
    plan: AssignmentPlan,
    context: AssignmentContext,
) -> tuple[list[str], _ModelBindings]:
    lead = plan.by_role("leader")[0]
    report_rows = plan.by_role("report-writer")
    report_writer = (
        report_rows[0]
        if report_rows
        else _model_for_role(
            context=context,
            provider=inp.report_writer_provider.strip().lower() or "claude",
            raw_value=inp.report_writer_model,
            role="report-writer",
        )
    )
    worker_rows = tuple(
        row for row in plan.assignments
        if row.role not in {"leader", "critic", "implementer", "report-writer"}
    )
    worker_assignments = tuple(
        _role_assignment(
            resolved=row,
            role=row.role,
            worker_id=_canonical_worker_id(row, worker_rows),
        )
        for row in worker_rows
    )
    if report_rows:
        worker_assignments = (
            *worker_assignments,
            _role_assignment(
                resolved=report_writer,
                role="report-writer",
                worker_id="report-writer",
            ),
        )
    models = _project_model_bindings(
        inp,
        plan,
        context,
        lead,
        report_writer,
        worker_assignments,
    )
    return [row.worker_id for row in worker_assignments], models


def _canonical_worker_id(
    assignment: ResolvedAssignment,
    rows: Sequence[ResolvedAssignment],
) -> str:
    prior = sum(
        row.provider_id == assignment.provider_id and row.ordinal < assignment.ordinal
        for row in rows
    )
    return assignment.provider_id if prior == 0 else f"{assignment.provider_id}-{prior + 1}"


def _project_model_bindings(
    inp: PrepareInputs,
    plan: AssignmentPlan,
    context: AssignmentContext,
    lead: ResolvedAssignment,
    report_writer: ResolvedAssignment,
    workers: tuple[RoleAssignment, ...],
) -> _ModelBindings:
    lead_assignment = _role_assignment(resolved=lead, role="lead")
    critic_rows = plan.by_role("critic")
    critic_assignment = (
        _role_assignment(resolved=critic_rows[0], role="critic")
        if critic_rows
        else None
    )
    executor_rows = plan.by_role("implementer")
    executor_assignment = (
        _role_assignment(
            resolved=executor_rows[0],
            role="executor",
            worker_id=executor_rows[0].provider_id,
        )
        if executor_rows
        else None
    )
    translator = _model_for_role(
        context=context,
        provider=report_writer.provider_id,
        raw_value=report_writer.model_id,
        role="translator",
        duty_id="translator",
    )
    projections = {
        provider: _legacy_projection_from_plan(provider, plan, context)
        for provider in ("claude", "codex", "antigravity")
    }
    _reject_executor_outside_roster(executor_assignment, workers)
    _reject_split_worker_models(executor_assignment, workers)
    invocation_assignments = _build_invocation_assignments(
        lead_assignment,
        workers,
        critic_assignment,
        _role_assignment(resolved=translator, role="translator"),
        inp.task_type,
    )
    worker_by_provider = {row.provider: row for row in workers}
    return _ModelBindings(
        lead=lead,
        cw=projections["claude"],
        co=projections["codex"],
        ge=projections["antigravity"],
        rw=report_writer,
        critic_choice=(critic_rows[0].provider_id if critic_rows else "off"),
        critic_model_execution=(
            critic_assignment.model_execution_value if critic_assignment else ""
        ),
        executor_display_name=(
            f"{executor_rows[0].display_name} executor" if executor_rows else ""
        ),
        codex_worker_execution=(
            worker_by_provider["codex"].model_execution_value
            if "codex" in worker_by_provider
            else projections["codex"].execution
        ),
        antigravity_worker_execution=(
            worker_by_provider["antigravity"].model_execution_value
            if "antigravity" in worker_by_provider
            else projections["antigravity"].execution
        ),
        executor_assignment=executor_assignment,
        lead_assignment=lead_assignment,
        worker_assignments=workers,
        invocation_assignments=invocation_assignments,
        translator=translator,
    )


def _legacy_projection_from_plan(
    provider: str,
    plan: AssignmentPlan,
    context: AssignmentContext,
) -> _LegacyModelProjection:
    selected = next(
        (
            row for row in plan.assignments
            if row.provider_id == provider and row.role != "leader"
        ),
        None,
    )
    if selected is not None:
        return _LegacyModelProjection(
            selected.model_id,
            selected.display_name,
            selected.execution,
        )
    model = resolve_model_selection(context.pool, provider, "analyser", "")
    return _LegacyModelProjection(
        model.model_ref.model_id,
        model.display_name,
        model.execution_value,
    )


def _selected_execution_provider_ids(
    inp: PrepareInputs,
    workers: list[str],
    requires_executor: bool,
) -> tuple[str, ...]:
    selected = [
        inp.lead_provider,
        *(worker for worker in workers if worker != "report-writer"),
        inp.report_writer_provider.strip().lower() or "claude",
    ]
    critic = inp.critic.strip().lower()
    if critic not in {"", "off"}:
        selected.append(critic)
    if requires_executor:
        selected.append(
            inp.executor.strip().lower()
            or _default("OKSTRA_DEFAULT_EXECUTOR", "claude")
        )
    return tuple(dict.fromkeys(selected))


def _reject_executor_outside_roster(
    executor: RoleAssignment | None,
    workers: tuple[RoleAssignment, ...],
) -> None:
    """Refuse an executor whose provider the roster never dispatches.

    An implementation run opens its executor with `--workers <provider>`, so a
    provider absent from the roster has no invocation to open. The legacy
    selection path says so outright; the canonical path derived its roster from
    role models alone, so `--executor <provider>` rendered fine and the run only
    failed later, at `requested worker(s) are not in this run roster`.
    """
    if executor is None:
        return
    roster = {row.worker_id for row in workers if row.worker_id}
    if executor.worker_id in roster:
        return
    raise PrepareError(
        f"--executor {executor.worker_id} is not in this run's roster "
        f"({', '.join(sorted(roster)) or 'empty'}); the executor is dispatched "
        f"as a worker, so give it a roster slot — "
        f"--role-model verifier={executor.worker_id}/<model> — or pick an "
        f"executor already in the roster."
    )


def _reject_split_worker_models(
    executor: RoleAssignment | None,
    workers: tuple[RoleAssignment, ...],
) -> None:
    """Refuse one worker id standing for two roles on two different models.

    The roster names a worker by provider, so an implementation run whose
    executor and verifier are the same provider shares that id — but
    `invocationAssignments["initial/<id>"]` can hold only one model, and it
    holds the verifier's. Dispatch then looks the executor's role execution up
    by that assignment, finds no row with a matching execution value, and stops
    at `v2 execution identity does not match a canonical role execution`. The
    render used to succeed and only the dispatch failed, by which point the run
    had already claimed its stage.
    """
    if executor is None:
        return
    peer = next(
        (row for row in workers if row.worker_id == executor.worker_id),
        None,
    )
    if peer is None or peer.model_execution_value == executor.model_execution_value:
        return
    raise PrepareError(
        f"worker {executor.worker_id!r} is the executor on "
        f"{executor.model_execution_value!r} and a verifier on "
        f"{peer.model_execution_value!r}; one worker id carries one model. "
        f"Give both roles the same model, or pick a different --executor "
        f"provider."
    )


def _resolve_executor_assignment(
    inp: PrepareInputs,
    workers: list[str],
    context: AssignmentContext,
    worker_model_inputs: dict[str, str],
) -> tuple[RoleAssignment, str]:
    executor_default = _default("OKSTRA_DEFAULT_EXECUTOR", "claude")
    executor_provider = (inp.executor or executor_default).strip().lower()
    allowed_executors = provider_ids("executor")
    if executor_provider not in allowed_executors:
        raise PrepareError(
            f"--executor must be one of: {', '.join(allowed_executors)} "
            f"(got: {executor_provider!r})"
        )
    if executor_provider not in workers:
        raise PrepareError(
            f"--executor {executor_provider} requires {executor_provider!r} in "
            f"--workers, but resolved roster is {workers!r}. "
            f"Add it explicitly, e.g. --workers {','.join(sorted(set(workers + [executor_provider])))}."
        )
    model = _model_for_role(
        context=context,
        provider=executor_provider,
        raw_value=worker_model_inputs.get(executor_provider, ""),
        role="implementer",
        duty_id="implementation-executor",
    )
    assignment = _role_assignment(
        resolved=model,
        role="executor",
        worker_id=executor_provider,
    )
    display_name = (
        f"{context.pool.resolve_provider(executor_provider).display_label} executor"
    )
    return assignment, display_name


def _build_worker_assignments(
    workers: list[str], models: dict,
) -> tuple[RoleAssignment, ...]:
    assignments: list[RoleAssignment] = []
    for worker_id in workers:
        is_report_writer = worker_id == "report-writer"
        provider = (
            models["report_writer_provider"] if is_report_writer else worker_id
        )
        selected = (
            models["rw"] if is_report_writer else models["providers"][provider]
        )
        assignments.append(_role_assignment(
            resolved=selected,
            role="report-writer" if is_report_writer else "analyser",
            worker_id=worker_id,
        ))
    return tuple(assignments)


def _role_assignment(
    *,
    resolved: ResolvedAssignment,
    role: str,
    worker_id: str = "",
) -> RoleAssignment:
    binding = resolved.binding
    if binding is None:
        if resolved.role != "leader" or resolved.entry_mode != "current-session":
            raise PrepareError("legacy assignment projection requires a model binding")
        return RoleAssignment(
            role=role,
            provider=resolved.provider_id,
            model_display=resolved.display_name,
            model_id=resolved.model_id,
            model_execution_value="unknown",
            runner="cli-wrapper",
            host_runtime=resolved.host_runtime,
            host_model_value=None,
            worker_id=worker_id,
        )
    return RoleAssignment(
        role=role,
        provider=resolved.provider_id,
        model_display=resolved.display_name,
        model_id=resolved.model_id,
        model_execution_value=binding.resolved_execution_value,
        runner=(
            "native-session"
            if binding.runner == "current-session"
            else binding.runner
        ),
        host_runtime=resolved.host_runtime,
        host_model_value=binding.host_model_value,
        worker_id=worker_id,
    )


def _build_invocation_assignments(
    lead: RoleAssignment,
    workers: tuple[RoleAssignment, ...],
    critic: RoleAssignment | None,
    translator: RoleAssignment,
    task_type: str,
) -> dict[str, dict[str, object]]:
    assignments = {"lead": lead.to_model_payload()}
    for assignment in workers:
        assignments[f"initial/{assignment.worker_id}"] = (
            assignment.to_model_payload()
        )
        assignments[f"reverify/{assignment.worker_id}"] = (
            assignment.to_model_payload()
        )
    if critic is not None:
        # 한 run 이 도달할 수 있는 critic 은 한 종류다. 둘 다 등록하면 실행될
        # 수 없는 로스터 행이 남고, `_optional_worker_roles` 가 provider 로만
        # role 라벨을 만들기 때문에 두 행의 이름이 겹쳐 validate-run 이
        # `duplicate worker role detected` 로 런을 실패시킨다.
        assignments[critic_assignment_ref(task_type)] = critic.to_model_payload()
    assignments["translator"] = translator.to_model_payload()
    return assignments


StageRunClaim = _implementation_stage.StageRunClaim


def _implementation_stage_prepare_error(
    exc: _implementation_stage.ImplementationStageError,
) -> PrepareError:
    return PrepareError(str(exc))


def _claim_implementation_stage_run(
    inp: PrepareInputs,
    ctx_stage_map: list,
    task_group_segment: str,
    task_id_segment: str,
    task_key: str,
    executor_worktree_status: str,
) -> StageRunClaim:
    """Compatibility wrapper for claiming an implementation stage run."""
    try:
        return _implementation_stage.claim_implementation_stage_run(
            inp,
            ctx_stage_map,
            task_group_segment,
            task_id_segment,
            task_key,
            executor_worktree_status,
        )
    except _implementation_stage.ImplementationStageError as exc:
        raise _implementation_stage_prepare_error(exc) from exc


def _publish_stage_run_claim(
    inp: PrepareInputs,
    ctx: dict,
    ctx_stage_map: list,
    claim: StageRunClaim,
) -> None:
    """Compatibility wrapper for publishing a Stage Run Claim."""
    _implementation_stage.publish_stage_run_claim(inp, ctx, ctx_stage_map, claim)


def _git_out(cwd, *args) -> str:
    r = _subprocess.run(["git", "-C", str(cwd), *args],
                        capture_output=True, text=True)
    return r.stdout.strip() if r.returncode == 0 else ""


def _single_stage_final_verification_worktree(inp: "PrepareInputs") -> WorktreeProvision:
    """Placeholder until the selected stage registry row is resolved."""
    return WorktreeProvision(
        status="deferred-final-verification",
        note=(
            "final-verification single-stage uses the selected implementation "
            "stage worktree from the registry"
        ),
    )


def _format_integration(integ) -> str:
    """stage 자동 통합 결과를 사람이 읽을 한국어 마크다운 블록으로 만든다."""
    def _csv(xs):
        return ", ".join(str(x) for x in xs) if xs else "없음"

    skipped = "; ".join(f"stage {n}({why})" for n, why in integ.teardown_skipped)
    lines = [
        "- **Stage 자동 통합 결과** (whole-task final-verification):",
        "- **머지된 stage:** " + _csv(integ.merged),
        "- **이미 머지됨:** " + _csv(integ.already_merged),
        "- **정리(teardown)된 stage:** " + _csv(integ.torn_down),
        "- **정리 보류:** " + (skipped or "없음"),
    ]
    for w in integ.warnings:
        lines.append("- **경고:** " + w)
    return "\n".join(lines)


def _final_verification_target_request(
    inp: "PrepareInputs",
    ctx_stage_map: list[dict],
) -> _stage_targets.FinalVerificationTargetRequest:
    stage = (
        int(inp.stage)
        if inp.stage and inp.stage != "auto"
        else None
    )
    return _stage_targets.FinalVerificationTargetRequest(
        project_root=Path(inp.project_root),
        project_id=inp.project_id,
        task_group=inp.task_group,
        task_id=inp.task_id,
        work_category=inp.work_category,
        approved_plan_path=Path(inp.approved_plan_path),
        stage=stage,
        stage_map=tuple(ctx_stage_map),
    )


def _apply_final_verification_target(
    ctx: dict,
    acquisition: _stage_targets.FinalVerificationTargetAcquisition,
) -> None:
    """Translate acquired domain facts into render-context fields."""
    target = acquisition.target
    if target.scope == "single-stage":
        stage = target.stages[0]
        ctx["EXECUTOR_WORKTREE_PATH"] = target.worktree_path
        ctx["EXECUTOR_WORKTREE_BRANCH"] = acquisition.worktree_branch
        ctx["EXECUTOR_WORKTREE_BASE_REF"] = target.base
        ctx["EXECUTOR_WORKTREE_STATUS"] = "reused-stage"
        ctx["EXECUTOR_WORKTREE_NOTE"] = (
            f"final-verification uses implementation stage {stage} worktree"
        )
    if acquisition.integration_result is not None:
        ctx["STAGE_INTEGRATION"] = _format_integration(
            acquisition.integration_result
        )
    diff_stat = _git_out(target.worktree_path, "diff", "--stat",
                         f"{target.base}..{target.head}")
    ctx["VERIFICATION_SCOPE"] = target.scope
    ctx["VERIFICATION_WORKTREE_PATH"] = target.worktree_path
    ctx["VERIFICATION_BASE_REF"] = target.base
    ctx["VERIFICATION_HEAD_REF"] = target.head
    ctx["VERIFICATION_TARGET"] = _format_verification_target(target, diff_stat)


def _format_verification_target(
    target: _stage_targets.FinalVerificationTarget,
    diff_stat: str,
) -> str:
    reports = "\n".join(
        f"  - stage {s}: `{rp or '(report_path unrecorded)'}`"
        for s, rp in zip(target.stages, target.reports)
    )
    return (
        f"- **Verification scope:** `{target.scope}`\n"
        f"- **Worktree:** `{target.worktree_path}`\n"
        f"- **Verification base ref:** `{target.base}`\n"
        f"- **Verification head ref:** `{target.head}`\n"
        f"- **Stages under verification:** {target.stages}\n"
        f"- **Source implementation reports:**\n{reports}\n"
        f"- **Verification diff stat:**\n```\n{diff_stat}\n```"
    )


def write_verification_target_snapshot(
    instruction_set: Path,
    target_markdown: str,
) -> tuple[Path, str]:
    """Persist one normalized final-verification target and return its digest."""
    normalized = target_markdown.replace("\r\n", "\n").replace("\r", "\n")
    normalized = normalized.rstrip() + "\n"
    digest = "sha256:" + hashlib.sha256(normalized.encode("utf-8")).hexdigest()
    path = instruction_set / "verification-target.md"
    path.write_text(
        normalized + f"- **Verification target digest:** `{digest}`\n",
        encoding="utf-8",
    )
    return path, digest


def _write_verification_target_artifact(
    inp: PrepareInputs,
    ctx: dict,
    instruction_set: Path,
) -> None:
    if inp.task_type != "final-verification":
        return
    target_markdown = str(ctx.get("VERIFICATION_TARGET") or "")
    if not target_markdown:
        raise PrepareError("final-verification target snapshot is missing")
    target_path, target_digest = write_verification_target_snapshot(
        instruction_set,
        target_markdown,
    )
    ctx["VERIFICATION_TARGET_PATH"] = str(target_path)
    ctx["VERIFICATION_TARGET_RELATIVE_PATH"] = (
        f"{ctx['INSTRUCTION_SET_RELATIVE_PATH']}/verification-target.md"
    )
    ctx["VERIFICATION_TARGET_DIGEST"] = target_digest
    ctx["VERIFICATION_TARGET"] = target_path.read_text(encoding="utf-8")


def _write_prior_run_error_digest(ctx: dict, instruction_set: Path) -> None:
    """Stage the earlier runs' actionable errors — only when there are any.

    No file at all when the task recorded none, rather than one saying "none":
    a staged file the lead must open to learn it is empty costs a read every
    run and teaches the lead to stop opening that path.
    """
    digest = prior_run_error_digest(Path(ctx["TASK_ROOT"]))
    if digest:
        (instruction_set / "prior-run-errors.md").write_text(digest, encoding="utf-8")


def _stage_or_clear(path: Path, body: str) -> None:
    """이 런의 값을 쓰거나, 값이 없으면 이전 런이 남긴 파일을 지운다.

    instruction-set 은 run 단위가 아니라 **task 단위** 디렉터리다
    (`paths.py`: `task_root / "instruction-set"`). 그래서 값이 있을 때만 쓰고
    없을 때 아무것도 안 하면, 이전 런의 파일이 그대로 남아 다음 런의 입력이 된다.
    directive 를 비우고 준비한 런에서 워커들이 예전 directive 를 읽고, 그 내용이
    현재 기록과 어긋나 반박에 시간을 쓴 사례가 보고됐다. carry-in 답변
    (`clarification-response.md`)은 같은 형태에 결과가 더 나쁘다 — 남은 파일이
    analysis packet 에 그대로 실린다.
    """
    if body:
        path.write_text(body, encoding="utf-8")
        return
    path.unlink(missing_ok=True)


def _write_instruction_set_sources(
    inp: PrepareInputs,
    ctx: dict,
    profile_content: str,
    review_material: str,
    host_rules_file: Path | None,
    selected_direction: SelectedDirection | None,
) -> Path:
    """instruction-set 디렉터리에 profile/material/brief/clarification/directive 와
    reference-expectations 를 기록하고 디렉터리 경로를 돌려준다."""
    instruction_set = Path(ctx["INSTRUCTION_SET_PATH"])
    instruction_set.mkdir(parents=True, exist_ok=True)
    if selected_direction is not None:
        write_selected_direction_snapshot(
            selected_direction,
            instruction_set / "selected-direction.json",
        )
    _write_analysis_evidence_artifact(ctx, instruction_set)
    _write_verification_target_artifact(inp, ctx, instruction_set)
    _write_prior_run_error_digest(ctx, instruction_set)
    profile_rendered = profile_content
    if inp.task_type == "implementation":
        profile_rendered += "\n\n{{DESIGN_PREP_CONTEXT}}\n\n{{FIX_RUN_CONTEXT}}"
    profile_tokens = (
        "EXECUTOR_WORKER_ID",
        "EXECUTOR_PROVIDER",
        "EXECUTOR_DISPLAY_NAME",
        "EXECUTOR_MODEL_DISPLAY",
        "EXECUTOR_MODEL_EXECUTION_VALUE",
        "EXECUTOR_HOST_MODEL_VALUE",
        "EXECUTOR_RUNNER",
        "EXECUTOR_DISPATCH_MODE",
        "EXECUTOR_WORKTREE_PATH",
        "EXECUTOR_WORKTREE_BRANCH",
        "EXECUTOR_WORKTREE_BASE_REF",
        "EXECUTOR_WORKTREE_STATUS",
        "EXECUTOR_WORKTREE_NOTE",
        "DESIGN_PREP_CONTEXT",
        "FIX_RUN_CONTEXT",
        "PHASE_FORBIDDEN_ACTIONS",
    )
    for key in profile_tokens:
        profile_rendered = profile_rendered.replace("{{" + key + "}}", ctx.get(key, ""))
    (instruction_set / "analysis-profile.md").write_text(profile_rendered, encoding="utf-8")
    if inp.task_type == "implementation":
        executor_source = (
            Path(ctx["WORKSPACE_ROOT"])
            / "prompts"
            / "profiles"
            / "_implementation-executor.md"
        )
        executor_rendered = executor_source.read_text(encoding="utf-8")
        for key in profile_tokens:
            executor_rendered = executor_rendered.replace(
                "{{" + key + "}}", ctx.get(key, "")
            )
        (instruction_set / "implementation-executor.md").write_text(
            executor_rendered,
            encoding="utf-8",
        )
    (instruction_set / "analysis-material.md").write_text(review_material, encoding="utf-8")
    shutil.copyfile(inp.brief_path, instruction_set / "task-brief.md")
    # A rule that lives only in the conversation is one compaction away from
    # gone, with no signal that it went. On disk it survives, and the session
    # conformance check can ask afterwards whether it was read. The token is
    # set unconditionally because render_template_with_ctx fail-fasts on a
    # token the launch template references but ctx lacks.
    host_rules_relative = ""
    if host_rules_file is not None:
        staged_host_rules = instruction_set / "host-orchestration-rules.md"
        shutil.copyfile(host_rules_file, staged_host_rules)
        host_rules_relative = relative_to_project_root(
            staged_host_rules, Path(ctx["PROJECT_ROOT"])
        )
    ctx["HOST_ORCHESTRATION_RULES_RELATIVE_PATH"] = host_rules_relative
    if inp.clarification_response_path:
        clarification_body = clarification_response_with_sidecars(
            Path(inp.clarification_response_path)
        )
    elif inp.task_type == "implementation" and inp.approved_plan_path:
        # implementation carry-in: the approved plan reaches this run by path
        # (executor re-reads it), so attach ONLY the planning HTML form answers
        # (`runs/implementation-planning/.../user-responses/`) — without these
        # the user's clarification answers never reach the implementation run.
        clarification_body = attached_user_responses_section(
            Path(inp.approved_plan_path)
        )
    else:
        clarification_body = ""
    _stage_or_clear(
        instruction_set / "clarification-response.md", clarification_body
    )
    _stage_or_clear(
        instruction_set / "directive.txt",
        inp.directive + "\n" if inp.directive else "",
    )
    render_reference_expectations(
        str(inp.brief_path), str(instruction_set / "reference-expectations.md"), ctx,
    )
    # 계획 phase 만 원장을 받는다. 원장은 "무엇이 이미 지어졌나" 를 계획 저작
    # 쪽에 알려 주는 것이므로, 계획을 쓰지 않는 phase 에는 실을 자리가 없다.
    stage_ledger = (
        build_stage_ledger(Path(ctx["TASK_MANIFEST_PATH"]).parent)
        if inp.task_type == "implementation-planning" else None
    )
    packet = build_analysis_packet(
        task_key=ctx["TASK_KEY"],
        task_type=ctx["TASK_TYPE"],
        task_brief_path=instruction_set / "task-brief.md",
        analysis_profile_path=instruction_set / "analysis-profile.md",
        reference_expectations_path=instruction_set / "reference-expectations.md",
        clarification_response_path=(
            instruction_set / "clarification-response.md"
            if (instruction_set / "clarification-response.md").is_file() else None
        ),
        directive=inp.directive,
        instruction_set_relative_path=ctx["INSTRUCTION_SET_RELATIVE_PATH"],
        fix_history_text=fix_cycles.packet_summary(
            fix_cycles.read_rows(Path(ctx["TASK_MANIFEST_PATH"]).parent)),
        stage_ledger_json=render_stage_ledger(stage_ledger),
        stage_ledger_notice=stage_ledger_notice(stage_ledger),
    )
    if inp.task_type in ANALYSIS_TASK_TYPES:
        packet += (
            "\n## Analysis Evidence\n\n"
            f"- Analysis evidence: `{ctx['INSTRUCTION_SET_RELATIVE_PATH']}/analysis-evidence.md`\n"
        )
    (instruction_set / "analysis-packet.md").write_text(packet, encoding="utf-8")
    return instruction_set


def _write_analysis_evidence_artifact(ctx: dict, instruction_set: Path) -> None:
    """Write selected evidence metadata without copying the original reports."""
    if ctx.get("TASK_TYPE") not in ANALYSIS_TASK_TYPES:
        return
    evidence = json.loads(ctx.get("EVIDENCE_INPUTS_JSON", "[]"))
    lines = [
        "# Analysis Evidence",
        "",
        "Revalidate every cited finding against the current code before relying on it.",
    ]
    for item in evidence:
        lines.extend([
            "",
            f"## {item['taskType']} / run {item['runSeq']}",
            "",
            f"- Metadata: task key `{item['taskKey']}`, source commit `{item['sourceCommit']}`, relation `{item['relation']}`",
            f"- Original report: `{item['reportPath']}`",
            f"- Review status: `{item['reviewStatus']}`",
            f"- Freshness: `{item['freshness']}`",
            "- Current-code obligation: revalidate cited findings against the current code.",
        ])
    if not evidence:
        lines.extend(["", "- No evidence reports were selected."])
    (instruction_set / "analysis-evidence.md").write_text(
        "\n".join(lines) + "\n", encoding="utf-8",
    )


def _render_lead_prompt_and_snapshot(
    inp: PrepareInputs,
    ctx: dict,
    instruction_set: Path,
    final_report_template: Path,
    prompt_template: Path,
) -> str:
    """Render lead instructions, then publish one verified lead invocation."""
    # inject populates ctx with compute + default tokens consumed by the lead
    # prompt render below (lead-execution-prompt.md). The final-report
    # template render is effectively a copy (Jinja2 `{{ var }}` syntax does
    # not match `_TOKEN_RE`); routed through render_template_with_ctx for SOT
    # consistency.
    inject_lead_prompt_computed_tokens(ctx)
    apply_lead_prompt_defaults(ctx)
    render_template_with_ctx(
        str(final_report_template), ctx["FINAL_REPORT_TEMPLATE_PATH"], ctx,
    )
    # Per-task-type schema excerpt for the report-writer worker. The full
    # schema validates the data.json post-hoc; the worker only
    # needs the common structure + this run's task-type block, so we write a
    # scoped excerpt into the instruction-set rather than make the worker read
    # the whole 44 KB / all-task-types schema (whose repo `schemas/...` path is
    # not resolvable from a consumer project's task bundle anyway).
    #
    # Guarded: a missing/unreadable schema must NOT break bundle preparation.
    # If the excerpt cannot be produced (e.g. an older install that predates
    # the schemas/ copy step), prep proceeds without it — the report-writer
    # still has the phase-stripped template + skill structure guide, and
    # validation runs against the full schema regardless.
    try:
        _excerpt = build_schema_excerpt(
            load_schema_version(CURRENT_REPORT_SCHEMA_VERSION),
            inp.task_type,
            installed_version(),
        )
        write_owned_object_atomic(
            Path(ctx["FINAL_REPORT_SCHEMA_PATH"]),
            _excerpt,
            artifact="final report schema excerpt",
        )
    except Exception:  # noqa: BLE001 — advisory artifact; never fail prep over it
        pass
    instructions_path = Path(ctx["LEAD_INSTRUCTIONS_PATH"])
    render_template_with_ctx(str(prompt_template), str(instructions_path), ctx)
    assignment = _agent_model_assignment(
        json.loads(ctx["INVOCATION_ASSIGNMENTS_JSON"])["lead"]
    )
    identity = invocation_execution_identity_from_manifest(
        load_owned_object(
            Path(ctx["RUN_MANIFEST_PATH"]), artifact="run manifest"
        ),
        assignment=assignment,
        assignment_ref="lead",
        duty_id="lead",
    )
    if identity is None:
        raise PrepareError("lead v2 execution identity is missing")
    invocation_id = f"{ctx['TASK_TYPE_SEGMENT']}-{ctx['RUN_PROMPTS_SEQ']}-lead"
    prepared = prepare_agent_invocation(AgentInvocationRequest(
        invocation_id=invocation_id,
        worker_id=None,
        audience="lead",
        assignment_ref="lead",
        purpose=None,
        assignment=assignment,
        instruction=AgentInstruction(
            anchor_lines=(),
            body=instructions_path.read_text(encoding="utf-8"),
            source_paths=(
                AgentInstructionSource(
                    "project", ctx["LEAD_INSTRUCTIONS_RELATIVE_PATH"]
                ),
                AgentInstructionSource("runtime", "prompts/launch.template.md"),
                AgentInstructionSource(
                    "runtime", "prompts/lead/okstra-lead-contract.md"
                ),
            ),
        ),
        project_root=Path(ctx["PROJECT_ROOT"]),
        run_manifest_path=Path(ctx["RUN_MANIFEST_PATH"]),
        duty_root=Path(ctx["DUTY_CONTRACT_ROOT"]),
        prompt_path=Path(ctx["RUN_PROMPT_SNAPSHOT_FILE"]),
        metadata_path=Path(ctx["LEAD_PROMPT_METADATA_PATH"]),
        dispatch_kind="lead",
        participant_ref=identity.participant_ref,
        role_execution_ref=identity.role_execution_ref,
        duty_id=identity.duty_id,
        invocation_ref=invocation_id,
        attempt=1,
    ))
    return prepared.prompt_path.read_text(encoding="utf-8")


def _agent_model_assignment(payload: object) -> AgentModelAssignment:
    try:
        return agent_model_assignment_from_payload(payload)
    except AgentInvocationError as exc:
        raise PrepareError(str(exc)) from exc


def _persist_run_inputs(
    inp: PrepareInputs,
    ctx: dict,
    models: _ModelBindings,
    selected_reviewers: str,
    brief_relative: str,
) -> None:
    """이 run 의 입력 스냅샷(run-inputs-*.json)을 run-manifests 디렉터리에 기록."""
    approved_plan_path = (
        str(Path(inp.approved_plan_path).resolve())
        if inp.approved_plan_path
        else ""
    )
    inputs = {
            "taskBriefPath": brief_relative,
            "taskBriefAbsolutePath": str(inp.brief_path),
            "directive": inp.directive,
            "workers": selected_reviewers,
            "leadProvider": models.lead_assignment.provider,
            "leadModel": models.lead.display,
            "claudeModel": models.cw.display,
            "codexModel": models.co.display,
            "antigravityModel": models.ge.display,
            "workerAssignments": [
                assignment.to_payload() for assignment in models.worker_assignments
            ],
            "reportWriterProvider": next(
                (
                    assignment.provider
                    for assignment in models.worker_assignments
                    if assignment.worker_id == "report-writer"
                ),
                inp.report_writer_provider or "claude",
            ),
            "reportWriterModel": models.rw.display,
            "executor": (
                models.executor_assignment.provider
                if models.executor_assignment is not None
                else ""
            ),
            "relatedTasks": inp.related_tasks_raw,
            "approvedPlanPath": approved_plan_path,
            "clarificationResponsePath": inp.clarification_response_path,
            "selectedDirectionPath": inp.selected_direction_path,
            "analysisTarget": json.loads(ctx.get("ANALYSIS_TARGET_JSON", "{}")).get(
                "requestedValue", ""
            ),
            "evidenceInputs": json.loads(ctx.get("EVIDENCE_INPUTS_JSON", "[]")),
            "renderOnly": inp.render_only,
    }
    run_inputs_path = _write_run_inputs_once(
        project_root=Path(inp.project_root),
        run_manifests_dir=Path(ctx["RUN_MANIFESTS_DIR"]),
        task_type_segment=ctx["TASK_TYPE_SEGMENT"],
        seq=ctx["RUN_MANIFESTS_SEQ"],
        inputs=inputs,
    )
    ctx["RUN_INPUTS_PATH"] = str(run_inputs_path)
    ctx["RUN_INPUTS_RELATIVE_PATH"] = relative_to_project_root(
        run_inputs_path, Path(inp.project_root)
    )


def _write_run_inputs_once(
    *,
    project_root: Path,
    run_manifests_dir: Path,
    task_type_segment: str,
    seq: str,
    inputs: dict,
) -> Path:
    path = run_manifests_dir / f"run-inputs-{task_type_segment}-{seq}.json"
    existing = _read_existing_manifest(path)
    if existing.get("schemaVersion") == "2.0":
        if existing.get("inputs") != inputs:
            raise PrepareError("input snapshot is immutable after v2 creation")
        return path
    return write_run_inputs(
        project_root=project_root,
        run_manifests_dir=run_manifests_dir,
        task_type_segment=task_type_segment,
        seq=seq,
        inputs=inputs,
    )


def _persist_pre_dispatch_run_manifest(ctx: dict) -> None:
    """Publish and re-read the complete invocation authority before prompts."""
    render_run_manifest(ctx["RUN_MANIFEST_PATH"], ctx)
    manifest = _read_existing_manifest(Path(ctx["RUN_MANIFEST_PATH"]))
    expected_contract = json.loads(ctx["AGENT_CONTRACT_JSON"])
    expected_assignments = json.loads(ctx["INVOCATION_ASSIGNMENTS_JSON"])
    if (
        manifest.get("agentContract") != expected_contract
        or manifest.get("invocationAssignments") != expected_assignments
    ):
        raise PrepareError("pre-dispatch run manifest is incomplete")


def _read_existing_manifest(manifest_path: Path) -> dict:
    if not manifest_path.exists():
        return {}
    try:
        return load_owned_object(manifest_path, artifact="run manifest")
    except (OSError, JsonBoundaryError):
        return {}


def _maybe_open_fix_cycle(inp: PrepareInputs, task_root: Path, existing: dict,
                          now: str):
    """--fix-cycle yes 검증 후 새 cycle 을 연다. 부적격이면 PrepareError."""
    if inp.task_type not in fix_cycles.FIX_CYCLE_ENTRY_PHASES:
        raise PrepareError(
            "--fix-cycle yes 는 entry phase("
            f"{', '.join(fix_cycles.FIX_CYCLE_ENTRY_PHASES)})에서만 허용됩니다: "
            f"{inp.task_type}")
    workflow = existing.get("workflow") or {}
    if workflow.get("lastCompletedPhase") != "release-handoff":
        raise PrepareError(
            "--fix-cycle yes 는 release-handoff 까지 완료된 task 에만 허용됩니다 "
            "(workflow.lastCompletedPhase 확인)")
    brief_text = Path(inp.brief_path).read_text(encoding="utf-8")
    fix_cycles.append_opened(
        task_root, target_report=str(existing.get("latestReportRecordPath", "")),
        symptom=fix_cycles.derive_symptom(brief_text), opened_at=now)
    return fix_cycles.open_cycle(fix_cycles.read_rows(task_root))


def _record_fix_cycle_events(inp: PrepareInputs, ctx: dict) -> str:
    """fix-cycles.jsonl 의 lazy-close → opened → run 기록. cycle id 반환.

    - lazy-close: open cycle 에 release-handoff run 이 부착됐고 디스크 manifest 의
      lastCompletedPhase 가 release-handoff 면 닫는다. manifest 는 prepare 가
      이번 run 으로 재작성하기 *전* 값이어야 하므로 finalize 직전에 호출한다.
    - opened: --fix-cycle yes 일 때만. 완료 task + entry phase 미충족이면
      PrepareError.
    - run: open cycle 이 있으면 이번 run 을 무조건 부착.
    """
    task_root = Path(ctx["TASK_MANIFEST_PATH"]).parent
    existing = _read_existing_manifest(Path(ctx["TASK_MANIFEST_PATH"]))
    workflow = existing.get("workflow") or {}
    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    open_c = fix_cycles.open_cycle(fix_cycles.read_rows(task_root))

    if open_c and workflow.get("lastCompletedPhase") == "release-handoff":
        rows = fix_cycles.read_rows(task_root)
        handoff_attached = any(
            r.get("event") == "run" and r.get("cycle") == open_c["cycle"]
            and r.get("task_type") == "release-handoff" for r in rows)
        if handoff_attached:
            fix_cycles.append_closed(
                task_root, cycle=open_c["cycle"], closed_by="release-handoff",
                report=str(existing.get("latestReportRecordPath", "")), closed_at=now)
            open_c = None

    if inp.fix_cycle == "yes" and open_c is None:
        open_c = _maybe_open_fix_cycle(inp, task_root, existing, now)

    if open_c is None:
        return ""
    run_manifest_rel = os.path.relpath(
        ctx["RUN_MANIFEST_PATH"], str(Path(inp.project_root)))
    fix_cycles.append_run(
        task_root, cycle=open_c["cycle"], task_type=inp.task_type,
        run_seq=int(ctx.get("RUN_MANIFESTS_SEQ", 0) or 0),
        run_manifest=run_manifest_rel)
    return open_c["cycle"]


def _finalize_status_and_render_manifests(
    inp: PrepareInputs, ctx: dict, task_index_template: Path,
    forbidden_by_phase: dict[str, str],
) -> None:
    """최종 task/run status 를 확정하고 workflow state 를 재계산한 뒤 team-state,
    task-manifest, task-index, run-manifest, timeline, discovery 산출물을 렌더한다."""
    if inp.render_only:
        ctx["CURRENT_TASK_STATUS"] = "ready-for-lead"
        ctx["CURRENT_RUN_STATUS"] = "prepared"
    else:
        ctx["CURRENT_TASK_STATUS"] = "lead-session-started"
        ctx["CURRENT_RUN_STATUS"] = "in-progress"
    ctx["LATEST_REPORT_PATH"] = ctx["FINAL_REPORT_PATH"]
    ctx["LATEST_REPORT_RECORD_RELATIVE_PATH"] = ctx["FINAL_REPORT_RECORD_RELATIVE_PATH"]
    ctx.update(compute_workflow_state(
        task_type=inp.task_type,
        current_run_status=ctx["CURRENT_RUN_STATUS"],
        current_task_status=ctx["CURRENT_TASK_STATUS"],
        render_only=inp.render_only,
        forbidden_by_phase=forbidden_by_phase,
        work_category=inp.work_category,
    ))
    render_team_state(ctx["TEAM_STATE_PATH"], ctx)
    render_active_run_context(ctx["ACTIVE_RUN_CONTEXT_PATH"], ctx)
    render_task_manifest(ctx["TASK_MANIFEST_PATH"], ctx)
    render_task_index(str(task_index_template), ctx["TASK_INDEX_PATH"], ctx)
    render_run_manifest(ctx["RUN_MANIFEST_PATH"], ctx)
    render_timeline(ctx["TIMELINE_PATH"], ctx)
    render_task_catalog_discovery(ctx["OKSTRA_TASK_CATALOG_FILE"], ctx)
    render_latest_task_discovery(ctx["OKSTRA_LATEST_TASK_FILE"], ctx)


def _provision_settings_symlink(inp: PrepareInputs) -> None:
    """non-render-only run 에서 project settings symlink 를 idempotent 하게 보장한다.
    실패해도 prepare 를 막지 않고 경고만 출력한다 (worker dispatch 권한 이슈)."""
    try:
        link = ensure_project_settings_symlink(project_root=Path(inp.project_root))
    except SettingsLinkError as exc:
        print(
            f"okstra-settings: failed to provision project settings symlink — "
            f"worker dispatch may be blocked by Claude Code permissions. ({exc})",
            file=sys.stderr,
        )
    else:
        if link is None:
            print(
                "okstra-settings: ~/.okstra/templates/settings.local.json missing — "
                "re-run 'npx okstra@latest install' (0.14.0+) to provision the symlink target.",
                file=sys.stderr,
            )


@dataclass
class _ProvisionedTask:
    """What one task-key's provisioning critical section hands to run-path compute."""

    worktree: WorktreeProvision
    stage_run_claim: object | None
    stage_arg: int | None


def _write_bundle_artifacts(
    inp: PrepareInputs,
    ctx: dict,
    project_root: Path,
    lead_runtime: str,
    claude_session_id: str,
) -> None:
    """Create the task/run directories and the artifacts the lead reads at launch.

    Order is contractual. The fix-cycle ledger runs LAST here but still *before*
    the instruction-set build in the caller, so a cycle opened by this run shows
    up in the analysis packet's Fix History; finalize rewrites the manifest after
    that, which keeps the lazy-close invariant that the on-disk manifest holds
    pre-rewrite values when the cycle is judged.
    """
    _ensure_task_directories(ctx)
    migrate_legacy_run_artifacts(ctx)
    cleanup_obsolete_generated_docs(
        project_root=project_root, instruction_set_dir=Path(ctx["INSTRUCTION_SET_PATH"]),
    )
    if lead_runtime == "claude-code":
        # Always materialise the resume command script for Claude Code. Even in
        # --render-only preparation flows the user (or a later non-interactive
        # runner) may invoke it manually; deferring its creation until
        # interactive launch leaves runs/<phase>/sessions/ empty and the
        # manifest pointing at a path that does not exist. Codex runs resume via
        # artifacts/checkpoints instead, so they intentionally leave this empty.
        write_claude_resume_command_file(
            resume_command_path=Path(ctx["CLAUDE_RESUME_COMMAND_PATH"]),
            project_root=project_root,
            claude_session_id=claude_session_id,
            task_key=ctx["TASK_KEY"],
            task_type=ctx["TASK_TYPE"],
            phase_state=ctx["CURRENT_RUN_STATUS"],
            worker_prompts_dir_relative=ctx["RUN_PROMPTS_RELATIVE_PATH"],
            prompt_seq=ctx["RUN_PROMPTS_SEQ"],
        )
    ctx["FIX_CYCLE_ID"] = _record_fix_cycle_events(inp, ctx)


def _prepare_agent_contract(
    inp: PrepareInputs,
    ctx: dict,
    workspace_root: Path,
    models: _ModelBindings,
) -> None:
    source = _duty_catalog_source(workspace_root)
    destination = Path(ctx["DUTY_CONTRACT_ROOT"])
    _snapshot_duty_catalog(source, destination)
    digest = digest_duty_catalog(destination)
    allowed_audiences = _allowed_agent_audiences(inp, models)
    contract = {
        "schemaVersion": 1,
        "dutyRootPath": ctx["DUTY_CONTRACT_ROOT_RELATIVE_PATH"],
        "catalogDigest": digest,
        "invocationReservationRootPath": (
            ctx["INVOCATION_RESERVATION_ROOT_RELATIVE_PATH"]
        ),
        "allowedAudiences": allowed_audiences,
        "authorizedPaths": {
            "instructionRoots": [
                ctx["RUN_PROMPTS_RELATIVE_PATH"],
                ctx["RUN_STATE_RELATIVE_PATH"],
            ],
            "promptRoots": [ctx["RUN_PROMPTS_RELATIVE_PATH"]],
            "resultRoots": [
                ctx["WORKER_RESULTS_RELATIVE_PATH"],
                ctx["RUN_REPORTS_RELATIVE_PATH"],
            ],
        },
    }
    ctx["DUTY_CATALOG_DIGEST"] = digest
    ctx["AGENT_CONTRACT_JSON"] = json.dumps(contract, ensure_ascii=False)
    ctx["INVOCATION_ASSIGNMENTS_JSON"] = json.dumps(
        models.invocation_assignments,
        ensure_ascii=False,
    )


def _duty_catalog_source(workspace_root: Path) -> Path:
    candidates = (
        workspace_root / "prompts" / "duties",
        okstra_home() / "prompts" / "duties",
    )
    for candidate in candidates:
        if (candidate / "common.md").is_file():
            return candidate
    raise PrepareError("agent duty catalog is missing from the runtime")


def _snapshot_duty_catalog(source: Path, destination: Path) -> None:
    if destination.exists():
        if digest_duty_catalog(destination) != digest_duty_catalog(source):
            raise PrepareError("existing run duty snapshot conflicts with runtime catalog")
        return
    destination.parent.mkdir(parents=True, exist_ok=True)
    temp_dir = Path(tempfile.mkdtemp(
        dir=destination.parent,
        prefix=f".{destination.name}.",
        suffix=".tmp",
    ))
    try:
        shutil.copytree(source, temp_dir, dirs_exist_ok=True)
        os.rename(temp_dir, destination)
    except FileExistsError as exc:
        if digest_duty_catalog(destination) != digest_duty_catalog(source):
            raise PrepareError("run duty snapshot publication conflict") from exc
    finally:
        if temp_dir.exists():
            shutil.rmtree(temp_dir)


def _allowed_agent_audiences(
    inp: PrepareInputs,
    models: _ModelBindings,
) -> list[str]:
    audiences = {"lead", "translator"}
    if models.worker_assignments:
        audiences.add("reverification-worker")
    if any(item.worker_id == "report-writer" for item in models.worker_assignments):
        audiences.add("report-writer")
    if inp.task_type == "implementation":
        audiences.update({"implementation-executor", "implementation-verifier"})
    elif inp.task_type == "final-verification":
        audiences.add("acceptance-verifier")
    else:
        # Same map the prompt policy resolves the duty from. Allowing a
        # different audience here than the one the policy will ask for makes
        # every worker prompt in the phase fail materialization.
        audiences.add(
            ANALYSIS_DUTY_BY_TASK_TYPE.get(inp.task_type, "analysis-worker")
        )
    if models.critic_choice not in {"", "off"}:
        audiences.update({"scope-critic", "acceptance-critic"})
    return sorted(audiences)


def _record_run_in_central_index(
    inp: PrepareInputs,
    ctx: dict,
    workspace_root: Path,
    run_seq_override: int | None,
) -> None:
    """Publish this run to ~/.okstra/{recent,active}.jsonl. Never fatal.

    A failure here leaves the central index incomplete but the bundle on disk is
    already usable, so prepare reports and continues. The rerun path pre-stamps a
    'reserving' row before spawning (OKSTRA_RUN_SEQ_OVERRIDE forces that seq); if
    the promotion to 'running' fails, that row would sit in active.jsonl forever
    and inflate activeCount, so it is cleaned up whenever the reserved seq is known.
    """
    try:
        _record_start(
            workspace_root=workspace_root,
            ctx=ctx,
            initial_status="prepared" if inp.render_only else "running",
            canonical_argv=_canonical_argv(inp, ctx),
            cwd=os.getcwd(),
            brief_sha256=_brief_sha256(inp.brief_path),
        )
    except Exception as exc:
        print(
            f"okstra-central: record_start failed; central index will be incomplete ({exc})",
            file=sys.stderr,
        )
        if run_seq_override is not None:
            _remove_leaked_reservation(ctx, run_seq_override)


def _initial_workflow_ctx(
    inp: PrepareInputs, forbidden_by_phase: dict
) -> dict[str, str]:
    """Workflow tokens for a run that has been prepared but not started.

    The run status is not known yet, so both statuses are the pre-launch
    constants; `_finalize_status_and_render_manifests` recomputes the block once
    the real status is settled.
    """
    task_status, run_status = "ready-for-lead", "not-run"
    return {
        "CURRENT_TASK_STATUS": task_status,
        "CURRENT_RUN_STATUS": run_status,
        **compute_workflow_state(
            task_type=inp.task_type,
            current_run_status=run_status,
            current_task_status=task_status,
            render_only=inp.render_only,
            forbidden_by_phase=forbidden_by_phase,
        ),
    }


def _related_tasks_ctx(ctx: dict, inp: PrepareInputs) -> dict[str, str]:
    """Render tokens for the tasks this run declares a relation to."""
    items = resolve_related_tasks(
        task_manifest_path=Path(ctx["TASK_MANIFEST_PATH"]),
        raw_related=inp.related_tasks_raw,
    )
    return {
        "RELATED_TASKS_JSON": json.dumps(items, ensure_ascii=False),
        "RELATED_TASKS_BULLETS": related_tasks_bullets(items),
        "RELATED_TASKS_INLINE": related_tasks_inline(items),
    }


def _reverify_scope_ctx(raw: str) -> dict[str, str]:
    """Render tokens for the re-verification scope the user pinned.

    Always emitted: the lead prompt reads both tokens unconditionally, and an
    absent one is a render failure rather than a silent `auto`.
    """
    scope = parse_user_reverify_scope(raw)
    return {
        "REVERIFY_SCOPE_MODE": scope.mode,
        "REVERIFY_SCOPE_STAGES": (
            ",".join(str(num) for num in scope.stages) or "(none)"
        ),
    }


def _model_ctx(models: "_ModelBindings") -> dict[str, str]:
    """Render tokens for every model binding this run resolved."""
    return {
        "LEAD_MODEL": models.lead.display,
        "LEAD_MODEL_EXECUTION_VALUE": models.lead_assignment.model_execution_value,
        "LEAD_PROVIDER": models.lead_assignment.provider,
        "LEAD_ASSIGNMENT_JSON": json.dumps(
            models.lead_assignment.to_payload(), ensure_ascii=False,
        ),
        "WORKER_ASSIGNMENTS_JSON": json.dumps(
            [assignment.to_payload() for assignment in models.worker_assignments],
            ensure_ascii=False,
        ),
        "INVOCATION_ASSIGNMENTS_JSON": json.dumps(
            models.invocation_assignments,
            ensure_ascii=False,
        ),
        "CLAUDE_WORKER_MODEL": models.cw.display,
        "CLAUDE_WORKER_MODEL_EXECUTION_VALUE": models.cw.execution,
        "CODEX_WORKER_MODEL": models.co.display,
        "CODEX_WORKER_MODEL_EXECUTION_VALUE": models.codex_worker_execution,
        "ANTIGRAVITY_WORKER_MODEL": models.ge.display,
        "ANTIGRAVITY_WORKER_MODEL_EXECUTION_VALUE": models.antigravity_worker_execution,
        "REPORT_WRITER_MODEL": models.rw.display,
        "REPORT_WRITER_MODEL_EXECUTION_VALUE": models.rw.execution,
        "REPORT_WRITER_PROVIDER": next(
            (
                assignment.provider for assignment in models.worker_assignments
                if assignment.worker_id == "report-writer"
            ),
            "",
        ),
        **_executor_model_ctx(models),
        "CRITIC_CHOICE": models.critic_choice,
        "CRITIC_MODEL_EXECUTION_VALUE": models.critic_model_execution,
    }


def _executor_model_ctx(models: "_ModelBindings") -> dict[str, str]:
    assignment = models.executor_assignment
    if assignment is None:
        return {
            "EXECUTOR_WORKER_ID": "",
            "EXECUTOR_PROVIDER": "",
            "EXECUTOR_DISPLAY_NAME": "",
            "EXECUTOR_MODEL_DISPLAY": "",
            "EXECUTOR_MODEL_EXECUTION_VALUE": "",
            "EXECUTOR_HOST_MODEL_VALUE": "",
            "EXECUTOR_RUNNER": "",
            "EXECUTOR_DISPATCH_MODE": "",
        }
    return {
        "EXECUTOR_WORKER_ID": assignment.worker_id,
        "EXECUTOR_PROVIDER": assignment.provider,
        "EXECUTOR_DISPLAY_NAME": models.executor_display_name,
        "EXECUTOR_MODEL_DISPLAY": assignment.model_display,
        "EXECUTOR_MODEL_EXECUTION_VALUE": assignment.model_execution_value,
        "EXECUTOR_HOST_MODEL_VALUE": assignment.host_model_value or "",
        "EXECUTOR_RUNNER": assignment.runner,
        "EXECUTOR_DISPATCH_MODE": (
            "host-native"
            if assignment.runner == "native-session"
            else "worker-dispatch"
        ),
    }


def _provision_task_and_stage(
    inp: PrepareInputs,
    project_root: Path,
    ctx_stage_map: list,
    task_group_segment: str,
    task_id_segment: str,
    task_key: str,
) -> _ProvisionedTask:
    """Reserve this run's worktree and, for implementation, its Stage Map stage.

    One worktree per task-key: requirements-discovery, error-analysis,
    implementation-planning and implementation phases of the same task all share
    this directory and branch. Runs BEFORE run-path compute: the worktree's
    degrade status (skipped-*) feeds the implementation Stage Run Claim, and the
    resolved stage namespaces the run path.

    The whole check→select→`git worktree add`→reserve sequence (task worktree AND
    implementation stage) is one critical section per task-key: the registry lock
    alone only covers the reserve row, so without this mutex two concurrent runs
    can pick the same stage or race the same path/branch at the git level (TOCTOU).
    """
    with worktree_provision_mutex(
        okstra_home(), inp.project_id, task_group_segment, task_id_segment,
    ):
        if inp.task_type == "final-verification" and inp.stage and inp.stage != "auto":
            worktree = _single_stage_final_verification_worktree(inp)
            # Single-stage final-verification namespaces its run path under
            # runs/final-verification/stage-<N> (same isolation as
            # implementation) so concurrent per-stage verifications never
            # share state/reports/worker-results.
            fv_stage_arg = int(inp.stage)
        else:
            fv_stage_arg = None
            try:
                worktree = provision_task_worktree(
                    task_type=inp.task_type,
                    project_root=project_root,
                    project_id=inp.project_id,
                    task_group_segment=task_group_segment,
                    task_id_segment=task_id_segment,
                    work_category=inp.work_category,
                    base_ref=inp.base_ref,
                    require_base_ref=True,
                )
            except RuntimeError as exc:
                raise PrepareError(
                    f"task worktree provisioning failed: {exc}"
                ) from exc

        # ---- implementation Stage Run Claim (path-independent) ----
        # Resolve + provision the stage BEFORE run-path compute so RUN_DIR
        # lands in runs/implementation/stage-<N>. The registry stage-key is
        # reserved exactly once here (inside provision_stage_worktree), and
        # the surrounding mutex makes the registry read in the claim
        # and that reserve atomic. Other task-types skip this claim;
        # single-stage final-verification threads its explicit stage via
        # fv_stage_arg, everything else keeps stage_arg=None (flat paths).
        if inp.task_type == "implementation":
            stage_run_claim = _claim_implementation_stage_run(
                inp, ctx_stage_map, task_group_segment, task_id_segment,
                task_key, worktree.status,
            )
            # Drop any stale waiver on this stage so the run actually verifies
            # conformance (kept inside the per-task-key mutex so concurrent
            # same-task runs don't race the manifest write).
            _clear_stale_stage_waiver(inp, project_root, stage_run_claim.stage)
            return _ProvisionedTask(worktree, stage_run_claim, stage_run_claim.stage)
        return _ProvisionedTask(worktree, None, fv_stage_arg)


def prepare_task_bundle(inp: PrepareInputs) -> PrepareOutputs:
    """Produce a complete okstra task bundle on disk. See module docstring."""
    workspace_root = Path(inp.workspace_root)
    project_root = Path(inp.project_root)
    lead_runtime = _normalize_lead_runtime(inp.lead_runtime)
    lead_runtime_request = (inp.lead_runtime_request or lead_runtime).strip()
    runtime_resolution_json = (inp.runtime_resolution_json or "{}").strip()
    try:
        json.loads(runtime_resolution_json or "{}")
    except json.JSONDecodeError as exc:
        raise PrepareError(f"invalid runtime resolution data: {exc}") from exc
    # Probed once here and reused below, so the gate and the manifest cannot
    # disagree about which backend this run is on.
    terminal_backend = detect_terminal_backend()
    # Outside cmux only claude-code has a dispatch backend of its own. Under
    # cmux okstra owns the panes for every lead, so the render-only restriction
    # no longer applies to any runtime.
    if (
        lead_runtime != "claude-code"
        and terminal_backend != BACKEND_CMUX_PANE
        and not inp.render_only
    ):
        raise PrepareError(
            f"lead runtime `{lead_runtime}` is currently render-only; "
            "use --render-only until a dispatch backend is enabled for this lead runtime."
        )

    # ---- input validation + asset resolution + roster/model 해소 ----
    # 각 단계는 명시적 입력/반환을 갖는 헬퍼로 분리되어 있다 (상단 phase 헬퍼
    # 블록 참조). 아래 orchestrator 는 그 결과를 지역 변수로 언팩해 이후 ctx
    # 조립 코드가 단일 흐름으로 읽히도록 유지한다.
    assets = _resolve_runtime_assets(workspace_root, inp)
    profile_file = assets.profile_file
    requires_executor = _profile_requires_executor(profile_file)
    selection, role_profile = _normalize_prepare_model_selection(inp, profile_file)
    prompt_template = assets.prompt_template
    task_index_template = assets.task_index_template
    final_report_template = assets.final_report_template
    ctx_stage_map = _validate_prepare_inputs(project_root, inp)
    if inp.task_type != "release-handoff":
        _validate_task_brief_preflight(
            project_root,
            inp.brief_path,
            assets.brief_validator,
        )
    selected_direction = _resolve_planning_direction(inp)
    if inp.task_type == "implementation":
        ctx_stage_map = _prepare_implementation_approved_plan(inp)

    # release-handoff: 검증 보고서 인용 input 문서를 생성해 brief 자리에 채운다.
    # 이후의 모든 brief 소비 경로(material/instruction-set 복사)는 그대로 동작한다.
    handoff_tokens = {"HANDOFF_MODE": "", "HANDOFF_STAGES": ""}
    if inp.task_type == "release-handoff":
        handoff_tokens = _materialize_release_handoff_input(
            workspace_root, project_root, inp)

    workers, selected_reviewers = _resolve_roster(inp, profile_file)
    pr_template_path_str, pr_template_source = _resolve_pr_template(inp)

    assignment_context, assignment_plan = _resolve_prepare_assignments(
        inp=inp,
        profile=role_profile,
        selection=selection,
        workers=workers,
        lead_runtime=lead_runtime,
        terminal_backend=terminal_backend,
        requires_executor=requires_executor,
    )
    if _uses_canonical_assignment_projection(inp):
        workers, models = _project_assignment_plan(
            inp,
            assignment_plan,
            assignment_context,
        )
        selected_reviewers = ",".join(workers)
    else:
        models = _resolve_model_bindings(
            inp,
            workers,
            assignment_context,
            requires_executor,
        )
    execution_manifest = _build_static_execution_manifest(
        assignment_plan,
        assignment_context,
        models.lead,
        models.translator,
    )
    dynamic_roles = tuple(
        requirement.role
        for requirement in role_profile.roles
        if requirement.dynamic
    )

    verify_installation(workspace_root)
    _register_and_check_project(project_root, inp)

    # ---- paths under per-task mutex (writes run-context-*.json) ----
    # OKSTRA_RUN_SEQ_OVERRIDE: okstra-ctl rerun / 테스트 hook 이 미리 reserve
    # 한 seq 를 강제하는 user-knob 환경 변수.
    raw_override = os.environ.get("OKSTRA_RUN_SEQ_OVERRIDE", "").strip()
    run_seq_override = int(raw_override) if raw_override else None

    # Identity segments derived with the SAME slugify rule compute_run_paths
    # uses, so they match ctx["TASK_GROUP_SEGMENT"]/["TASK_ID_SEGMENT"]
    # byte-for-byte. We need them BEFORE run-path compute because
    # implementation Stage Run Claim depends on the task-worktree degrade
    # status, and the run path itself is stage-namespaced.
    task_group_segment = slugify(inp.task_group)
    task_id_segment = slugify(inp.task_id)
    task_key = f"{inp.project_id}:{inp.task_group}:{inp.task_id}"

    # Worktree branch namespace, workflow state and every rendered manifest read
    # the same resolved value, so a phase invoked without --work-category still
    # lands on the branch namespace the task was classified with.
    inp.work_category = resolve_work_category(
        inp.work_category,
        project_root=project_root,
        task_group=inp.task_group,
        task_id=inp.task_id,
    )

    provisioned = _provision_task_and_stage(
        inp, project_root, ctx_stage_map,
        task_group_segment, task_id_segment, task_key,
    )
    worktree = provisioned.worktree
    stage_run_claim = provisioned.stage_run_claim
    stage_arg = provisioned.stage_arg

    analysis_source_commit = ""
    resolved_evidence = ()
    resolved_target: dict[str, object] = {}
    if inp.task_type in ANALYSIS_TASK_TYPES:
        analysis_root = Path(worktree.path or project_root)
        evidence_paths = parse_evidence_paths(inp.evidence_inputs_raw, project_root)
        try:
            analysis_source_commit = resolve_analysis_head(analysis_root)
            resolved_evidence = resolve_evidence_inputs(
                project_root, inp.task_type, evidence_paths, analysis_source_commit,
            )
            if inp.task_type == "feature-analysis":
                resolved_target = resolve_analysis_target(
                    inp.analysis_target,
                    resolved_evidence,
                    load_candidate_map(project_root, evidence_paths),
                )
        except AnalysisInputError as exc:
            raise PrepareError(str(exc)) from exc

    ctx = compute_and_write_run_context(
        workspace_root=workspace_root, project_root=project_root,
        project_id=inp.project_id, task_group=inp.task_group, task_id=inp.task_id,
        task_type=inp.task_type, run_seq_override=run_seq_override,
        stage=stage_arg,
    )

    ctx.update({
        "TERMINAL_BACKEND": terminal_backend,
        "EXECUTOR_WORKTREE_PATH": worktree.path,
        "EXECUTOR_WORKTREE_BRANCH": worktree.branch,
        "EXECUTOR_WORKTREE_BASE_REF": worktree.base_ref,
        "EXECUTOR_WORKTREE_STATUS": worktree.status,
        "EXECUTOR_WORKTREE_NOTE": worktree.note,
        # Phase 6 plan-body verification toggle, read by
        # `render._build_convergence_block` when emitting the manifest's
        # `convergence.planBodyVerification.enabled` field. Default ("")
        # is treated as enabled.
        "OKSTRA_PLAN_VERIFICATION": (
            "false" if not inp.plan_verification_enabled else ""
        ),
        "ANALYSIS_SOURCE_COMMIT": analysis_source_commit,
        "ANALYSIS_TARGET_JSON": json.dumps(resolved_target, ensure_ascii=False),
        "EVIDENCE_INPUTS_JSON": json.dumps(
            [item.to_dict() for item in resolved_evidence], ensure_ascii=False,
        ),
    })

    # implementation: override the task-worktree fields with the claimed
    # STAGE worktree. Must run
    # AFTER the task-worktree fields above (mirrors the original ordering
    # where stage reservation overrode them post task-provision).
    if inp.task_type == "implementation":
        _publish_stage_run_claim(
            inp, ctx, ctx_stage_map, stage_run_claim,
        )
        carry = derive_stage_fix_carry(
            Path(ctx["RUN_DIR"]), ctx.get("EXECUTOR_WORKTREE_PATH", ""),
        )
        ctx["FIX_RUN_CONTEXT"] = carry.as_prompt_markdown() if carry else ""

    if inp.render_only:
        # render-only entry path (e.g. okstra-run skill, in-session takeover):
        # the calling Claude session itself becomes the lead, so we must NOT
        # mint a fresh UUID — instead, best-effort detect the live session's
        # jsonl under ~/.claude/projects/<encoded-cwd>/. Leaving this blank
        # caused Phase 7 token-usage collection to fail (lead jsonl not
        # found) and the validator to report `sessionId=`.
        claude_session_id = resolve_inproc_lead_session_id(project_root)
    else:
        claude_session_id = generate_claude_session_id()

    # ---- material + related-tasks ----
    profile_content = _expand_profile_includes(profile_file)
    review_material = build_analysis_material(inp.brief_path, inp.directive)

    # ---- relative paths for brief + clarification ----
    brief_relative = relative_to_project_root(inp.brief_path, project_root)
    clarification_relative = (
        relative_to_project_root(Path(inp.clarification_response_path), project_root)
        if inp.clarification_response_path else ""
    )
    selected_direction_relative = (
        relative_to_project_root(Path(inp.selected_direction_path), project_root)
        if inp.selected_direction_path else ""
    )

    forbidden_by_phase = load_phase_forbidden(workspace_root)

    # ---- assemble full ctx (the values render functions expect) ----
    ctx.update({
        "ANALYSIS_PROFILE": inp.task_type,
        "RECOMMENDED_ANALYSERS": selected_reviewers,
        "HOST_RUNTIME": lead_runtime,
        "LEAD_RUNTIME": lead_runtime,
        "LEAD_RUNTIME_REQUEST": lead_runtime_request,
        "RUNTIME_RESOLUTION_JSON": runtime_resolution_json or "{}",
        "PR_TEMPLATE_PATH": pr_template_path_str,
        "PR_TEMPLATE_SOURCE": pr_template_source,
        **handoff_tokens,
        "CLAUDE_SESSION_ID": claude_session_id,
        "CLARIFICATION_RESPONSE_PATH": inp.clarification_response_path,
        "CLARIFICATION_RESPONSE_RELATIVE_PATH": clarification_relative,
        "SELECTED_DIRECTION_PATH": inp.selected_direction_path,
        "SELECTED_DIRECTION_RELATIVE_PATH": selected_direction_relative,
        **_reverify_scope_ctx(inp.reverify_scope),
        "BRIEF_FILE_PATH": str(inp.brief_path),
        "BRIEF_RELATIVE_PATH": brief_relative,
        **_model_ctx(models),
        **_related_tasks_ctx(ctx, inp),
        **_initial_workflow_ctx(inp, forbidden_by_phase),
        "VALIDATION_STATUS": "not-run",
        "VALIDATION_UPDATED_AT": "",
        "VALIDATION_FAILURES_JSON": "[]",
        "LATEST_REPORT_PATH": "",
        "LATEST_REPORT_RECORD_RELATIVE_PATH": "",
        "RENDER_ONLY": "true" if inp.render_only else "false",
        "OKSTRA_VERSION": installed_version(),
        "EXECUTION_IDENTITY_JSON": json.dumps(
            execution_manifest.to_payload(), ensure_ascii=False,
        ),
        "DYNAMIC_EXECUTION_ROLES_JSON": json.dumps(
            dynamic_roles, ensure_ascii=False,
        ),
    })
    refresh_run_context_snapshot(ctx)
    if lead_runtime == "codex":
        ctx["CLAUDE_RESUME_COMMAND_PATH"] = ""
        ctx["CLAUDE_RESUME_COMMAND_RELATIVE_PATH"] = ""
        ctx["CLAUDE_RESUME_COMMAND_FILENAME"] = ""
    if inp.task_type == "final-verification":
        acquisition = _stage_targets.acquire_final_verification_target(
            _final_verification_target_request(inp, ctx_stage_map)
        )
        _apply_final_verification_target(ctx, acquisition)

    _write_bundle_artifacts(
        inp, ctx, project_root, lead_runtime, claude_session_id,
    )
    _prepare_agent_contract(inp, ctx, workspace_root, models)

    # ---- write instruction-set scaffolding ----
    instruction_set = _write_instruction_set_sources(
        inp,
        ctx,
        profile_content,
        review_material,
        assets.host_rules_file,
        selected_direction,
    )
    # ---- run-inputs persistence ----
    _persist_run_inputs(inp, ctx, models, selected_reviewers, brief_relative)
    _persist_pre_dispatch_run_manifest(ctx)

    # ---- lead prompt publication (manifest is now the external authority) ----
    prompt_text = _render_lead_prompt_and_snapshot(
        inp, ctx, instruction_set, final_report_template, prompt_template
    )

    # ---- final status + manifest/discovery renders ----
    _finalize_status_and_render_manifests(
        inp, ctx, task_index_template, forbidden_by_phase
    )
    persist_static_execution_identity(
        run_manifest_path=Path(ctx["RUN_MANIFEST_PATH"]),
        task_manifest_path=Path(ctx["TASK_MANIFEST_PATH"]),
        input_snapshot_path=Path(ctx["RUN_INPUTS_PATH"]),
        manifest=execution_manifest,
        task_key=ctx["TASK_KEY"],
        run_ref=ctx["RUN_MANIFEST_RELATIVE_PATH"],
        dynamic_roles=dynamic_roles,
    )
    _record_artifact_runtime_render_only_event(inp, ctx)

    _record_run_in_central_index(inp, ctx, workspace_root, run_seq_override)

    if not inp.render_only:
        _provision_settings_symlink(inp)

    concurrent_run: dict = {}
    if stage_run_claim is not None and stage_run_claim.concurrent_stages:
        concurrent_run = {
            "detected": True,
            "active_stages": stage_run_claim.concurrent_stages,
        }
    return PrepareOutputs(
        ctx=ctx,
        prompt_text=prompt_text,
        extras={
            "profile_content": profile_content,
            "concurrent_run": concurrent_run,
        },
    )


class _StoreRuntimeSingleton(Action):
    def __call__(
        self,
        parser: ArgumentParser,
        namespace: Namespace,
        values: object,
        option_string: str | None = None,
    ) -> None:
        seen = getattr(namespace, "_runtime_singletons_seen", frozenset())
        if self.dest in seen:
            raise ArgumentError(
                self, f"{option_string} may be supplied only once"
            )
        setattr(namespace, self.dest, True if self.nargs == 0 else values)
        setattr(namespace, "_runtime_singletons_seen", seen | {self.dest})


def _add_runtime_resolution_arguments(parser: ArgumentParser) -> None:
    runtime_ids = (*default_host_registry().ids(), "all")
    parser.add_argument(
        "--lead-runtime",
        action=_StoreRuntimeSingleton,
        default="claude-code",
        choices=runtime_ids,
        dest="lead_runtime",
        help=(
            "Lead runtime adapter. Default: claude-code. `codex` renders "
            "Codex CLI artifact bundles; `antigravity` renders Antigravity CLI "
            "artifact bundles; `external` renders neutral bootstrap bundles for "
            "manifest inspection until a dispatch backend is enabled."
        ),
    )
    parser.add_argument(
        "--lead-runtime-request", action=_StoreRuntimeSingleton, default=""
    )
    parser.add_argument(
        "--runtime-resolution-launch-mode",
        action=_StoreRuntimeSingleton,
        choices=("lead", "team", "catalog"),
        default="",
    )
    parser.add_argument(
        "--runtime-resolution-install-target", action="append", default=[]
    )
    for flag in (
        "--runtime-resolution-host",
        "--runtime-resolution-reason",
        "--runtime-resolution-fallback-from",
    ):
        parser.add_argument(flag, action=_StoreRuntimeSingleton, default="")
    parser.add_argument(
        "--runtime-resolution-confidence",
        action=_StoreRuntimeSingleton,
        choices=("none", "low", "medium", "high"),
        default="",
    )
    parser.add_argument("--runtime-resolution-warning", action="append", default=[])
    parser.add_argument(
        "--runtime-resolution-catalog-hosts-present",
        action=_StoreRuntimeSingleton,
        nargs=0,
        default=False,
    )
    parser.add_argument(
        "--runtime-resolution-catalog-host-id", action="append", default=[]
    )


def build_prepare_argument_parser():
    """Build the shared flat-argument parser used by public prepare shims."""
    p = ArgumentParser()
    p.add_argument("--workspace-root", required=True)
    p.add_argument("--project-root", required=True)
    p.add_argument("--project-id", required=True)
    p.add_argument("--task-group", required=True)
    p.add_argument("--task-id", required=True)
    p.add_argument("--task-type", required=True)
    p.add_argument(
        "--task-brief", default="", dest="task_brief",
        help=(
            "Required for every task-type except release-handoff (whose input "
            "document is generated by prepare from the cited "
            "final-verification reports)."
        ),
    )
    p.add_argument("--directive", default="")
    p.add_argument("--analysis-target", default="", dest="analysis_target")
    p.add_argument("--evidence-inputs", default="", dest="evidence_inputs_raw")
    p.add_argument(
        "--fix-cycle", default="", choices=["", "yes", "no"], dest="fix_cycle",
        help=(
            "done(release-handoff)까지 완료된 task 에 entry phase 로 재진입할 때 "
            "이번 작업을 버그 픽스 사이클로 기록할지 결정. 'yes' = 새 cycle "
            "open + run 부착, 'no'/'' = 기록 안 함."
        ),
    )
    p.add_argument("--workers", default="", dest="workers_override")
    p.add_argument("--role-count", action="append", default=[], dest="role_counts_raw")
    p.add_argument("--role-model", action="append", default=[], dest="role_models_raw")
    p.add_argument(
        "--host-session-context-json",
        default="",
        dest="host_session_context_json",
    )
    p.add_argument("--lead-provider", default="")
    p.add_argument("--lead-model", default="")
    p.add_argument("--claude-model", default="")
    p.add_argument("--codex-model", default="")
    p.add_argument("--antigravity-model", default="")
    p.add_argument("--worker-model", default="", dest="worker_models_raw")
    p.add_argument("--report-writer-provider", default="")
    p.add_argument("--report-writer-model", default="")
    _add_runtime_resolution_arguments(p)
    p.add_argument("--executor", default="")
    p.add_argument("--critic", default="")
    p.add_argument("--related-tasks", default="", dest="related_tasks_raw")
    p.add_argument("--approved-plan", default="", dest="approved_plan_path")
    p.add_argument(
        "--qa-waiver",
        default="",
        dest="qa_waiver",
        help=(
            'User-recorded stage conformance waiver: "<stageKey>:<reason>". '
            "Normally used only for blocking local io entries; external QA is advisory."
        ),
    )
    p.add_argument(
        "--stage", default="auto", dest="stage",
        help=(
            "implementation task only. Which Stage Map entry to execute. "
            "'auto' (default) = lowest-numbered stage whose depends-on are all "
            "consumers.jsonl status:done. Numeric '<N>' = force that stage."
        ),
    )
    p.add_argument(
        "--stages", default="", dest="stages",
        help=(
            "release-handoff only. Comma-separated stage numbers to bundle "
            "into one PR (stage-group mode); empty = whole-task mode. "
            "Distinct from --stage (implementation/final-verification "
            "Stage Map selection)."
        ),
    )
    p.add_argument(
        "--approve",
        action="store_true",
        dest="approve_plan_ack",
        help=(
            "Treat the CLI invocation itself as the plan approval signal. "
            "Sets the report record `frontmatter.approved` to true and refreshes "
            "the full reading copy."
        ),
    )
    p.add_argument(
        "--implementation-option",
        default="",
        dest="implementation_option",
        help=(
            "implementation task only. Name of the Option Candidate the user "
            "chose from the implementation-planning final-report. Written into "
            "the report record `frontmatter.implementationOption` (schema-v1: "
            "the markdown `implementation-option:` line). When omitted, implementation falls back to the plan's "
            "`Recommended Option`."
        ),
    )
    p.add_argument("--clarification-response", default="", dest="clarification_response_path")
    p.add_argument("--selected-direction", default="", dest="selected_direction_path")
    p.add_argument(
        "--reverify-scope",
        default="",
        dest="reverify_scope",
        help=(
            "implementation-planning 재실행 전용. 사용자가 고른 재검증 범위. "
            "'' / 'auto' = 리드의 incremental-scope 판정에 맡김(기본), "
            "'full' = 전체 재검증 강제, '<stage csv>' (예: '2,3') = 그 stage 를 "
            "impacted 로 지정."
        ),
    )
    p.add_argument(
        "--pr-template-path",
        default="",
        dest="pr_template_path",
        help=(
            "release-handoff 전용 1회성 PR 본문 템플릿 경로. 빈 값이면 "
            "project.json → ~/.okstra/config.json → 스킬 디폴트 순으로 해석."
        ),
    )
    p.add_argument("--render-only", action="store_true", dest="render_only")
    p.add_argument(
        "--no-plan-verification",
        action="store_false",
        dest="plan_verification_enabled",
        default=True,
        help=(
            "Disable the Phase 6 plan-body verification round for "
            "`--task-type implementation-planning`. Default: enabled. "
            "When disabled, the top-of-report `User Approval Request` "
            "marker line is rendered unconditionally (legacy behaviour)."
        ),
    )
    p.add_argument(
        "--work-category",
        default="",
        dest="work_category",
        help=(
            "Work-category classification for this task "
            "(bugfix / feature / refactor / ops / improvement). "
            "When omitted, falls back to the category already recorded in "
            "task-manifest.json, then to `feature`."
        ),
    )
    p.add_argument(
        "--base-ref",
        default="",
        dest="base_ref",
        help=(
            "Git ref (branch name, tag, or commit SHA) used as the base of "
            "the task worktree on the first phase of this task-key. "
            "Required for first-phase prepare; ignored on subsequent phases "
            "(the registered worktree is reused). Mirrors the PR-base picker "
            "used by the `release-handoff` phase: typical values are "
            "`main` / `dev` / `staging` / `preprod` / `prod` or any local ref."
        ),
    )
    return p


def _runtime_resolution_json_from_args(args: Namespace) -> str:
    values = {
        "launch mode": args.runtime_resolution_launch_mode,
        "host": args.runtime_resolution_host,
        "confidence": args.runtime_resolution_confidence,
        "reason": args.runtime_resolution_reason,
    }
    optional_values = (
        args.runtime_resolution_install_target,
        args.runtime_resolution_fallback_from,
        args.runtime_resolution_warning,
        args.runtime_resolution_catalog_hosts_present,
        args.runtime_resolution_catalog_host_id,
    )
    if not any(values.values()) and not any(optional_values):
        return "{}"
    missing = [name for name, value in values.items() if not value]
    if missing:
        raise PrepareError(
            "incomplete runtime resolution arguments: " + ", ".join(missing)
        )
    if (
        args.runtime_resolution_catalog_host_id
        and not args.runtime_resolution_catalog_hosts_present
    ):
        raise PrepareError(
            "runtime resolution catalog host IDs require the presence marker"
        )
    payload: dict[str, object] = {
        "ok": True,
        "requestedRuntime": args.lead_runtime_request or args.lead_runtime,
        "resolvedRuntime": args.lead_runtime,
        "launchMode": args.runtime_resolution_launch_mode,
        "installTargets": args.runtime_resolution_install_target,
        "host": args.runtime_resolution_host,
        "confidence": args.runtime_resolution_confidence,
        "reason": args.runtime_resolution_reason,
        "fallbackFrom": args.runtime_resolution_fallback_from or None,
        "warnings": args.runtime_resolution_warning,
    }
    if args.runtime_resolution_catalog_hosts_present:
        payload["catalogHostIds"] = args.runtime_resolution_catalog_host_id
    return json.dumps(payload, ensure_ascii=False)


def main(argv: list[str]) -> int:
    """Parse one public prepare invocation and emit its rendered bundle."""
    args = build_prepare_argument_parser().parse_args(argv)

    project_root = Path(args.project_root).expanduser().resolve()
    if args.task_type == "release-handoff":
        if args.task_brief:
            print(
                "--task-brief is not accepted for --task-type release-handoff "
                "— the input document is generated from the cited "
                "final-verification reports",
                file=__import__("sys").stderr,
            )
            return 1
        brief_abs = Path("")  # prepare 가 input 문서를 생성해 채운다
    else:
        brief_abs = resolve_user_file(args.task_brief, project_root)
        if brief_abs is None:
            print(f"task brief not found: {args.task_brief}",
                  file=__import__("sys").stderr)
            return 1
    clarification_abs = ""
    if args.clarification_response_path:
        cr = _resolve_planning_input_path(
            args.clarification_response_path, project_root
        )
        if cr is None:
            print(
                f"clarification response file not found: {args.clarification_response_path}",
                file=__import__("sys").stderr,
            )
            return 1
        clarification_abs = str(cr)
    selected_direction_abs = ""
    if args.selected_direction_path:
        selected = _resolve_planning_input_path(
            args.selected_direction_path, project_root
        )
        if selected is None:
            print(
                f"selected direction report not found: {args.selected_direction_path}",
                file=__import__("sys").stderr,
            )
            return 1
        selected_direction_abs = str(selected)

    try:
        host_session_context = deserialize_host_session_context(
            args.host_session_context_json
        )
        runtime_resolution_json = _runtime_resolution_json_from_args(args)
    except (ModelSelectionInputError, PrepareError) as exc:
        print(str(exc), file=__import__("sys").stderr)
        return 1
    if not args.host_session_context_json:
        adapter = default_host_registry().resolve(args.lead_runtime)
        host_session_context = HostSessionContext(
            host_id=adapter.descriptor.id,
            entry_mode="spawn-process",
            available_functions=frozenset(),
            interaction_surface="unavailable",
            current_model=CurrentSessionModelAttestation.unknown(
                adapter.descriptor.native_provider_id
            ),
        )

    inputs = PrepareInputs(
        workspace_root=Path(args.workspace_root).resolve(),
        project_root=project_root,
        project_id=args.project_id,
        task_group=args.task_group,
        task_id=args.task_id,
        task_type=args.task_type,
        brief_path=brief_abs,
        analysis_target=args.analysis_target,
        evidence_inputs_raw=args.evidence_inputs_raw,
        directive=args.directive,
        workers_override=args.workers_override,
        role_counts_raw=tuple(args.role_counts_raw),
        role_models_raw=tuple(args.role_models_raw),
        host_session_context=host_session_context,
        lead_provider=args.lead_provider,
        lead_model=args.lead_model,
        claude_model=args.claude_model,
        codex_model=args.codex_model,
        antigravity_model=args.antigravity_model,
        worker_models_raw=args.worker_models_raw,
        report_writer_provider=args.report_writer_provider,
        report_writer_model=args.report_writer_model,
        lead_runtime=args.lead_runtime,
        lead_runtime_request=args.lead_runtime_request,
        runtime_resolution_json=runtime_resolution_json,
        executor=args.executor,
        critic=args.critic,
        related_tasks_raw=args.related_tasks_raw,
        work_category=args.work_category,
        base_ref=args.base_ref,
        approved_plan_path=args.approved_plan_path,
        qa_waiver=args.qa_waiver,
        stage=args.stage,
        stages=args.stages,
        clarification_response_path=clarification_abs,
        selected_direction_path=selected_direction_abs,
        reverify_scope=args.reverify_scope,
        pr_template_path=args.pr_template_path,
        render_only=args.render_only,
        approve_plan_ack=args.approve_plan_ack,
        implementation_option=args.implementation_option,
        plan_verification_enabled=args.plan_verification_enabled,
        fix_cycle=args.fix_cycle,
    )
    try:
        out = prepare_task_bundle(inputs)
    except PrepareError as exc:
        print(str(exc), file=__import__("sys").stderr)
        return 1

    ctx = out.ctx
    # summary block — bash wrapper consumes (and may pipe to user).
    print(f"okstra task key: {ctx['TASK_KEY']}")
    print(f"okstra task root: {ctx['TASK_ROOT']}")
    print(f"okstra latest task discovery file: {ctx['OKSTRA_LATEST_TASK_FILE']}")
    print(f"okstra task catalog file: {ctx['OKSTRA_TASK_CATALOG_FILE']}")
    print(f"okstra instruction-set: {ctx['INSTRUCTION_SET_PATH']}")
    print(f"okstra run manifest: {ctx['RUN_MANIFEST_PATH']}")
    print(f"okstra reference expectations: {ctx['REFERENCE_EXPECTATIONS_FILE']}")
    print(f"okstra final report template: {ctx['FINAL_REPORT_TEMPLATE_PATH']}")
    cr = out.extras.get("concurrent_run", {})
    if cr.get("detected"):
        stages_csv = ",".join(str(s) for s in cr["active_stages"])
        print(f"okstra concurrent-run stages: {stages_csv}")
    if inputs.render_only:
        print()
        print(out.prompt_text, end="")
    else:
        print(f"okstra current run dir: {ctx['RUN_DIR']}")
        print(f"final report path: {ctx['FINAL_REPORT_PATH']}")
        lead_provider = ctx.get("LEAD_PROVIDER", "")
        launch = lead_launch_spec(lead_provider)
        print(f"lead model: {ctx['LEAD_MODEL']}")
        print(f"claude session id: {ctx['CLAUDE_SESSION_ID']}")
        print(f"resume command file: {ctx['CLAUDE_RESUME_COMMAND_PATH']}")
        print(f"launch mode: interactive {launch.executable} handoff")
        print(f"lead working directory: {ctx['PROJECT_ROOT']}")
        print()
        # In non-render-only mode emit the JSON a front end needs to exec the
        # lead. The argv is assembled here rather than in the caller so provider
        # launch knowledge stays in the catalog and every front end — the bash
        # wrapper and the Node CLI — starts the lead identically.
        machine = _lead_launch_payload(ctx)
        print(f"__OKSTRA_LAUNCH__ {json.dumps(machine)}")
    return 0


def _lead_launch_payload(ctx: dict[str, object]) -> dict[str, object]:
    lead_runtime = str(ctx.get("LEAD_RUNTIME", "claude-code"))
    project_root = Path(str(ctx["PROJECT_ROOT"]))
    run_manifest_path = Path(str(ctx["RUN_MANIFEST_PATH"]))
    try:
        manifest = load_owned_object(run_manifest_path, artifact="run manifest")
        resources = manifest["resources"]
        assignment = agent_model_assignment_from_payload(
            manifest["invocationAssignments"]["lead"]
        )
        prompt_file = project_root / resources["leadExecutionPromptPath"]
        metadata_path = project_root / resources["leadPromptMetadataPath"]
        metadata = load_owned_object(metadata_path, artifact="lead prompt metadata")
        if metadata["prompt"]["path"] != resources["leadExecutionPromptPath"]:
            raise KeyError("lead prompt resource does not match metadata")
    except (OSError, JsonBoundaryError, KeyError, TypeError, AgentInvocationError) as exc:
        raise PrepareError("lead invocation authority is incomplete") from exc
    errors = verify_agent_invocation(
        metadata_path,
        project_root=project_root,
        expected_run_manifest_path=run_manifest_path,
        expected_assignment=assignment,
        expected_worker_id="lead",
        expected_assignment_ref="lead",
        expected_audience="lead",
    )
    if errors:
        raise PrepareError("lead invocation verification failed: " + "; ".join(errors))
    lead_provider = assignment.provider
    launch = lead_launch_spec(lead_provider)
    launch_argv = lead_launch_argv(
        lead_provider,
        model=assignment.model_execution_value,
        session_id=str(ctx["CLAUDE_SESSION_ID"]),
        prompt=prompt_file.read_text(encoding="utf-8"),
    )
    launch_prefix_size = 1 + len(launch.sandbox_waiver)
    return {
        "leadRuntime": lead_runtime,
        "leadProvider": lead_provider,
        "leadExecutable": launch.executable,
        "leadSessionId": ctx["CLAUDE_SESSION_ID"],
        "leadModelExecutionValue": assignment.model_execution_value,
        "projectRoot": str(project_root),
        "promptFile": str(prompt_file),
        "runManifestPath": str(run_manifest_path),
        "leadPromptMetadataPath": str(metadata_path),
        "sandboxWaiverNote": launch.sandbox_waiver_note,
        "launchArgv": launch_argv,
        "launchRequestArgv": launch_argv[launch_prefix_size:],
    }


if __name__ == "__main__":
    import sys
    raise SystemExit(main(sys.argv[1:]))