"""보고서 작성자에게 전달할 동결된 합성 입력 묶음."""
from __future__ import annotations

import copy
import json
import os
import re
import tempfile
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping

from okstra_token_usage.blocks import accounting_workers

from .analysis_packet import reference_source_extracts
from .json_boundary import JsonBoundaryError, load_owned_object, write_owned_object_atomic
from .implementation_options import (
    CANDIDATE_COMPARISON_ROUTING,
    EVALUATION_CRITERIA,
    MAX_CRITERION_VALUE,
    MAX_RANKED_OPTIONS,
    MAX_RAW_CANDIDATES_PER_ANALYSER,
    MIN_CRITERION_VALUE,
    MIN_FEASIBLE_VOTES,
    NO_VALID_OPTIONS_ROUTING,
)
from .report_narrative import writer_owned_data, writer_owned_schema
from .schema_excerpt import build_schema_excerpt
from .scope_provenance import brief_end_state_id_sequence

from .exact_coverage import COVERAGE_VERDICT_PRECEDENCE
from .final_report_schema import follow_up_task_rules, task_block_rules, verdict_token_rule
from .report_contract import TASK_TYPE_DATA_PROPERTY
from .report_markdown import humanise
from .report_narrative import NarrativeContractError, allowed_top_level_fields
from .report_narrative import NARRATIVE_GRAMMAR_INSTRUCTIONS


@dataclass(frozen=True)
class ReportSynthesisPacketIssue:
    owner: str
    label: str
    path: Path
    reason: str


class ReportSynthesisPacketError(ValueError):
    def __init__(self, issues: tuple[ReportSynthesisPacketIssue, ...]) -> None:
        self.issues = issues
        super().__init__("; ".join(issue.reason for issue in issues))


@dataclass(frozen=True)
class ReportSynthesisSource:
    label: str
    owner: str
    path: str
    content: str

    def to_dict(self) -> dict[str, str]:
        return {
            "label": self.label,
            "owner": self.owner,
            "path": self.path,
            "content": self.content,
        }


