"""Read a final-report's clarification rows, whichever schema wrote them.

A clarification is what a run owes the user — a decision, a file attachment,
a single data point. Each row carries a ``Blocks`` value out of
``{approval, next-phase, none}``. Rows with ``Blocks=approval`` are the
approval gate: they MUST resolve before the user flips the frontmatter
``approved`` field to ``true`` and starts the next ``implementation`` run.
A ``request-revision`` / ``reject`` answer still blocks the report that has
not yet incorporated it. The report that recorded the same id in
``supersessionLedger`` has already absorbed that return and does not block.

The two schemas store those rows in different places, and that is why the
read functions take a report **path**, not its text:

* schema-v1 — the ``## 1. Clarification Items`` markdown table (introduced
  when §4.5.9 / §5.1 / §5.2 collapsed into a single section).
* schema-v2 — ``clarificationItems[]`` in the ``.data.json`` sibling. Its AI
  handoff markdown renders them as one section per row under
  ``## Clarification and User Decisions``, not as a §1 table, so the §1 table
  walk finds nothing there by construction.

Every gate goes through ``scan_approval_gate`` / ``scan_open_user_input`` so
run-prep (``_validate_approved_plan``), the wizard, and the user-response CLI
cannot disagree about what is still open.

Gate semantics are fail-closed: when the rows cannot be read with confidence
(§1 heading missing/drifted, table header unrecognized, a body row whose
metadata cell fails to parse, or a v2 row missing id/blocks/status), the scan
reports an ``unreadable_reason`` and callers must refuse approval instead of
soft-passing. ``parse_clarification_items`` keeps the lenient
None-on-absence contract for the schema-v1 HTML-view renderers, which only
need best-effort row extraction.
"""
from __future__ import annotations

import json
import re
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

from okstra_ctl.final_report_paths import final_report_data_path
from okstra_ctl.json_boundary import load_owned_object
from okstra_ctl.md_table import is_separator_row, split_pipe_row, to_cell_text


# The final-report renderer (render_final_report.py:_inject_anchors) appends a
# ` <a id="slug"></a>` scroll anchor to every heading. The §1 slice MUST tolerate
# that trailing anchor — otherwise `_section_1_slice` fails on every *rendered*
# report, parse returns None, and the Blocks=approval approval gate is silently
# bypassed (run.py:_validate_approved_plan soft-passes None).
SECTION_HEADING_PATTERN = re.compile(
    r'^##\s+1\.\s+Clarification Items\s*(?:<a id="[^"]+"></a>\s*)?$',
    re.MULTILINE,
)
NEXT_TOP_LEVEL_HEADING_PATTERN = re.compile(r"^##\s+(?!1\.)", re.MULTILINE)


@dataclass(frozen=True)
class ClarificationItem:
    """One row of the §1 table.

    ``raw_*`` fields preserve the exact cell text (after backtick stripping)
    for diagnostics; canonical lowercased versions live in ``blocks`` /
    ``status`` for predicate use.
    """
    row_id: str
    kind: str  # "material" | "decision" | "data-point" | other
    blocks: str  # canonical lowercase: "approval" | "next-phase" | "none" | other
    status: str  # canonical lowercase: "open" | "answered" | "resolved" | "obsolete" | other
    raw_blocks: str
    raw_status: str


# The final-report renderer injects a scroll anchor into ID-defining first
# cells — either leading (`<a id="c-001"></a>C-001`) or inside the bold marker
# (`**<a id="e-001"></a>E-001**`). Strip every such empty anchor during cell
# normalization so the ID parses as a bare token for clarification parsing AND
# the HTML view's `C-\d+` form detection, and so the anchor never leaks into
# the HTML view as html-escaped literal text.
_CELL_ANCHOR_RE = re.compile(r'<a id="([^"]*)"></a>')


def _strip_backticks(cell: str) -> str:
    s = _CELL_ANCHOR_RE.sub("", cell.strip()).strip()
    if s.startswith("`") and s.endswith("`") and len(s) >= 2:
        s = s[1:-1].strip()
    return s


def _split_pipe_row(line: str) -> list[str]:
    """``md_table.split_pipe_row`` + §1-specific cell normalization
    (scroll-anchor removal, outer-backtick unwrap)."""
    return [_strip_backticks(c) for c in split_pipe_row(line)]


def _section_1_slice(report_text: str) -> Optional[str]:
    """Return the substring spanning the §1 section (heading exclusive of the
    next ``##`` heading), or None if §1 is absent."""
    start_match = SECTION_HEADING_PATTERN.search(report_text)
    if not start_match:
        return None
    rest = report_text[start_match.end():]
    end_match = NEXT_TOP_LEVEL_HEADING_PATTERN.search(rest)
    return rest[: end_match.start()] if end_match else rest


_META_ID_RE = re.compile(r"([A-Za-z][A-Za-z0-9]*-\d+)")


def _meta_field(cell: str, key: str) -> str:
    """Extract a ``<key>: value`` field from a stacked §1 meta cell.

    The §1 table collapses the short columns into one ``<br>``-delimited
    metadata cell (``**C-101**<br>Ticket: `DEV-1`<br>Kind: `decision`<br>
    Blocks: `approval`<br>Status: open``). Values may be backtick-wrapped.
    """
    m = re.search(
        rf"{re.escape(key)}:\s*`?\s*([^`<|]+?)\s*`?\s*(?:<br\s*/?>|$)",
        cell,
        re.IGNORECASE,
    )
    return m.group(1).strip() if m else ""


