"""역할별 리포트 입력을 검증해 계약 3.0 정본을 한 번 게시한다."""
from __future__ import annotations

import copy
import json
import os
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Mapping, Sequence

from .agent.activity import agent_activity_rows
from .clarification_items import (
    clarification_disposition,
    incorporated_clarification_ids,
    row_blocks_progress,
)
from .final_report_schema import load_schema_version, validate
from .report_inputs import ReportInputPath, report_input_paths, uses_report_contract_v3
from .json_boundary import JsonBoundaryError, load_owned_object, serialize_owned_object
from .report_narrative import parse_narrative, task_narrative_errors
from .report_synthesis_packet import (
    report_synthesis_packet_paths,
    verify_report_synthesis_packet_sources,
)
from .conformance import strip_stage_declaration_label
from .design_prep import DesignPrepError, materialize_design_prep_requests
from .implementation_options import validate_blocked_answer_channel
from .technical_verification import validate_technical_verification_report
from .implementation_direction import (
    load_selected_direction_snapshot,
    validate_selected_direction_plan,
)
from .paths import task_dir
from .report_projections import (
    ReportProjectionError,
    project_convergence,
    project_design,
    project_execution,
    project_token_usage,
)
from .scope_provenance import parse_source
from .verification_target import read_verification_target
from .write_policy import planned_path_declaration_errors


@dataclass(frozen=True)
class AssemblyIssue:
    owner: str
    artifact_path: str
    field_path: str
    reason: str


class ReportAssemblyError(ValueError):
    """게시 전 입력 검증 실패와 실제 소유자를 전달한다."""

    def __init__(self, issues: AssemblyIssue | Sequence[AssemblyIssue]) -> None:
        rows = (issues,) if isinstance(issues, AssemblyIssue) else tuple(issues)
        if not rows:
            raise ValueError("report assembly error requires at least one issue")
        self.issues = rows
        self.issue = rows[0]
        super().__init__("; ".join(_issue_message(issue) for issue in rows))


def _issue_message(issue: AssemblyIssue) -> str:
    return (
        f"owner={issue.owner} artifactPath={issue.artifact_path} "
        f"fieldPath={issue.field_path} reason={issue.reason}"
    )


def _fail(owner: str, path: Path, field: str, reason: object) -> None:
    raise ReportAssemblyError(AssemblyIssue(owner, str(path), field, str(reason)))


def _read_json(row: ReportInputPath) -> dict[str, Any]:
    if not row.path.is_file():
        _fail(row.owner, row.path, "$", "required input is missing")
    try:
        value = load_owned_object(row.path, artifact=f"{row.owner} report input")
    except JsonBoundaryError as exc:
        _fail(row.owner, row.path, "$", exc)
    if not isinstance(value, dict):
        _fail(row.owner, row.path, "$", "input must be a JSON object")
    return value


def _input_map(project_root: Path, manifest: Mapping[str, Any]) -> dict[str, ReportInputPath]:
    try:
        return {row.key: row for row in report_input_paths(project_root, manifest)}
    except ValueError as exc:
        _fail("orchestrator", project_root, "manifest", exc)


def _input_preflight_issues(
    inputs: Mapping[str, ReportInputPath],
    schema: Mapping[str, Any],
    task_type: str,
) -> tuple[AssemblyIssue, ...]:
    issues: list[AssemblyIssue] = []
    for row in inputs.values():
        if not row.path.is_file():
            issues.append(
                AssemblyIssue(
                    row.owner,
                    str(row.path),
                    "$",
                    "required input is missing",
                )
            )
            continue
        try:
            if row.key == "narrative":
                narrative = parse_narrative(row.path.read_text(encoding="utf-8"), schema)
                issues.extend(
                    AssemblyIssue(
                        row.owner, str(row.path), error.split(":", 1)[0], error,
                    )
                    for error in task_narrative_errors(narrative, schema, task_type)
                )
            elif row.key == "agent-activity":
                row.path.read_text(encoding="utf-8")
            else:
                value = load_owned_object(
                    row.path,
                    artifact=f"{row.owner} report input",
                )
                if not isinstance(value, dict):
                    raise ValueError("input must be a JSON object")
        except (JsonBoundaryError, OSError, UnicodeError, ValueError) as exc:
            issues.append(AssemblyIssue(row.owner, str(row.path), "$", str(exc)))
    return tuple(issues)