@dataclass(frozen=True)
class ReportSynthesisPacket:
    task_key: str
    task_type: str
    result_path: str
    sources: tuple[ReportSynthesisSource, ...]
    accounting_snapshot: dict[str, Any]
    original_requirement_ids: tuple[str, ...] = ()
    participating_analysers: tuple[str, ...] = ()
    # 증분 판정. 저작 계약이 이번 run 의 stage 번호를 말할 수 있는 유일한 출처다.
    incremental_decision: dict[str, Any] | None = None
    # 넘겨받은 완성 리포트 스키마에서 뽑은 값. 작성자는 스키마 가지를 읽지 않으므로
    # 여기 적어 줘야 도달한다(2026-09-02 실측: `Human Summary` 절 누락, `Verdict
    # Token` 에 `analysis-complete` — 둘 다 HTML 렌더에서야 거절됐다).
    required_top_level: tuple[str, ...] = ()
    verdict_tokens: tuple[str, ...] = ()
    # 이 task type 의 데이터 블록과, 스키마가 그 블록 안쪽에 못 박은 모양(객체당
    # 한 줄). 최상위 필드까지만 싣던 동안 리드는 하위 구조의 필수 필드·식별자
    # 패턴을 스키마 JSON 에서 손으로 뽑았고, 놓친 만큼 리포트를 다시 썼다
    # (2026-09-03 실측: implementation-option-selection 네 회차).
    block_key: str = ""
    block_rules: tuple[str, ...] = ()
    # 비종결 task type 의 `followUpTasks` 행 규칙(최소 행 수, phase-continuation
    # 행, 그 행의 autoSpawn). 스키마의 if/then 가지라 작성자에게 도달하지 않았고
    # 빈 배열이 조립에서야 거절됐다(2026-09-09 실측, dev-10642).
    follow_up_rules: tuple[str, ...] = ()
    # 작성자가 최상위에 쓸 수 있는 필드 전체(서사 스키마의 properties). 필수만
    # 적던 동안 리드가 어느 task type 에도 없는 절을 지시했고, 작성자는 조립이
    # 거절할 때까지 그 지시를 거를 근거가 없었다(2026-09-03 실측).
    allowed_top_level: tuple[str, ...] = ()

    def _carry_instructions(self) -> list[str]:
        """이번 run 의 이월 지시. 증분이 아니면 빈 목록.

        일반 규칙으로 두면 계획 분석자 프로필의 한 줄이 되고, 그 줄은 이번
        run 의 `carry_stages` 값을 말하지 못한다. 번호를 박아 넣는 자리는
        판정을 아는 여기뿐이다.
        """
        decision = self.incremental_decision
        if not decision:
            return []
        carried = ", ".join(str(stage) for stage in decision.get("carryStages", []))
        reverify = ", ".join(str(stage) for stage in decision.get("reverifyStages", []))
        return [
            "This run is an incremental re-plan. Record the decision verbatim as "
            "`implementationPlanning.incrementalDecision` with keys mode, "
            "reverifyStages, carryStages, reason.",
            f"Copy stage rows {carried} from the \"{CARRIED_STAGES_LABEL}\" source "
            f"into `implementationPlanning.stages` unchanged - every field, byte for "
            f"byte. Author only stage(s) {reverify}. `okstra incremental-carry` "
            f"rejects a carried stage that changed or is missing.",
        ]

    def _schema_instructions(self) -> list[str]:
        """스키마가 이 task type 에 못 박은 값을 저작 계약 문장으로 옮긴다."""
        lines: list[str] = []
        if self.required_top_level:
            labels = ", ".join(f"`{label}`" for label in self.required_top_level)
            lines.append(
                f"Required top-level fields for this task type: {labels}. "
                "Report assembly refuses a narrative that omits any of them."
            )
        if len(self.verdict_tokens) == 1:
            lines.append(
                f"`Final Verdict` -> `Verdict Token`: write exactly "
                f"`{self.verdict_tokens[0]}`; the schema pins it for task type "
                f"`{self.task_type}` and refuses every other value."
            )
        elif self.verdict_tokens:
            allowed = ", ".join(f"`{token}`" for token in self.verdict_tokens)
            lines.append(
                f"`Final Verdict` -> `Verdict Token`: one of {allowed} for task type "
                f"`{self.task_type}`."
            )
        if self.allowed_top_level:
            labels = ", ".join(f"`{label}`" for label in self.allowed_top_level)
            lines.append(
                f"Writer-owned top-level fields, the complete set a narrative may "
                f"contain: {labels}. Report assembly refuses any other top-level "
                "field, whichever instruction asked for it."
            )
        lines.extend(self.follow_up_rules)
        if self.block_rules:
            lines.append(
                f"Shape of the `{self.block_key}` block, one line per object, from "
                "the frozen schema. Paths are schema keys: write each key as its "
                "Title Case label (`rankedOptions` -> `Ranked Options`), and `[]` "
                "marks the items of an `- Item N` list. Report assembly refuses a "
                "narrative that breaks any line."
            )
            lines.extend(self.block_rules)
        return lines

    def _carry_validation_rules(self) -> list[str]:
        return ["carried-stages-unchanged"] if self.incremental_decision else []

    def _task_instructions(self) -> list[str]:
        if self.task_type == "technical-verification":
            return [
                "Copy sourceReport, sourceDataSha256, scope and every fact identity "
                "from the frozen technical input embedded in the Analysis packet.",
                "Inspect the run-local plans and command logs cited by settled worker "
                "results. These cited experiment artifacts are part of this task's "
                "evidence scope. Record one check per fact and retain limitations.",
                "Route only to implementation-option-selection with one matching "
                "phase-continuation. Evidence does not approve adoption or replace "
                "independent candidate feasibility votes.",
            ]
        if self.task_type == "implementation-planning":
            return self._planning_requirement_instructions()
        if self.task_type != "implementation-option-selection":
            return []
        criteria = ", ".join(f"`{item}`" for item in EVALUATION_CRITERIA)
        requirement_ids = ", ".join(
            f"`{item}`" for item in self.original_requirement_ids
        )
        analysers = ", ".join(
            f"`{item}`" for item in self.participating_analysers
        )
        return [
            f"Original requirement ids, in order: {requirement_ids}.",
            f"Participating analysers: {analysers}.",
            f"Evaluation criteria, in exact order: {criteria}.",
            "Every `rankedOptions` and `candidateAudit` row must contain one "
            "`requirementCoverage` row per original requirement id and one "
            "`criterionScores` row per evaluation criterion. Weights and scores "
            f"are integers from {MIN_CRITERION_VALUE} through {MAX_CRITERION_VALUE}; "
            "`weightedScore` is the recalculated weighted mean.",
            "`proposedBy` names exactly one participating analyser id, never a "
            "list. Counting `rankedOptions` and `candidateAudit` together, each "
            f"analyser may propose at most {MAX_RAW_CANDIDATES_PER_ANALYSER} raw "
            "candidates and the run may hold at most "
            f"{MAX_RAW_CANDIDATES_PER_ANALYSER} times the analyser count.",
            "`coverageSummary` is recalculated from the row's `requirementCoverage` "
            "statuses and `scopeCommitments`, and every field must equal the "
            "recalculation: `totalCount` = number of original requirement ids; "
            "`coveredCount` = rows with status `covered`; `coveragePercent` = "
            "round(coveredCount / totalCount * 100, 2); `unmappedCommitments` = "
            "commitment ids whose `requirementIds` is empty or names an id outside "
            "the original set; `scopePrecisionPercent` = round((commitments - "
            "unmapped) / commitments * 100, 2); `contradictedRequirements` = "
            "original ids with status `contradicted`, in original order; "
            "`coverageVerdict` = the first that applies of "
            + ", ".join(f"`{verdict}`" for verdict in COVERAGE_VERDICT_PRECEDENCE)
            + " (any contradicted; coveredCount below totalCount; any unmapped; "
            "otherwise).",
            "A ranked option is valid only when every participating analyser "
            "supplied one feasibility vote, at least "
            f"{MIN_FEASIBLE_VOTES} votes are `feasible`, and both `safetyBlockers` "
            "and `unresolvedFeasibilityFacts` are empty.",
            ("Classify each `unresolvedFeasibilityFacts` entry with `resolutionKind`: "
            "`user-decision` requires nonempty `clarificationRefs` linking that fact "
            "to actual C-NNN records, including answered records; `technical-verification` "
            "requires no user question. Preserve the fact, whyItMatters, and evidence. "
            "Historical unclassified facts remain readable but require classification "
            "before blocked report reassembly. Saving does not lift blocked routing."),
            ("After a linked decision is answered, preserve its disposition and selected "
            "value. Update humanSummary.actions, verdictCard.nextStep, and selection "
            "guidance to name remaining technical verification separately from unanswered "
            "user decisions; never reopen an answered question to permit saving."),
            "Each `feasibilityVotes` row states that analyser's own verdict, "
            "rationale, and strongest counterevidence as its result gives them. "
            "Two non-`uncertain` votes with identical rationale and "
            "counterevidence are rejected as one worker's text copied under "
            "another's name.",
            f"Rank at most {MAX_RANKED_OPTIONS} valid options by descending "
            "`requirement-fit`, `correctness-risk`, `architecture-fit`, "
            "`weightedScore`, then id. With valid options set routing to "
            f"`{CANDIDATE_COMPARISON_ROUTING}`; without a valid option use "
            f"`{NO_VALID_OPTIONS_ROUTING}`.",
            "In `candidate-comparison`, keep `preselectedDirection` null and do "
            "not route directly to `implementation-planning`. In "
            "`preselected-validation`, emit exactly one validated option, an empty "
            "`candidateAudit`, a cited `preselectedDirection`, and routing "
            "`implementation-planning`.",
        ]

    def _planning_requirement_instructions(self) -> list[str]:
        """계획 저작 계약에 브리프 요구사항 id 순서를 박아 넣는다.

        `validate_selected_direction_plan` 은 `requirementCoverage` 행 순서가
        taskBriefPath 의 end-state id 순서와 정확히 한 번씩 일치할 것을
        요구한다 (`implementation_direction._coverage_summary_errors`). 그
        순서가 저작 계약에 없으면 작성자는 브리프 원문에서 재유도해야 하고,
        오차는 §5.5.9 라운드 낭비와 조립 거부로 돌아온다. 레거시 브리프
        (end-state id 없음)는 빈 목록이므로 지시도 싣지 않는다 — 검증기도
        같은 조건으로 자기 비활성화한다.
        """
        if not self.original_requirement_ids:
            return []
        requirement_ids = ", ".join(
            f"`{item}`" for item in self.original_requirement_ids
        )
        return [
            f"Original requirement ids, in order: {requirement_ids}.",
            "`implementationPlanning.requirementCoverage` must contain exactly "
            "one row per original requirement id (`originalRequirementId`), in "
            "that exact order.",
            "`endStateCoverage` must contain exactly one row per original "
            "requirement id; an `addressed` row names its `coveredBy` anchor.",
            "Coverage references use existing identifiers: `stageRefs` contains stage numbers, "
            "`stepRefs` uses `<stage>.<step>` (for example `1.2`), `validationRefs` uses "
            "validation checklist ids, and `fileRefs` uses exact changed file paths without annotations. "
            "Map QA script changes to their requirements too. A future follow-up is not a stage or validation id.",
            "Derive `coverageSummary` from the coverage rows; do not declare exact 100% "
            "or `plan-ready` while requirements are uncovered or file changes are unmapped. "
            "Preserve recorded user decisions when describing deferred work.",
        ]

    def to_dict(self) -> dict[str, Any]:
        return {
            "schemaVersion": "1.0",
            "taskKey": self.task_key,
            "taskType": self.task_type,
            "authoringContract": {
                "resultPath": self.result_path,
                "format": "report-narrative-v3.0",
                "sourcePolicy": "read-only-synthesis-packet",
                "instructions": [
                    *NARRATIVE_GRAMMAR_INSTRUCTIONS,
                    "Write the complete human-readable report narrative.",
                    "Preserve settled source values, identities, dissent, and user responses.",
                    "Do not invent a value when a source is missing or contradictory.",
                    "Write only the narrative, pointer, and audit artifacts named by the dispatch.",
                    *_INTERNAL_IDENTIFIER_INSTRUCTION,
                    *self._schema_instructions(),
                    *self._task_instructions(),
                    *self._carry_instructions(),
                ],
                "runtimeOwnedContent": [
                    "session identifiers",
                    "token usage",
                    "estimated cost",
                    "user response carry-in",
                ],
                "validationRules": [
                    "writer-owned-fields-only",
                    "all-defects-collected",
                    *self._carry_validation_rules(),
                ],
            },
            "accountingSnapshot": self.accounting_snapshot,
            "sources": [source.to_dict() for source in self.sources],
        }

    def to_markdown(self) -> str:
        lines = [
            f"# OKSTRA Report Synthesis Packet - {self.task_key}",
            "",
            "## Authoring Contract",
            "",
            f"- Task type: `{self.task_type}`",
            f"- Result path: `{self.result_path}`",
            "- Output format: `report-narrative-v3.0`",
            *(f"- {text}" for text in NARRATIVE_GRAMMAR_INSTRUCTIONS),
            "- Input policy: read this synthesis packet as the dispatched source set",
            "- Responsibility: write the complete human-readable narrative while preserving settled values",
            "- Runtime-owned values: session identifiers, token usage, estimated cost, "
            "user response carry-in",
            "- Accounting details stay in the JSON snapshot; assembly supplies them.",
            "- Validation: writer-owned fields, all defects collected",
        ]
        # 작성자가 실제로 읽는 것은 마크다운이다. JSON 정본에만 실으면 지시가
        # 도달하지 않는다.
        lines.extend(f"- {text}" for text in _INTERNAL_IDENTIFIER_INSTRUCTION)
        lines.extend(f"- {text}" for text in self._schema_instructions())
        lines.extend(f"- {text}" for text in self._task_instructions())
        lines.extend(f"- {text}" for text in self._carry_instructions())
        sections = [("Authoring Contract", "\n".join(lines[1:]).rstrip() + "\n")]
        sections.extend(
            (source.label, "\n".join(_source_markdown(
                source, self.task_type, self.sources,
            )).rstrip() + "\n")
            for source in self.sources
        )
        return _indexed_markdown(lines[0], sections)