def _meta_id(cell: str) -> str:
    """The bold headline ID of a §1 meta cell (the part before the first <br>)."""
    headline = re.split(r"<br\s*/?>", cell, maxsplit=1)[0]
    m = _META_ID_RE.search(headline)
    return m.group(1) if m else ""


def parse_meta_cell(cell: str) -> Optional[ClarificationItem]:
    """Parse one §1 stacked meta cell into a ``ClarificationItem``, or ``None``
    when the cell is not a §1 meta cell (no ``Blocks:``/``Status:`` markers —
    e.g. a header or an unrelated table). Shared by the approval-gate parser
    and the HTML view's form-attach pass so both read the cell identically.
    """
    raw_blocks = _meta_field(cell, "Blocks")
    raw_status = _meta_field(cell, "Status")
    if not (raw_blocks and raw_status):
        return None
    return ClarificationItem(
        row_id=_meta_id(cell),
        kind=_meta_field(cell, "Kind").lower(),
        blocks=raw_blocks.lower(),
        status=raw_status.lower(),
        raw_blocks=raw_blocks,
        raw_status=raw_status,
    )


@dataclass(frozen=True)
class _Section1Table:
    """Outcome of walking the §1 slice for its data table.

    ``items`` is ``None`` when no recognizable table header exists among the
    pipe lines. ``unparsed_row_count`` counts body rows whose metadata cell
    failed ``parse_meta_cell`` (all-empty filler rows excluded).
    ``has_pipe_lines`` distinguishes the renderer's legitimate table-less
    placeholder (emptyState bullet) from a table whose header drifted.
    """
    items: Optional[list[ClarificationItem]]
    unparsed_row_count: int
    has_pipe_lines: bool


def _walk_section_1_table(section: str) -> _Section1Table:
    lines = section.splitlines()
    has_pipe_lines = any(line.lstrip().startswith("|") for line in lines)
    # Locate the §1 data table by its header. The merged-meta layout collapses
    # ID/Ticket/Kind/Blocks/Status into one metadata cell and keeps the
    # English `Statement` + `User input` columns; detect on those two (any
    # other table — intro, legacy 5.1/5.2 — is rejected).
    header_idx = -1
    for idx, line in enumerate(lines):
        if not line.lstrip().startswith("|"):
            continue
        cells = [c.lower() for c in _split_pipe_row(line)]
        if "user input" in cells and any(c.startswith("statement") for c in cells):
            header_idx = idx
            break
    if header_idx < 0:
        return _Section1Table(None, 0, has_pipe_lines)

    items: list[ClarificationItem] = []
    unparsed = 0
    body_started = False
    for line in lines[header_idx + 1:]:
        if not line.lstrip().startswith("|"):
            if body_started:
                break
            continue
        if is_separator_row(line):
            body_started = True
            continue
        if not body_started:
            continue
        cells = _split_pipe_row(line)
        if not any(cells):
            continue
        item = parse_meta_cell(cells[0])
        if item is None:
            unparsed += 1
            continue
        items.append(item)
    return _Section1Table(items, unparsed, True)


def parse_clarification_items(report_text: str) -> Optional[list[ClarificationItem]]:
    """Return the list of §1 rows. ``None`` means "no §1 meta table detected"
    (missing section or unrecognized table header) — caller must NOT treat
    that as "table is empty".

    Lenient view-renderer contract: rows whose metadata cell fails to parse
    are skipped, not surfaced. The approval gate must use
    ``scan_approval_gate`` instead, which fail-closes on those rows.
    """
    section = _section_1_slice(report_text)
    if section is None:
        return None
    return _walk_section_1_table(section).items


# schema-v2 는 clarification 을 §1 마크다운 테이블이 아니라 data.json 의
# `clarificationItems[]` 로 들고, AI 핸드오프 Markdown 은 그것을
# `## Clarification and User Decisions` 아래 JSON 블록으로 직렬화한다. §1 테이블
# 워크는 거기서 아무것도 못 찾으므로, 게이트가 v1 파서만 쓰면 열린 항목이 있는
# v2 리포트를 "읽을 수 없음"으로 떨어뜨린다. data.json 은 두 렌더러가 이미 읽는
# SSOT 다 — clarification 상태도 여기서 읽는다.
# 리포트 정본의 구조화 판본. 2.0 과 3.0 은 `clarificationItems[]` 를 같은 자리에
# 같은 모양으로 들고 있으므로 이 좁히기에는 차이가 없다. ADR-0019 가 2.0 을 계속
# 판독하라고 정했으므로 둘 다 받는다 — `stage_map.structured_report` 와 같은 규칙.
STRUCTURED_REPORT_VERSIONS = frozenset({"2.0", "3.0"})


def _read_report_text(report_path: Path) -> str:
    return report_path.read_text(encoding="utf-8", errors="replace")


def _structured_report_data(report_path: Path) -> Optional[dict]:
    """``report_path`` 의 구조화 리포트 레코드, 아니면 ``None``.

    파일이 없거나 JSON 이 깨졌거나 구조화 판본이 아니면 ``None`` — 호출자는 v1
    마크다운 경로로 폴백하고, 진짜 구조화 리포트인데 레코드가 깨진 경우는 그
    폴백이 "§1 없음" fail-closed 로 잡는다.

    판본을 `2.0` 하나로 못박아 두던 동안 3.0 리포트가 여기서 ``None`` 이 됐고,
    그러면 호출자가 JSON 본문에서 `## 1.` 헤딩을 찾다 실패해 **리포트 전문을
    그대로 복사**했다 — 이 좁히기가 막으려던 바로 그 중복이다. 실측: 66K 짜리
    추출본이 나와야 할 자리에 693K data.json 이 스테이징됐고, 패킷의
    `## Clarification Carry-In Extract` 가 0행이 되어 워커가 사용자 답변을
    전혀 받지 못했다.
    """
    data_path = final_report_data_path(report_path)
    if not data_path.is_file():
        return None
    try:
        data = load_owned_object(data_path, artifact="final report record")
    except (OSError, ValueError):
        return None
    if (
        not isinstance(data, dict)
        or data.get("schemaVersion") not in STRUCTURED_REPORT_VERSIONS
    ):
        return None
    return data


