"""All run-artifact rendering, in one module.

이 모듈은 기존에 `scripts/lib/okstra/{template,team-state,manifest,discovery,
directories}.sh` 의 python heredoc 으로 흩어져 있던 렌더 로직을 한 곳으로 모은다.
bash 측은 이제 heredoc 없이 `python3 -m okstra_ctl.render <subcommand> <args>`
형태의 한 줄 호출로 같은 산출물을 만든다.

설계 원칙:
  - 각 함수는 (ctx dict, 추가 path 인자) 만 받아서 파일을 쓴다.
  - 환경 변수에 의존하지 않는다 (ctx 가 권위).
  - 단일 진입점 `main()` 이 subcommand 를 dispatcher 로 라우팅한다.

ctx dict 의 schema 는 `okstra_ctl.paths.compute_run_paths()` 의 반환값을
기본으로, 호출자가 추가 키 (workflow state / model display / related tasks /
session id 등) 를 덧붙여 전달한다.
"""

from __future__ import annotations

import hashlib
import json
import re
import sys
from collections.abc import Mapping
from pathlib import Path

from okstra_project.dirs import TASK_MANIFEST_FILENAME, OKSTRA_DIR_NAME, project_json_path

# phase 시퀀스 / 기본 next-phase 매핑의 SSOT 는 workflow 모듈이다. 과거
# render_task_manifest 가 동일한 리스트/딕셔너리를 로컬에 중복 정의했는데,
# 이는 silent drift 위험이 있어 SSOT import 로 통합한다.
from . import fix_cycles
from . import next_phase
from .analysis_inputs import ANALYSIS_TASK_TYPES
from .application.resolve_assignment import (
    resolve_assignment_runner,
    resolve_lead_provider,
)
from .application.dispatch_assignments import dispatch_assignments
from .adapters.dispatch import default_worker_dispatch_port, with_worker_dispatch
from .adapters.dispatch.cmux import dispatch_port_for_terminal_backend
from .paths import okstra_home
from .models import UnknownProviderError, provider_ids, provider_spec
from .ports.worker_dispatch import WorkerDispatchRequest
from .registry.host_registry import default_host_registry
from .registry.provider_registry import default_provider_registry
from .path_hints import compact_active_run_context, hydrate_run_context
from .paths import task_timeline_file
from .json_boundary import JsonBoundaryError, load_owned_object, write_owned_object_atomic
from .report_contract import CURRENT_REPORT_SCHEMA_VERSION
from .worker_artifacts import artifacts_from_context
from .workflow import PHASE_SEQUENCE


class TokenRenderError(Exception):
    """Raised when a template references a `{{TOKEN}}` not present in ctx.

    Specific to the pure-token renderer in this module. Distinct from
    `okstra_ctl.render_final_report.FinalReportRenderError` which wraps
    jinja2 / IO failures during final-report rendering.
    """


# --------------------------------------------------------------------------- #
# helpers
# --------------------------------------------------------------------------- #


def _load_ctx(ctx_arg: str) -> dict:
    """Accept either an inline JSON string or a path to a JSON file.

    bash 는 보통 `build_render_context_json` 의 결과를 그대로 inline 으로 넘기고,
    skill 은 디스크의 run-context 파일 경로를 넘긴다.
    """
    s = ctx_arg.lstrip()
    if s.startswith("{"):
        return hydrate_run_context(json.loads(ctx_arg))
    return hydrate_run_context(
        load_owned_object(Path(ctx_arg), artifact="render context")
    )


def _write_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")


def _write_json(path: Path, payload: dict) -> None:
    write_owned_object_atomic(path, payload, artifact="rendered runtime object")


def _execution_identity(ctx: Mapping[str, object]) -> dict:
    raw = ctx.get("EXECUTION_IDENTITY_JSON")
    if not isinstance(raw, str) or not raw.strip():
        return {}
    payload = json.loads(raw)
    if not isinstance(payload, dict):
        raise TokenRenderError("EXECUTION_IDENTITY_JSON must contain an object")
    return payload


def _lead_runtime(ctx: dict) -> str:
    return ctx.get("LEAD_RUNTIME", "") or "claude-code"


def _lead_descriptor(ctx: dict):
    return default_host_registry().resolve(_lead_runtime(ctx)).descriptor


def _worker_dispatch_plan(ctx: dict):
    host_adapter = default_host_registry().resolve(_lead_runtime(ctx))
    fallback_port = default_worker_dispatch_port(host_adapter.descriptor)
    port = dispatch_port_for_terminal_backend(
        ctx.get("TERMINAL_BACKEND", ""),
        fallback_port,
    )
    decorated = with_worker_dispatch(host_adapter, port)
    request = WorkerDispatchRequest(
        project_root=Path(ctx.get("PROJECT_ROOT", "")),
        run_manifest=Path(ctx.get("RUN_MANIFEST_PATH", "")),
    )
    return dispatch_assignments(request, decorated.worker_dispatch())


def _lead_agent(ctx: dict) -> str:
    return _lead_descriptor(ctx).agent_id


def _lead_agent_label(ctx: dict) -> str:
    return _lead_descriptor(ctx).agent_label


def _lead_role(ctx: dict) -> str:
    return _lead_descriptor(ctx).role


def _lead_adapter(ctx: dict, plan) -> dict:
    descriptor = default_host_registry().resolve(_lead_runtime(ctx)).descriptor
    return {
        "name": plan.adapter_name or descriptor.id,
        "dispatchMode": plan.dispatch_mode or descriptor.dispatch_mode,
        "sessionAccounting": descriptor.session_accounting,
    }


def _lead_adapter_contract(ctx: dict, plan) -> str:
    descriptor = default_host_registry().resolve(_lead_runtime(ctx)).descriptor
    return plan.relay_contract or descriptor.relay_contract


def _runtime_resolution(ctx: dict) -> dict:
    raw = ctx.get("RUNTIME_RESOLUTION_JSON", "") or "{}"
    try:
        payload = json.loads(raw)
    except Exception:
        return {}
    return payload if isinstance(payload, dict) else {}


# The phases a `{% if header.taskType ... %}` block may target: the lifecycle
# sequence itself, kept as an allowlist rather than `[a-z-]+` so a typo'd phase
# name fails to match and the block survives the copy — caught by eye instead of
# silently vanishing from every phase's excerpt. Longest-first so
# `implementation-planning` is never matched as `implementation`.
_PHASE_ALT = "|".join(
    re.escape(phase) for phase in sorted(PHASE_SEQUENCE, key=len, reverse=True)
)
# Two forms: a single-phase block (`== 'X'`) and a multi-phase one
# (`in ('X', 'Y')`). The multi form exists because a section can belong to a
# set of phases — `### 2.3 End State Coverage` is written by the three
# pre-implementation phases and by no other.
_PHASE_BLOCK_RE = re.compile(
    r"\{% if header\.taskType (?:"
    rf"== '(?P<one>{_PHASE_ALT})'"
    rf"|in \((?P<many>'(?:{_PHASE_ALT})'(?:\s*,\s*'(?:{_PHASE_ALT})')*),?\s*\)"
    r") %\}\n(?P<body>.*?)\{% endif %\}\n",
    re.DOTALL,
)
_PHASE_NAME_RE = re.compile(r"'([a-z-]+)'")


def _strip_phase_blocks(text: str, current_phase: str) -> str:
    """Resolve phase-conditional blocks (`{% if header.taskType == 'X' %}
    ... {% endif %}`) against *current_phase*.

    Blocks whose target set contains *current_phase* keep their body (jinja
    markers dropped); blocks targeting only other phases are removed
    entirely. A phase that no block names keeps none of them — and when
    *current_phase* is empty that is every block. The `## 5.5` / `5.6` /
    `5.7` / `5.8` deliverable sections stay single-phase, so they still
    apply to exactly one task-type each.

    Observed (fontsninja-classifier-v2 RD run): the raw final-report
    template copied into instruction-set/final-report-template.md was
    43 KB / 631 lines; ~30 KB / ~330 lines belonged to the four other
    phases' deliverables and was never relevant to that run. Stripping
    at copy time cuts the lead/report-writer's baseline by ~7 K tokens
    per phase entry.

    Inline conditionals (those that begin and end on the same line) are
    intentionally untouched — the regex only matches block-form
    `{% if ... %}\\n ... \\n{% endif %}\\n`.
    """

    def repl(m: "re.Match[str]") -> str:
        one = m.group("one")
        targets = {one} if one else set(_PHASE_NAME_RE.findall(m.group("many")))
        return m.group("body") if current_phase in targets else ""

    return _PHASE_BLOCK_RE.sub(repl, text)


_FM_DEFAULT = "no-classification"

_FM_TAGS_BASE = []

_FM_TAGS_CATALOG: dict[str, list[str]] = {
    "task-brief": ["task-brief"],
    "task-index": ["task-index"],
    "error-analysis-input": ["error-analysis", "input"],
    "implementation-input": ["implementation", "input"],
    "implementation-planning-input": ["implementation-planning", "input"],
    "final-verification-input": ["final-verification", "input"],
    "release-handoff-input": ["release-handoff", "input"],
    "quick-input": ["quick", "input"],
    "final-report": ["final-report"],
    "schedule": ["schedule"],
}


def _fm_scalar(value: str, default: str = _FM_DEFAULT) -> str:
    v = (value or "").strip()
    return v if v else default


def _fm_array(values: list[str], extras: list[str] | None = None) -> str:
    items = [str(v).strip() for v in values if v and str(v).strip()]
    if extras:
        items.extend(str(e).strip() for e in extras if str(e).strip())
    if not items:
        return "[]"
    return "[" + ", ".join(f'"{v}"' for v in items) + "]"


def _fm_tags(doc_type: str) -> str:
    extra = _FM_TAGS_CATALOG.get((doc_type or "").strip(), [])
    return _fm_array(_FM_TAGS_BASE + list(extra))


def _doc_type_from_template_path(template_path: str) -> str:
    name = Path(template_path).name
    if name.endswith(".template.md"):
        stem = name[: -len(".template.md")]
    else:
        stem = Path(name).stem
    return stem


def _frontmatter_id_from_task_key(task_key: str) -> str:
    """task_key (`project_id:task_group:task_id`) 를 ID 형식으로 변환.

    `:` 를 `-` 로 치환한 단일 문자열. 예시:
        ``fontsninja-classifier-v2:DEV-9388:DEV-9429``
        -> ``fontsninja-classifier-v2-DEV-9388-DEV-9429``
    """
    return (task_key or "").strip().replace(":", "-")


def _frontmatter_mapping(ctx: dict) -> dict:
    task_id = (ctx.get("TASK_ID") or "").strip()
    project_id = (ctx.get("PROJECT_ID") or "").strip()
    task_group = (ctx.get("TASK_GROUP") or "").strip()
    task_key = (ctx.get("TASK_KEY") or "").strip()
    task_date = (ctx.get("TASK_DATE") or "").strip()
    doc_type = (ctx.get("DOC_TYPE") or "").strip()
    task_type = (ctx.get("TASK_TYPE") or "").strip()

    fm_id = _frontmatter_id_from_task_key(task_key)
    fm_id_scalar = f'"{fm_id}"' if fm_id else f'"{_FM_DEFAULT}"'
    alias_value = f"{fm_id}-{task_type}" if (fm_id and task_type) else fm_id
    return {
        "{{TASK_KEY}}": _fm_scalar(task_key),
        "{{TASK_ID}}": _fm_scalar(task_id),
        "{{PROJECT_ID}}": _fm_scalar(project_id),
        "{{TASK_GROUP}}": _fm_scalar(task_group),
        "{{TASK_DATE}}": _fm_scalar(task_date),
        # task_key 의 `:` 를 `-` 로 치환한 단일 스칼라.
        # 예: "fontsninja-classifier-v2-DEV-9388-DEV-9429"
        "{{FM_ID}}": fm_id_scalar,
        # id 와 task-type 을 `-` 로 연결한 단일 alias 를 array 한 칸에 담는다
        # (Obsidian aliases 컨벤션).
        "{{FM_ALIASES}}": _fm_array([alias_value]) if alias_value else "[]",
        "{{FM_TAGS}}": _fm_tags(doc_type),
        # 신규: 모든 okstra 산출물의 frontmatter 가 task type 을 명시한다.
        "{{FM_TASK_TYPE}}": _fm_scalar(task_type),
    }


def _resolve_workers(ctx: dict) -> list[str]:
    return [
        w.strip() for w in ctx.get("RECOMMENDED_ANALYSERS", "").split(",") if w.strip()
    ]


def _assignment_runner_for_ctx(
    ctx: dict,
    provider: str,
    role: str,
    requested_runner: str = "",
) -> str:
    host = default_host_registry().resolve(_lead_runtime(ctx)).descriptor
    spec = default_provider_registry().resolve(provider)
    return resolve_assignment_runner(
        host=host,
        provider=spec,
        role=role,
        requested_runner=requested_runner,
    ).runner


def _resolved_lead_assignment(
    ctx: dict,
    requested_provider: str,
    requested_runner: str = "",
):
    host_registry = default_host_registry()
    provider_registry = default_provider_registry()
    assignment = resolve_lead_provider(
        host_id=_lead_runtime(ctx),
        requested_provider=requested_provider,
        host_registry=host_registry,
        provider_registry=provider_registry,
    )
    if requested_runner:
        resolution = resolve_assignment_runner(
            host=host_registry.resolve(_lead_runtime(ctx)).descriptor,
            provider=provider_registry.resolve(assignment.provider),
            role="lead",
            requested_runner=requested_runner,
        )
        return assignment.provider, resolution.runner
    return assignment.provider, assignment.runner


def _lead_assignment(ctx: dict) -> dict:
    parsed = None
    raw = ctx.get("LEAD_ASSIGNMENT_JSON", "")
    if raw:
        try:
            candidate = json.loads(raw)
            if isinstance(candidate, dict):
                parsed = candidate
        except json.JSONDecodeError:
            pass
    source = parsed or {}
    provider, runner = _resolved_lead_assignment(
        ctx,
        source.get("provider", "") or ctx.get("LEAD_PROVIDER", ""),
        source.get("runner", ""),
    )
    return {
        **source,
        "role": "lead",
        "provider": provider,
        "model": source.get("model", ctx.get("LEAD_MODEL", "")),
        "modelExecutionValue": source.get(
            "modelExecutionValue", ctx.get("LEAD_MODEL_EXECUTION_VALUE", "")
        ),
        "runner": runner,
    }


def _invocation_assignments(ctx: dict) -> dict:
    raw = ctx.get("INVOCATION_ASSIGNMENTS_JSON", "")
    try:
        parsed = json.loads(raw) if raw else {}
    except json.JSONDecodeError as exc:
        raise ValueError("invalid invocation assignments JSON") from exc
    if not isinstance(parsed, dict):
        raise ValueError("invocation assignments must be an object")
    return parsed


def _agent_contract(ctx: dict) -> dict:
    raw = ctx.get("AGENT_CONTRACT_JSON", "")
    try:
        parsed = json.loads(raw) if raw else {}
    except json.JSONDecodeError as exc:
        raise ValueError("invalid agent contract JSON") from exc
    if not isinstance(parsed, dict):
        raise ValueError("agent contract must be an object")
    return parsed


def _executor_contract(ctx: dict) -> dict | None:
    runner = ctx.get("EXECUTOR_RUNNER", "")
    dispatch_mode = ctx.get("EXECUTOR_DISPATCH_MODE", "")
    host_model_value = ctx.get("EXECUTOR_HOST_MODEL_VALUE") or None
    if not runner and not ctx.get("EXECUTOR_PROVIDER"):
        if ctx.get("TASK_TYPE") == "implementation":
            raise ValueError("implementation executor assignment is required")
        return None
    expected_mode = {
        "native-session": "host-native",
        "cli-wrapper": "worker-dispatch",
    }.get(runner)
    if expected_mode is None or dispatch_mode != expected_mode:
        raise ValueError(
            "executor runner and dispatch mode must be native-session/host-native "
            "or cli-wrapper/worker-dispatch"
        )
    if runner == "native-session" and not host_model_value:
        raise ValueError("native-session executor requires host model value")
    if runner == "cli-wrapper" and host_model_value is not None:
        raise ValueError("cli-wrapper executor must not define host model value")
    return {
        "workerId": ctx.get("EXECUTOR_WORKER_ID", ""),
        "provider": ctx.get("EXECUTOR_PROVIDER", ""),
        "displayName": ctx.get("EXECUTOR_DISPLAY_NAME", ""),
        "model": ctx.get("EXECUTOR_MODEL_DISPLAY", ""),
        "modelExecutionValue": ctx.get("EXECUTOR_MODEL_EXECUTION_VALUE", ""),
        "hostModelValue": host_model_value,
        "runner": runner,
        "dispatchMode": dispatch_mode,
        "appliesTo": "implementation",
    }