_SOURCE_FIELDS = (
    ("Analysis packet", "orchestrator", "instructionSet", "analysisPacketPath", True),
    ("Task brief", "reporter", "instructionSet", "taskBriefPath", True),
    ("Analysis profile", "orchestrator", "instructionSet", "analysisProfilePath", False),
    ("Analysis material", "reporter", "instructionSet", "analysisMaterialPath", False),
    ("Reference expectations", "reporter", "instructionSet", "referenceExpectationsPath", False),
    ("Clarification response", "user", "instructionSet", "clarificationResponsePath", False),
    ("Final report template", "orchestrator", "instructionSet", "reportTemplatePath", True),
    ("Final report schema", "orchestrator", "instructionSet", "finalReportSchemaPath", True),
    ("Convergence state", "convergence", "run", "convergenceStatePath", True),
)


_READ_CHUNK_BYTES = 16_000
_SHARED_TEXT_MIN_BYTES = 256
_ANALYSER_OPERATION_SECTIONS = frozenset({
    "Required workers", "Optional workers", "Team contract",
    "Worker interaction model", "Tooling — read-only MCP availability",
    "Run-scoped worker-resource lifecycle",
})


def _indexed_markdown(title: str, sections: list[tuple[str, str]]) -> str:
    """긴 한 줄과 다중 바이트 문자도 손실 없이 나눌 읽기 위치를 싣는다."""
    ranges: list[tuple[str, int, int]] = []
    offset = 0
    for label, content in sections:
        raw = content.encode("utf-8")
        start = 0
        while start < len(raw):
            end = min(start + _READ_CHUNK_BYTES, len(raw))
            while end < len(raw) and raw[end] & 0xC0 == 0x80:
                end -= 1
            ranges.append((label, offset + start, end - start))
            start = end
        offset += len(raw)
    prefix = f"{title}\n\n## Read Index\n\n"
    prefix += (
        "Read every range once, in order. Offsets are zero-based UTF-8 bytes "
        "in this Markdown file; each range is at most 16000 bytes. Use "
        "`dd if='<packet-path>' bs=1 skip=<start> count=<bytes> 2>/dev/null`. "
        "Continue with the next range after a successful read; do not reread "
        "the beginning. Source paths identify originals; the sibling "
        "`.data.json` retains every frozen source verbatim.\n\n"
        "| Source | Start byte | Bytes |\n|---|---:|---:|\n"
    )
    prefix_size = 0
    while True:
        index = prefix + "".join(
            f"| {label.replace('|', '/')} | {start + prefix_size} | {size} |\n"
            for label, start, size in ranges
        ) + "\n<!-- END READ INDEX -->\n"
        size = len(index.encode("utf-8"))
        if size == prefix_size:
            break
        prefix_size = size
    return index + "".join(content for _, content in sections)