def _v2_row(entry: dict) -> Optional[dict]:
    """``clarificationItems[]`` 한 행을 §1 행과 같은 shape 으로. 필수 필드가
    빠졌으면 ``None`` — 호출자가 fail-closed 로 셀 수 있게 한다."""
    row_id = entry.get("id")
    raw_blocks = entry.get("blocks")
    raw_status = entry.get("status")
    if not (isinstance(row_id, str) and row_id):
        return None
    if not (isinstance(raw_blocks, str) and raw_blocks):
        return None
    if not (isinstance(raw_status, str) and raw_status):
        return None
    kind = entry.get("kind")
    item = ClarificationItem(
        row_id=row_id,
        kind=kind.lower() if isinstance(kind, str) else "",
        blocks=raw_blocks.lower(),
        status=raw_status.lower(),
        raw_blocks=raw_blocks,
        raw_status=raw_status,
    )
    options = entry.get("options")
    return {
        "item": item,
        "statement": str(entry.get("statement") or ""),
        "expected_form": str(entry.get("expectedForm") or ""),
        "options": (
            [option for option in options if isinstance(option, dict)]
            if isinstance(options, list)
            else []
        ),
        "disposition": clarification_disposition(entry),
    }


def _v2_clarification_rows(report_path: Path) -> Optional[list[dict]]:
    """schema-v2 clarification 행들, 이 리포트가 v2 가 아니면 ``None``.
    필수 필드가 빠진 행은 건너뛰는 lenient 계약(§1 파서와 동일)."""
    data = _structured_report_data(report_path)
    if data is None:
        return None
    entries = data.get("clarificationItems")
    if not isinstance(entries, list):
        return []
    rows = [_v2_row(e) for e in entries if isinstance(e, dict)]
    return [row for row in rows if row is not None]


def read_clarification_rows(report_path: Path) -> list[dict]:
    """리포트 한 건의 clarification 행 — schema-v2 는 data.json 사이드카에서,
    schema-v1 은 §1 테이블에서. 스키마 버전을 아는 유일한 읽기 지점이다.

    ``parse_section_1_rows`` 와 같은 lenient 계약: 어느 쪽에서도 행을 못 찾으면
    ``[]``.
    """
    v2_rows = _v2_clarification_rows(report_path)
    if v2_rows is not None:
        return v2_rows
    return parse_section_1_rows(_read_report_text(report_path))


def parse_section_1_rows(report_text: str) -> list[dict]:
    """§1 테이블 한 행마다 메타(ClarificationItem) + 원문 Statement / Expected
    form 셀.

    입력은 §1 을 담은 마크다운 본문이다 — 리포트 전문일 수도 있고, 다음 run 에
    첨부되는 carry-in 본문(``clarification_response_with_sidecars``)일 수도
    있다. §1 셀 추출의 단일 참조점 — 다른 모듈이 §1 레이아웃을 다시 훑지 않도록
    이 함수로 통일한다.
    """
    section = _section_1_slice(report_text)
    if section is None:
        return []
    lines = section.splitlines()
    header_idx = -1
    for idx, line in enumerate(lines):
        if not line.lstrip().startswith("|"):
            continue
        cells = [c.lower() for c in _split_pipe_row(line)]
        if "user input" in cells and any(c.startswith("statement") for c in cells):
            header_idx = idx
            break
    if header_idx < 0:
        return []
    header = [c.lower() for c in _split_pipe_row(lines[header_idx])]
    s_col = next((j for j, h in enumerate(header) if h.startswith("statement")), -1)
    e_col = next((j for j, h in enumerate(header) if h.startswith("expected form")), -1)
    rows: list[dict] = []
    body = False
    for line in lines[header_idx + 1:]:
        if not line.lstrip().startswith("|"):
            if body:
                break
            continue
        if is_separator_row(line):
            body = True
            continue
        if not body:
            continue
        cells = _split_pipe_row(line)
        item = parse_meta_cell(cells[0]) if cells else None
        if item is None:
            continue
        rows.append({
            "item": item,
            "statement": cells[s_col] if 0 <= s_col < len(cells) else "",
            "expected_form": cells[e_col] if 0 <= e_col < len(cells) else "",
            # v1 cells are strings; the structured options live only in v2's
            # data sibling. The key stays so callers never branch on schema.
            "options": [],
        })
    return rows


# 아직 사용자 판단이 없는 상태. `answered` 는 판단 기록이며 다시 묻지 않는다.
UNRESOLVED_STATUSES = {"open"}

# `Blocks` values that gate the user's `approved: true` flip.
APPROVAL_BLOCKS = frozenset({"approval"})
# `Blocks` values that owe the user an answer at all. `next-phase` rows never
# gate approval, but the next phase cannot start with them open either, so the
# user-response skill must list them — counting only `approval` made a report
# whose open items were all `next-phase` look like it had none.
USER_INPUT_BLOCKS = frozenset({"approval", "next-phase"})
ANSWER_DISPOSITIONS = frozenset({
    "answer",
    "select",
    "accept-risk",
    "request-revision",
    "reject",
})
# 사용자가 진행을 고른 처분. DISAGREE 표와 행은 증거로 남고 게이트는 내리다.
PROCEEDING_DISPOSITIONS = frozenset({"answer", "select", "accept-risk"})
# 사용자가 이 계획으로 진행하지 않겠다고 고른 처분.
RETURN_DISPOSITIONS = frozenset({"request-revision", "reject"})
# 이 런이 그 답을 본문에 반영했다고 원장이 적은 처분.
INCORPORATED_LEDGER_DISPOSITIONS = frozenset(
    {"superseded", "no-dependent-statement"}
)