def _worker_assignments(ctx: dict) -> list[dict]:
    raw = ctx.get("WORKER_ASSIGNMENTS_JSON", "")
    if raw:
        try:
            parsed = json.loads(raw)
            if isinstance(parsed, list) and all(isinstance(row, dict) for row in parsed):
                return [_resolved_worker_assignment(ctx, row) for row in parsed]
        except json.JSONDecodeError:
            pass
    catalog = _worker_catalog(ctx, include_assignments=False)
    return [
        {
            "workerId": worker_id,
            "role": "report-writer" if worker_id == "report-writer" else "analyser",
            "provider": catalog[worker_id]["agent"],
            "model": catalog[worker_id]["model"],
            "modelExecutionValue": catalog[worker_id]["modelExecutionValue"],
            "runner": _assignment_runner_for_ctx(
                ctx,
                catalog[worker_id]["agent"],
                "report-writer" if worker_id == "report-writer" else "analyser",
            ),
        }
        for worker_id in _resolve_workers(ctx)
    ]


def _resolved_worker_assignment(ctx: dict, source: dict) -> dict:
    worker_id = source.get("workerId", "")
    role = source.get("role", "") or (
        "report-writer" if worker_id == "report-writer" else "analyser"
    )
    provider = source.get("provider", "") or _fallback_worker_provider(
        ctx, worker_id
    )
    runner = _assignment_runner_for_ctx(
        ctx,
        provider,
        role,
        source.get("runner", ""),
    )
    return {
        **source,
        "workerId": worker_id,
        "role": role,
        "provider": provider,
        "runner": runner,
    }


def _worker_catalog(ctx: dict, *, include_assignments: bool = True) -> dict:
    assignments = {} if not include_assignments else {
        row.get("workerId"): row for row in _worker_assignments(ctx)
        if row.get("workerId")
    }
    catalog = {}
    for worker_id, paths in artifacts_from_context(ctx, relative=True).items():
        assignment = assignments.get(worker_id, {})
        provider = assignment.get("provider") or _fallback_worker_provider(ctx, worker_id)
        provider_info = provider_spec(provider)
        model, execution = _legacy_worker_model(ctx, worker_id)
        catalog[worker_id] = {
            "workerId": worker_id,
            "role": (
                "Report writer worker" if worker_id == "report-writer"
                else f"{provider_spec(worker_id).display_label} worker"
            ),
            "agent": provider,
            "agentLabel": provider_info.display_label,
            "provider": provider,
            "runner": assignment.get(
                "runner", _assignment_runner_for_ctx(
                    ctx,
                    provider,
                    "report-writer" if worker_id == "report-writer" else "analyser",
                ),
            ),
            "model": assignment.get("model", model),
            "modelExecutionValue": assignment.get("modelExecutionValue", execution),
            **paths,
        }
    return catalog


def _fallback_worker_provider(ctx: dict, worker_id: str) -> str:
    if worker_id == "report-writer":
        return ctx.get("REPORT_WRITER_PROVIDER", "") or "claude"
    return worker_id


def _legacy_worker_model(ctx: dict, worker_id: str) -> tuple[str, str]:
    token = "REPORT_WRITER" if worker_id == "report-writer" else (
        worker_id.upper().replace("-", "_")
    )
    return (
        ctx.get(f"{token}_WORKER_MODEL", ctx.get(f"{token}_MODEL", "")),
        ctx.get(
            f"{token}_WORKER_MODEL_EXECUTION_VALUE",
            ctx.get(f"{token}_MODEL_EXECUTION_VALUE", ""),
        ),
    )


def _active_workers(ctx: dict) -> list[dict]:
    catalog = _worker_catalog(ctx)
    workers = []
    for worker_id in _resolve_workers(ctx):
        item = catalog[worker_id]
        workers.append({
            "workerId": item["workerId"],
            "role": item["role"],
            "agent": item["agent"],
            "agentLabel": item["agentLabel"],
            "provider": item["provider"],
            "runner": item["runner"],
            "model": item["model"],
            "modelExecutionValue": item["modelExecutionValue"],
            "promptPath": item["promptPath"],
            "resultPath": item["resultPath"],
            "attemptRequired": True,
        })
    return workers


def _active_task(ctx: dict) -> dict:
    return {
        "projectId": ctx.get("PROJECT_ID", ""),
        "taskGroup": ctx.get("TASK_GROUP", ""),
        "taskId": ctx.get("TASK_ID", ""),
        "taskKey": ctx.get("TASK_KEY", ""),
        "taskType": ctx.get("TASK_TYPE", ""),
        "workCategory": ctx.get("WORKFLOW_WORK_CATEGORY", "unknown"),
        "projectRoot": ctx.get("PROJECT_ROOT", ""),
        "taskRootPath": ctx.get("TASK_ROOT_RELATIVE_PATH", ""),
    }


def _active_workflow(ctx: dict) -> dict:
    return {
        "currentPhase": ctx.get("WORKFLOW_CURRENT_PHASE", ""),
        "currentPhaseState": ctx.get("WORKFLOW_CURRENT_PHASE_STATE", ""),
        "lastCompletedPhase": ctx.get("WORKFLOW_LAST_COMPLETED_PHASE", ""),
        "nextRecommendedPhase": next_phase.promote(
            ctx.get("WORKFLOW_NEXT_RECOMMENDED_PHASE")
        ),
        "awaitingApproval": ctx.get("WORKFLOW_AWAITING_APPROVAL", "false") == "true",
        "allowedOutputs": ctx.get("PHASE_ALLOWED_OUTPUTS", ""),
        "forbiddenActions": ctx.get("PHASE_FORBIDDEN_ACTIONS", ""),
    }