def _writer_profile_view(content: str) -> str:
    """저작 규칙과 미지 절을 유지하고 분석자 배치 절만 참조로 바꾼다."""
    lines: list[str] = []
    skipping = False
    fence = ""
    for number, line in enumerate(content.splitlines(), 1):
        marker = re.match(r"^\s*(`{3,}|~{3,})", line)
        if marker:
            token = marker.group(1)
            if not fence:
                fence = token
            elif token[0] == fence[0] and len(token) >= len(fence):
                fence = ""
        if not fence and line.startswith("- "):
            label = line[2:].split(":", 1)[0].split(" (", 1)[0].strip("*")
            skipping = label in _ANALYSER_OPERATION_SECTIONS
            if skipping:
                lines.append(f"- {label}: analyser operation; original source line {number}.")
        elif not fence and line.startswith("#"):
            skipping = False
        if not skipping:
            lines.append(line)
    return "\n".join(lines)


def _replace_shared_text(value: Any, references: Mapping[str, str]) -> Any:
    if isinstance(value, str):
        return {"$sharedText": references[value]} if value in references else value
    if isinstance(value, list):
        return [_replace_shared_text(item, references) for item in value]
    if isinstance(value, dict):
        return {key: _replace_shared_text(item, references) for key, item in value.items()}
    return value


def _convergence_view(content: str) -> str:
    """반복된 근거 문자열만 공유하고 모든 회차·표·미지 필드는 유지한다."""
    try:
        state = json.loads(content)
    except json.JSONDecodeError:
        return content
    counts: Counter[str] = Counter()
    pending = [state]
    while pending:
        value = pending.pop()
        if isinstance(value, dict):
            # 원자료의 같은 모양과 새 참조를 혼동하지 않는다.
            if "$sharedText" in value:
                return content
            pending.extend(value.values())
        elif isinstance(value, list):
            pending.extend(value)
        elif isinstance(value, str) and len(value.encode("utf-8")) >= _SHARED_TEXT_MIN_BYTES:
            counts[value] += 1
    repeated = sorted(text for text, count in counts.items() if count > 1)
    if not repeated:
        return content
    references = {text: f"T{index}" for index, text in enumerate(repeated, 1)}
    view = {
        "sharedText": {ref: text for text, ref in references.items()},
        "state": _replace_shared_text(state, references),
    }
    return json.dumps(view, ensure_ascii=False, indent=1)