def clarification_disposition(row: Mapping[str, object]) -> str:
    """행에 기록된 사용자 처분. 없으면 빈 문자열."""
    resolution = row.get("resolution")
    if not isinstance(resolution, Mapping):
        context = row.get("approvalContext")
        if isinstance(context, Mapping):
            resolution = context.get("resolution")
    if isinstance(resolution, Mapping):
        disposition = str(resolution.get("disposition") or "").strip()
        if disposition:
            return disposition
    user_input = str(row.get("userInput") or "").strip()
    options = row.get("options")
    if user_input and isinstance(options, list):
        for option in options:
            if (
                isinstance(option, Mapping)
                and str(option.get("answer") or "").strip() == user_input
            ):
                return str(option.get("disposition") or "").strip()
    return ""


def incorporated_clarification_ids(
    report_data: Mapping[str, object] | None,
) -> frozenset[str]:
    """이 런이 답을 본문에 반영했다고 원장에 적은 C-id.

    `superseded` 와 `no-dependent-statement` 만 센다. 원장에 없는
    `request-revision` / `reject` 는 아직 다음 계획을 막는 되돌림이다.
    """
    if not isinstance(report_data, Mapping):
        return frozenset()
    planning = report_data.get("implementationPlanning")
    if not isinstance(planning, Mapping):
        return frozenset()
    ledger = planning.get("supersessionLedger")
    if not isinstance(ledger, list):
        return frozenset()
    ids: set[str] = set()
    for entry in ledger:
        if not isinstance(entry, Mapping):
            continue
        row_id = str(entry.get("clarificationId") or "").strip()
        disposition = str(entry.get("disposition") or "").strip().lower()
        if row_id and disposition in INCORPORATED_LEDGER_DISPOSITIONS:
            ids.add(row_id)
    return frozenset(ids)


def row_blocks_progress(
    status: str,
    disposition: str = "",
    *,
    incorporated: bool = False,
) -> bool:
    """이 행이 승인·다음 단계 진입을 막는가.

    진행 처분(`accept-risk` / `select` / `answer`)은 고치지 않은 DISAGREE 를
    행과 투표에 남긴 채로 게이트만 내린다. `request-revision` / `reject` 는
    사용자가 진행을 거절한 것이므로 막는다. 다만 이 보고서 원장이 그 답을
    이미 반영했으면(`incorporated`) 같은 되돌림이 다음 계획 런을 강제하지
    않는다. 처분이 없는 `answered` 도 판단 기록이므로 막지 않는다.
    """
    normalized_status = status.strip().lower()
    normalized_disposition = disposition.strip().lower()
    if normalized_status == "obsolete":
        return False
    if normalized_disposition in PROCEEDING_DISPOSITIONS:
        return False
    if normalized_disposition in RETURN_DISPOSITIONS:
        return not incorporated
    return normalized_status not in {"answered", "resolved"}


def progress_blocking_ids(
    rows: object,
    blocking_values: frozenset[str] = APPROVAL_BLOCKS,
    *,
    report_data: Mapping[str, object] | None = None,
) -> list[str]:
    """게이트를 아직 막는 행 id. 사용자 진행 처분이 있는 행은 빠진다.

    ``report_data`` 가 있으면 원장에 반영된 되돌림 행도 빠진다.
    """
    if not isinstance(rows, list):
        return []
    incorporated = incorporated_clarification_ids(report_data)
    ids: list[str] = []
    for row in rows:
        if not isinstance(row, Mapping):
            continue
        blocks = str(row.get("blocks") or "").strip().lower()
        status = str(row.get("status") or "").strip()
        row_id = row.get("id")
        if (
            blocks in blocking_values
            and row_blocks_progress(
                status,
                clarification_disposition(row),
                incorporated=isinstance(row_id, str) and row_id in incorporated,
            )
            and isinstance(row_id, str)
            and row_id
        ):
            ids.append(row_id)
    return ids


@dataclass(frozen=True)
class ClarificationScan:
    """Fail-closed read of the clarification rows blocking on one `Blocks`
    value set.

    ``unreadable_reason`` is ``None`` only when the scan is confident: the
    rows parsed cleanly (or the report is the legitimate table-less
    placeholder) and ``blockers`` is therefore authoritative. A non-None
    reason means the caller must refuse to act — never soft-pass.
    """
    blockers: list[ClarificationItem]
    unreadable_reason: Optional[str]


def scan_approval_gate(report_path: Path) -> ClarificationScan:
    """Scan for ``Blocks=approval`` rows that still block progress.

    A recorded user proceeding disposition (`accept-risk` / `select` /
    `answer`), including one that lives only in the sidecar, does not block.
    ``request-revision`` / ``reject`` still block unless this report's
    ``supersessionLedger`` already incorporated that id. The scan refuses to
    guess whenever the schema drifted.
    """
    return scan_clarification_blockers(
        report_path,
        APPROVAL_BLOCKS,
        honor_sidecar_answers=True,
        sidecar_unblocks_proceeding_only=True,
    )