def _narrative(row: ReportInputPath, schema: Mapping[str, Any]) -> dict[str, Any]:
    if not row.path.is_file():
        _fail(row.owner, row.path, "$", "required input is missing")
    try:
        return parse_narrative(row.path.read_text(encoding="utf-8"), schema)
    except (OSError, UnicodeError, ValueError) as exc:
        _fail(row.owner, row.path, "$", exc)


def _clarifications(
    ledger: Mapping[str, Any], activities: list[dict[str, Any]], path: Path,
) -> list[dict[str, Any]]:
    if ledger.get("owner") != "lead":
        _fail("lead", path, "owner", "approval ledger owner must be lead")
    active = ledger.get("activeClarifications")
    if not isinstance(active, list):
        _fail("lead", path, "activeClarifications", "must be an array")
    for index, row in enumerate(active):
        if not isinstance(row, Mapping):
            _fail("lead", path, f"activeClarifications[{index}]", "must be an object")
    activity_by_id = {row.get("activityId"): row for row in activities}
    rows = [
        _clarification_row(row, activity_by_id, path)
        for row in active
    ]
    rows.extend(_carried_clarification_rows(ledger, {row.get("id") for row in rows}, path))
    return rows


def _carried_clarification_rows(
    ledger: Mapping[str, Any], seen_ids: set[object], path: Path,
) -> list[dict[str, Any]]:
    """이월 결정은 active 질문이 아니다. 이번 런 활동 원장을 요구하지 않는다."""
    carried = ledger.get("carriedDecisions")
    if not isinstance(carried, list):
        return []
    rows: list[dict[str, Any]] = []
    seen = set(seen_ids)
    for index, entry in enumerate(carried):
        if not isinstance(entry, Mapping):
            _fail("lead", path, f"carriedDecisions[{index}]", "must be an object")
        decision = entry.get("decision")
        if not isinstance(decision, Mapping):
            _fail(
                "lead", path, f"carriedDecisions[{index}].decision",
                "must be an object",
            )
        cid = decision.get("id")
        if cid in seen:
            continue
        rows.append(_carried_clarification_row(decision))
        seen.add(cid)
    return rows


def _carried_clarification_row(source: Mapping[str, Any]) -> dict[str, Any]:
    row = {
        key: source[key]
        for key in (
            "id", "ticketId", "kind", "statement", "expectedForm", "blocks",
            "origin", "userConfirmation", "userInput", "options",
        )
        if key in source
    }
    if "approval" in source:
        row["approvalContext"] = source["approval"]
    resolution = source.get("resolutionInput")
    if not isinstance(resolution, Mapping):
        # `resolved` 가 아니라 `answered` 다. 스키마는 `resolved` 에 `resolution`
        # 을 요구하고 그 `checkRefs` 는 **이번 run** 의 활동 행(`A-NNN`)인데,
        # 이월된 답에는 이번 run 의 활동이 없다 — `resolved` 로 쓰면 통과 가능한
        # 값이 없는 행이 된다. 답이 있었다는 사실은 `userInput` 이 나른다.
        row["status"] = "answered"
        return _neutralize_carried_approval_block(row)
    row.update(
        status="resolved",
        userInput=resolution.get("userText", ""),
        resolution={
            "disposition": resolution.get("disposition"),
            "userText": resolution.get("userText"),
            "checkRefs": list(resolution.get("checkRefs") or []),
        },
    )
    return _neutralize_carried_approval_block(row)


def _neutralize_carried_approval_block(row: dict[str, Any]) -> dict[str, Any]:
    """이월 행은 이미 반영된 답이다. Blocks=approval 을 그대로 올리면
    역추적이 없는 승인 차단이 다시 생긴다."""
    if row.get("blocks") == "approval":
        row["blocks"] = "none"
        row.pop("approvalContext", None)
    return row


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


def _clarification_response_value(
    project_root: Path,
    manifest: Mapping[str, Any],
) -> str:
    direct = manifest.get("clarificationResponsePath")
    if isinstance(direct, str) and direct.strip():
        return direct.strip()
    context_value = manifest.get("activeRunContextPath")
    if isinstance(context_value, str) and context_value.strip():
        context_path = Path(context_value)
        if not context_path.is_absolute():
            context_path = project_root / context_path
        context = _read_json(
            ReportInputPath("active-run-context", "orchestrator", context_path)
        )
        instruction_set = context.get("instructionSet")
        value = (
            instruction_set.get("clarificationResponsePath")
            if isinstance(instruction_set, Mapping)
            else ""
        )
        if isinstance(value, str) and value.strip():
            return value.strip()
    return _latest_user_response_value(project_root, manifest)