def _writer_schema_view(content: str, task_type: str) -> tuple[str, list[str]]:
    try:
        schema = json.loads(content)
    except json.JSONDecodeError:
        return content, ["- View: schema is not JSON; the original text is retained."]
    if not isinstance(schema, dict) or not schema.get("properties"):
        return content, []
    owned = writer_owned_schema(schema)
    # 부분 서사 검증에서 빠지는 루트 조건도 최종 조립의 저작 지침에는 남긴다.
    owned["allOf"] = schema.get("allOf", [])
    owned["required"] = [
        key for key in schema.get("required", []) if key in owned["properties"]
    ]
    excerpt = build_schema_excerpt(owned, task_type)
    return json.dumps(excerpt, ensure_ascii=False, indent=1), [
        "- View: writer-owned schema excerpt. Task-specific requirements are listed above.",
    ]


def _source_markdown(
    source: ReportSynthesisSource, task_type: str,
    sources: tuple[ReportSynthesisSource, ...],
) -> list[str]:
    content = source.content.rstrip("\n")
    view = []
    if source.label == "Analysis profile":
        content = _writer_profile_view(content)
    if source.label == "Analysis material":
        brief = next((item.content.strip() for item in sources if item.label == "Task brief"), "")
        if len(brief.encode("utf-8")) >= _SHARED_TEXT_MIN_BYTES:
            content = content.replace(brief, 'Read the complete "Source: Task brief" above.')
    if source.label == "Convergence state":
        content = _convergence_view(content)
        if content != source.content.rstrip("\n"):
            view = [
                "- View: lossless shared-text references. Replace each "
                '`{"$sharedText":"Tn"}` with `sharedText.Tn` when reading `state`. '
                "Every finding, vote, round, condition, dissent, and unknown field remains.",
            ]
    if source.label == "Analysis packet":
        texts = {item.label: item.content for item in sources}
        content = reference_source_extracts(
            source.content, task_type,
            brief_text=texts.get("Task brief", ""),
            profile_text=texts.get("Analysis profile", ""),
            reference_text=texts.get("Reference expectations", ""),
            clarification_text=texts.get("Clarification response", ""),
        ).rstrip("\n")
        if content != source.content.rstrip("\n"):
            view = [
                "- View: repeated extracts refer to frozen sources below; "
                "the original line index is omitted.",
            ]
    if source.label == "Final report schema":
        content, view = _writer_schema_view(content, task_type)
    longest = max((len(run) for run in re.findall(r"`+", content)), default=0)
    fence = "`" * max(3, longest + 1)
    return [
        "",
        f"## Source: {source.label}",
        "",
        f"- Owner: `{source.owner}`",
        f"- Path: `{source.path}`",
        *view,
        "",
        f"{fence}text",
        content,
        fence,
    ]


def _string(value: object) -> str:
    return value.strip() if isinstance(value, str) else ""


def _context_path(
    manifest: Mapping[str, Any],
    active_context: Mapping[str, Any],
    section: str,
    key: str,
) -> str:
    block = active_context.get(section)
    if isinstance(block, Mapping):
        value = _string(block.get(key))
        if value:
            return value
    return _string(manifest.get(key))


# 패킷은 워커 결과를 통째로 나른다. 그 안의 finding id(`F-001`)는 수렴 상태에만
# 있고 리포트는 그 id 를 정의하는 객체를 싣지 않는다 — 인용하면 독자가 해소할 수
# 없는 참조가 된다. 규칙은 여기에 둔다: 워커 결과를 보여 주는 자리가 여기뿐이다.
# 집행 코드는 없다(리포트 산문은 자유 서술이다). 저자 지시로만 전달한다.
_INTERNAL_IDENTIFIER_INSTRUCTION = (
    "Do not cite worker-internal identifiers in reader-facing prose - finding "
    "ids (`F-NNN`), worker ids, round labels. This report defines none of them, "
    "so the reader cannot resolve them. State the substance instead, and cite "
    "only identifiers the report itself defines (`R-NNN`, `C-NNN`, `A-NNN`, "
    "stage numbers).",
)


CARRIED_STAGES_LABEL = "Carried plan stages"


def read_incremental_decision(
    project_root: Path, manifest: Mapping[str, Any],
) -> dict | None:
    """이 run 의 증분 판정. 없거나 증분이 아니면 ``None``.

    레코드가 없으면 full 로 읽는다 — 그것이 이 기능이 없던 동안의 동작이고,
    "판정을 못 찾았으니 좁혀도 된다"는 반대 방향의 실패보다 안전하다.
    """
    value = _string(manifest.get("incrementalDecisionPath"))
    if not value:
        return None
    path = _resolve(project_root, value)
    if not path.is_file():
        return None
    try:
        record = load_owned_object(path, artifact="incremental scope decision")
    except (OSError, UnicodeError, JsonBoundaryError):
        return None
    if not isinstance(record, Mapping) or record.get("mode") != "incremental":
        return None
    stages = record.get("carryStages")
    if not isinstance(stages, list) or not stages:
        return None
    return dict(record)