def scan_open_user_input(report_path: Path) -> ClarificationScan:
    """Scan for every unresolved row that still owes the user an answer
    (``Blocks`` in ``{approval, next-phase}``)."""
    return scan_clarification_blockers(
        report_path, USER_INPUT_BLOCKS, honor_sidecar_answers=True
    )


def scan_clarification_blockers(
    report_path: Path,
    blocking_values: frozenset[str],
    *,
    honor_sidecar_answers: bool,
    sidecar_unblocks_proceeding_only: bool = False,
) -> ClarificationScan:
    """Shared fail-closed clarification walk for both gates above — schema-v2
    reads its rows from the data sibling and schema-v1 from the §1 table.
    ``honor_sidecar_answers`` hides rows the sidecar already answered.
    ``sidecar_unblocks_proceeding_only`` keeps an unincorporated
    ``request-revision`` / ``reject`` as a blocker so a return choice cannot
    start the next phase. A ledger entry for that id means this report already
    absorbed the return.
    """
    v2_scan = _scan_v2_blockers(report_path, blocking_values)
    scan = (
        v2_scan if v2_scan is not None
        else scan_section_1_blockers(_read_report_text(report_path), blocking_values)
    )
    return (
        _resolve_blockers_answered_by_user(
            report_path,
            scan,
            proceeding_only=sidecar_unblocks_proceeding_only,
        )
        if honor_sidecar_answers
        else scan
    )


def _resolve_blockers_answered_by_user(
    report_path: Path,
    scan: ClarificationScan,
    *,
    proceeding_only: bool = False,
) -> ClarificationScan:
    """사용자가 사이드카로 답한 행을 blocker 에서 뺀 스캔.

    답의 정본은 사용자의 `user-responses/` 사이드카다. 리포트의 `Status` 는 그
    run 이 스스로 적어둔 값이고, 답이 사이드카로만 들어오는 경로(HTML 뷰의
    `Export user response`, `okstra user-response write`)에서는 갱신되지 않는다.
    사용자 입력 목록은 사이드카 답변이 있는 항목을 다시 묻지 않는다. 승인
    게이트는 진행 처분만 차단에서 뺀다.

    fail-closed 는 그대로다: 행 자체를 못 읽은 스캔(`unreadable_reason`)은
    어떤 id 가 blocker 인지 모르는 상태이므로 사이드카로 덮지 않는다.
    """
    if scan.unreadable_reason is not None or not scan.blockers:
        return scan
    dispositions = sidecar_dispositions(report_path)
    if not dispositions:
        return scan
    answered_ids = {
        row_id
        for row_id, disposition in dispositions.items()
        if (not proceeding_only) or disposition in PROCEEDING_DISPOSITIONS
    }
    if not answered_ids:
        return scan
    return ClarificationScan(
        [b for b in scan.blockers if b.row_id not in answered_ids], None
    )


def scan_section_1_blockers(
    report_text: str, blocking_values: frozenset[str]
) -> ClarificationScan:
    """Fail-closed §1 walk over any markdown carrying the table — a schema-v1
    report, or the carry-in body a reconciliation just produced."""
    section = _section_1_slice(report_text)
    if section is None:
        if _LOOSE_SECTION_1_RE.search(report_text):
            reason = (
                "`## 1. Clarification Items` heading exists but does not match "
                "the schema heading format (anchor/format drift)"
            )
        else:
            expected = "/".join(sorted(blocking_values))
            reason = (
                "report has no `## 1. Clarification Items` section — the gate "
                f"cannot confirm there are no unresolved `Blocks={expected}` rows"
            )
        return ClarificationScan([], reason)
    table = _walk_section_1_table(section)
    if table.items is None:
        if table.has_pipe_lines:
            return ClarificationScan([], (
                "§1 contains a table but its header row is not the schema "
                "header (`| ... | Statement | Expected form | User input |`)"
            ))
        # Renderer's emptyState placeholder: heading is intact and no table
        # was emitted — confidently "no blocking items".
        return ClarificationScan([], None)
    if table.unparsed_row_count:
        return ClarificationScan([], (
            f"§1 table has {table.unparsed_row_count} row(s) whose metadata "
            "cell could not be parsed (Blocks/Status markers missing or "
            "malformed)"
        ))
    blockers = [
        it for it in table.items
        if it.blocks in blocking_values and row_blocks_progress(it.status)
    ]
    return ClarificationScan(blockers, None)


def _scan_v2_blockers(
    report_path: Path, blocking_values: frozenset[str]
) -> Optional[ClarificationScan]:
    """schema-v2 data.json 기준 스캔, 이 리포트가 v2 가 아니면 ``None``.
    필수 필드가 빠진 행은 §1 의 unparsed row 와 같이 fail-closed 로 다룬다."""
    data = _structured_report_data(report_path)
    if data is None:
        return None
    entries = data.get("clarificationItems")
    if entries is None:
        return ClarificationScan([], None)
    if not isinstance(entries, list):
        return ClarificationScan([], (
            "schema-v2 data.json `clarificationItems` is not an array — the "
            "gate cannot read the clarification rows"
        ))
    rows = [_v2_row(e) if isinstance(e, dict) else None for e in entries]
    unparsed = sum(1 for row in rows if row is None)
    if unparsed:
        return ClarificationScan([], (
            f"schema-v2 data.json has {unparsed} `clarificationItems` row(s) "
            "missing id/blocks/status"
        ))
    incorporated = incorporated_clarification_ids(data)
    blockers = [
        row["item"] for row in rows
        if row["item"].blocks in blocking_values
        and row_blocks_progress(
            row["item"].status,
            str(row.get("disposition") or ""),
            incorporated=row["item"].row_id in incorporated,
        )
    ]
    return ClarificationScan(blockers, None)