def _active_run(ctx: dict) -> dict:
    return {
        # `implementation` binds one run to one Stage Map stage, and the
        # executor prompt has to name it: its sidecar forbids recomputing the
        # stage from `consumers.jsonl`. Feeds the implementation prompt anchor
        # in `initial_prompt_materialization`.
        "stage": ctx.get("RUN_STAGE", ""),
        # A fix run's carried findings decide what the executor fixes and what
        # the verifier MUST cite as resolved / still-failing. The rendered
        # analysis profile holds the same block, but no CLI worker can read that
        # file, so this copy is what `initial_prompt_materialization` inlines
        # into their prompts. Derived once at prepare: re-deriving it later would
        # read a worktree HEAD the executor may have already moved.
        "fixRunCarry": ctx.get("FIX_RUN_CONTEXT", ""),
        "runDirectoryPath": ctx.get("RUN_DIR_RELATIVE_PATH", ""),
        "runManifestPath": ctx.get("RUN_MANIFEST_RELATIVE_PATH", ""),
        "teamStatePath": ctx.get("TEAM_STATE_RELATIVE_PATH", ""),
        "promptSnapshotPath": ctx.get("RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""),
        "finalReportRecordPath": ctx.get("FINAL_REPORT_RECORD_RELATIVE_PATH", ""),
        "convergenceStatePath": ctx.get("CONVERGENCE_STATE_RELATIVE_PATH", ""),
        "finalStatusPath": ctx.get("FINAL_STATUS_RELATIVE_PATH", ""),
        "validatorScriptPath": ctx.get("RUN_VALIDATOR_RELATIVE_PATH", ""),
        "resumeCommandPath": ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", ""),
        "workerPromptsDirectoryPath": ctx.get("RUN_PROMPTS_RELATIVE_PATH", ""),
        "workerResultsDirectoryPath": ctx.get("WORKER_RESULTS_RELATIVE_PATH", ""),
    }


def _active_instruction_set(ctx: dict) -> dict:
    instruction_set = ctx.get("INSTRUCTION_SET_RELATIVE_PATH", "")
    clarification = (
        instruction_set + "/clarification-response.md"
        if ctx.get("CLARIFICATION_RESPONSE_RELATIVE_PATH", "") else ""
    )
    return {
        "path": instruction_set,
        "analysisPacketPath": ctx.get("ANALYSIS_PACKET_RELATIVE_PATH", ""),
        "taskBriefPath": instruction_set + "/task-brief.md",
        "analysisProfilePath": instruction_set + "/analysis-profile.md",
        "analysisMaterialPath": instruction_set + "/analysis-material.md",
        "referenceExpectationsPath": ctx.get("REFERENCE_EXPECTATIONS_RELATIVE_PATH", ""),
        "clarificationResponsePath": clarification,
        "finalReportTemplatePath": ctx.get("FINAL_REPORT_TEMPLATE_RELATIVE_PATH", ""),
        "finalReportSchemaPath": ctx.get("FINAL_REPORT_SCHEMA_RELATIVE_PATH", ""),
        "verificationTargetPath": ctx.get("VERIFICATION_TARGET_RELATIVE_PATH", ""),
        "verificationTargetDigest": ctx.get("VERIFICATION_TARGET_DIGEST", ""),
    }


def _active_error_logs(ctx: dict) -> dict:
    artifacts = artifacts_from_context(ctx, relative=True)
    return {
        "runErrorsLogPath": ctx.get("RUN_ERRORS_LOG_RELATIVE_PATH", ""),
        "sidecarsByWorkerId": {
            worker_id: paths["errorsSidecarPath"]
            for worker_id, paths in artifacts.items()
        },
    }


def _active_runtime_resources(ctx: dict) -> dict:
    runtime_home = okstra_home()
    return {
        "codingPreflightDir": ctx.get("OKSTRA_CODING_PREFLIGHT_DIR")
        or str(runtime_home / "prompts" / "coding-preflight"),
        "workerPreamblePathByAudience": {
            "analysis": ctx.get("ANALYSIS_WORKER_PREAMBLE_PATH")
            or str(runtime_home / "templates" / "worker-prompt-preamble.md"),
            "implementation-executor": ctx.get(
                "IMPLEMENTATION_WORKER_PREAMBLE_PATH"
            )
            or str(runtime_home / "templates" / "implementation-worker-preamble.md"),
            "implementation-verifier": ctx.get(
                "IMPLEMENTATION_WORKER_PREAMBLE_PATH"
            )
            or str(runtime_home / "templates" / "implementation-worker-preamble.md"),
            "report-writer": ctx.get("REPORT_WRITER_PREAMBLE_PATH")
            or str(runtime_home / "templates" / "report-writer-prompt-preamble.md"),
        },
        "workerErrorContractPath": ctx.get("WORKER_ERROR_CONTRACT_PATH")
        or str(runtime_home / "templates" / "worker-error-contract.md"),
    }


def _active_executor_worktree(ctx: dict) -> dict:
    return {
        "status": ctx.get("EXECUTOR_WORKTREE_STATUS", ""),
        "path": ctx.get("EXECUTOR_WORKTREE_PATH", ""),
        "branch": ctx.get("EXECUTOR_WORKTREE_BRANCH", ""),
        "baseRef": ctx.get("EXECUTOR_WORKTREE_BASE_REF", ""),
        "note": ctx.get("EXECUTOR_WORKTREE_NOTE", ""),
    }


def _active_verification_target(ctx: dict) -> dict:
    if ctx.get("TASK_TYPE") != "final-verification":
        return {}
    return {
        "scope": ctx.get("VERIFICATION_SCOPE", ""),
        "worktreePath": ctx.get("VERIFICATION_WORKTREE_PATH", ""),
        "baseRef": ctx.get("VERIFICATION_BASE_REF", ""),
        "headRef": ctx.get("VERIFICATION_HEAD_REF", ""),
        "path": ctx.get("VERIFICATION_TARGET_RELATIVE_PATH", ""),
        "digest": ctx.get("VERIFICATION_TARGET_DIGEST", ""),
    }


def _active_source_artifacts(ctx: dict) -> dict:
    return {
        "taskManifestPath": ctx.get("TASK_MANIFEST_RELATIVE_PATH", ""),
        "runContextPath": ctx.get("RUN_CONTEXT_RELATIVE_PATH", ""),
        "runInputsPath": ctx.get("RUN_INPUTS_RELATIVE_PATH", ""),
        "historyTimelinePath": ctx.get("TIMELINE_RELATIVE_PATH", ""),
    }


def _active_lazy_read_plan() -> dict:
    return {
        "leadPhase1Primary": True,
        "readTaskIndexOnlyForHumanSummary": True,
        "readHistoryTimelineOnlyForHistoryOrCarryInDisambiguation": True,
        "readFinalReportTemplateOnlyForReportWriter": True,
    }


# --------------------------------------------------------------------------- #
# team-state
# --------------------------------------------------------------------------- #


def render_active_run_context(active_context_path: str, ctx: dict) -> None:
    """Write the compact lead intake surface for the current run.

    This is an interface file, not a replacement for task/run manifests. The
    source manifests remain the audit/replay authority; this file concentrates
    the fields the lead needs at Phase 1 so it does not have to recompose the
    current run from several shallow artifacts.
    """
    payload = {
        "schemaVersion": "1.0",
        "kind": "active-run-context",
        "task": _active_task(ctx),
        "workflow": _active_workflow(ctx),
        "run": _active_run(ctx),
        "agentContract": _agent_contract(ctx),
        "instructionSet": _active_instruction_set(ctx),
        "workers": _active_workers(ctx),
        "errorLogs": _active_error_logs(ctx),
        "runtimeResources": _active_runtime_resources(ctx),
        "executorWorktree": _active_executor_worktree(ctx),
        "verificationTarget": _active_verification_target(ctx),
        "sourceArtifacts": _active_source_artifacts(ctx),
        "lazyReadPlan": _active_lazy_read_plan(),
    }
    _write_json(Path(active_context_path), compact_active_run_context(ctx, payload))


def render_team_state(team_state_path: str, ctx: dict) -> None:
    selected = _resolve_workers(ctx)
    catalog = _worker_catalog(ctx)
    worker_dispatch_plan = _worker_dispatch_plan(ctx)
    workers = []
    for row in _optional_worker_roles(ctx):
        workers.append({
            **{key: row[key] for key in (
                "workerId", "role", "agent", "provider", "runner",
                "model", "modelExecutionValue", "resultPath", "promptPath",
            )},
            "status": "not-run",
            "reason": "",
        })
    for w in selected:
        m = catalog[w]
        workers.append(
            {
                "workerId": m["workerId"],
                "role": m["role"],
                "agent": m["agent"],
                "provider": m["provider"],
                "runner": m["runner"],
                "model": m["model"],
                "modelExecutionValue": m["modelExecutionValue"],
                "status": "not-run",
                "resultPath": m["resultPath"],
                "promptPath": m["promptPath"],
                "reason": "",
            }
        )
    payload = {
        "schemaVersion": "1.0",
        "taskKey": ctx.get("TASK_KEY", ""),
        "taskType": ctx.get("TASK_TYPE", ""),
        "hostRuntime": ctx.get("HOST_RUNTIME", "") or _lead_runtime(ctx),
        "leadRuntime": _lead_runtime(ctx),
        "leadAssignment": _lead_assignment(ctx),
        "workerAssignments": _worker_assignments(ctx),
        "agentContract": _agent_contract(ctx),
        "invocationAssignments": _invocation_assignments(ctx),
        "leadAdapter": _lead_adapter(ctx, worker_dispatch_plan),
        "leadEventsPath": ctx.get("LEAD_EVENTS_RELATIVE_PATH", ""),
        "runDirectoryPath": ctx.get("RUN_DIR_RELATIVE_PATH", ""),
        "workflowState": ctx.get("CURRENT_RUN_STATUS", ""),
        "lead": {
            "role": _lead_role(ctx),
            "agent": _lead_agent(ctx),
            "provider": _lead_assignment(ctx).get("provider", ""),
            "runner": _lead_assignment(ctx).get("runner", ""),
            "model": ctx.get("LEAD_MODEL", ""),
            "modelExecutionValue": ctx.get("LEAD_MODEL_EXECUTION_VALUE", ""),
            "status": ctx.get("CURRENT_RUN_STATUS", ""),
            "sessionId": ctx.get("CLAUDE_SESSION_ID", ""),
        },
        "workers": workers,
        "validator": {
            "scriptPath": ctx.get("RUN_VALIDATOR_RELATIVE_PATH", ""),
            "status": ctx.get("VALIDATION_STATUS", "not-run"),
            "lastValidatedAt": ctx.get("VALIDATION_UPDATED_AT", ""),
            "failures": json.loads(ctx.get("VALIDATION_FAILURES_JSON", "[]")),
        },
        "artifacts": {
            "leadPromptSnapshotPath": ctx.get("RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""),
            "workerPromptsDirectoryPath": ctx.get("RUN_PROMPTS_RELATIVE_PATH", ""),
            "finalReportRecordPath": ctx.get("FINAL_REPORT_RECORD_RELATIVE_PATH", ""),
            "finalStatusPath": ctx.get("FINAL_STATUS_RELATIVE_PATH", ""),
            "workerResultsDirectoryPath": ctx.get("WORKER_RESULTS_RELATIVE_PATH", ""),
        },
    }
    _write_json(Path(team_state_path), payload)


# --------------------------------------------------------------------------- #
# reference expectations + discovery
# --------------------------------------------------------------------------- #


def render_reference_expectations(brief_path: str, output_path: str, ctx: dict) -> None:
    section_map = {
        "Configuration References and Expected Values": "config",
        "Deployment Manifests and Expected Values": "deployment",
    }
    captured = {"config": [], "deployment": []}
    current_section = None
    for line in Path(brief_path).read_text(encoding="utf-8").splitlines():
        if line.startswith("## "):
            current_section = section_map.get(line[3:].strip())
            continue
        if current_section:
            captured[current_section].append(line)

    config_text = "\n".join(captured["config"]).strip()
    deployment_text = "\n".join(captured["deployment"]).strip()
    brief_relative = ctx.get("INSTRUCTION_SET_RELATIVE_PATH", "") + "/task-brief.md"

    parts = [
        "# Task Reference Expectations",
        "",
        f"- Task Key: `{ctx.get('TASK_KEY', '')}`",
        f"- Task Type: `{ctx.get('TASK_TYPE', '')}`",
        f"- Source brief snapshot: `{brief_relative}`",
        "",
        "## Usage Rules",
        "",
        "- Treat this file as the canonical task-level reference for config files and deployment manifests that carry expected values for the current task.",
        "- If a section below is empty, that means the task brief did not provide explicit expected-state guidance for that category.",
        "- Missing expectations are missing information, not confirmed current state.",
        "",
        "## Configuration References and Expected Values",
        "",
    ]
    parts.append(
        config_text
        or "- No explicit configuration-file expectations were provided in the task brief."
    )
    parts.extend(["", "## Deployment Manifests and Expected Values", ""])
    parts.append(
        deployment_text
        or "- No explicit deployment-manifest expectations were provided in the task brief."
    )
    _write_text(Path(output_path), "\n".join(parts).rstrip() + "\n")


def render_task_catalog_discovery(output_path: str, ctx: dict) -> None:
    project_root = Path(ctx["PROJECT_ROOT"])
    tasks_root = Path(ctx["OKSTRA_TASKS_ROOT"])

    def s(payload, key):
        if not isinstance(payload, dict):
            return ""
        v = payload.get(key, "")
        return v if isinstance(v, str) else ""

    def rel(p):
        try:
            return p.relative_to(project_root).as_posix()
        except ValueError:
            return str(p)

    entries = []
    for manifest_path in sorted(tasks_root.rglob(TASK_MANIFEST_FILENAME)):
        try:
            manifest = load_owned_object(manifest_path, artifact="task manifest")
        except Exception:
            continue
        if not isinstance(manifest, dict):
            continue
        task_key = s(manifest, "taskKey").strip()
        if not task_key:
            continue
        task_root = manifest_path.parent
        timeline_relative = s(manifest, "historyTimelinePath").strip()
        timeline_file = (
            (project_root / timeline_relative)
            if timeline_relative
            else task_timeline_file(task_root)
        )
        latest_run = {}
        if timeline_file.is_file():
            try:
                payload = load_owned_object(timeline_file, artifact="task timeline")
            except Exception:
                payload = {}
            runs = payload.get("runs", []) if isinstance(payload, dict) else []
            if isinstance(runs, list):
                for item in reversed(runs):
                    if isinstance(item, dict):
                        latest_run = item
                        break
        workflow = (
            manifest.get("workflow")
            if isinstance(manifest.get("workflow"), dict)
            else {}
        )
        entries.append(
            {
                "taskKey": task_key,
                "taskGroup": s(manifest, "taskGroup"),
                "taskId": s(manifest, "taskId"),
                "taskGroupPathSegment": s(manifest, "taskGroupPathSegment"),
                "taskIdPathSegment": s(manifest, "taskIdPathSegment"),
                "taskType": s(manifest, "taskType"),
                "workCategory": s(manifest, "workCategory"),
                "currentStatus": s(manifest, "currentStatus"),
                "workStatus": s(manifest, "workStatus"),
                "workStatusUpdatedAt": s(manifest, "workStatusUpdatedAt"),
                "workStatusNote": s(manifest, "workStatusNote"),
                "updatedAt": s(manifest, "updatedAt"),
                "currentPhase": (workflow or {}).get("currentPhase", "")
                if isinstance(workflow, dict)
                else "",
                "currentPhaseState": (workflow or {}).get("currentPhaseState", "")
                if isinstance(workflow, dict)
                else "",
                "lastCompletedPhase": (workflow or {}).get("lastCompletedPhase", "")
                if isinstance(workflow, dict)
                else "",
                "nextRecommendedPhase": next_phase.promote(
                    (workflow or {}).get("nextRecommendedPhase")
                    if isinstance(workflow, dict)
                    else None
                ),
                "awaitingApproval": (workflow or {}).get("awaitingApproval", False)
                if isinstance(workflow, dict)
                else False,
                "taskRootPath": s(manifest, "taskRootPath") or rel(task_root),
                "taskManifestPath": s(manifest, "taskManifestPath")
                or rel(manifest_path),
                "taskIndexPath": s(manifest, "taskIndexPath"),
                "instructionSetPath": s(manifest, "instructionSetPath"),
                "referenceExpectationsPath": s(manifest, "referenceExpectationsPath"),
                "taskBriefPath": s(manifest, "taskBriefPath"),
                "latestRunPath": s(manifest, "latestRunPath")
                or s(latest_run, "runDirectoryPath"),
                "latestRunManifestPath": s(latest_run, "runManifestPath"),
                "latestRunPromptsPath": s(manifest, "latestRunPromptsPath")
                or s(latest_run, "workerPromptDirectoryPath"),
                "latestPromptSnapshotPath": s(latest_run, "promptSnapshotPath"),
                "latestTeamStatePath": s(latest_run, "teamStatePath"),
                "latestRunStatus": s(manifest, "latestRunStatus")
                or s(latest_run, "status"),
                "latestReportRecordPath": s(manifest, "latestReportRecordPath")
                or s(latest_run, "reportPath"),
                "latestResumeCommandPath": s(manifest, "latestResumeCommandPath")
                or s(latest_run, "resumeCommandPath"),
                "historyTimelinePath": timeline_relative or rel(timeline_file),
                "fixCycles": manifest.get("fixCycles")
                or {"count": 0, "openCycleId": None, "latest": None},
            }
        )
    entries.sort(
        key=lambda x: (x.get("updatedAt", ""), x.get("taskKey", "")), reverse=True
    )
    payload = {
        "schemaVersion": "1.0",
        "projectId": ctx.get("PROJECT_ID", ""),
        "updatedAt": ctx.get("RUN_TIMESTAMP_ISO", ""),
        "latestTaskKey": ctx.get("TASK_KEY", ""),
        "latestTaskDiscoveryPath": ctx.get("OKSTRA_LATEST_TASK_RELATIVE_PATH", ""),
        "taskCount": len(entries),
        "tasks": entries,
    }
    _write_json(Path(output_path), payload)


def render_latest_task_discovery(output_path: str, ctx: dict) -> None:
    task_manifest_path = Path(ctx.get("TASK_MANIFEST_PATH", ""))
    task_manifest = {}
    if task_manifest_path.exists():
        try:
            task_manifest = load_owned_object(
                task_manifest_path, artifact="task manifest"
            )
        except Exception:
            task_manifest = {}
    workflow = (
        task_manifest.get("workflow")
        if isinstance(task_manifest.get("workflow"), dict)
        else {}
    )
    payload = {
        "schemaVersion": "1.0",
        "updatedAt": ctx.get("RUN_TIMESTAMP_ISO", ""),
        "taskKey": ctx.get("TASK_KEY", ""),
        "taskType": ctx.get("TASK_TYPE", ""),
        "workCategory": task_manifest.get(
            "workCategory", ctx.get("WORKFLOW_WORK_CATEGORY", "unknown")
        ),
        "currentStatus": task_manifest.get(
            "currentStatus", ctx.get("CURRENT_TASK_STATUS", "")
        ),
        "latestRunStatus": task_manifest.get(
            "latestRunStatus", ctx.get("CURRENT_RUN_STATUS", "")
        ),
        "workflow": workflow,
        "taskRootPath": ctx.get("TASK_ROOT_RELATIVE_PATH", ""),
        "taskManifestPath": ctx.get("TASK_MANIFEST_RELATIVE_PATH", ""),
        "taskIndexPath": ctx.get("TASK_INDEX_RELATIVE_PATH", ""),
        "instructionSetPath": ctx.get("INSTRUCTION_SET_RELATIVE_PATH", ""),
        "taskCatalogPath": ctx.get("OKSTRA_TASK_CATALOG_RELATIVE_PATH", ""),
        "referenceExpectationsPath": ctx.get(
            "REFERENCE_EXPECTATIONS_RELATIVE_PATH", ""
        ),
        "latestRunPath": ctx.get("LATEST_RUN_RELATIVE_PATH", ""),
        "latestRunManifestPath": ctx.get("RUN_MANIFEST_RELATIVE_PATH", ""),
        "latestRunPromptsPath": ctx.get("RUN_PROMPTS_RELATIVE_PATH", ""),
        "latestPromptSnapshotPath": ctx.get("RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""),
        "latestTeamStatePath": ctx.get("TEAM_STATE_RELATIVE_PATH", ""),
        "latestReportRecordPath": ctx.get("LATEST_REPORT_RECORD_RELATIVE_PATH", ""),
        "expectedReportRecordPath": ctx.get("FINAL_REPORT_RECORD_RELATIVE_PATH", ""),
        "expectedStatusPath": ctx.get("FINAL_STATUS_RELATIVE_PATH", ""),
        "latestResumeCommandPath": ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", ""),
        "historyTimelinePath": ctx.get("TIMELINE_RELATIVE_PATH", ""),
    }
    _write_json(Path(output_path), payload)


# --------------------------------------------------------------------------- #
# directories migration
# --------------------------------------------------------------------------- #


_PATH_BOUNDARY_CHARS = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-/")


def _rewrite_path_refs(text: str, path_rewrites: dict) -> str:
    """저장된 경로 문자열을 새 경로로 교체하되, 경계를 검사해 더 긴 무관한
    경로 안에 부분 문자열로 끼어든 경우는 건드리지 않는다.

    `old_rel` 앞뒤가 경로를 잇는 문자(영숫자/`.`/`-`/`_`/`/`)면 그 위치는
    다른 경로의 일부이므로 교체 대상에서 제외한다."""
    for old_rel, new_rel in path_rewrites.items():
        if not old_rel:
            continue
        out: list[str] = []
        idx = 0
        while True:
            hit = text.find(old_rel, idx)
            if hit == -1:
                out.append(text[idx:])
                break
            before = text[hit - 1] if hit > 0 else ""
            after_pos = hit + len(old_rel)
            after = text[after_pos] if after_pos < len(text) else ""
            on_boundary = (
                before not in _PATH_BOUNDARY_CHARS
                and after not in _PATH_BOUNDARY_CHARS
            )
            out.append(text[idx:hit])
            out.append(new_rel if on_boundary else old_rel)
            idx = after_pos
        text = "".join(out)
    return text


def migrate_legacy_run_artifacts(ctx: dict) -> None:
    import shutil

    project_root = Path(ctx["PROJECT_ROOT"])
    task_root = Path(ctx["TASK_ROOT"])
    run_dir = Path(ctx["RUN_DIR"])
    legacy_targets = [
        ("run-manifest-", ".json", Path(ctx["RUN_MANIFESTS_DIR"])),
        ("team-state-", ".json", Path(ctx["RUN_STATE_DIR"])),
        ("lead-execution-prompt-", ".md", Path(ctx["RUN_PROMPTS_DIR"])),
        ("claude-execution-prompt-", ".md", Path(ctx["RUN_PROMPTS_DIR"])),
        ("final-report-", ".md", Path(ctx["RUN_REPORTS_DIR"])),
        ("final-", ".status", Path(ctx["RUN_STATUS_DIR"])),
        ("claude-resume-", ".sh", Path(ctx["RUN_SESSIONS_DIR"])),
    ]
    if not run_dir.is_dir():
        return
    path_rewrites = {}
    for entry in run_dir.iterdir():
        if not entry.is_file():
            continue
        for prefix, suffix, target_dir in legacy_targets:
            if not entry.name.startswith(prefix) or not entry.name.endswith(suffix):
                continue
            target_dir.mkdir(parents=True, exist_ok=True)
            destination = target_dir / entry.name
            old_relative = str(entry.relative_to(project_root))
            new_relative = str(destination.relative_to(project_root))
            if destination.exists():
                try:
                    if entry.read_bytes() == destination.read_bytes():
                        entry.unlink()
                        path_rewrites[old_relative] = new_relative
                except Exception:
                    pass
                break
            shutil.move(str(entry), str(destination))
            path_rewrites[old_relative] = new_relative
            break
    if not path_rewrites:
        return
    text_suffixes = {".json", ".md", ".txt", ".sh"}
    for path in task_root.rglob("*"):
        if not path.is_file() or path.suffix not in text_suffixes:
            continue
        try:
            original = path.read_text(encoding="utf-8")
        except Exception:
            continue
        updated = _rewrite_path_refs(original, path_rewrites)
        if updated != original:
            path.write_text(updated, encoding="utf-8")


# --------------------------------------------------------------------------- #
# task / run manifest + timeline + task-index
# --------------------------------------------------------------------------- #


def _required_worker_roles(ctx: dict, reviewers: list[str]) -> list[dict]:
    catalog = _worker_catalog(ctx)
    return [
        {
            "workerId": catalog[item]["workerId"],
            "role": catalog[item]["role"],
            "agent": catalog[item]["agent"],
            "provider": catalog[item]["provider"],
            "runner": catalog[item]["runner"],
            "model": catalog[item]["model"],
            "modelExecutionValue": catalog[item]["modelExecutionValue"],
            "resultPath": catalog[item]["resultPath"],
            "promptPath": catalog[item]["promptPath"],
            "attemptRequired": True,
        }
        for item in reviewers
    ]


def _optional_worker_roles(ctx: dict) -> list[dict]:
    """Roles this run may dispatch but does not require — today, the critics.

    A critic is opt-in, so it never belonged in `requiredWorkerRoles`. But it
    was absent from the roster entirely, and both `okstra team dispatch` and
    the liveness reader locate a worker by its team-state row: dispatching a
    critic failed with `team-state has no workerId=acceptance`, while adding
    the row by hand failed validation as an `unexpected worker role`. Declaring
    it here makes the roster say what the run may run, so both sides agree.

    Keyed off `invocationAssignments`, which is where the run records the
    critic it actually resolved — an absent `critic/*` entry means no critic.
    """
    assignments = _invocation_assignments(ctx)
    roles: list[dict] = []
    for assignment_ref, assignment in sorted(assignments.items()):
        if not assignment_ref.startswith("critic/"):
            continue
        provider = str(assignment.get("provider", ""))
        roles.append({
            # `okstra team dispatch` projects a v2 worker's state key off the
            # assignment ref's last segment; the row it looks up must use it.
            "workerId": assignment_ref.rsplit("/", 1)[-1],
            "role": f"{provider_spec(provider).display_label} critic",
            "agent": provider,
            "provider": provider,
            "runner": str(assignment.get("runner", "")),
            "model": str(assignment.get("model", "")),
            "modelExecutionValue": str(assignment.get("modelExecutionValue", "")),
            # The lead names these when it materializes the invocation; a
            # critic has no prompt or result until it is dispatched.
            "resultPath": "",
            "promptPath": "",
            "attemptRequired": False,
        })
    return roles


def _reporter_confirmation_status(brief_bytes: bytes) -> str:
    try:
        lines = brief_bytes.decode("utf-8").splitlines()
    except UnicodeDecodeError:
        return ""
    if not lines or lines[0].strip() != "---":
        return ""
    for line in lines[1:]:
        if line.strip() == "---":
            break
        key, separator, value = line.partition(":")
        if separator and key.strip() == "reporter-confirmations":
            return value.strip().strip("'\"").lower()
    return ""


def _analysis_scope_confirmation_snapshot(ctx: dict) -> dict | None:
    if ctx.get("TASK_TYPE") not in ANALYSIS_TASK_TYPES:
        return None
    brief_path = Path(ctx.get("BRIEF_FILE_PATH", ""))
    try:
        brief_bytes = brief_path.read_bytes() if brief_path.is_file() else None
    except OSError:
        brief_bytes = None
    return {
        "taskBriefPath": ctx.get("BRIEF_RELATIVE_PATH", ""),
        "status": (
            _reporter_confirmation_status(brief_bytes)
            if brief_bytes is not None
            else ""
        ),
        "briefSha256": (
            hashlib.sha256(brief_bytes).hexdigest()
            if brief_bytes is not None
            else ""
        ),
    }


def _derive_phase_states(existing_workflow: dict, ctx: dict) -> tuple[dict, str, str]:
    """phaseStates dict + (current_phase, current_phase_state) 를 도출한다.

    기존 manifest 의 phaseStates 를 보존하면서 PHASE_SEQUENCE 의 모든 phase 를
    not-started 로 채우고 current_phase 만 현재 상태로 덮어쓴다."""
    phase_states = (
        existing_workflow.get("phaseStates", {})
        if isinstance(existing_workflow.get("phaseStates"), dict)
        else {}
    )
    current_phase = ctx.get("WORKFLOW_CURRENT_PHASE", ctx.get("TASK_TYPE", ""))
    current_phase_state = ctx.get("WORKFLOW_CURRENT_PHASE_STATE", "not-started")
    for phase in PHASE_SEQUENCE:
        phase_states.setdefault(phase, "not-started")
    if current_phase:
        phase_states[current_phase] = current_phase_state
    return phase_states, current_phase, current_phase_state


def _derive_next_recommended_phase(
    existing_workflow: dict, current_phase_state: str
) -> dict:
    """prepare 시점의 포인터를 만든다.

    대상 phase 는 리드가 정한다. prepare 는 그 값을 보존하고 진행 가능 여부만
    내린다 — 실행이 아직 안 끝났는데 다음을 제안하면 위저드가 현재 phase 를
    끝난 것으로 오인한다. (기록된 사례: implementation 이 `prepared` 로
    준비됐을 때 위저드가 `final-verification` 을 기본 task_type 으로 추천했다.)

    현재 phase 가 `completed` 면 그 오인이 성립하지 않으므로 포인터를 승격만 해서
    그대로 돌려준다. 내리는 대상도 `ready` 하나뿐이다 — `pending` · `blocked` ·
    `terminal` 은 애초에 착수를 부르지 않는 상태라 내려봐야 얻는 것이 없고,
    리드가 판단해 적은 `blocked`(사람이 봐야 하는 상태)와 `terminal`(생애주기
    종료)을 `pending`("아직 아무도 정하지 않음")으로 지워버린다.
    """
    pointer = next_phase.promote(existing_workflow.get("nextRecommendedPhase"))
    if current_phase_state == "completed":
        return pointer
    if pointer["status"] != next_phase.STATUS_READY:
        return pointer
    return next_phase.make(
        phase=pointer["phase"],
        status=next_phase.STATUS_PENDING,
        rationale=pointer["rationale"],
    )


def _derive_latest_pointers(existing: dict, ctx: dict, current_report_relative: str) -> dict:
    """latest run/report/team 포인터 + lastSafeCheckpoint 를 도출한다.

    render-only 경로는 기존 manifest 의 포인터를 보존(checkpoint 가 진행 중인
    실제 run 을 덮어쓰지 않도록)하고, 일반 경로는 이번 run 의 ctx 값으로 갱신한다."""
    render_only = ctx.get("RENDER_ONLY", "") == "true"
    existing_workflow = (
        existing.get("workflow", {}) if isinstance(existing.get("workflow"), dict) else {}
    )
    existing_checkpoint = existing_workflow.get("lastSafeCheckpoint", {})
    if not isinstance(existing_checkpoint, dict):
        existing_checkpoint = {}
    if render_only:
        return {
            "lastSafeCheckpoint": {
                "label": existing_checkpoint.get("label", ""),
                "taskManifestPath": existing_checkpoint.get(
                    "taskManifestPath", ctx.get("TASK_MANIFEST_RELATIVE_PATH", "")
                ),
                "taskIndexPath": existing_checkpoint.get(
                    "taskIndexPath", ctx.get("TASK_INDEX_RELATIVE_PATH", "")
                ),
                "latestRunPath": existing_checkpoint.get(
                    "latestRunPath", existing.get("latestRunPath", "")
                ),
                "latestRunManifestPath": existing_checkpoint.get(
                    "latestRunManifestPath", ""
                ),
                "latestTeamStatePath": existing_checkpoint.get(
                    "latestTeamStatePath", existing.get("teamStatePath", "")
                ),
                "latestReportRecordPath": existing_checkpoint.get(
                    "latestReportRecordPath", existing.get("latestReportRecordPath", "")
                ),
                "latestResumeCommandPath": existing_checkpoint.get(
                    "latestResumeCommandPath", existing.get("latestResumeCommandPath", "")
                ),
            },
            "latestRunPath": existing.get("latestRunPath", "")
            or ctx.get("LATEST_RUN_RELATIVE_PATH", ""),
            "latestRunStatus": existing.get("latestRunStatus", "")
            or ctx.get("CURRENT_RUN_STATUS", ""),
            "latestRunPromptsPath": existing.get("latestRunPromptsPath", "")
            or ctx.get("RUN_PROMPTS_RELATIVE_PATH", ""),
            "latestReportRecordPath": existing.get("latestReportRecordPath", "")
            or current_report_relative,
            "latestTeamStatePath": existing.get("teamStatePath", "")
            or ctx.get("TEAM_STATE_RELATIVE_PATH", ""),
            "latestResumeCommandPath": existing.get("latestResumeCommandPath", ""),
        }
    return {
        "lastSafeCheckpoint": {
            "label": ctx.get("WORKFLOW_LAST_SAFE_CHECKPOINT_LABEL", ""),
            "taskManifestPath": ctx.get("TASK_MANIFEST_RELATIVE_PATH", ""),
            "taskIndexPath": ctx.get("TASK_INDEX_RELATIVE_PATH", ""),
            "latestRunPath": ctx.get("LATEST_RUN_RELATIVE_PATH", ""),
            "latestRunManifestPath": ctx.get("RUN_MANIFEST_RELATIVE_PATH", ""),
            "latestTeamStatePath": ctx.get("TEAM_STATE_RELATIVE_PATH", ""),
            "latestReportRecordPath": current_report_relative,
            "latestResumeCommandPath": ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", ""),
        },
        "latestRunPath": ctx.get("LATEST_RUN_RELATIVE_PATH", ""),
        "latestRunStatus": ctx.get("CURRENT_RUN_STATUS", ""),
        "latestRunPromptsPath": ctx.get("RUN_PROMPTS_RELATIVE_PATH", ""),
        "latestReportRecordPath": current_report_relative
        or existing.get("latestReportRecordPath", ""),
        "latestTeamStatePath": ctx.get("TEAM_STATE_RELATIVE_PATH", ""),
        "latestResumeCommandPath": ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", "")
        or existing.get("latestResumeCommandPath", ""),
    }


def render_task_manifest(manifest_path: str, ctx: dict) -> None:
    path = Path(manifest_path)
    existing = {}
    if path.exists():
        try:
            existing = load_owned_object(path, artifact="task manifest")
        except Exception:
            existing = {}
    reviewers = _resolve_workers(ctx)
    worker_dispatch_plan = _worker_dispatch_plan(ctx)
    catalog = _worker_catalog(ctx)
    required_worker_roles = _required_worker_roles(ctx, reviewers)
    worker_prompt_paths = {item: catalog[item]["promptPath"] for item in reviewers}
    lead_agent = _lead_agent(ctx)
    lead_role = _lead_role(ctx)
    required_agent_status_entries = [lead_role] + [
        catalog[item]["role"] for item in reviewers
    ]
    related_tasks = json.loads(ctx.get("RELATED_TASKS_JSON", "[]"))
    current_report_relative = ctx.get("LATEST_REPORT_RECORD_RELATIVE_PATH") or ctx.get(
        "FINAL_REPORT_RECORD_RELATIVE_PATH", ""
    )
    existing_workflow = (
        existing.get("workflow", {})
        if isinstance(existing.get("workflow"), dict)
        else {}
    )
    phase_states, current_phase, current_phase_state = _derive_phase_states(
        existing_workflow, ctx
    )
    work_category = existing.get("workCategory") or ctx.get(
        "WORKFLOW_WORK_CATEGORY", "unknown"
    )
    next_recommended_phase = _derive_next_recommended_phase(
        existing_workflow, current_phase_state
    )
    last_completed_phase = existing_workflow.get("lastCompletedPhase") or ctx.get(
        "WORKFLOW_LAST_COMPLETED_PHASE", ""
    )
    awaiting_approval = existing_workflow.get("awaitingApproval")
    if not isinstance(awaiting_approval, bool):
        awaiting_approval = ctx.get("WORKFLOW_AWAITING_APPROVAL", "false") == "true"
    pointers = _derive_latest_pointers(existing, ctx, current_report_relative)
    last_safe_checkpoint = pointers["lastSafeCheckpoint"]
    latest_run_relative = pointers["latestRunPath"]
    latest_run_status = pointers["latestRunStatus"]
    latest_run_prompts_relative = pointers["latestRunPromptsPath"]
    latest_report_relative = pointers["latestReportRecordPath"]
    latest_team_state_relative = pointers["latestTeamStatePath"]
    latest_resume_command_relative = pointers["latestResumeCommandPath"]
    convergence_block = _build_convergence_block(ctx)
    execution_identity = _execution_identity(ctx)
    existing_v2 = (
        existing.get("schemaVersion") == "2.0"
        and existing.get("executionIdentityVersion") == 2
    )
    writes_v2 = execution_identity.get("executionIdentityVersion") == 2
    payload = {
        "schemaVersion": "2.0" if existing_v2 or writes_v2 else "1.0",
        "projectId": ctx.get("PROJECT_ID", ""),
        "taskGroup": ctx.get("TASK_GROUP", ""),
        "taskId": ctx.get("TASK_ID", ""),
        "taskKey": ctx.get("TASK_KEY", ""),
        "taskGroupPathSegment": ctx.get("TASK_GROUP_SEGMENT", ""),
        "taskIdPathSegment": ctx.get("TASK_ID_SEGMENT", ""),
        "projectRoot": ctx.get("PROJECT_ROOT", ""),
        "taskType": ctx.get("TASK_TYPE", ""),
        "hostRuntime": ctx.get("HOST_RUNTIME", "") or _lead_runtime(ctx),
        "leadRuntime": _lead_runtime(ctx),
        "leadAssignment": _lead_assignment(ctx),
        "workerAssignments": _worker_assignments(ctx),
        "leadAdapter": _lead_adapter(ctx, worker_dispatch_plan),
        "workCategory": work_category,
        # user-managed status (set-work-status CLI) — carried across re-renders
        "workStatus": existing.get("workStatus", ""),
        "workStatusUpdatedAt": existing.get("workStatusUpdatedAt", ""),
        "workStatusNote": existing.get("workStatusNote", ""),
        "taskBriefPath": ctx.get("BRIEF_RELATIVE_PATH", ""),
        "recommendedWorkers": reviewers,
        "relatedTasks": related_tasks,
        "currentStatus": ctx.get("CURRENT_TASK_STATUS", ""),
        "renderOnly": ctx.get("RENDER_ONLY", ""),
        "taskRootPath": ctx.get("TASK_ROOT_RELATIVE_PATH", ""),
        "instructionSetPath": ctx.get("INSTRUCTION_SET_RELATIVE_PATH", ""),
        "taskManifestPath": ctx.get("TASK_MANIFEST_RELATIVE_PATH", ""),
        "taskIndexPath": ctx.get("TASK_INDEX_RELATIVE_PATH", ""),
        "runsPath": ctx.get("RUNS_RELATIVE_PATH", ""),
        "historyPath": ctx.get("HISTORY_RELATIVE_PATH", ""),
        "historyTimelinePath": ctx.get("TIMELINE_RELATIVE_PATH", ""),
        "taskCatalogPath": ctx.get("OKSTRA_TASK_CATALOG_RELATIVE_PATH", ""),
        "referenceExpectationsPath": ctx.get(
            "REFERENCE_EXPECTATIONS_RELATIVE_PATH", ""
        ),
        "latestRunPath": latest_run_relative,
        "latestRunStatus": latest_run_status,
        "latestRunPromptsPath": latest_run_prompts_relative,
        "latestReportRecordPath": latest_report_relative,
        "latestResumeCommandPath": latest_resume_command_relative,
        "teamStatePath": latest_team_state_relative,
        "fixCycles": fix_cycles.summarize(
            fix_cycles.read_rows(Path(manifest_path).parent)
        ),
        "workflow": {
            "phaseSequence": PHASE_SEQUENCE,
            "currentPhase": current_phase,
            "currentPhaseState": phase_states.get(current_phase, current_phase_state),
            "phaseStates": phase_states,
            "lastCompletedPhase": last_completed_phase,
            "nextRecommendedPhase": next_recommended_phase,
            "awaitingApproval": awaiting_approval,
            "lastSafeCheckpoint": last_safe_checkpoint,
        },
        "artifacts": {
            "analysisProfilePath": ctx.get("INSTRUCTION_SET_RELATIVE_PATH", "")
            + "/analysis-profile.md",
            "analysisMaterialPath": ctx.get("INSTRUCTION_SET_RELATIVE_PATH", "")
            + "/analysis-material.md",
            "analysisPacketPath": ctx.get("ANALYSIS_PACKET_RELATIVE_PATH", ""),
            "verificationTargetPath": ctx.get(
                "VERIFICATION_TARGET_RELATIVE_PATH", ""
            ),
            "verificationTargetDigest": ctx.get("VERIFICATION_TARGET_DIGEST", ""),
            "taskBriefCopyPath": ctx.get("INSTRUCTION_SET_RELATIVE_PATH", "")
            + "/task-brief.md",
            "referenceExpectationsPath": ctx.get(
                "REFERENCE_EXPECTATIONS_RELATIVE_PATH", ""
            ),
            "leadExecutionPromptPath": ctx.get("RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""),
            "claudeExecutionPromptPath": ctx.get("RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""),
            "leadPromptSnapshotPath": ctx.get("RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""),
            "workerPromptsDirectoryPath": ctx.get("RUN_PROMPTS_RELATIVE_PATH", ""),
            "workerPromptPathByWorkerId": worker_prompt_paths,
            "finalReportTemplatePath": ctx.get(
                "FINAL_REPORT_TEMPLATE_RELATIVE_PATH", ""
            ),
            "resumeCommandPath": ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", ""),
            "teamStatePath": ctx.get("TEAM_STATE_RELATIVE_PATH", ""),
            "activeRunContextPath": ctx.get("ACTIVE_RUN_CONTEXT_RELATIVE_PATH", ""),
            "leadEventsPath": ctx.get("LEAD_EVENTS_RELATIVE_PATH", ""),
            "workerResultsDirectoryPath": ctx.get("WORKER_RESULTS_RELATIVE_PATH", ""),
            "validatorScriptPath": ctx.get("RUN_VALIDATOR_RELATIVE_PATH", ""),
        },
        "resultContract": {
            "leadAgent": lead_agent,
            "leadRole": lead_role,
            "leadModel": ctx.get("LEAD_MODEL", ""),
            "leadModelExecutionValue": ctx.get("LEAD_MODEL_EXECUTION_VALUE", ""),
            "leadExecutionMode": "synthesis-only",
            "finalSynthesisOwner": lead_role,
            "artifactFirst": True,
            "resultCollectionMode": "lead-managed",
            "finalReportFormat": "markdown",
            "finalReportFilename": ctx.get("FINAL_REPORT_FILENAME", ""),
            "finalStatusFilename": ctx.get("FINAL_STATUS_FILENAME", ""),
            "minimumPreferredWorkerResults": len(reviewers),
            "requiredWorkerAttempts": reviewers,
            "requiredWorkerRoles": required_worker_roles,
            "optionalWorkerRoles": _optional_worker_roles(ctx),
            "requiredAgentStatusEntries": required_agent_status_entries,
            "requireDistinctLeadFromWorkerSession": True,
            "requireAllRequiredWorkerAttempts": True,
            "requireAntigravityWorkerAttempt": "antigravity" in reviewers,
            "requireCollectedWorkerStatusesBeforeFinalVerdict": True,
            "disallowLeadSoloAnalysisAsWorkerResult": True,
            "disallowGenericParallelOnlyExecution": True,
            "workerOutputSections": [
                "Findings",
                "Missing Information or Assumptions",
                "Safe or Reasonable Areas",
                "Uncertain Points",
                "Recommended Next Actions",
            ],
            "finalReportSections": [
                "Problem or Validation Summary",
                "Agent Execution Status",
                "Cross Verification Result",
                "Final Verdict",
                "Evidence and Detailed Analysis",
                "Missing Information and Risk",
                "Recommended Next Actions",
            ],
            "statusLabels": [
                "prepared",
                "team-created",
                "workers-dispatched",
                "worker-results-collected",
                "synthesis-written",
                "in-progress",
                "completed",
                "contract-violated",
                "timeout",
                "error",
                "not-run",
            ],
        },
        "contractValidation": {
            "required": True,
            "status": ctx.get("VALIDATION_STATUS", "not-run"),
            "lastCheckedAt": ctx.get("VALIDATION_UPDATED_AT", ""),
            "passed": ctx.get("VALIDATION_STATUS", "not-run") == "passed",
            "failures": json.loads(ctx.get("VALIDATION_FAILURES_JSON", "[]")),
            "teamStatePath": ctx.get("TEAM_STATE_RELATIVE_PATH", ""),
            "validatorScriptPath": ctx.get("RUN_VALIDATOR_RELATIVE_PATH", ""),
        },
        "leadSession": {
            "sessionId": ctx.get("CLAUDE_SESSION_ID", ""),
            "resumeCommandPath": ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", ""),
            "accounting": _lead_descriptor(ctx).session_accounting,
        },
        "convergence": convergence_block,
        "createdAt": existing.get("createdAt") or ctx.get("RUN_TIMESTAMP_ISO", ""),
        "updatedAt": ctx.get("RUN_TIMESTAMP_ISO", ""),
    }
    if _lead_runtime(ctx) == "claude-code":
        payload["claudeSession"] = {
            "sessionId": ctx.get("CLAUDE_SESSION_ID", ""),
            "resumeCommandPath": ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", ""),
        }
    if existing_v2 or writes_v2:
        existing_index = existing.get("roleExecutionIndex")
        payload["executionIdentityVersion"] = 2
        payload["roleExecutionIndex"] = (
            list(existing_index) if isinstance(existing_index, list) else []
        )
    _write_json(path, payload)


def _build_convergence_block(ctx: dict) -> dict:
    """Resolve the `convergence` sub-tree written into task-manifest.json.

    Defaults follow `prompts/lead/convergence.md`:
    - `enabled` default True
    - `maxRounds` default 1 for `requirements-discovery`, 2 otherwise
    - `verificationMode` default "lightweight"
    - `adversarial` default True for discovery, planning, and analysis task types
      (forces `verificationMode` to "full-reanalysis"), False otherwise
    - `planBodyVerification` is implementation-planning specific; the key is
      always emitted (dead-letter on other phases) so the schema stays stable.
      Its `selfFixMaxRounds` default 1 bounds the report-writer self-fix loop
      that runs before a planner-fixable defect is promoted to the user.
      `gating` is true here because the plan does not exist yet. After the
      report-writer draft, `okstra plan-items prepare` flips it to false when
      `designPreparation.mode` is `no-design-inputs` and the Stage Map has
      exactly one row.

    ctx knobs honoured:
    - `OKSTRA_PLAN_VERIFICATION`: "true" | "false" | "" (empty → default True).
      Wired from CLI `--no-plan-verification` (sets "false").
    - `CRITIC_CHOICE`: "" | "off" | a provider with the critic capability — critic
      backing provider (enabled only for requirements-discovery / error-analysis /
      implementation-planning / final-verification); model taken from that
      provider's execution value.
    """
    task_type = ctx.get("TASK_TYPE", "")
    default_max_rounds = 1 if task_type == "requirements-discovery" else 2
    adversarial_phases = {
        "requirements-discovery",
        "error-analysis",
        "implementation-option-selection",
        "implementation-planning",
        "project-analysis",
        "feature-analysis",
        "change-impact-analysis",
    }
    is_adversarial = task_type in adversarial_phases
    raw_plan_verify = (ctx.get("OKSTRA_PLAN_VERIFICATION", "") or "").strip().lower()
    plan_verify_enabled = raw_plan_verify != "false"
    critic_choice = (ctx.get("CRITIC_CHOICE", "") or "").strip().lower()
    # Independent of `adversarial_phases` above (they answer different questions and
    # may diverge): the coverage critic is opt-in for the finding-producing phases.
    critic_phases = {"requirements-discovery", "error-analysis", "implementation-planning", "final-verification"}
    critic_enabled = critic_choice in provider_ids("critic") and task_type in critic_phases
    assignment = next(
        (
            row for row in _worker_assignments(ctx)
            if row.get("provider") == critic_choice and row.get("role") != "report-writer"
        ),
        {},
    )
    _, legacy_execution = _legacy_worker_model(ctx, critic_choice)
    critic_block = {
        "enabled": critic_enabled,
        "provider": critic_choice if critic_enabled else None,
        "modelExecutionValue": (
            ctx.get("CRITIC_MODEL_EXECUTION_VALUE")
            or assignment.get("modelExecutionValue")
            or legacy_execution
            or None
        ) if critic_enabled else None,
    }
    return {
        "enabled": True,
        "adversarial": is_adversarial,
        "maxRounds": default_max_rounds,
        "verificationMode": "full-reanalysis" if is_adversarial else "lightweight",
        "critic": critic_block,
        "planBodyVerification": {
            "enabled": plan_verify_enabled,
            "maxRounds": 1,
            "selfFixMaxRounds": 1,
            "gating": True,
        },
    }


def render_run_manifest(run_manifest_path: str, ctx: dict) -> None:
    run_manifest_file = Path(run_manifest_path)
    run_manifest_exists = run_manifest_file.is_file()
    existing_run_manifest = {}
    if run_manifest_exists:
        try:
            loaded_run_manifest = load_owned_object(
                run_manifest_file, artifact="run manifest"
            )
            existing_run_manifest = (
                loaded_run_manifest
                if isinstance(loaded_run_manifest, dict)
                else {}
            )
        except JsonBoundaryError:
            existing_run_manifest = {}
    task_manifest_path = Path(ctx.get("TASK_MANIFEST_PATH", ""))
    task_manifest = {}
    if task_manifest_path.exists():
        try:
            task_manifest = load_owned_object(
                task_manifest_path, artifact="task manifest"
            )
        except Exception:
            task_manifest = {}
    reviewers = _resolve_workers(ctx)
    worker_dispatch_plan = _worker_dispatch_plan(ctx)
    catalog = _worker_catalog(ctx)
    required_worker_roles = _required_worker_roles(ctx, reviewers)
    worker_prompt_paths = {item: catalog[item]["promptPath"] for item in reviewers}
    lead_agent = _lead_agent(ctx)
    lead_role = _lead_role(ctx)
    related_tasks = json.loads(ctx.get("RELATED_TASKS_JSON", "[]"))
    workflow = (
        task_manifest.get("workflow", {})
        if isinstance(task_manifest.get("workflow"), dict)
        else {}
    )
    # prepare 가 감지한 동시-run 사실의 영속 앵커. validator 는 이 prepare-측
    # 기록이 있을 때만 no-team(teamCreate skipped) 경로를 legal 로 인정한다 —
    # lead 의 team-state 자기 선언만으로는 열리지 않는다.
    concurrent_run_stages = [
        int(s)
        for s in str(ctx.get("CONCURRENT_RUN_STAGES", "") or "").split(",")
        if s.strip().isdigit()
    ]
    execution_identity = _execution_identity(ctx)
    payload = {
        "schemaVersion": execution_identity.get("schemaVersion", "1.0"),
        "reportContractVersion": CURRENT_REPORT_SCHEMA_VERSION,
        "okstraVersion": ctx.get("OKSTRA_VERSION", ""),
        "projectId": ctx.get("PROJECT_ID", ""),
        "taskGroup": ctx.get("TASK_GROUP", ""),
        "taskId": ctx.get("TASK_ID", ""),
        "taskKey": ctx.get("TASK_KEY", ""),
        "taskType": ctx.get("TASK_TYPE", ""),
        "runTimestamp": ctx.get("RUN_TIMESTAMP_ISO", ""),
        "reportLanguage": ctx.get("REPORT_LANGUAGE", "en") or "en",
        "leadModel": ctx.get("LEAD_MODEL", ""),
        # Every other path in this manifest is project-relative; this is the
        # anchor they resolve against. Convergence reads it to place run
        # artifacts, so a manifest without it cannot seed a round.
        "projectRoot": ctx.get("PROJECT_ROOT", ""),
        "hostRuntime": ctx.get("HOST_RUNTIME", "") or _lead_runtime(ctx),
        "leadRuntime": _lead_runtime(ctx),
        "leadRuntimeRequest": ctx.get("LEAD_RUNTIME_REQUEST", "") or _lead_runtime(ctx),
        "terminalBackend": ctx.get("TERMINAL_BACKEND", ""),
        "runtimeResolution": _runtime_resolution(ctx),
        "leadAssignment": _lead_assignment(ctx),
        "workerAssignments": _worker_assignments(ctx),
        "agentContract": _agent_contract(ctx),
        "invocationAssignments": _invocation_assignments(ctx),
        "leadAdapter": _lead_adapter(ctx, worker_dispatch_plan),
        "workCategory": task_manifest.get(
            "workCategory", ctx.get("WORKFLOW_WORK_CATEGORY", "unknown")
        ),
        "taskBriefPath": ctx.get("BRIEF_RELATIVE_PATH", ""),
        "relatedTasks": related_tasks,
        "recommendedWorkers": reviewers,
        "taskRootPath": ctx.get("TASK_ROOT_RELATIVE_PATH", ""),
        "taskManifestPath": ctx.get("TASK_MANIFEST_RELATIVE_PATH", ""),
        "instructionSetPath": ctx.get("INSTRUCTION_SET_RELATIVE_PATH", ""),
        "taskCatalogPath": ctx.get("OKSTRA_TASK_CATALOG_RELATIVE_PATH", ""),
        "referenceExpectationsPath": ctx.get(
            "REFERENCE_EXPECTATIONS_RELATIVE_PATH", ""
        ),
        "runDirectoryPath": ctx.get("RUN_DIR_RELATIVE_PATH", ""),
        "runDateTimeSegment": ctx.get("RUN_DATETIME_SEGMENT", ""),
        "runSequencesByCategory": {
            "manifests": ctx.get("RUN_MANIFESTS_SEQ", ""),
            "prompts": ctx.get("RUN_PROMPTS_SEQ", ""),
            "reports": ctx.get("RUN_REPORTS_SEQ", ""),
            "status": ctx.get("RUN_STATUS_SEQ", ""),
            "state": ctx.get("RUN_STATE_SEQ", ""),
            "sessions": ctx.get("RUN_SESSIONS_SEQ", ""),
            "workerResults": ctx.get("WORKER_RESULTS_SEQ", ""),
        },
        "runManifestPath": ctx.get("RUN_MANIFEST_RELATIVE_PATH", ""),
        "leadEventsPath": ctx.get("LEAD_EVENTS_RELATIVE_PATH", ""),
        "promptSnapshotPath": ctx.get("RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""),
        "leadInstructionsPath": ctx.get("LEAD_INSTRUCTIONS_RELATIVE_PATH", ""),
        "leadExecutionPromptPath": ctx.get(
            "RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""
        ),
        "leadPromptMetadataPath": ctx.get(
            "LEAD_PROMPT_METADATA_RELATIVE_PATH", ""
        ),
        "resources": {
            "leadInstructionsPath": ctx.get(
                "LEAD_INSTRUCTIONS_RELATIVE_PATH", ""
            ),
            "leadExecutionPromptPath": ctx.get(
                "RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""
            ),
            "leadPromptMetadataPath": ctx.get(
                "LEAD_PROMPT_METADATA_RELATIVE_PATH", ""
            ),
            "leadPromptSnapshotPath": ctx.get(
                "RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""
            ),
            "claudeExecutionPromptPath": ctx.get(
                "RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""
            ),
        },
        "invocationReservationRootPath": ctx.get(
            "INVOCATION_RESERVATION_ROOT_RELATIVE_PATH", ""
        ),
        "workerPromptsDirectoryPath": ctx.get("RUN_PROMPTS_RELATIVE_PATH", ""),
        "workerPromptPathByWorkerId": worker_prompt_paths,
        "expectedReportRecordPath": ctx.get("FINAL_REPORT_RECORD_RELATIVE_PATH", ""),
        "reportNarrativePath": ctx.get("REPORT_WRITER_NARRATIVE_RELATIVE_PATH", ""),
        "approvalDecisionsPath": ctx.get("APPROVAL_DECISIONS_RELATIVE_PATH", ""),
        "designPreparationPath": ctx.get("DESIGN_PREPARATION_RELATIVE_PATH", ""),
        "planBodyVerificationPath": ctx.get("PLAN_BODY_VERIFICATION_STATE_RELATIVE_PATH", ""),
        "expectedStatusPath": ctx.get("FINAL_STATUS_RELATIVE_PATH", ""),
        "teamStatePath": ctx.get("TEAM_STATE_RELATIVE_PATH", ""),
        "activeRunContextPath": ctx.get("ACTIVE_RUN_CONTEXT_RELATIVE_PATH", ""),
        "convergenceStatePath": ctx.get("CONVERGENCE_STATE_RELATIVE_PATH", ""),
        "analysisPacketPath": ctx.get("ANALYSIS_PACKET_RELATIVE_PATH", ""),
        "analysisEvidencePath": (
            ctx.get("INSTRUCTION_SET_RELATIVE_PATH", "") + "/analysis-evidence.md"
            if ctx.get("TASK_TYPE") in {"project-analysis", "feature-analysis", "change-impact-analysis"}
            else ""
        ),
        "analysisSourceCommit": ctx.get("ANALYSIS_SOURCE_COMMIT", ""),
        "analysisTarget": json.loads(ctx.get("ANALYSIS_TARGET_JSON", "{}")),
        "evidenceInputs": json.loads(ctx.get("EVIDENCE_INPUTS_JSON", "[]")),
        "verificationTargetPath": ctx.get("VERIFICATION_TARGET_RELATIVE_PATH", ""),
        "verificationTargetDigest": ctx.get("VERIFICATION_TARGET_DIGEST", ""),
        "workerResultsDirectoryPath": ctx.get("WORKER_RESULTS_RELATIVE_PATH", ""),
        "reportTemplatePath": ctx.get("FINAL_REPORT_TEMPLATE_RELATIVE_PATH", ""),
        "validatorScriptPath": ctx.get("RUN_VALIDATOR_RELATIVE_PATH", ""),
        "leadSessionId": ctx.get("CLAUDE_SESSION_ID", ""),
        "resumeCommandPath": ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", ""),
        "concurrentRun": {
            "detected": bool(concurrent_run_stages),
            "activeStages": concurrent_run_stages,
        },
        "workflowSnapshot": {
            "phaseSequence": workflow.get("phaseSequence", []),
            "currentPhase": workflow.get(
                "currentPhase", ctx.get("WORKFLOW_CURRENT_PHASE", "")
            ),
            "currentPhaseState": workflow.get(
                "currentPhaseState", ctx.get("WORKFLOW_CURRENT_PHASE_STATE", "")
            ),
            "phaseStates": workflow.get("phaseStates", {}),
            "lastCompletedPhase": workflow.get(
                "lastCompletedPhase", ctx.get("WORKFLOW_LAST_COMPLETED_PHASE", "")
            ),
            "nextRecommendedPhase": next_phase.promote(
                workflow.get("nextRecommendedPhase")
            ),
            "awaitingApproval": workflow.get(
                "awaitingApproval",
                ctx.get("WORKFLOW_AWAITING_APPROVAL", "false") == "true",
            ),
            "lastSafeCheckpoint": workflow.get("lastSafeCheckpoint", {}),
        },
        "teamContract": {
            "leadAgent": lead_agent,
            "leadRole": lead_role,
            "leadModel": ctx.get("LEAD_MODEL", ""),
            "leadModelExecutionValue": ctx.get("LEAD_MODEL_EXECUTION_VALUE", ""),
            "leadExecutionMode": "synthesis-only",
            "finalSynthesisOwner": lead_role,
            "requiredWorkerAttempts": reviewers,
            "requiredWorkerRoles": required_worker_roles,
            "optionalWorkerRoles": _optional_worker_roles(ctx),
            "requiredAgentStatusEntries": [lead_role]
            + [catalog[item]["role"] for item in reviewers],
            "requireDistinctLeadFromWorkerSession": True,
            "requireAllRequiredWorkerAttempts": True,
            "requireAntigravityWorkerAttempt": "antigravity" in reviewers,
            "requireCollectedWorkerStatusesBeforeFinalVerdict": True,
            "disallowLeadSoloAnalysisAsWorkerResult": True,
            "disallowGenericParallelOnlyExecution": True,
            "preferredCompletedWorkerResults": len(reviewers),
            "executor": _executor_contract(ctx),
        },
        "validation": {
            "required": True,
            "status": ctx.get("VALIDATION_STATUS", "not-run"),
            "lastCheckedAt": ctx.get("VALIDATION_UPDATED_AT", ""),
            "passed": ctx.get("VALIDATION_STATUS", "not-run") == "passed",
            "failures": json.loads(ctx.get("VALIDATION_FAILURES_JSON", "[]")),
        },
        "status": ctx.get("CURRENT_RUN_STATUS", ""),
        "renderOnly": ctx.get("RENDER_ONLY", ""),
        "createdAt": ctx.get("RUN_TIMESTAMP_ISO", ""),
    }
    scope_confirmation = existing_run_manifest.get("analysisScopeConfirmation")
    if "analysisScopeConfirmation" not in existing_run_manifest:
        scope_confirmation = _analysis_scope_confirmation_snapshot(ctx)
    if scope_confirmation is not None:
        payload["analysisScopeConfirmation"] = scope_confirmation
    if ctx.get("FIX_CYCLE_ID"):
        payload["fixCycleId"] = ctx["FIX_CYCLE_ID"]
    if (
        CURRENT_REPORT_SCHEMA_VERSION == "3.0"
        or ctx.get("TASK_TYPE") == "implementation-planning"
    ):
        if not run_manifest_exists:
            payload["activityContractVersion"] = 1
        elif "activityContractVersion" in existing_run_manifest:
            payload["activityContractVersion"] = existing_run_manifest[
                "activityContractVersion"
            ]
    payload["reportContracts"] = (
        ["implementation-design-prep-v1"]
        if ctx.get("TASK_TYPE") == "implementation-planning"
        else []
    )
    if _lead_runtime(ctx) == "claude-code":
        payload["claudeSessionId"] = ctx.get("CLAUDE_SESSION_ID", "")
    payload.update(execution_identity)
    _guard_invocation_manifest_rewrite(existing_run_manifest, payload)
    _write_json(Path(run_manifest_path), payload)
    if CURRENT_REPORT_SCHEMA_VERSION == "3.0":
        _initialize_report_ledgers(ctx, payload)


def _initialize_report_ledgers(ctx: Mapping[str, Any], manifest: Mapping[str, Any]) -> None:
    approval_value = str(ctx.get("APPROVAL_DECISIONS_PATH") or "")
    approval_path = Path(approval_value)
    if approval_value and not approval_path.is_file():
        _write_json(approval_path, {
            "schemaVersion": "1.0",
            "owner": "lead",
            "taskKey": manifest.get("taskKey"),
            "taskType": manifest.get("taskType"),
            "runSeq": (manifest.get("runSequencesByCategory") or {}).get("manifests"),
            "activeClarifications": [],
            "carriedDecisions": [],
        })
    activity_value = str(ctx.get("LEAD_EVENTS_PATH") or "")
    activity_path = Path(activity_value)
    if activity_value and not activity_path.is_file():
        _write_text(activity_path, "")


_MODEL_ASSIGNMENT_KEYS = {
    "provider",
    "model",
    "modelExecutionValue",
    "runner",
    "hostRuntime",
    "hostModelValue",
}
_V2_INVOCATION_IDENTITY_KEYS = {
    "participantRef",
    "roleExecutionRef",
    "dutyId",
}
_AGENT_CONTRACT_KEYS = {
    "schemaVersion",
    "dutyRootPath",
    "catalogDigest",
    "invocationReservationRootPath",
    "allowedAudiences",
    "authorizedPaths",
}
_IMMUTABLE_INVOCATION_KEYS = (
    "agentContract",
    "invocationAssignments",
    "leadAssignment",
    "workerAssignments",
    "runDirectoryPath",
    "runManifestPath",
    "leadInstructionsPath",
    "leadExecutionPromptPath",
    "promptSnapshotPath",
    "leadPromptMetadataPath",
    "resources",
    "invocationReservationRootPath",
    "workerPromptsDirectoryPath",
    "workerPromptPathByWorkerId",
    "workerResultsDirectoryPath",
    "expectedReportRecordPath",
    "reportNarrativePath",
    "approvalDecisionsPath",
    "designPreparationPath",
    "planBodyVerificationPath",
    "reportTemplatePath",
)
_V3_REPORT_PATH_KEYS = frozenset({
    "reportNarrativePath",
    "approvalDecisionsPath",
    "designPreparationPath",
    "planBodyVerificationPath",
})


def _guard_invocation_manifest_rewrite(
    existing: Mapping[str, object],
    proposed: Mapping[str, object],
) -> None:
    existing_has_contract = bool(existing.get("agentContract"))
    proposed_has_contract = bool(proposed.get("agentContract"))
    if existing_has_contract:
        _validate_invocation_manifest(existing)
    if proposed_has_contract:
        _validate_invocation_manifest(proposed)
    elif existing_has_contract:
        raise ValueError("incomplete invocation manifest: proposed contract missing")
    if not existing_has_contract:
        return
    drift = [
        key
        for key in _IMMUTABLE_INVOCATION_KEYS
        if existing.get(key) != proposed.get(key)
    ]
    if drift:
        raise ValueError(
            "immutable invocation manifest fields changed: " + ", ".join(drift)
        )


def _validate_invocation_manifest(payload: Mapping[str, object]) -> None:
    contract = payload.get("agentContract")
    assignments = payload.get("invocationAssignments")
    if (
        not isinstance(contract, Mapping)
        or set(contract) != _AGENT_CONTRACT_KEYS
        or contract.get("schemaVersion") != 1
        or not isinstance(assignments, Mapping)
        or not assignments
    ):
        raise ValueError("incomplete invocation manifest")
    _invocation_manifest_identity_version(payload)
    _validate_invocation_assignment_shapes(assignments)
    _validate_compatibility_projections(payload, assignments)
    for key in (
        item
        for item in _IMMUTABLE_INVOCATION_KEYS[4:]
        if item != "workerPromptPathByWorkerId"
        and (
            item not in _V3_REPORT_PATH_KEYS
            or payload.get("reportContractVersion") == "3.0"
        )
    ):
        if not payload.get(key):
            raise ValueError(f"incomplete invocation manifest path: {key}")
    workers = payload.get("workerAssignments")
    worker_prompt_paths = payload.get("workerPromptPathByWorkerId")
    expected_worker_ids = {
        str(worker.get("workerId") or "")
        for worker in workers
        if isinstance(worker, Mapping)
    }
    if (
        not isinstance(worker_prompt_paths, Mapping)
        or set(worker_prompt_paths) != expected_worker_ids
        or any(
            not isinstance(path, str) or not path
            for path in worker_prompt_paths.values()
        )
    ):
        raise ValueError(
            "incomplete invocation manifest path: workerPromptPathByWorkerId"
        )


def _validate_compatibility_projections(
    payload: Mapping[str, object],
    assignments: Mapping[str, object],
) -> None:
    lead = payload.get("leadAssignment")
    workers = payload.get("workerAssignments")
    if not isinstance(lead, Mapping) or not isinstance(workers, list):
        raise ValueError("incomplete invocation manifest compatibility projection")
    if _model_projection(lead) != _model_projection_from_assignment(
        assignments.get("lead")
    ):
        raise ValueError("immutable invocation manifest lead projection drift")
    for worker in workers:
        if not isinstance(worker, Mapping):
            raise ValueError("incomplete invocation manifest worker projection")
        reference = f"initial/{worker.get('workerId', '')}"
        if _model_projection(worker) != _model_projection_from_assignment(
            assignments.get(reference)
        ):
            raise ValueError("immutable invocation manifest worker projection drift")


def _model_projection(value: Mapping[str, object]) -> dict[str, object]:
    return {key: value.get(key) for key in _MODEL_ASSIGNMENT_KEYS}


def _model_projection_from_assignment(value: object) -> dict[str, object] | None:
    if not isinstance(value, Mapping) or not _MODEL_ASSIGNMENT_KEYS.issubset(value):
        return None
    return _model_projection(value)


def _invocation_manifest_identity_version(payload: Mapping[str, object]) -> int:
    schema = payload.get("schemaVersion")
    identity = payload.get("executionIdentityVersion")
    if schema == "2.0" and identity == 2:
        return 2
    if schema == "2.0" or identity is not None:
        raise ValueError("invocation manifest mixes v1 and v2 execution identity")
    if schema not in (None, 1, "1", "1.0"):
        raise ValueError("unsupported invocation manifest execution identity version")
    return 1


def _validate_invocation_assignment_shapes(
    assignments: Mapping[str, object],
) -> None:
    allowed = {frozenset(_MODEL_ASSIGNMENT_KEYS)}
    for item in assignments.values():
        if not isinstance(item, Mapping):
            raise ValueError("incomplete invocation manifest assignments")
        keys = set(item)
        if keys & _V2_INVOCATION_IDENTITY_KEYS:
            raise ValueError(
                "invocation assignment duplicates v2 execution identity"
            )
        if frozenset(keys) not in allowed:
            raise ValueError("incomplete invocation manifest assignments")


def render_timeline(timeline_path: str, ctx: dict) -> None:
    task_manifest_path = Path(ctx.get("TASK_MANIFEST_PATH", ""))
    task_manifest = {}
    if task_manifest_path.exists():
        try:
            task_manifest = load_owned_object(
                task_manifest_path, artifact="task manifest"
            )
        except Exception:
            task_manifest = {}
    reviewers = _resolve_workers(ctx)
    catalog = _worker_catalog(ctx)
    path = Path(timeline_path)
    existing = {}
    if path.exists():
        try:
            existing = load_owned_object(path, artifact="task timeline")
        except Exception:
            existing = {}
    runs = existing.get("runs", [])
    current_run_manifest_path = ctx.get("RUN_MANIFEST_PATH", "")
    current_run_manifest_relative_path = ctx.get("RUN_MANIFEST_RELATIVE_PATH", "")
    filtered = [
        item
        for item in runs
        if item.get("runManifestPath") != current_run_manifest_relative_path
        and item.get("runManifestPath") != current_run_manifest_path
    ]
    workflow = (
        task_manifest.get("workflow")
        if isinstance(task_manifest.get("workflow"), dict)
        else {}
    )
    workflow = workflow or {}
    entry = {
            "runTimestamp": ctx.get("RUN_TIMESTAMP_ISO", ""),
            "runDirectoryPath": ctx.get("RUN_DIR_RELATIVE_PATH", ""),
            "runManifestPath": ctx.get("RUN_MANIFEST_RELATIVE_PATH", ""),
            "runDateTimeSegment": ctx.get("RUN_DATETIME_SEGMENT", ""),
            "runSequencesByCategory": {
                "manifests": ctx.get("RUN_MANIFESTS_SEQ", ""),
                "prompts": ctx.get("RUN_PROMPTS_SEQ", ""),
                "reports": ctx.get("RUN_REPORTS_SEQ", ""),
                "status": ctx.get("RUN_STATUS_SEQ", ""),
                "state": ctx.get("RUN_STATE_SEQ", ""),
                "sessions": ctx.get("RUN_SESSIONS_SEQ", ""),
                "workerResults": ctx.get("WORKER_RESULTS_SEQ", ""),
            },
            "taskType": ctx.get("TASK_TYPE", ""),
            "workCategory": task_manifest.get(
                "workCategory", ctx.get("WORKFLOW_WORK_CATEGORY", "unknown")
            ),
            "status": ctx.get("CURRENT_RUN_STATUS", ""),
            "taskBriefPath": ctx.get("BRIEF_RELATIVE_PATH", ""),
            "promptSnapshotPath": ctx.get("RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""),
            "workerPromptDirectoryPath": ctx.get("RUN_PROMPTS_RELATIVE_PATH", ""),
            "workerPromptPathByWorkerId": {
                item: catalog[item]["promptPath"] for item in reviewers
            },
            "reportRecordPath": ctx.get("LATEST_REPORT_RECORD_RELATIVE_PATH")
            or ctx.get("FINAL_REPORT_RECORD_RELATIVE_PATH", ""),
            "teamStatePath": ctx.get("TEAM_STATE_RELATIVE_PATH", ""),
            "resumeCommandPath": ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", ""),
            "relatedTasks": json.loads(ctx.get("RELATED_TASKS_JSON", "[]")),
            "workflowSnapshot": {
                "phaseSequence": workflow.get("phaseSequence", []),
                "currentPhase": workflow.get("currentPhase", ""),
                "currentPhaseState": workflow.get("currentPhaseState", ""),
                "phaseStates": workflow.get("phaseStates", {}),
                "lastCompletedPhase": workflow.get("lastCompletedPhase", ""),
                "nextRecommendedPhase": next_phase.promote(
                    workflow.get("nextRecommendedPhase")
                ),
                "awaitingApproval": workflow.get("awaitingApproval", False),
                "lastSafeCheckpoint": workflow.get("lastSafeCheckpoint", {}),
            },
    }
    if ctx.get("FIX_CYCLE_ID"):
        entry["fixCycleId"] = ctx["FIX_CYCLE_ID"]
    filtered.append(entry)
    payload = {
        "schemaVersion": "1.0",
        "projectId": ctx.get("PROJECT_ID", ""),
        "taskGroup": ctx.get("TASK_GROUP", ""),
        "taskId": ctx.get("TASK_ID", ""),
        "taskKey": ctx.get("TASK_KEY", ""),
        "runs": filtered,
    }
    _write_json(path, payload)


def _fix_cycles_index_line(manifest: dict) -> str:
    fc = manifest.get("fixCycles") or {}
    if not fc.get("count"):
        return "none"
    latest = fc.get("latest") or {}
    state = (
        f"open: {fc.get('openCycleId')}" if fc.get("openCycleId") else "all closed"
    )
    return (
        f"{fc.get('count')} cycle(s), {state} — "
        f"latest `{latest.get('cycle', '')}`: {latest.get('symptom', '')}"
    )


def render_task_index(template_path: str, output_path: str, ctx: dict) -> None:
    template = Path(template_path).read_text(encoding="utf-8")
    task_manifest_path = Path(ctx["TASK_MANIFEST_PATH"])
    task_manifest = {}
    if task_manifest_path.exists():
        try:
            task_manifest = load_owned_object(
                task_manifest_path, artifact="task manifest"
            )
        except Exception:
            task_manifest = {}
    workflow = (
        task_manifest.get("workflow", {})
        if isinstance(task_manifest.get("workflow"), dict)
        else {}
    )
    phase_states = (
        workflow.get("phaseStates", {})
        if isinstance(workflow.get("phaseStates"), dict)
        else {}
    )
    phase_order = workflow.get("phaseSequence", [])
    if not isinstance(phase_order, list) or not phase_order:
        phase_order = list(PHASE_SEQUENCE)
    phase_state_lines = [
        f"- `{phase}`: `{phase_states.get(phase, 'not-started')}`"
        for phase in phase_order
    ]
    checkpoint = (
        workflow.get("lastSafeCheckpoint", {})
        if isinstance(workflow.get("lastSafeCheckpoint"), dict)
        else {}
    )
    checkpoint_lines = [
        f"- Label: `{checkpoint.get('label', 'unknown')}`",
        f"- Run manifest: `{checkpoint.get('latestRunManifestPath', ctx.get('RUN_MANIFEST_RELATIVE_PATH', ''))}`",
        f"- Team state: `{checkpoint.get('latestTeamStatePath', ctx.get('TEAM_STATE_RELATIVE_PATH', ''))}`",
        f"- Report: `{checkpoint.get('latestReportRecordPath', task_manifest.get('latestReportRecordPath', '--')) or '--'}`",
        f"- Resume command: `{checkpoint.get('latestResumeCommandPath', task_manifest.get('latestResumeCommandPath', '--')) or '--'}`",
    ]
    rc = (
        task_manifest.get("resultContract")
        if isinstance(task_manifest.get("resultContract"), dict)
        else {}
    )
    cv = (
        task_manifest.get("contractValidation")
        if isinstance(task_manifest.get("contractValidation"), dict)
        else {}
    )
    art = (
        task_manifest.get("artifacts")
        if isinstance(task_manifest.get("artifacts"), dict)
        else {}
    )
    lead_role = rc.get("leadRole", _lead_role(ctx))
    latest_resume_command = (
        task_manifest.get("latestResumeCommandPath")
        or ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", "")
        or "--"
    )
    model_assignment_lines = [
        f"- `{lead_role}`: `{rc.get('leadModel', ctx.get('LEAD_MODEL', ''))}`"
    ]
    for assignment in task_manifest.get("workerAssignments", []):
        if not isinstance(assignment, dict):
            continue
        worker_id = assignment.get("workerId", "")
        provider = assignment.get("provider", "")
        if worker_id == "report-writer":
            label = "Report writer worker"
        else:
            try:
                label = f"{provider_spec(provider).display_label} worker"
            except (UnknownProviderError, TypeError):
                label = f"{worker_id} worker"
        model_assignment_lines.append(
            f"- `{label}`: `{assignment.get('model', '')}`"
        )
    mapping = {
        "{{TASK_KEY}}": task_manifest.get("taskKey", ctx.get("TASK_KEY", "")),
        "{{TASK_TYPE}}": task_manifest.get("taskType", ctx.get("TASK_TYPE", "")),
        "{{TASK_DATE}}": ctx.get("TASK_DATE", ""),
        "{{PROJECT_ID}}": ctx.get("PROJECT_ID", ""),
        "{{TASK_GROUP}}": ctx.get("TASK_GROUP", ""),
        "{{TASK_ID}}": ctx.get("TASK_ID", ""),
        "{{CURRENT_TASK_STATUS}}": task_manifest.get(
            "currentStatus", ctx.get("CURRENT_TASK_STATUS", "")
        ),
        "{{CURRENT_RUN_STATUS}}": task_manifest.get(
            "latestRunStatus", ctx.get("CURRENT_RUN_STATUS", "")
        ),
        "{{RELATED_TASKS_INLINE}}": ctx.get("RELATED_TASKS_INLINE", "None"),
        "{{RECOMMENDED_ANALYSERS}}": ", ".join(
            task_manifest.get("recommendedWorkers", [])
        ),
        "{{LEAD_MODEL}}": rc.get("leadModel", ctx.get("LEAD_MODEL", "")),
        "{{OKSTRA_VERSION}}": ctx.get("OKSTRA_VERSION", ""),
        "{{LATEST_RUN_RELATIVE_PATH}}": task_manifest.get(
            "latestRunPath", ctx.get("LATEST_RUN_RELATIVE_PATH", "")
        ),
        "{{LATEST_REPORT_RECORD_RELATIVE_PATH}}": task_manifest.get(
            "latestReportRecordPath", ctx.get("LATEST_REPORT_RECORD_RELATIVE_PATH", "")
        ),
        "{{TEAM_STATE_RELATIVE_PATH}}": task_manifest.get(
            "teamStatePath", ctx.get("TEAM_STATE_RELATIVE_PATH", "")
        ),
        "{{VALIDATION_STATUS}}": cv.get(
            "status", ctx.get("VALIDATION_STATUS", "not-run")
        ),
        "{{CLAUDE_RESUME_COMMAND_RELATIVE_PATH}}": latest_resume_command,
        "{{MODEL_ASSIGNMENT_LINES}}": "\n".join(model_assignment_lines),
        "{{TASK_MANIFEST_RELATIVE_PATH}}": task_manifest.get(
            "taskManifestPath", ctx.get("TASK_MANIFEST_RELATIVE_PATH", "")
        ),
        "{{OKSTRA_LATEST_TASK_RELATIVE_PATH}}": ctx.get(
            "OKSTRA_LATEST_TASK_RELATIVE_PATH", ""
        ),
        "{{INSTRUCTION_SET_RELATIVE_PATH}}": task_manifest.get(
            "instructionSetPath", ctx.get("INSTRUCTION_SET_RELATIVE_PATH", "")
        ),
        "{{REFERENCE_EXPECTATIONS_RELATIVE_PATH}}": task_manifest.get(
            "referenceExpectationsPath",
            ctx.get("REFERENCE_EXPECTATIONS_RELATIVE_PATH", ""),
        ),
        "{{FINAL_REPORT_TEMPLATE_RELATIVE_PATH}}": art.get(
            "finalReportTemplatePath",
            ctx.get("FINAL_REPORT_TEMPLATE_RELATIVE_PATH", ""),
        ),
        "{{RUN_MANIFESTS_RELATIVE_PATH}}": ctx.get("RUN_MANIFESTS_RELATIVE_PATH", ""),
        "{{RUN_STATE_RELATIVE_PATH}}": ctx.get("RUN_STATE_RELATIVE_PATH", ""),
        "{{RUN_PROMPTS_RELATIVE_PATH}}": task_manifest.get(
            "latestRunPromptsPath", ctx.get("RUN_PROMPTS_RELATIVE_PATH", "")
        ),
        "{{RUN_REPORTS_RELATIVE_PATH}}": ctx.get("RUN_REPORTS_RELATIVE_PATH", ""),
        "{{RUN_STATUS_RELATIVE_PATH}}": ctx.get("RUN_STATUS_RELATIVE_PATH", ""),
        "{{RUN_SESSIONS_RELATIVE_PATH}}": ctx.get("RUN_SESSIONS_RELATIVE_PATH", ""),
        "{{WORKER_RESULTS_RELATIVE_PATH}}": art.get(
            "workerResultsDirectoryPath", ctx.get("WORKER_RESULTS_RELATIVE_PATH", "")
        ),
        "{{RUN_VALIDATOR_RELATIVE_PATH}}": cv.get(
            "validatorScriptPath", ctx.get("RUN_VALIDATOR_RELATIVE_PATH", "")
        ),
        "{{WORK_CATEGORY}}": task_manifest.get(
            "workCategory", ctx.get("WORKFLOW_WORK_CATEGORY", "unknown")
        ),
        "{{WORKFLOW_CURRENT_PHASE}}": workflow.get(
            "currentPhase", ctx.get("WORKFLOW_CURRENT_PHASE", "")
        ),
        "{{WORKFLOW_CURRENT_PHASE_STATE}}": workflow.get(
            "currentPhaseState", ctx.get("WORKFLOW_CURRENT_PHASE_STATE", "")
        ),
        "{{WORKFLOW_LAST_COMPLETED_PHASE}}": workflow.get(
            "lastCompletedPhase", ctx.get("WORKFLOW_LAST_COMPLETED_PHASE", "")
        )
        or "--",
        # 토큰은 사람이 읽는 한 줄이라 문자열이어야 한다. 포인터의 `status` 는
        # 같은 페이지의 다른 줄이 아니라 소비자 로직이 읽는다.
        "{{WORKFLOW_NEXT_RECOMMENDED_PHASE}}": next_phase.promote(
            workflow.get("nextRecommendedPhase")
        )["phase"]
        or "--",
        "{{WORKFLOW_AWAITING_APPROVAL}}": "yes"
        if workflow.get("awaitingApproval", False)
        else "no",
        "{{WORKFLOW_PHASE_STATE_LINES}}": "\n".join(phase_state_lines),
        "{{WORKFLOW_LAST_SAFE_CHECKPOINT_LINES}}": "\n".join(checkpoint_lines),
        "{{FIX_CYCLES_SUMMARY}}": _fix_cycles_index_line(task_manifest),
    }
    fm_ctx = dict(ctx)
    fm_ctx.setdefault("DOC_TYPE", _doc_type_from_template_path(template_path))
    mapping.update(_frontmatter_mapping(fm_ctx))
    rendered = template
    for k, v in mapping.items():
        rendered = rendered.replace(k, v)
    rendered = _strip_phase_blocks(rendered, ctx.get("TASK_TYPE", ""))
    _write_text(Path(output_path), rendered.rstrip() + "\n")


# --------------------------------------------------------------------------- #
# Available MCP servers block
# --------------------------------------------------------------------------- #


_NO_MCP_SERVERS_LINE = (
    f"- No MCP servers are declared in `{OKSTRA_DIR_NAME}/project.json`'s "
    "`mcpServers` array. Treat MCP tools as unavailable for this run. To enable "
    "them, add entries shaped `{name, description, tools, notes?}` to that array "
    "and re-render the bundle."
)


def build_available_mcp_servers_block(project_root: Path) -> str:
    """Render the `## Available MCP Servers` first bullet from project.json.

    The MCP server list used to be hardcoded for one specific environment.
    It now comes from the project's okstra project.json (`mcpServers` array),
    so each user/project declares the MCP surface available to their
    lead+workers. Missing file or empty array yields a generic "none declared"
    fallback.
    """
    config_path = project_json_path(project_root)
    try:
        raw = load_owned_object(config_path, artifact="project config")
    except JsonBoundaryError:
        return _NO_MCP_SERVERS_LINE
    servers = raw.get("mcpServers") if isinstance(raw, dict) else None
    if not isinstance(servers, list) or not servers:
        return _NO_MCP_SERVERS_LINE
    lines: list[str] = []
    for entry in servers:
        if not isinstance(entry, dict):
            continue
        name = str(entry.get("name", "")).strip()
        if not name:
            continue
        description = str(entry.get("description", "")).strip()
        tools = entry.get("tools") or []
        notes = str(entry.get("notes", "")).strip()
        parts = [f"`mcp__{name}`"]
        if description:
            parts.append(description)
        if isinstance(tools, list) and tools:
            tool_names = ", ".join(
                f"`{str(t).strip()}`" for t in tools if str(t).strip()
            )
            if tool_names:
                parts.append(f"Tools: {tool_names}")
        if notes:
            parts.append(notes)
        lines.append("- " + ". ".join(parts) + ".")
    return "\n".join(lines) if lines else _NO_MCP_SERVERS_LINE


# --------------------------------------------------------------------------- #
# launch.template.md rendering
# --------------------------------------------------------------------------- #


def sanitize_team_name(base: str, suffix: str = "") -> str:
    """`base`(+`suffix`)를 teamName 라벨 / Agent name 패턴에 맞춘다.

    Agent name·teamName 은 `^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$` 를 따른다(콜론
    불가·최대 64자). 그런데 okstra 의 base 는 `okstra-<TASK_KEY>` 이고 TASK_KEY 는
    `project:group:task-id` 라 콜론을 포함하며 64자를 쉽게 넘는다. 불허문자를 '-' 로
    치환하고, 초과 시 stage 격리용 `suffix`(`-s<N>` / `-fv-s<N>`)는 보존하면서 원본
    해시 8자로 절단해 유일성을 유지한다(suffix 만으로 stage 를 구분하는 호출부 계약).
    """
    safe = re.sub(r"[^A-Za-z0-9_-]+", "-", base).strip("-")
    limit = 64 - len(suffix)
    if len(safe) > limit:
        digest = hashlib.sha1(base.encode("utf-8")).hexdigest()[:8]
        safe = safe[: max(0, limit - 9)].rstrip("-") + "-" + digest
    return safe + suffix


def inject_lead_prompt_computed_tokens(ctx: dict) -> None:
    """Populate ctx in-place with derived lead-prompt tokens.

    Tokens that are not 1:1 with a ctx key (TEAM_CREATION_GATE, lead identity
    blocks, worker dispatch guidance, worker result/status summaries, etc.) are
    computed deterministically from ctx so the pure-lookup renderer
    (`render_template_with_ctx`) can resolve them via plain `ctx[token]` lookup.

    Always overwrites — caller-supplied values for these computed keys are replaced
    on every call. For optional defaults (VALIDATION_STATUS etc.) use the
    companion `apply_lead_prompt_defaults` which preserves caller values.
    """
    selected = _resolve_workers(ctx)
    catalog = _worker_catalog(ctx)
    lead_runtime = _lead_runtime(ctx)
    worker_dispatch_plan = _worker_dispatch_plan(ctx)
    lead_role = _lead_role(ctx)
    lead_agent_label = _lead_agent_label(ctx)
    lead_model = ctx.get("LEAD_MODEL", "")
    lead_model_execution = ctx.get("LEAD_MODEL_EXECUTION_VALUE", "")
    home_prompts = okstra_home() / "prompts"
    lead_contract_path = ctx.get("OKSTRA_LEAD_CONTRACT_PATH") or str(
        home_prompts / "lead" / "okstra-lead-contract.md"
    )
    adapter_contract_path = ctx.get("OKSTRA_LEAD_ADAPTER_CONTRACT_PATH") or str(
        home_prompts / _lead_adapter_contract(ctx, worker_dispatch_plan)
    )
    context_loader_path = ctx.get("OKSTRA_CONTEXT_LOADER_PATH") or str(
        home_prompts / "lead" / "context-loader.md"
    )
    team_contract_path = ctx.get("OKSTRA_TEAM_CONTRACT_PATH") or str(
        home_prompts / "lead" / "team-contract.md"
    )
    convergence_path = ctx.get("OKSTRA_CONVERGENCE_PATH") or str(
        home_prompts / "lead" / "convergence.md"
    )
    plan_body_verification_path = ctx.get("OKSTRA_PLAN_BODY_VERIFICATION_PATH") or str(
        home_prompts / "lead" / "plan-body-verification.md"
    )
    report_writer_path = ctx.get("OKSTRA_REPORT_WRITER_PATH") or str(
        home_prompts / "lead" / "report-writer.md"
    )
    coding_preflight_dir = ctx.get("OKSTRA_CODING_PREFLIGHT_DIR") or str(
        home_prompts / "coding-preflight"
    )
    runtime_templates = okstra_home() / "templates"
    analysis_preamble = ctx.get("ANALYSIS_WORKER_PREAMBLE_PATH") or str(
        runtime_templates / "worker-prompt-preamble.md"
    )
    implementation_preamble = ctx.get("IMPLEMENTATION_WORKER_PREAMBLE_PATH") or str(
        runtime_templates / "implementation-worker-preamble.md"
    )
    report_writer_preamble = ctx.get("REPORT_WRITER_PREAMBLE_PATH") or str(
        runtime_templates / "report-writer-prompt-preamble.md"
    )
    worker_error_contract = ctx.get("WORKER_ERROR_CONTRACT_PATH") or str(
        runtime_templates / "worker-error-contract.md"
    )
    okstra_runtime_resources_block = (
        "## Okstra Runtime Resources\n"
        "\n"
        f"- Lifecycle core contract: `{lead_contract_path}`\n"
        f"- Selected runtime adapter contract: `{adapter_contract_path}`\n"
        "- Read the lifecycle core first, then the selected adapter. "
        "Only the selected adapter may define host tool spelling.\n"
        f"- Context loader: `{context_loader_path}`\n"
        f"- Team contract: `{team_contract_path}`\n"
        f"- Convergence contract: `{convergence_path}`\n"
        f"- Plan-body verification contract (implementation-planning Phase 6 sub-step only): `{plan_body_verification_path}`\n"
        f"- Report writer contract: `{report_writer_path}`\n"
        f"- Coding preflight pack: `{coding_preflight_dir}`\n"
        f"- Analysis worker preamble: `{analysis_preamble}`\n"
        f"- Implementation worker preamble: `{implementation_preamble}`\n"
        f"- Report writer preamble: `{report_writer_preamble}`\n"
        f"- Shared worker error contract: `{worker_error_contract}`"
    )

    def fmt_assignment(role: str, model: str, execution: str) -> str:
        if execution and execution != model:
            return f"- `{role}`: `{model}` (launch value: `{execution}`)"
        return f"- `{role}`: `{model}`"

    worker_result_lines: list[str] = []
    team_role_lines = [f"  1. `{lead_role}` (assigned model: `{lead_model}`)"]
    model_assignment_lines = [
        fmt_assignment(lead_role, lead_model, lead_model_execution)
    ]
    worker_role_labels: list[str] = []
    execution_status_entries = [f"`{lead_role}`"]
    execution_status_table_lines = [
        "| 에이전트 | 역할 | 모델 | 상태 | 핵심 발견 요약 |",
        "|----------|------|------|------|----------------|",
        f"| {lead_agent_label} | {lead_role} | {lead_model} | completed / timeout / error / not-run | 최종 synthesis 작성 상태와 핵심 판단 |",
    ]
    for index, worker in enumerate(selected, start=2):
        m = catalog[worker]
        worker_result_lines.append(
            f"- {m['role']} result path: `{m['resultPath']}` (assigned model: `{m['model']}`)"
        )
        team_role_lines.append(
            f"  {index}. `{m['role']}` (assigned model: `{m['model']}`)"
        )
        model_assignment_lines.append(
            fmt_assignment(m["role"], m["model"], m["modelExecutionValue"])
        )
        worker_role_labels.append(f"`{m['role']}`")
        execution_status_entries.append(f"`{m['role']}`")
        execution_status_table_lines.append(
            f"| {m['agentLabel']} | {m['role']} | {m['model']} | completed / timeout / error / not-run | {m['role']}의 핵심 발견 요약 |"
        )

    if worker_role_labels:
        if len(worker_role_labels) == 1:
            worker_role_sentence = (
                f"- {worker_role_labels[0]} is the required worker role."
            )
        else:
            worker_role_sentence = (
                f"- {', '.join(worker_role_labels[:-1])}, "
                f"and {worker_role_labels[-1]} are the required worker roles."
            )
        preferred_results_sentence = (
            f"- Aim to collect completed results from all "
            f"{len(worker_role_labels)} required workers."
        )
    else:
        worker_role_sentence = "- No worker roles were selected for this run."
        preferred_results_sentence = "- No worker results are expected for this run."
    worker_attempt_sentence = (
        "- `Antigravity worker` is mandatory to attempt for this workflow."
        if "antigravity" in selected
        else "- `Antigravity worker` is not selected for this run, so no Antigravity attempt is required."
    )

    task_type = ctx.get("TASK_TYPE", "")
    adapter_setup_facts: list[str] = []
    if lead_runtime == "claude-code":
        base_name = f'okstra-{ctx.get("TASK_KEY", "")}'
        impl_stage = str(ctx.get("EFFECTIVE_STAGES", "") or "").strip()
        fv_stage = str(ctx.get("RUN_STAGE", "") or "").strip()
        stage_suffix = ""
        if task_type == "implementation" and impl_stage:
            stage_suffix = f"-s{impl_stage}"
        elif task_type == "final-verification" and fv_stage:
            stage_suffix = f"-fv-s{fv_stage}"
        adapter_setup_facts.append(
            f"- Adapter audit label: `{sanitize_team_name(base_name, stage_suffix)}`"
        )
    concurrent_stages = str(ctx.get("CONCURRENT_RUN_STAGES", "") or "").strip()
    if concurrent_stages:
        adapter_setup_facts.append(
            f"- Prepare-recorded concurrent stages: `{concurrent_stages}`"
        )
    adapter_setup_block = (
        "\n".join(adapter_setup_facts)
        if adapter_setup_facts
        else "- Adapter setup facts: none"
    )

    if task_type == "release-handoff" or not selected:
        team_creation_gate_block = (
            "## Single-Lead Phase (no worker dispatch)\n"
            "\n"
            "This run has no worker roster or convergence loop. The lead performs "
            "the phase inline and follows the selected adapter only for user prompts, "
            "artifact operations, usage collection, and cleanup."
        )
    else:
        team_creation_gate_block = (
            "## Runtime Adapter Dispatch Gate (BLOCKING)\n"
            "\n"
            f"- Selected lead runtime: `{lead_runtime}`\n"
            f"- Selected adapter contract: `{adapter_contract_path}`\n"
            "- Read that adapter before Phase 3. Execute `prompt_user`, "
            "`dispatch_worker`, `await_workers`, `redispatch_worker`, "
            "`shutdown_workers`, `record_lead_event`, and `collect_usage` only "
            "through its mapping.\n"
            "- Do not call a primitive documented by an unselected adapter.\n"
            "- Record the adapter-required Phase 3 state and emit the canonical "
            "`PROGRESS: phase-3-team-create ...` checkpoint before dispatch.\n"
            f"{adapter_setup_block}\n"
            "- The run manifest is authoritative for concurrent-run metadata, "
            "dispatch backend, and artifact paths."
        )

    lead_intro_line = f"You are `{lead_role}` for project `{ctx.get('PROJECT_ID', '')}`."
    lead_bootstrap_instruction = (
        f"Read the Lifecycle core contract at `{lead_contract_path}`, then read the "
        f"selected runtime adapter at `{adapter_contract_path}`. Read the manifests "
        "below for all task metadata, paths, model assignments, and worker roster. "
        "Execute `write_artifact`, `dispatch_worker`, and `await_workers` only through "
        "the selected adapter's mapping. Follow the lifecycle core's lazy-read rules "
        "for support resources; do not invoke hidden/internal skills."
    )
    worker_dispatch_guidance = (
        "## Worker Dispatch Contract\n"
        "\n"
        "- Worker prompt paths in `task-manifest.json` are assigned prompt-history "
        "locations, not pre-rendered completion markers. An absent file means its "
        "dispatch history has not been materialized yet.\n"
        "- Code-backed `dispatch_worker` mappings leave missing roster prompts to "
        "`okstra_ctl.initial_prompt_materialization` and its canonical "
        "`materialize_initial_prompts()` entry point. The selected adapter supplies "
        "its declared delivery mode; do not construct or overwrite those prompts "
        "manually.\n"
        "- For native in-process dispatch, use the same canonical initial-prompt headers, "
        "persist the assigned prompt history through `write_artifact`, and apply "
        "the adapter-declared `lazy-path-reference` delivery mode before starting the "
        "worker. This native path does not apply to reverify or critic dispatches.\n"
        "- Use `await_workers` through the selected adapter and confirm terminal state "
        "plus every required Result Path. File presence alone is never a signal to "
        "skip; worker selection and skipped status come from `team-state`.\n"
        "- Resolve the worker's PromptPlan audience and inject exactly one "
        "`**Worker Preamble Path:**` from the audience map: "
        f"analysis=`{analysis_preamble}`, implementation executor/verifier="
        f"`{implementation_preamble}`, report-writer=`{report_writer_preamble}`.\n"
        "- Every initial worker prompt also injects "
        f"`**Worker Error Contract Path:** {worker_error_contract}`. The selected "
        "preamble owns audience procedure; the shared error contract owns error "
        "schema and write rules. Do not re-inline either contract."
    )

    lead_descriptor = _lead_descriptor(ctx)
    if lead_descriptor.has_claude_session:
        lead_session_block = (
            "## Session\n"
            "\n"
            f"- Session ID: `{ctx.get('CLAUDE_SESSION_ID', '')}`\n"
            f"- Resume: `{ctx.get('CLAUDE_RESUME_COMMAND_RELATIVE_PATH', '')}`"
        )
    else:
        lead_session_block = (
            "## Session\n"
            "\n"
            f"- Lead runtime: `{lead_descriptor.id}`\n"
            f"- Lead session accounting: {lead_descriptor.session_accounting}\n"
            "- Resume behavior is provided by the selected host lead-session port."
        )

    # Compute results (deterministic from ctx, 덮어쓰기)
    ctx["LEAD_INTRO_LINE"] = lead_intro_line
    ctx["LEAD_BOOTSTRAP_INSTRUCTION"] = lead_bootstrap_instruction
    ctx["LEAD_SESSION_BLOCK"] = lead_session_block
    ctx["TEAM_CREATION_GATE"] = team_creation_gate_block
    ctx["WORKER_DISPATCH_GUIDANCE"] = worker_dispatch_guidance
    ctx["OKSTRA_RUNTIME_RESOURCES"] = okstra_runtime_resources_block
    ctx["WORKER_RESULT_PATH_LINES"] = "\n".join(worker_result_lines)
    ctx["MODEL_ASSIGNMENT_LINES"] = "\n".join(model_assignment_lines)
    ctx["TEAM_ROLE_LINES"] = "\n".join(team_role_lines)
    ctx["REQUIRED_WORKER_ROLE_SENTENCE"] = worker_role_sentence
    ctx["ANTIGRAVITY_ATTEMPT_SENTENCE"] = worker_attempt_sentence
    ctx["PREFERRED_WORKER_RESULTS_SENTENCE"] = preferred_results_sentence
    ctx["EXECUTION_STATUS_EXACT_ENTRIES"] = ", ".join(execution_status_entries)
    ctx["EXECUTION_STATUS_TABLE_ROWS"] = "\n".join(execution_status_table_lines)


def apply_lead_prompt_defaults(ctx: dict) -> None:
    """Apply default values for optional lead-prompt ctx fields.

    Sets the optional tokens that the lead prompt template references but
    which callers may legitimately leave unset (e.g., no validation has run
    yet, no related tasks were declared, the run is not an implementation
    batch). Caller-supplied values are preserved via `setdefault` / `if-not-in`
    semantics — this function only fills gaps, never overwrites.

    Companion to `inject_lead_prompt_computed_tokens` (which always
    overwrites with deterministically-derived values). The two functions
    are kept separate so each has a single clear responsibility:
    inject = compute-and-overwrite, apply_defaults = fill-if-missing.
    """
    ctx.setdefault("VALIDATION_STATUS", "not-run")
    ctx.setdefault("RELATED_TASKS_BULLETS", "- None recorded")
    ctx.setdefault("RELATED_TASKS_INLINE", "None")
    # Empty for non-implementation runs; the implementation prepare path
    # overwrites it with the resolved stage-batch directive.
    ctx.setdefault("STAGE_BATCH_DIRECTIVE", "")
    # Empty for non-final-verification runs; the final-verification prepare
    # path overwrites it with the resolved verification target block.
    ctx.setdefault("VERIFICATION_TARGET", "")
    # Empty except for whole-task final-verification, where the prepare path
    # overwrites it with the stage auto-integration summary.
    ctx.setdefault("STAGE_INTEGRATION", "")
    runtime_home = okstra_home()
    ctx.setdefault(
        "ANALYSIS_WORKER_PREAMBLE_PATH",
        str(runtime_home / "templates" / "worker-prompt-preamble.md"),
    )
    ctx.setdefault(
        "IMPLEMENTATION_WORKER_PREAMBLE_PATH",
        str(runtime_home / "templates" / "implementation-worker-preamble.md"),
    )
    ctx.setdefault(
        "REPORT_WRITER_PREAMBLE_PATH",
        str(runtime_home / "templates" / "report-writer-prompt-preamble.md"),
    )
    ctx.setdefault(
        "WORKER_ERROR_CONTRACT_PATH",
        str(runtime_home / "templates" / "worker-error-contract.md"),
    )
    # Lead resource paths the launch template references directly. paths.py
    # seeds the production values; setdefault backfills callers that render the
    # template without the full path ctx.
    ctx.setdefault(
        "OKSTRA_LEAD_CONTRACT_PATH",
        str(runtime_home / "prompts" / "lead" / "okstra-lead-contract.md"),
    )
    ctx.setdefault(
        "OKSTRA_TEAM_CONTRACT_PATH",
        str(runtime_home / "prompts" / "lead" / "team-contract.md"),
    )
    if "AVAILABLE_MCP_SERVERS" not in ctx:
        ctx["AVAILABLE_MCP_SERVERS"] = build_available_mcp_servers_block(
            Path(ctx.get("PROJECT_ROOT", "."))
        )


_TOKEN_RE = re.compile(r"\{\{([A-Z][A-Z0-9_]*)\}\}")


def render_template_with_ctx(template_path: str, output_path: str, ctx: dict) -> None:
    """Render a `{{TOKEN}}` template with pure ctx[token] lookup.

    - Tokens match regex `_TOKEN_RE` (uppercase snake).
    - Each token MUST exist in ctx. Missing → `TokenRenderError` (fail-fast).
    - Phase block stripping (`{% if header.taskType == 'X' %} ... {% endif %}`)
      is applied per `ctx['TASK_TYPE']`.
    - Frontmatter mapping (`_frontmatter_mapping`) is overlaid (same as legacy
      renderer).

    Callers that need computed tokens (team_creation_gate etc.) MUST call
    `inject_lead_prompt_computed_tokens(ctx)` BEFORE invoking this function.
    Optional defaults (VALIDATION_STATUS etc.) should be filled by calling
    `apply_lead_prompt_defaults(ctx)` in the same setup step.
    """
    template = Path(template_path).read_text(encoding="utf-8")

    fm_ctx = dict(ctx)
    fm_ctx.setdefault("DOC_TYPE", _doc_type_from_template_path(template_path))
    fm_overlay = _frontmatter_mapping(fm_ctx)   # {"{{DOC_TITLE}}": "...", ...}

    # frontmatter overlay 가 채우는 키들도 lookup 대상 — 단일 lookup 으로 통일
    lookup: dict[str, str] = {}
    for tok_with_braces, value in fm_overlay.items():
        key = tok_with_braces[2:-2]   # "{{X}}" -> "X"
        lookup[key] = value

    missing: list[str] = []

    def _resolve(match: "re.Match[str]") -> str:
        token = match.group(1)
        if token in lookup:
            return lookup[token]
        if token in ctx:
            return str(ctx[token])
        missing.append(token)
        return match.group(0)

    # 단일 패스 치환: 토큰을 _TOKEN_RE 매칭으로 한 번만 훑어 치환값으로 바꾼다.
    # 치환값 안에 `{{TOKEN}}`/`{% if %}` 같은 마커가 들어 있어도 이미 소비된
    # 구간이라 재치환되지 않는다 (데이터를 통한 토큰/마커 주입 방지).
    rendered = _TOKEN_RE.sub(_resolve, template)

    if missing:
        names = ", ".join(sorted(set(missing)))
        raise TokenRenderError(
            f"undefined lead-prompt token(s): {names} (template={template_path}). "
            f"Add the key(s) to ctx in run.py / "
            f"inject_lead_prompt_computed_tokens() / apply_lead_prompt_defaults()."
        )

    rendered = _strip_phase_blocks(rendered, ctx.get("TASK_TYPE", ""))
    _write_text(Path(output_path), rendered.rstrip() + "\n")


# --------------------------------------------------------------------------- #
# CLI dispatcher
# --------------------------------------------------------------------------- #


def main(argv: list[str]) -> int:
    if not argv:
        print(
            "usage: python3 -m okstra_ctl.render <subcommand> ...\n"
            "  (requires PYTHONPATH=$(okstra paths --field python); "
            "normal callers go through scripts/okstra.sh instead)",
            file=sys.stderr,
        )
        return 2
    sub = argv[0]
    rest = argv[1:]
    try:
        if sub == "team-state":
            ctx_path, team_state_path = rest
            render_team_state(team_state_path, _load_ctx(ctx_path))
        elif sub == "reference-expectations":
            ctx_path, brief_path, output_path = rest
            render_reference_expectations(brief_path, output_path, _load_ctx(ctx_path))
        elif sub == "task-catalog-discovery":
            ctx_path, output_path = rest
            render_task_catalog_discovery(output_path, _load_ctx(ctx_path))
        elif sub == "latest-task-discovery":
            ctx_path, output_path = rest
            render_latest_task_discovery(output_path, _load_ctx(ctx_path))
        elif sub == "migrate-legacy":
            (ctx_path,) = rest
            migrate_legacy_run_artifacts(_load_ctx(ctx_path))
        elif sub == "task-manifest":
            ctx_path, manifest_path = rest
            render_task_manifest(manifest_path, _load_ctx(ctx_path))
        elif sub == "run-manifest":
            ctx_path, run_manifest_path = rest
            render_run_manifest(run_manifest_path, _load_ctx(ctx_path))
        elif sub == "timeline":
            ctx_path, timeline_path = rest
            render_timeline(timeline_path, _load_ctx(ctx_path))
        elif sub == "task-index":
            ctx_path, template_path, output_path = rest
            render_task_index(template_path, output_path, _load_ctx(ctx_path))
        elif sub == "template":
            ctx_path, template_path, output_path = rest
            ctx = _load_ctx(ctx_path)
            inject_lead_prompt_computed_tokens(ctx)
            apply_lead_prompt_defaults(ctx)
            render_template_with_ctx(template_path, output_path, ctx)
        else:
            print(f"unknown subcommand: {sub}", file=sys.stderr)
            return 2
    except Exception as exc:
        print(f"render {sub} failed: {exc}", file=sys.stderr)
        return 1
    return 0


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