def _latest_user_response_value(
    project_root: Path,
    manifest: Mapping[str, Any],
) -> str:
    record_value = manifest.get("expectedReportRecordPath")
    if not isinstance(record_value, str) or not record_value.strip():
        return ""
    record_path = Path(record_value)
    if not record_path.is_absolute():
        record_path = project_root / record_path
    if record_path.parent.name != "reports":
        return ""
    task_type = str(manifest.get("taskType") or "").strip()
    if not task_type:
        return ""
    responses = sorted(
        path
        for path in (record_path.parent.parent / "user-responses").glob(
            f"user-response-{task_type}-*.md"
        )
        if path.is_file()
    )
    if not responses:
        return ""
    return _project_relative(project_root, responses[-1])


def _run_inputs_clarification_response(manifest_path: Path) -> str:
    """이 run 을 띄운 carry-in 경로를 per-run `run-inputs-*.json` 에서 읽는다.

    같은 사용자 입력을 두 소비자가 서로 다른 파일에서 찾고 있었다.
    `validate-run` 의 `_carry_in_source_for_run` 은 run-inputs 를 보며 그
    독스트링이 "run-inputs 기록이 유일한 per-run 증거"라고 적어 두는데, 조립은
    매니페스트와 active-run-context 만 봤다. 두 곳 다 비어 있는 phase 에서는
    조립이 `clarificationCarryIn` 을 만들지 않고 검증기는 없다고 실패시킨다 —
    실측(2026-08-26, `fontsninja-nlpvibe` final-verification): 매니페스트에
    키 없음, active-run-context 의 `instructionSet` 이 `null`, run-inputs 에는
    값 있음. carry-in 을 붙여 실행한 run 이 구조적으로 실패했다. 그 필드는
    조립 소유라 작성자가 교정 라운드로도 못 넘는다.
    """
    name = manifest_path.name
    if not name.startswith("run-manifest-") or not name.endswith(".json"):
        return ""
    inputs_path = manifest_path.with_name(
        name.replace("run-manifest-", "run-inputs-", 1)
    )
    if not inputs_path.is_file():
        return ""
    try:
        payload = load_owned_object(inputs_path, artifact="run inputs")
    except (JsonBoundaryError, OSError):
        return ""
    inputs = payload.get("inputs") if isinstance(payload, Mapping) else None
    if not isinstance(inputs, Mapping):
        return ""
    # 새 계획은 후보비교 레코드를, 구현은 계획 레코드를 물려받는다 — 답변은 그
    # 레코드의 행이다. 첫 키만 보던 동안 두 phase 의 `clarificationCarryIn` 은
    # 비었고, 열람본의 C-NNN 은 이전 run 의 페이지로 이어지지 못했다.
    for key in ("clarificationResponsePath", "selectedDirectionPath", "approvedPlanPath"):
        value = inputs.get(key)
        if isinstance(value, str) and value.strip():
            return value.strip()
    return ""


def _verification_scope(project_root: Path, manifest: Mapping[str, Any]) -> str:
    """이 run 이 검증한 범위 — 준비된 target 스냅샷이 정본이다.

    이 필드는 읽는 쪽만 있고 쓰는 쪽이 없었다. `handoff record-verified` 와
    `report_finalize` 의 teardown 분기가 읽고, `validate-run` 은 target 과
    대조한다. 그런데 서술문 스키마에 속성이 없어 작성자가 저작할 수 없고
    조립도 만들지 않았다 — 그래서 값은 언제나 비었고, 세 소비자가 전부 조용히
    엉뚱하게 동작했다: `record-verified` 는 항상 거부하고, teardown 은 항상
    건너뛰고, target 대조는 `actual` 이 비어 아예 비교되지 않았다. 실측
    (2026-08-26, `fontsninja-nlpvibe` final-verification): 리포트는 수락인데
    `consumers.jsonl` 의 verified 행은 0건인 채로 run 이 `passed` 로 닫혔다.

    작성자의 것이 아니라는 것은 프로필이 이미 못박고 있다 — "`verificationScope`
    in particular gates both stage-group eligibility and release-handoff routing,
    so it is not the report's to restate". 그래서 조립이 target 에서 읽어 싣는다.
    다이제스트가 안 맞는 스냅샷은 `read_verification_target` 이 ``None`` 을
    돌려주므로 그때는 값을 만들지 않는다 — 변조된 근거로 라우팅을 여는 것보다
    비어 있는 편이 낫다.
    """
    # run 매니페스트는 target 경로를 최상위 `verificationTargetPath` 로 싣는다
    # (render.py). 종전에는 `manifest["instructionSet"]` 아래를 읽었는데 그
    # 블록은 active-run-context 에만 있어 값이 언제나 비었다 — 실측(2026-09-06,
    # fontsninja-v3-site final-verification 001): 리포트에 `verificationScope`
    # 가 없어 Phase 7 teardown 이 "single-stage" 로 건너뛰고 `record-verified`
    # 도 거부 대상이었다.
    relative = str(manifest.get("verificationTargetPath") or "").strip()
    if not relative:
        return ""
    target = read_verification_target(project_root, relative)
    return str((target or {}).get("scope") or "").strip()