# 느슨한 §1 헤딩 탐지: 엄격한 SECTION_HEADING_PATTERN 이 실패해도 이게 매칭되면
# "§1 헤딩은 있는데 형태가 어긋나 파싱에 실패" 한 상태다. trailing 부분을 보지
# 않으므로 앵커 변형·수동 편집·미래 렌더 변경 어디서든 헤딩의 존재만 잡는다.
_LOOSE_SECTION_1_RE = re.compile(r"^##\s+1\.\s+Clarification Items\b", re.MULTILINE)


def section_1_present_but_unparsed(report_text: str) -> bool:
    """§1 헤딩이 느슨 탐지엔 잡히지만 엄격 SECTION_HEADING_PATTERN 에는 매칭하지
    못하는 경우 True — 헤딩 형태가 어긋나(앵커·포맷 drift) §1 슬라이스 자체가
    실패하는 상태다.

    이때 ``_section_1_slice`` 가 None 을 반환해 parse 가 통째로 None 이 되고 승인
    게이트가 "schema 없음 → soft-pass" 로 조용히 열린다. §1 앵커 버그가 정확히 이
    메커니즘으로 터졌다. 헤딩 자체가 없는 legacy 리포트(둘 다 불매칭)와, 엄격
    매칭에 성공하는 정상 헤딩(테이블이 없는 emptyState placeholder 포함)은 False —
    placeholder 는 헤딩이 멀쩡하므로 fail-closed 로 오인하지 않는다. 정규식만 넓혀 온
    과거 수정과 달리, 이 판별은 "헤딩 형태 drift" 자체를 차단해 재발 클래스를 닫는다."""
    if SECTION_HEADING_PATTERN.search(report_text):
        return False
    return bool(_LOOSE_SECTION_1_RE.search(report_text))


def user_response_sidecars(source: Path) -> list[Path]:
    """``source`` 형제 ``user-responses/`` 의 ``user-response-*.md`` 목록(이름 순).

    ``source`` 가 ``runs/<task-type>/reports/final-report-*.md`` 레이아웃일 때만
    형제 ``user-responses/`` 디렉토리를 찾는다(HTML 뷰의 `Export user response`
    가 내려준 파일을 사용자가 거기 저장). 그 외 경로·디렉토리 부재 시 빈 목록.
    carry-in 첨부 본문(``clarification_response_with_sidecars``)과 위저드 안내
    문구의 사이드카 카운트가 같은 한 곳을 보도록 하는 단일 참조점이다.
    """
    responses_dir = source.parent.parent / "user-responses"
    if source.parent.name != "reports" or not responses_dir.is_dir():
        return []
    return sorted(
        p for p in responses_dir.glob("user-response-*.md") if p.is_file()
    )


_ANALYSIS_REPORT_NAME_RE = re.compile(
    r"^final-report-(?P<task_type>project-analysis|feature-analysis|"
    r"change-impact-analysis)-(?P<seq>\d{3})\.md$"
)
_SIDECAR_FRONTMATTER_RE = re.compile(
    r"\A---[ \t]*\r?\n(?P<body>.*?)(?:\r?\n)---[ \t]*(?:\r?\n|\Z)",
    re.DOTALL,
)
_ANALYSIS_REVIEW_HEADING_RE = re.compile(
    r"^## ANALYSIS REVIEW\s*$", re.MULTILINE
)


def _sidecar_frontmatter_value(text: str, key: str) -> str:
    match = _SIDECAR_FRONTMATTER_RE.match(text)
    if match is None:
        return ""
    value = re.search(
        rf"^{re.escape(key)}:\s*(\S.*?)\s*$",
        match.group("body"),
        re.MULTILINE,
    )
    return value.group(1) if value else ""


def _sidecars_for_attachment(source: Path) -> list[Path]:
    sidecars = user_response_sidecars(source)
    report_match = _ANALYSIS_REPORT_NAME_RE.fullmatch(source.name)
    if report_match is None:
        return sidecars
    expected_source = (
        f"runs/{report_match.group('task_type')}/reports/{source.name}"
    )
    ordinary: list[Path] = []
    candidates: list[tuple[Path, str]] = []
    for sidecar in sidecars:
        try:
            text = sidecar.read_text(encoding="utf-8")
        except OSError:
            ordinary.append(sidecar)
            continue
        if _ANALYSIS_REVIEW_HEADING_RE.search(text) is None:
            ordinary.append(sidecar)
            continue
        source_report = _sidecar_frontmatter_value(text, "source-report")
        seq = _sidecar_frontmatter_value(text, "seq")
        if source_report and source_report != expected_source:
            continue
        if seq and seq != report_match.group("seq"):
            continue
        candidates.append((sidecar, text))

    from okstra_ctl.user_response import UserResponseError, parse_analysis_review

    valid: list[Path] = []
    malformed: list[Path] = []
    for sidecar, text in candidates:
        try:
            review = parse_analysis_review(text)
        except UserResponseError:
            malformed.append(sidecar)
            continue
        if review is not None and (
            review.source_report == expected_source
            and review.seq == report_match.group("seq")
        ):
            valid.append(sidecar)
        else:
            malformed.append(sidecar)
    selected = valid if valid else malformed
    return sorted([*ordinary, *selected])