def carried_stage_rows(prior_data: Mapping[str, Any], carry_stages: list) -> list[dict]:
    """직전 리포트에서 이월 대상 stage 의 **본문 행**을 뽑는다.

    Stage Ledger 가 싣는 것은 `stage`/`title`/`status`/`dependsOn`/`headCommit`
    뿐이다. narrative 의 `stages[]` 한 행은 `sliceValue`·`acceptance`·`carryIn`·
    `stepwiseExecution`·`exitContract`·`stageValidation` 을 요구하므로, 원장만
    받은 작성자는 이월 행을 옮길 수 없다.
    """
    # 작성자에게 넘기는 행은 작성자 소유 필드만이어야 한다. 완성 리포트의
    # stage 행에는 `designSurfaceCoverage` 같은 assembly 소유 필드가 함께 있고,
    # 그것까지 실으면 "옮겨 적으라"는 지시가 곧 소유권 위반 지시가 된다.
    planning = writer_owned_data(prior_data).get("implementationPlanning")
    rows = planning.get("stages") if isinstance(planning, Mapping) else None
    if not isinstance(rows, list):
        return []
    wanted = {int(stage) for stage in carry_stages if isinstance(stage, int)}
    return [
        dict(row)
        for row in rows
        if isinstance(row, Mapping) and row.get("stage") in wanted
    ]


def carried_stages_path(narrative_path: Path) -> Path:
    _, markdown_path = report_synthesis_packet_paths(narrative_path)
    return markdown_path.with_name(
        markdown_path.name.replace(
            "report-writer-synthesis-packet", "report-writer-carried-stages"
        )
    ).with_suffix(".json")


def materialize_carried_stages(
    *,
    project_root: Path,
    manifest: Mapping[str, Any],
    narrative_path: Path,
) -> tuple[str, dict] | None:
    """이월 stage 본문을 파일로 남기고 (프로젝트 상대 경로, 판정) 을 돌려준다.

    묶음의 source 는 파일과 그 다이제스트다. 파생 내용을 특례로 인라인하지 않고
    실제 파일로 만드는 이유는 두 가지다 — 다이제스트가 다른 source 와 같은
    의미를 갖고, 어긋났을 때 사람이 열어 볼 수 있다.
    """
    decision = read_incremental_decision(project_root, manifest)
    if decision is None:
        return None
    prior = _resolve(project_root, _string(decision.get("prevDataPath")))
    if not prior.is_file():
        return None
    try:
        prior_data = load_owned_object(prior, artifact="prior planning report")
    except (OSError, UnicodeError, JsonBoundaryError):
        return None
    rows = carried_stage_rows(prior_data, decision["carryStages"])
    if not rows:
        return None
    path = carried_stages_path(narrative_path)
    write_owned_object_atomic(
        path,
        {
            "schemaVersion": "1.0",
            "sourcePlanPath": _string(decision.get("prevDataPath")),
            "carryStages": list(decision["carryStages"]),
            "stages": rows,
        },
        artifact="carried plan stages",
    )
    try:
        return str(path.resolve().relative_to(project_root.resolve())), decision
    except ValueError:
        return str(path), decision


def _source_specs(
    project_root: Path,
    manifest: Mapping[str, Any],
    active_context: Mapping[str, Any],
    team_state: Mapping[str, Any],
    narrative_path: Path,
    carried_source: str,
) -> list[tuple[str, str, str, bool]]:
    specs: list[tuple[str, str, str, bool]] = []
    for label, owner, section, key, required in _SOURCE_FIELDS:
        path = _context_path(manifest, active_context, section, key)
        if label == "Final report template" and not path:
            path = _context_path(
                manifest, active_context, section, "finalReportTemplatePath"
            )
        if label == "Final report schema" and not path:
            instruction_set = _context_path(
                manifest, active_context, section, "instructionSetPath"
            )
            if instruction_set:
                path = str(Path(instruction_set) / "final-report-schema.json")
        if path or required:
            specs.append((label, owner, path, required))
    if carried_source:
        specs.append((CARRIED_STAGES_LABEL, "orchestrator", carried_source, True))
    specs.extend(_user_response_specs(project_root, narrative_path))
    attempt_specs = _attempt_result_specs(manifest)
    if attempt_specs:
        specs.extend(attempt_specs)
        return specs
    workers = team_state.get("workers")
    if not isinstance(workers, list):
        return specs
    for worker in workers:
        if not isinstance(worker, Mapping) or worker.get("workerId") == "report-writer":
            continue
        path = _string(worker.get("resultPath"))
        if path:
            worker_id = _string(worker.get("workerId")) or "unknown"
            specs.append((f"Analysis result ({worker_id})", worker_id, path, True))
    return specs


def _user_response_specs(
    project_root: Path,
    narrative_path: Path,
) -> list[tuple[str, str, str, bool]]:
    response_dir = narrative_path.parent.parent / "user-responses"
    if not response_dir.is_dir():
        return []
    return [
        (
            f"User response ({path.name})",
            "user",
            _relative(project_root, path),
            True,
        )
        for path in sorted(response_dir.glob("user-response-*.md"))
        if path.is_file()
    ]