def _clarification_carry_in(
    project_root: Path,
    manifest: Mapping[str, Any],
    manifest_path: Path,
) -> dict[str, str] | None:
    value = _clarification_response_value(project_root, manifest) or (
        _run_inputs_clarification_response(manifest_path)
    )
    if not value:
        return None
    path = Path(value)
    if not path.is_absolute():
        path = project_root / path
    if not path.is_file():
        _fail("user", path, "$", "attached clarification response is missing")
    return {"sourceFile": _project_relative(project_root, path)}


def _clarification_row(
    source: Mapping[str, Any], activities: Mapping[object, Mapping[str, Any]], path: Path,
) -> dict[str, Any]:
    row = {
        key: source[key]
        for key in (
            "id", "ticketId", "kind", "statement", "expectedForm", "blocks",
            "origin", "userConfirmation", "options",
        )
        if key in source
    }
    if "approval" in source:
        row["approvalContext"] = source["approval"]
    resolution = source.get("resolutionInput")
    if not isinstance(resolution, Mapping):
        row["status"] = "open"
        return row
    clarification_id = source.get("id")
    check_refs = resolution.get("checkRefs")
    if not isinstance(check_refs, list) or not check_refs:
        _fail("lead", path, f"{clarification_id}.resolutionInput.checkRefs", "must not be empty")
    for activity_id in check_refs:
        activity = activities.get(activity_id)
        if activity is None:
            _fail("lead", path, f"{clarification_id}.resolutionInput.checkRefs", f"unknown activity {activity_id}")
        if clarification_id not in (activity.get("clarificationRefs") or []):
            _fail("activity-ledger", path, str(activity_id), f"missing clarificationRefs {clarification_id}")
    row.update(
        status="resolved",
        userInput=resolution.get("userText", ""),
        resolution={
            "disposition": resolution.get("disposition"),
            "userText": resolution.get("userText"),
            "checkRefs": list(check_refs),
        },
    )
    return row


def _attach_plan_backlinks(data: dict[str, Any], activities: list[dict[str, Any]]) -> None:
    planning = data.get("implementationPlanning")
    if not isinstance(planning, dict):
        return
    verification = planning.get("planBodyVerification")
    if not isinstance(verification, dict):
        return
    refs_by_item: dict[str, set[str]] = {}
    for activity in activities:
        for item_id in activity.get("planItemIds") or []:
            refs_by_item.setdefault(str(item_id), set()).update(
                str(ref) for ref in activity.get("clarificationRefs") or []
            )
    for item in verification.get("planItems") or []:
        if isinstance(item, dict):
            item["clarificationRefs"] = sorted(refs_by_item.get(str(item.get("id")), set()))