def _sidecar_answer_records(source: Path) -> dict[str, tuple[str, str]]:
    """사이드카 답을 `{id: (value, disposition)}` 로 모은다.

    `disposition` 이 `ANSWER_DISPOSITIONS` 에 속하는 항목만 답으로 센다.
    `reframe` 은 답이 아니므로 집합에서 빠진다. 같은 id 는 이름순 마지막이
    이긴다 — 최신이 reframe 이거나 값이 비면 앞선 답을 지운다.
    """
    from okstra_ctl.user_response import parse_user_response_entries

    answers: dict[str, tuple[str, str]] = {}
    for sidecar in user_response_sidecars(source):
        for entry in parse_user_response_entries(
            sidecar.read_text(encoding="utf-8")
        ):
            if entry.value and entry.disposition in ANSWER_DISPOSITIONS:
                answers[entry.response_id] = (entry.value, entry.disposition)
            else:
                answers.pop(entry.response_id, None)
    return answers


def sidecar_answers(source: Path) -> dict[str, str]:
    """`user-responses/` 사이드카들의 답변을 `{clarification-id: value}` 로 모은다.

    사용자가 답한 항목이 무엇인지 아는 단일 참조점 — carry-in 병합도, 승인
    게이트도, 스킬의 열린 항목 목록도 전부 이 한 곳을 본다.
    """
    return {
        row_id: value
        for row_id, (value, _disposition) in _sidecar_answer_records(source).items()
    }


def sidecar_dispositions(source: Path) -> dict[str, str]:
    """사이드카 답의 처분을 `{clarification-id: disposition}` 로 모은다."""
    return {
        row_id: disposition
        for row_id, (_value, disposition) in _sidecar_answer_records(source).items()
    }


def attached_user_responses_section(source: Path) -> str:
    """`source` 형제 `user-responses/` 사이드카만 모은 `# Attached User Responses`
    섹션 본문. 사이드카 부재 시 빈 문자열.

    plan 본문이 이미 별도 경로(`--approved-plan`)로 참조되는 implementation
    carry-in 에서, 원문을 중복 복사하지 않고 사용자 답변만 instruction-set 에
    첨부할 때 쓴다. `clarification_response_with_sidecars` 와 같은 직렬화 포맷을
    한 곳에서 만들어 두 carry-in 경로가 갈라지지 않게 한다.
    """
    sidecars = _sidecars_for_attachment(source)
    if not sidecars:
        return ""
    parts = ["# Attached User Responses\n"]
    for sidecar in sidecars:
        parts.append(
            f"\n## {sidecar.name}\n\n"
            f"{sidecar.read_text(encoding='utf-8').strip()}\n"
        )
    return "".join(parts)


def clarification_response_with_sidecars(source: Path) -> str:
    """clarification-response 본문에 `user-responses/` 사이드카를 덧붙인 본문.

    `resume-clarification` 의 설계된 입력은 사용자가 §1 의 `User input` 열을
    채운 **직전 final-report 자체**다. 그 전문을 그대로 복사하면 리포트가
    instruction-set 안에 두 번째로 존재하게 되고, report-writer 는 같은 내용을
    `clarification-response.md` 로 한 번, 직전 리포트 경로로 다시 한 번 읽는다.
    리포트는 run 이 누적될수록 커지므로 이 중복은 스스로 악화된다(실측: 283K
    리포트가 892K 의 중복 읽기를 만들어 report-writer 를 timeout 시켰다).

    그래서 소스가 final-report 일 때는 답변이 실린 clarification 행만 잘라내고
    원문은 경로로 가리킨다 — `attached_user_responses_section` 이 implementation
    carry-in 에서 이미 쓰는 "원문은 경로로, 답변만 첨부" 규칙과 같다. 행을 어디서
    읽는지는 스키마가 정한다(v1 은 §1 표, v2 는 data.json). clarification 을
    아예 담지 않은 소스(사용자가 직접 쓴 답변 파일)만 원문 그대로 복사한다.
    """
    text = source.read_text(encoding="utf-8")
    section = attached_user_responses_section(source)
    answers = sidecar_answers(source)
    body = _clarification_carry_body(source, text, answers)
    if not section:
        return body
    return body.rstrip("\n") + "\n\n---\n\n" + section


SECTION_1_HEADING = "## 1. Clarification Items"
_SECTION_1_TABLE_HEADER = (
    "| Record | Statement | Expected form | User input |\n"
    "|---|---|---|---|"
)
_SECTION_1_EMPTY_STATE = "- The source report recorded no clarification items."


def _clarification_carry_body(
    source: Path, text: str, answers: dict[str, str]
) -> str:
    """final-report 소스는 §1 + 원문 포인터로 좁히고, 그 외는 원문 그대로.

    §1 이 있으면 사이드카 답변을 그 표의 `User input` 열에 병합해, 답이 표 안에
    자리하도록 한다(파일 헤더가 선언하는 "답은 User input 열에" 계약을 실제로
    참으로 만든다)."""
    carried = _carry_section_1(source, text)
    if carried is None:
        return text
    heading, section_body = carried
    if answers:
        section_body = _reconcile_user_input(section_body, answers)
    return (
        "# Clarification Response (carry-in)\n\n"
        f"- Source report: `{source}`\n"
        "- This file carries **only** the source report's Clarification Items "
        "section; the report itself is read from the path above when a phase "
        "needs it. Do not re-read the source report to find the answers — they "
        "are in the `User input` column below.\n\n"
        f"{heading}\n{section_body}\n"
    )