def _attempt_result_specs(
    manifest: Mapping[str, Any],
) -> list[tuple[str, str, str, bool]]:
    invocations_value = manifest.get("invocations")
    attempts_value = manifest.get("attempts")
    if not isinstance(invocations_value, list) or not isinstance(attempts_value, list):
        return []
    invocations = {
        _string(row.get("invocationRef")): row
        for row in invocations_value
        if isinstance(row, Mapping) and _string(row.get("invocationRef"))
    }
    excluded_duties = {"lead", "report-writer", "translator"}
    specs: list[tuple[str, str, str, bool]] = []
    seen: set[str] = set()
    for attempt in attempts_value:
        if not isinstance(attempt, Mapping) or attempt.get("status") != "ok":
            continue
        path = _string(attempt.get("resultPath"))
        invocation_ref = _string(attempt.get("invocationRef"))
        invocation = invocations.get(invocation_ref, {})
        duty = _string(invocation.get("dutyId"))
        if not path or path in seen or duty in excluded_duties:
            continue
        seen.add(path)
        attempt_number = attempt.get("attempt")
        specs.append(
            (
                f"Settled result ({invocation_ref}, attempt {attempt_number})",
                duty or invocation_ref,
                path,
                True,
            )
        )
    return specs


def _accounting_snapshot(team_state: Mapping[str, Any]) -> dict[str, Any]:
    workers = accounting_workers(team_state)
    worker_usage = []
    if isinstance(workers, list):
        worker_usage = [
            {
                "workerId": _string(worker.get("workerId")) or "unknown",
                "role": _string(worker.get("role")),
                "usage": copy.deepcopy(worker.get("usage"))
                if isinstance(worker.get("usage"), Mapping)
                else {},
            }
            for worker in workers
            if isinstance(worker, Mapping)
        ]
    lead_sessions = team_state.get("leadSessionIds")
    return {
        "leadSessionIds": copy.deepcopy(lead_sessions)
        if isinstance(lead_sessions, list)
        else [],
        "leadUsage": copy.deepcopy(team_state.get("leadUsage"))
        if isinstance(team_state.get("leadUsage"), Mapping)
        else {},
        "workerUsage": worker_usage,
        "usageSummary": copy.deepcopy(team_state.get("usageSummary"))
        if isinstance(team_state.get("usageSummary"), Mapping)
        else {},
    }


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


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


def _read_sources(
    project_root: Path,
    specs: list[tuple[str, str, str, bool]],
) -> tuple[ReportSynthesisSource, ...]:
    issues: list[ReportSynthesisPacketIssue] = []
    sources: list[ReportSynthesisSource] = []
    for label, owner, value, required in specs:
        if not value:
            if required:
                issues.append(
                    ReportSynthesisPacketIssue(
                        owner,
                        label,
                        project_root,
                        "required source path is not configured",
                    )
                )
            continue
        path = _resolve(project_root, value)
        if not path.is_file():
            issues.append(
                ReportSynthesisPacketIssue(
                    owner, label, path, "required source is missing"
                )
            )
            continue
        try:
            content = path.read_text(encoding="utf-8")
        except (OSError, UnicodeError) as exc:
            issues.append(ReportSynthesisPacketIssue(owner, label, path, str(exc)))
            continue
        sources.append(
            ReportSynthesisSource(
                label=label,
                owner=owner,
                path=_relative(project_root, path),
                content=content,
            )
        )
    if issues:
        raise ReportSynthesisPacketError(tuple(issues))
    return tuple(sources)


def build_report_synthesis_packet(
    *,
    project_root: Path,
    manifest: Mapping[str, Any],
    active_context: Mapping[str, Any],
    team_state: Mapping[str, Any],
    narrative_path: Path,
) -> ReportSynthesisPacket:
    carried = materialize_carried_stages(
        project_root=project_root, manifest=manifest, narrative_path=narrative_path,
    )
    sources = _read_sources(
        project_root,
        _source_specs(
            project_root,
            manifest,
            active_context,
            team_state,
            narrative_path,
            carried[0] if carried else "",
        ),
    )
    task_type = _string(manifest.get("taskType"))
    original_requirement_ids: tuple[str, ...] = ()
    participating_analysers: tuple[str, ...] = ()
    # 계획 저작도 같은 id 순서를 계약으로 받는다 — 커버리지 행 순서 검증
    # (`validate_selected_direction_plan`)의 기준값이 이 목록이다.
    if task_type in ("implementation-option-selection", "implementation-planning"):
        brief_value = _context_path(
            manifest, active_context, "instructionSet", "taskBriefPath"
        )
        original_requirement_ids = brief_end_state_id_sequence(
            _resolve(project_root, brief_value)
        )
    if task_type == "implementation-option-selection":
        roster = manifest.get("recommendedWorkers")
        if isinstance(roster, list):
            participating_analysers = tuple(
                _string(worker)
                for worker in roster
                if _string(worker) and _string(worker) != "report-writer"
            )
    required_top_level, verdict_tokens, block_rules, follow_up_rules = (
        _schema_authoring_rules(sources, task_type)
    )
    return ReportSynthesisPacket(
        task_key=_string(manifest.get("taskKey")),
        task_type=task_type,
        result_path=_relative(project_root, narrative_path),
        required_top_level=required_top_level,
        verdict_tokens=verdict_tokens,
        follow_up_rules=follow_up_rules,
        block_key=TASK_TYPE_DATA_PROPERTY.get(task_type, "") if block_rules else "",
        block_rules=block_rules,
        allowed_top_level=_allowed_top_level_labels(),
        sources=sources,
        accounting_snapshot=_accounting_snapshot(team_state),
        original_requirement_ids=original_requirement_ids,
        participating_analysers=participating_analysers,
        incremental_decision=carried[1] if carried else None,
    )