def _compose(
    project_root: Path,
    manifest_path: Path,
    manifest: Mapping[str, Any],
    inputs: Mapping[str, ReportInputPath],
    schema: Mapping[str, Any],
) -> dict[str, Any]:
    narrative = _narrative(inputs["narrative"], schema)
    ledger = _read_json(inputs["approval-decisions"])
    team_state = _read_json(inputs["execution-status"])
    convergence = _read_json(inputs["convergence"])
    if not inputs["agent-activity"].path.is_file():
        _fail("activity-ledger", inputs["agent-activity"].path, "$", "required input is missing")
    try:
        activities = list(agent_activity_rows(project_root, manifest_path))
    except ValueError as exc:
        _fail("activity-ledger", inputs["agent-activity"].path, "$", exc)
    try:
        convergence_data = project_convergence(convergence)
    except ReportProjectionError as exc:
        _fail("convergence", inputs["convergence"].path, "$", exc)
    try:
        execution = project_execution(manifest, team_state)
        usage = project_token_usage(team_state)
    except (TypeError, ValueError) as exc:
        _fail("team-state", inputs["execution-status"].path, "$", exc)
    data = {"schemaVersion": "3.0", **narrative, **execution}
    data["tokenUsage"] = usage
    data["crossVerification"] = convergence_data["crossVerification"]
    data["clarificationItems"] = _clarifications(ledger, activities, inputs["approval-decisions"].path)
    # 작성자의 사실 분류와 리드 원장의 질문 연결은 양쪽 입력을 합친 뒤 검사한다.
    for reason in validate_blocked_answer_channel(data):
        _fail(
            "lead",
            inputs["approval-decisions"].path,
            "activeClarifications",
            reason,
        )
    carry_in = _clarification_carry_in(project_root, manifest, manifest_path)
    if carry_in is not None:
        data["clarificationCarryIn"] = carry_in
    scope = _verification_scope(project_root, manifest)
    if scope:
        data["verificationScope"] = scope
    data["agentActivity"] = activities
    data["evidence"] = {"primary": _promoted_evidence(convergence_data)}
    data["missingInformation"] = []
    # 스냅샷 복사가 _attach_metadata 보다 먼저다 — frontmatter 가
    # selectedDirectionRef.snapshotPath 를 읽는다.
    _apply_selected_direction_snapshot(data, project_root, manifest)
    _attach_metadata(data, manifest)
    _attach_machine_planning(data, inputs)
    _attach_plan_backlinks(data, activities)
    _fill_end_state_coverage(data)
    return data


def _selected_direction_inputs(
    project_root: Path, manifest: Mapping[str, Any]
) -> tuple[Path, Path]:
    """구현 진입(run.py `_validate_selected_implementation_plan`)과 같은 경로."""
    _project_id, task_group, task_id = _identity(manifest)
    instruction_set = task_dir(project_root, task_group, task_id) / "instruction-set"
    return (
        instruction_set / "selected-direction.json",
        instruction_set / "task-brief.md",
    )


def _apply_selected_direction_snapshot(
    data: dict[str, Any], project_root: Path, manifest: Mapping[str, Any]
) -> None:
    """selected-direction 계획의 기계 복사 필드를 스냅샷 원문으로 채운다.

    `directionRealization` 의 4필드는 `!=` 완전 일치가 계약인데
    (implementation_direction `_direction_realization_errors`) writer 가
    의역하는 실측 실패가 있었다(dev-10341, 4필드 전부). 판단이 없는 복사는
    조립 소유다 — `_attach_machine_planning` 이 design-prep 투영을 소유하는
    것과 같은 근거. `conformanceTests` 의 `stage-<N> — ` 라벨 잔류도 같은
    계급이라 여기서 뗀다.
    """
    planning = data.get("implementationPlanning")
    if (
        not isinstance(planning, dict)
        or planning.get("planningContract") != "selected-direction"
    ):
        return
    snapshot_path, _brief_path = _selected_direction_inputs(project_root, manifest)
    snapshot, load_failures = load_selected_direction_snapshot(snapshot_path)
    if snapshot is None:
        _fail(
            "orchestrator", snapshot_path,
            "implementationPlanning.selectedDirectionRef",
            "; ".join(load_failures),
        )
    # selectedDirectionRef 5필드도 전부 기계 유도값이다 — sourceReport /
    # sourceDataSha256 / optionId 는 스냅샷 본문에, 경로·해시는 스냅샷
    # 실물에 있다. run 001 실측에서 writer 가 snapshotPath 를 틀렸다.
    planning["selectedDirectionRef"] = {
        "sourceReport": snapshot.data.get("sourceReport"),
        "sourceDataSha256": snapshot.data.get("sourceDataSha256"),
        "optionId": snapshot.data.get("optionId"),
        "snapshotPath": snapshot.relative_path,
        "snapshotSha256": snapshot.sha256,
    }
    realization = planning.get("directionRealization")
    direction = snapshot.data.get("direction")
    if isinstance(realization, dict) and isinstance(direction, Mapping):
        realization["coreMechanism"] = direction.get("coreMechanism")
        realization["architectureBoundaries"] = direction.get(
            "architectureBoundaries"
        )
        realization["planningInvariants"] = snapshot.data.get("planningInvariants")
        realization["userConstraints"] = snapshot.data.get("userConstraints")
    for stage in planning.get("stages") or []:
        if not isinstance(stage, dict):
            continue
        declared = stage.get("conformanceTests")
        number = stage.get("stage")
        if (
            isinstance(declared, str)
            and isinstance(number, int)
            and not isinstance(number, bool)
            and number >= 1
        ):
            stage["conformanceTests"] = strip_stage_declaration_label(
                number, declared
            )