def _carry_section_1(source: Path, text: str) -> Optional[tuple[str, str]]:
    """carry-in 본문에 실을 (헤딩, §1 본문). clarification 을 담지 않은 소스면
    ``None``.

    schema-v2 는 행을 data.json 에 들고 AI 마크다운에는 §1 표가 없다. §1 슬라이스
    만 보던 동안 v2 소스는 "좁힐 것이 없다" 로 판정돼 **리포트 전문이 그대로
    복사**됐다 — 이 좁히기가 막으려던 바로 그 중복이다. carry-in 은 파생 문서이고
    다운스트림(승인 게이트·프롬프트 빌더·검증 워커)이 §1 표 하나만 읽으므로, v2
    행도 같은 표로 렌더한다."""
    data = _structured_report_data(source)
    if data is not None:
        entries = data.get("clarificationItems")
        return SECTION_1_HEADING, _structured_section_1_body(
            entries if isinstance(entries, list) else []
        )
    slice_ = _section_1_slice(text)
    if slice_ is None:
        return None
    heading = SECTION_HEADING_PATTERN.search(text)
    assert heading is not None  # _section_1_slice returned a slice
    return heading.group(0), slice_.rstrip()


def _structured_section_1_body(entries: list) -> str:
    """schema-v2 `clarificationItems[]` 를 §1 표 본문으로.

    메타 셀은 렌더러가 쓰는 모양 그대로다 — `Status:` 는 따옴표 없이 써야
    `_reconcile_user_input` 이 답을 병합하면서 상태를 answered 로 넘길 수 있다."""
    rows = [e for e in entries if isinstance(e, dict) and e.get("id")]
    if not rows:
        return f"\n{_SECTION_1_EMPTY_STATE}"
    lines = ["", _SECTION_1_TABLE_HEADER]
    for entry in rows:
        meta = (
            f"**{entry['id']}**"
            f"<br>Ticket: `{to_cell_text(entry.get('ticketId'))}`"
            f"<br>Kind: `{to_cell_text(entry.get('kind'))}`"
            f"<br>Blocks: `{to_cell_text(entry.get('blocks'))}`"
            f"<br>Status: {to_cell_text(entry.get('status'))}"
        )
        lines.append(
            f"| {meta} | {to_cell_text(entry.get('statement'))} "
            f"| {to_cell_text(entry.get('expectedForm'))} "
            f"| {to_cell_text(entry.get('userInput'))} |"
        )
    return "\n".join(lines)


# The final-report renderer writes `Status: open` / `Status: answered` unquoted
# in the stacked meta cell; only those two are unresolved. Resolve in place so
# the meta cell's other fields (ID, Ticket, Kind, Blocks) are left untouched.
_STATUS_ANSWER_RE = re.compile(r"(Status:\s*)(?:open|answered)\b", re.IGNORECASE)


def _locate_user_input_column(lines: list[str]) -> tuple[int, int]:
    """§1 데이터 표의 헤더 줄 인덱스와 `User input` 열 인덱스. 표가 없으면 (-1, -1)."""
    for idx, line in enumerate(lines):
        if not line.lstrip().startswith("|"):
            continue
        cells = [c.lower() for c in _split_pipe_row(line)]
        if "user input" in cells and any(c.startswith("statement") for c in cells):
            return idx, cells.index("user input")
    return -1, -1


def _reconcile_row(line: str, ui_col: int, answers: dict[str, str]) -> str:
    """답이 있고 open/answered 인 행이면 `User input` 칸을 그 답으로 채우고 Status 를
    answered 로 바꾼 줄을, 그 외에는 원본 줄을 그대로 돌려준다.

    칸에 이미 값이 있어도 사용자의 사이드카 답이 이긴다. 그 칸을 채우는 것은
    run 자신(직전 렌더가 옮겨 적은 값)이고, 사용자가 나중에 답을 바꾸면 둘이
    갈라진다 — 사용자가 쓴 쪽을 정본으로 삼지 않으면 run 이 자기가 적어둔 값으로
    계속 되돌아간다.

    판정은 앵커/백틱을 벗긴 셀(`_split_pipe_row`)로 — 그래야 `_meta_id` 가 스크롤
    앵커의 소문자 slug 대신 진짜 대문자 ID 를 읽는다. 재조립은 원본 셀
    (`split_pipe_row`)로 해서 앵커를 보존한다."""
    norm = _split_pipe_row(line)
    item = parse_meta_cell(norm[0]) if norm else None
    if item is None or item.row_id not in answers:
        return line
    if item.status not in UNRESOLVED_STATUSES:
        return line
    raw = split_pipe_row(line)
    if not 0 <= ui_col < len(raw):
        return line
    raw[ui_col] = answers[item.row_id]
    raw[0] = _STATUS_ANSWER_RE.sub(r"\1answered", raw[0])
    return "| " + " | ".join(to_cell_text(c) for c in raw) + " |"


def _reconcile_user_input(section: str, answers: dict[str, str]) -> str:
    """§1 표에서 사이드카 답이 있는 미해결 행의 `User input` 칸을 답으로 채우고
    Status 를 answered 로 바꾼 §1 본문을 돌려준다.

    답의 정본 위치를 §1 표 안으로 옮긴다 — 표만 읽는 승인 게이트·프롬프트
    빌더·검증 워커가 모두 답을 보게 하려는 것. 사이드카는 §1 표 밖 별도 섹션에만
    있어서 표만 신뢰하는 소비자는 그 답을 놓쳤다."""
    lines = section.splitlines()
    header_idx, ui_col = _locate_user_input_column(lines)
    if header_idx < 0:
        return section
    out = list(lines)
    body = False
    for i in range(header_idx + 1, len(lines)):
        line = lines[i]
        if not line.lstrip().startswith("|"):
            if body:
                break
            continue
        if is_separator_row(line):
            body = True
            continue
        if body:
            out[i] = _reconcile_row(line, ui_col, answers)
    return "\n".join(out)