def _schema_authoring_rules(
    sources: tuple[ReportSynthesisSource, ...], task_type: str,
) -> tuple[tuple[str, ...], tuple[str, ...], tuple[str, ...], tuple[str, ...]]:
    """넘겨받은(동결된) 완성 리포트 스키마에서 작성자 몫의 규칙 네 가지를 뽑는다.

    최상위 `required` 가운데 작성자 소유 이름(서사 스키마의 properties)만
    사람이 읽는 라벨로, 이 task type 의 `Verdict Token` 허용값, 이 task type 의
    데이터 블록 안쪽 모양(`task_block_rules`), 그리고 `followUpTasks` 행 규칙
    (`follow_up_task_rules`). 스키마 소스가 없거나
    JSON 이 아니면 빈 값이다 — 이 함수는 조립 검증을 대신하지 않고 도달하지
    못하던 규칙을 저작 계약에 옮길 뿐이다.
    """
    schema_text = next(
        (source.content for source in sources if source.label == "Final report schema"),
        "",
    )
    try:
        schema = json.loads(schema_text) if schema_text else {}
    except ValueError:
        schema = {}
    if not isinstance(schema, dict):
        schema = {}
    try:
        allowed = allowed_top_level_fields()
    except (NarrativeContractError, JsonBoundaryError):
        allowed = frozenset()
    required = schema.get("required")
    required_top_level = tuple(
        humanise(key)
        for key in (required if isinstance(required, list) else [])
        if isinstance(key, str) and key in allowed
    )
    block_key = TASK_TYPE_DATA_PROPERTY.get(task_type, "")
    block_rules = task_block_rules(writer_owned_schema(schema), block_key) if block_key else ()
    return (
        required_top_level,
        verdict_token_rule(schema, task_type),
        block_rules,
        follow_up_task_rules(schema, task_type),
    )


def _allowed_top_level_labels() -> tuple[str, ...]:
    """작성자 소유 최상위 필드 전체 — 서사 스키마 부재는 빈 값(조립이 판정한다)."""
    try:
        allowed = allowed_top_level_fields()
    except (NarrativeContractError, JsonBoundaryError):
        return ()
    return tuple(humanise(key) for key in sorted(allowed))


def report_synthesis_packet_paths(narrative_path: Path) -> tuple[Path, Path]:
    name = narrative_path.name
    prefix = "report-writer-narrative-"
    if name == "report-writer-narrative.md":
        packet_name = "report-writer-synthesis-packet.md"
    elif name.startswith(prefix) and name.endswith(".md"):
        packet_name = "report-writer-synthesis-packet-" + name[len(prefix):]
    else:
        raise ValueError(f"report narrative path has an unsupported name: {narrative_path}")
    markdown_path = narrative_path.with_name(packet_name)
    data_path = markdown_path.with_suffix(".data.json")
    return data_path, markdown_path


def materialize_report_synthesis_packet(
    *,
    project_root: Path,
    manifest: Mapping[str, Any],
    active_context: Mapping[str, Any],
    team_state: Mapping[str, Any],
    narrative_path: Path,
) -> tuple[Path, Path]:
    packet = build_report_synthesis_packet(
        project_root=project_root,
        manifest=manifest,
        active_context=active_context,
        team_state=team_state,
        narrative_path=narrative_path,
    )
    data_path, markdown_path = report_synthesis_packet_paths(narrative_path)
    write_owned_object_atomic(
        data_path,
        packet.to_dict(),
        artifact="report synthesis packet",
    )
    _write_text_atomic(markdown_path, packet.to_markdown())
    return data_path, markdown_path


def _write_text_atomic(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary = tempfile.mkstemp(
        prefix=f".{path.name}.",
        suffix=".tmp",
        dir=path.parent,
    )
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            handle.write(text)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)


def verify_report_synthesis_packet_sources(
    project_root: Path,
    data_path: Path,
) -> tuple[ReportSynthesisPacketIssue, ...]:
    try:
        payload = load_owned_object(data_path, artifact="report synthesis packet")
    except JsonBoundaryError as exc:
        return (
            ReportSynthesisPacketIssue(
                "orchestrator",
                "Report synthesis packet",
                data_path,
                str(exc),
            ),
        )
    sources = payload.get("sources") if isinstance(payload, Mapping) else None
    if not isinstance(sources, list):
        return (
            ReportSynthesisPacketIssue(
                "orchestrator",
                "Report synthesis packet",
                data_path,
                "sources must be an array",
            ),
        )
    issues: list[ReportSynthesisPacketIssue] = []
    for source in sources:
        if not isinstance(source, Mapping):
            continue
        label = _string(source.get("label")) or "Unknown source"
        owner = _string(source.get("owner")) or "orchestrator"
        path = _resolve(project_root, _string(source.get("path")))
        if not path.is_file():
            issues.append(
                ReportSynthesisPacketIssue(
                    owner,
                    label,
                    path,
                    "frozen source is missing",
                )
            )
            continue
        try:
            path.read_text(encoding="utf-8")
        except (OSError, UnicodeError) as exc:
            issues.append(ReportSynthesisPacketIssue(owner, label, path, str(exc)))
            continue
    return tuple(issues)