def selected_direction_plan_errors(
    data: Mapping[str, Any], project_root: Path, manifest: Mapping[str, Any]
) -> list[str]:
    """게시 직전에 구현 진입 검증을 그대로 돌린다.

    같은 검증이 말미 validate-run 에서는 advisory 로만 나와, 통과 발행된
    계획이 구현 진입(run.py:1445)에서 처음 하드 거부되는 실측 wedge 가
    있었다(dev-10341, 17건). 게시 시점 실패는 writer/lead 가 run 안에서
    고칠 수 있다.
    """
    planning = data.get("implementationPlanning")
    if (
        not isinstance(planning, Mapping)
        or planning.get("planningContract") != "selected-direction"
    ):
        return []
    snapshot_path, brief_path = _selected_direction_inputs(project_root, manifest)
    return validate_selected_direction_plan(data, brief_path, snapshot_path)


def validate_plan_draft(
    data: Mapping[str, Any], project_root: Path, manifest: Mapping[str, Any]
) -> list[str]:
    """작성자 입력을 게시하지 않고 같은 기계 투영과 의미 검사로 검증한다."""
    draft = copy.deepcopy(dict(data))
    _apply_selected_direction_snapshot(draft, project_root, manifest)
    _attach_metadata(draft, manifest)
    return [
        *task_narrative_errors(
            draft, load_schema_version("3.0"), str(manifest.get("taskType", "")),
        ),
        *planned_path_declaration_errors(draft),
        *[
            f"implementationPlanning: {error}"
            for error in selected_direction_plan_errors(draft, project_root, manifest)
        ],
    ]


def _identity(manifest: Mapping[str, Any]) -> tuple[str, str, str]:
    task_key = str(manifest.get("taskKey") or "")
    parts = task_key.split(":")
    if len(parts) >= 3:
        return parts[0], parts[-2], parts[-1]
    return (
        str(manifest.get("projectId") or "unknown"),
        str(manifest.get("taskGroup") or "unknown"),
        str(manifest.get("taskId") or task_key or "unknown"),
    )


def _attach_metadata(data: dict[str, Any], manifest: Mapping[str, Any]) -> None:
    project_id, task_group, task_id = _identity(manifest)
    task_type = str(manifest.get("taskType") or "")
    task_key = str(manifest.get("taskKey") or "")
    created = str(manifest.get("runTimestamp") or manifest.get("createdAt") or "unknown")
    data["meta"] = {"reportLanguage": str(manifest.get("reportLanguage") or "en")}
    clarifications = data.get("clarificationItems") or []
    incorporated = incorporated_clarification_ids(data)
    blocked = any(
        isinstance(row, Mapping)
        and row_blocks_progress(
            str(row.get("status") or ""),
            clarification_disposition(row),
            incorporated=str(row.get("id") or "") in incorporated,
        )
        for row in clarifications
    )
    frontmatter = {
        "title": f"OKSTRA Final Report - {task_key}",
        "id": task_key.replace(":", "-"),
        "tags": ["final-report"],
        "status": "in-progress" if blocked else "completed",
        "aliases": [f"{task_id}-{task_type}"],
        "date": created,
        "taskId": task_id,
        "taskGroup": task_group,
        "projectId": project_id,
        "taskType": task_type,
        "workerId": "report-writer",
        "approved": False,
    }
    planning = data.get("implementationPlanning")
    if isinstance(planning, Mapping) and planning.get("planningContract") == "selected-direction":
        selected = planning.get("selectedDirectionRef")
        frontmatter["selectedDirectionRef"] = str(
            selected.get("snapshotPath") if isinstance(selected, Mapping) else ""
        )
    else:
        frontmatter["implementationOption"] = ""
    data["frontmatter"] = frontmatter
    data["header"] = {
        "taskKey": task_key,
        "createdAt": created,
        "taskType": task_type,
        "reportOwner": "Okstra lead",
        "reportAuthor": "Report writer worker",
        "leadModel": str(manifest.get("leadModel") or "unknown"),
        "okstraVersion": str(manifest.get("okstraVersion") or "unknown"),
    }


def _promoted_evidence(convergence: Mapping[str, Any]) -> list[dict[str, Any]]:
    return [
        {
            "id": f"E-{index:03d}",
            "ticketId": row["ticketId"],
            "evidence": row["evidence"],
            "sourceItems": row["sourceItems"],
            "source": "convergence",
        }
        for index, row in enumerate(convergence.get("promotedEvidence") or [], start=1)
    ]


def _attach_machine_planning(
    data: dict[str, Any], inputs: Mapping[str, ReportInputPath],
) -> None:
    planning = data.get("implementationPlanning")
    if not isinstance(planning, dict):
        return
    design_row = inputs.get("design-preparation")
    if design_row is None:
        return
    try:
        design = project_design(planning, _read_json(design_row))
    except ReportProjectionError as exc:
        _fail(design_row.owner, design_row.path, "$", exc)
    planning["designPreparation"] = design["designPreparation"]
    stages = {row.get("stage"): row for row in planning.get("stages") or [] if isinstance(row, dict)}
    for coverage in design["stageCoverage"]:
        stage = stages.get(coverage.get("stage"))
        if stage is not None:
            stage["designSurfaceCoverage"] = coverage.get("rows") or []
    verification_row = inputs.get("plan-body-verification")
    if verification_row is None:
        return
    verification = _read_json(verification_row)
    if verification.get("owner") != "convergence":
        _fail(
            verification_row.owner, verification_row.path, "owner",
            "plan-body verification owner must be convergence",
        )
    projection = verification.get("planBodyVerification")
    if not isinstance(projection, Mapping):
        _fail(
            verification_row.owner, verification_row.path,
            "planBodyVerification", "must be an object",
        )
    planning["planBodyVerification"] = dict(projection)
    _sync_human_summary_with_plan_body_gate(data, projection)


def _sync_human_summary_with_plan_body_gate(
    data: dict[str, Any], projection: Mapping[str, Any],
) -> None:
    """게이트가 통과했는데 내러티브가 unseeded 로 남으면 사람용 칸을 맞춘다.

    report-writer 는 PBV 전에 쓰고, 조립이 게이트를 덮어씌운다. blockers 와
    nextStep 이 그때의 문장을 그대로 두면 통과한 계획이 아직 검증 전으로 보인다.
    """
    gate = str(projection.get("gateResult") or "")
    if gate not in {"passed", "passed-with-dissent"}:
        return
    summary = data.get("humanSummary")
    if isinstance(summary, dict) and isinstance(summary.get("blockers"), list):
        summary["blockers"] = [
            row for row in summary["blockers"]
            if "unseeded" not in str(row).lower()
        ]
    card = data.get("verdictCard")
    if not isinstance(card, dict):
        return
    next_step = str(card.get("nextStep") or "")
    lowered = next_step.lower()
    if "unseeded" not in lowered and "seed" not in lowered:
        return
    next_step = (
        "Plan-body verification passed. Do not start implementation "
        "until `frontmatter.approved` is true."
    )
    card["nextStep"] = next_step
    final = data.get("finalVerdict")
    if isinstance(final, dict):
        final["nextStep"] = next_step


def _fill_end_state_coverage(data: dict[str, Any]) -> None:
    """writer 가 못 쓴 endStateCoverage 를 requirementCoverage brief:EB-001 에서 채운다.

    내러티브 스키마가 이 칸을 열어 주면 writer 가 쓸 수 있다. 비어 있으면
    validate-run 이 brief id 마다 한 행을 요구하므로 조립이 채운다.
    """
    existing = data.get("endStateCoverage")
    if isinstance(existing, list) and existing:
        return
    planning = data.get("implementationPlanning")
    if not isinstance(planning, dict):
        return
    derived = _end_state_rows_from_requirement_coverage(planning)
    if derived:
        data["endStateCoverage"] = derived


def _end_state_rows_from_requirement_coverage(
    planning: Mapping[str, Any],
) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    seen: set[str] = set()
    for item in planning.get("requirementCoverage") or []:
        if not isinstance(item, Mapping):
            continue
        ref = parse_source(str(item.get("source") or ""))
        if not ref.is_end_state_id or ref.value in seen:
            continue
        seen.add(ref.value)
        status = str(item.get("status") or "")
        row: dict[str, Any] = {
            "id": ref.value,
            "disposition": "addressed" if status == "covered" else "deferred",
            "coveredBy": str(item.get("id") or item.get("coveredBy") or ""),
        }
        if row["disposition"] != "addressed":
            row["rationale"] = (
                f"requirementCoverage `{item.get('id')}` status is `{status or 'empty'}`."
            )
        rows.append(row)
    return rows


def _schema_owner(error: str) -> str:
    field = error.split(" ", 1)[0]
    owners = {
        "executionStatus": "team-state",
        "tokenUsage": "team-state",
        "crossVerification": "convergence",
        "clarificationItems": "lead",
        "agentActivity": "activity-ledger",
        "executionRoles": "run-manifest",
        "frontmatter": "orchestrator",
        "header": "orchestrator",
        "meta": "orchestrator",
    }
    if "designPreparation" in error or "designSurfaceCoverage" in error:
        return "design-surface-detector"
    if "planBodyVerification" in error:
        return "convergence"
    if "selectedDirectionRef" in error:
        return "report-writer"
    return owners.get(field.split(".", 1)[0], "report-writer")


def _publish(
    target: Path, data: Mapping[str, Any], replace: Callable[[str, str], None],
) -> None:
    target.parent.mkdir(parents=True, exist_ok=True)
    serialized = serialize_owned_object(target, data, artifact="final report record")
    descriptor, name = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            handle.write(serialized)
            handle.flush()
            os.fsync(handle.fileno())
        replace(name, str(target))
    finally:
        if os.path.exists(name):
            os.unlink(name)


def _materialize_planning_requests(target: Path) -> None:
    """게시한 계획 기록에서 Okstra 소유 design-prep 요청을 만든다.

    계약 3.0 은 token-usage 에서 `--substitute-data` 를 빼므로, v2 가
    populate_token_cells 안에서 하던 생성을 조립이 맡는다.
    """
    if not target.name.startswith("final-report-implementation-planning-"):
        return
    try:
        materialize_design_prep_requests(target)
    except DesignPrepError as exc:
        _fail("orchestrator", target, "designPreparation", exc)


def assemble_report(
    project_root: Path, manifest_path: Path, *,
    replace: Callable[[str, str], None] = os.replace,
) -> dict[str, Any]:
    """계약 3.0 정본을 조립해 게시한다.

    입력이 없어 조립 자체가 안 되면 기존 파일을 건드리지 않는다. 조립은
    됐는데 스키마만 깨진 기록은 게시한 뒤 거부한다. 게시를 건너뛰면
    `validate-run` 이 이번 런 내용을 보지 못한다.
    """
    manifest_row = ReportInputPath("run-manifest", "orchestrator", manifest_path)
    manifest = _read_json(manifest_row)
    if not uses_report_contract_v3(manifest):
        _fail("orchestrator", manifest_path, "reportContractVersion", "requires 3.0")
    schema = load_schema_version("3.0")
    inputs = _input_map(project_root, manifest)
    input_issues = list(_input_preflight_issues(inputs, schema, str(manifest.get("taskType", ""))))
    packet_data_path, _ = report_synthesis_packet_paths(inputs["narrative"].path)
    if packet_data_path.is_file():
        input_issues.extend(
            AssemblyIssue(
                issue.owner,
                str(issue.path),
                f"sources.{issue.label}",
                issue.reason,
            )
            for issue in verify_report_synthesis_packet_sources(
                project_root,
                packet_data_path,
            )
        )
    if input_issues:
        raise ReportAssemblyError(tuple(input_issues))
    data = _compose(project_root, manifest_path, manifest, inputs, schema)
    errors = validate(data, schema)
    path_errors = planned_path_declaration_errors(data)
    if path_errors:
        raise ReportAssemblyError(tuple(
            AssemblyIssue("report-writer", str(inputs["narrative"].path), error.split(":", 1)[0], error)
            for error in path_errors
        ))
    direction_errors = selected_direction_plan_errors(
        data, project_root, manifest
    )
    value = manifest.get("expectedReportRecordPath")
    if not isinstance(value, str) or not value:
        _fail("orchestrator", manifest_path, "expectedReportRecordPath", "required")
    target = Path(value)
    target = target if target.is_absolute() else project_root / target
    verification_errors = validate_technical_verification_report(data, target, project_root)
    if verification_errors:
        raise ReportAssemblyError(tuple(
            AssemblyIssue("report-writer", str(target), "technicalVerification", error)
            for error in verification_errors
        ))
    _publish(target, data, replace)
    if errors or direction_errors:
        raise ReportAssemblyError(
            tuple(
                AssemblyIssue(
                    _schema_owner(error),
                    str(target),
                    error.split(" ", 1)[0],
                    error,
                )
                for error in errors
            )
            + tuple(
                AssemblyIssue(
                    "report-writer", str(target), "implementationPlanning", error
                )
                for error in direction_errors
            )
        )
    _materialize_planning_requests(target)
    return data