"""Post-hoc checks for neutral lead lifecycle and live-activity contracts.

Design: docs/superpowers/specs/2026-06-10-blocking-contract-posthoc-conformance-design.md

| Check | Contract | Evidence |
|-------|----------|----------|
| 1. lead PROGRESS checkpoints | prompts/lead/okstra-lead-contract.md "Progress reporting (BLOCKING)" | selected adapter evidence source |
| 2. worker-provider heartbeat | worker provider contract | audit sidecar `- PROGRESS: <stage> <ISO>` lines |
| 3. implementation entry guard | prompts/lead/okstra-lead-contract.md "Entry guard (BLOCKING)" + prompts/launch.template.md "Host Orchestration Rules" (only when the run staged the rules file) | selected adapter evidence source |

Evidence rules prevent false passes:
- `claude-jsonl` accepts only assistant text and file-read tool-use records (a
  `Read` call, or a shell command naming the file), excluding injected skill text
  and sidechain records.
- Session evidence is scoped to the current run window so a previous run cannot
  satisfy the neutral lead lifecycle contract.
- Checkpoints emitted after validation begins (`phase-7-teardown`, `complete`)
  cannot be observed and are not required.
"""
from __future__ import annotations

import json
import os
import re
import sys
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Mapping, NamedTuple

# scripts/ (repo) and python/ (installed under ~/.okstra/lib) are not packages;
# insert whichever exists so okstra_ctl is importable directly.
_VALIDATORS_DIR = Path(__file__).resolve().parent
for _ssot_dir in (_VALIDATORS_DIR.parent / "scripts", _VALIDATORS_DIR.parent / "python"):
    if _ssot_dir.is_dir() and str(_ssot_dir) not in sys.path:
        sys.path.insert(0, str(_ssot_dir))

from okstra_ctl.worker_heartbeat import (  # noqa: E402
    HEARTBEAT_LINE_RE,
    IN_STAGE_PREFIX,
    max_gap_seconds_after,
)
from okstra_ctl.lead_events import (  # noqa: E402
    LeadEvent,
    LeadEventParseError,
    read_lead_events,
)
from okstra_ctl.agent_activity import ACTIVITY_FIELDS  # noqa: E402
from okstra_ctl.domain.host import HostNotRegistered  # noqa: E402
from okstra_ctl.final_report_paths import final_report_data_path  # noqa: E402
from okstra_ctl.registry.host_registry import default_host_registry  # noqa: E402
from okstra_ctl.wrapper_status import read_wrapper_status  # noqa: E402

_DISPATCHED_STATUSES = {"completed", "timeout", "error", "in-progress"}
_WORKER_DISPATCH_MODES = {"cli-wrapper", "mixed"}
_ATTEMPTED_STATUSES = {"completed", "timeout", "error"}
_LIVENESS_AUDIT_HEARTBEAT = "audit-heartbeat"
_LIVENESS_WRAPPER_STATUS = "wrapper-status"
_REPORT_WRITER_HEARTBEAT_STAGES = {
    "started",
    "required-reading-complete",
    "synthesis-start",
    "data-json-write-start",
    "render-start",
    "write-result-start",
}

# lead 의 체크포인트 라인 — assistant text 블록 안에서 line-anchored 로만 인정.
# lead 는 계약상 raw 로 emit 해야 하지만 실측(dev-9902)에서 `` `PROGRESS: ...` ``
# 인라인 코드나 코드펜스로 감싸 emit 하는 경우가 있어, 라인 양끝의 backtick 과
# 선행 들여쓰기를 허용한다. phase 는 backtick 을 포함하지 않는다.
_PROGRESS_LINE_RE = re.compile(
    r"^[ \t]*`*[ \t]*PROGRESS:[ \t]+(?P<phase>[^\s`]+)(?P<rest>[^`\n]*)`*[ \t]*$",
    re.MULTILINE,
)
_ACTIVITY_ID_RE = re.compile(r"^A-(\d{3,})$")
_TARGETED_REVERIFICATION_REF_RE = re.compile(
    r"^plan-body-verification:round-(\d+)$"
)
_CLARIFICATION_ID_RE = re.compile(r"^C-\d{3,}$")
_WORKER_FIELD_RE = re.compile(
    r"(?:^|\s)worker=(?P<worker>.+?)(?=\s+[a-zA-Z][a-zA-Z0-9_.-]*=|$)"
)
_COMPLETED_STATUS_RE = re.compile(r"(?:^|\s)status=completed(?:\s|$)")

# heartbeat 라인 shape 과 cadence 예산은 okstra_ctl.worker_heartbeat 정본을 쓴다 —
# `okstra worker-liveness` 가 run 도중 같은 판정을 내리므로 정의가 갈리면 안 된다.
_HEARTBEAT_LINE_RE = HEARTBEAT_LINE_RE
_IN_STAGE_PREFIX = IN_STAGE_PREFIX

_LEAD_CONTRACT_CITE = "prompts/lead/okstra-lead-contract.md 'Entry guard (BLOCKING)'"
_HOST_RULES_CITE = "prompts/launch.template.md 'Host Orchestration Rules'"


class _EntryGuardRead(NamedTuple):
    """implementation entry guard 가 Read 를 요구하는 파일 한 건."""

    basename: str
    # (이 Read 가 앞서야 할 PROGRESS 체크포인트, 실패 메시지에 쓸 시점 표현).
    # None 이면 순서 제약이 없고 run 윈도우 안의 존재만 검사한다.
    anchor: tuple[str, str] | None
    cite: str
    # True 면 run 이 이 파일을 instruction-set 에 스테이징했을 때만 요구한다.
    staged_per_run: bool = False


_ENTRY_GUARD_READS = (
    _EntryGuardRead(
        "_implementation-executor.md", ("phase-6-synthesis", "at Phase 5"), _LEAD_CONTRACT_CITE
    ),
    _EntryGuardRead(
        "_implementation-verifier.md", ("phase-6-synthesis", "at Phase 5"), _LEAD_CONTRACT_CITE
    ),
    _EntryGuardRead(
        "_implementation-deliverable.md", ("phase-7-persist", "at Phase 6"), _LEAD_CONTRACT_CITE
    ),
    # lead 프로파일 sidecar 가 아니라, run 이 instruction-set 에 스테이징한 host
    # orchestration 규칙이다. 세 가지가 sidecar 와 다르다.
    #
    # anchor 없음: 이 파일은 `render-bundle` 이 만든다(run.py `_write_instruction_set_sources`).
    #   그 앞에서 발화하는 gate(Step 5.1 waiver 제안)에는 아직 파일이 없고, 뒤에
    #   오는 gate 는 run 의 PROGRESS 체크포인트와 시간 관계가 고정돼 있지 않다.
    #   증거가 뒷받침하는 주장은 "run 윈도우 안에 Read 가 최소 한 번" 뿐이다.
    # staged_per_run: 파일이 없는 번들 — 이 변경 전에 렌더된 run, 또는 payload 에
    #   prompts/host-orchestration/<task-type>.md 가 없는 설치본 — 은 읽을 수 없는
    #   파일 때문에 막히면 안 된다. 규칙이 없는 task-type 에 conformance 가 규칙을
    #   요구하지 않는다는 원칙과 같은 판단이다.
    _EntryGuardRead(
        "host-orchestration-rules.md", None, _HOST_RULES_CITE, staged_per_run=True
    ),
)

# 환경으로 선택되는 어댑터라 lead 의 런타임이 이 파일을 가르쳐 주지 않는다.
# 읽지 않은 lead 는 자기 런타임이 아는 방식 — cmux 경로에서는 okstra 가 소유한
# 디스패치를 host 네이티브로 가로채는 방식 — 으로 되돌아간다.
CMUX_ADAPTER_BASENAME = "cmux.md"
CMUX_ADAPTER_NAME = "cmux"
_CMUX_ADAPTER_CITE = "prompts/lead/adapters/cmux.md"

# Read 증거는 basename 으로 거른다 — 절대 경로는 레이어(repo / runtime / 설치본)
# 마다 다르지만 basename 은 동일하다. 목록이 갈리지 않도록 기대치에서 파생한다.
_TRACKED_READ_BASENAMES = (
    *(row.basename for row in _ENTRY_GUARD_READS),
    CMUX_ADAPTER_BASENAME,
)


# 셸 read 도 증거다. 계약이 요구하는 것은 "이 run 안에서 그 파일의 내용을 실제로
# 적재했는가" 이고, `cat` / `sed -n` 는 Read 도구와 동일하게 그것을 만족한다.
# 도구 이름만 인정하면, 호스트가 "가능하면 Bash 로 파일을 읽으라"고 지시하는 모드
# (Claude Code auto 모드)에서는 계약을 그대로 이행한 lead 가 매번
# `contract-violated` 로 끝난다 — 두 지시가 정면으로 충돌한다.
#
# basename 이 경로 토큰으로 등장할 때만 인정한다. 트리 전체를 훑는 명령
# (`grep -rn "..." prompts/`)은 그 파일을 지목하지 않았으므로 매칭되지 않고,
# 파일을 인자로 준 명령은 매칭된다 — 후자는 실제로 그 파일을 읽는다.
_SHELL_READ_TOOL_NAME = "Bash"
_SHELL_READ_PATTERNS = {
    base: re.compile(rf"(?:^|[\s'\"`=(/]){re.escape(base)}(?:$|[\s'\"`);:,])")
    for base in _TRACKED_READ_BASENAMES
}


def _read_evidence_basenames(block: dict) -> tuple[str, ...]:
    """tool_use 블록 하나가 증거로 인정되는 tracked basename 들."""
    name = block.get("name")
    payload = block.get("input") or {}
    if name == "Read":
        base = Path(str(payload.get("file_path") or "")).name
        return (base,) if base in _TRACKED_READ_BASENAMES else ()
    if name == _SHELL_READ_TOOL_NAME:
        command = str(payload.get("command") or "")
        if not command:
            return ()
        return tuple(
            base for base, pattern in _SHELL_READ_PATTERNS.items()
            if pattern.search(command)
        )
    return ()


@dataclass
class SessionConformanceResult:
    errors: list[str] = field(default_factory=list)
    # Conformance findings that are reported without failing the run: the
    # PROGRESS narration lines. A missing checkpoint line means the run is
    # harder to follow after the fact, not that any of its work is wrong, and
    # by the time this validator runs the line can no longer be emitted — the
    # session that would have written it has ended. Checks that verify a
    # *claim* (a user confirmation the report says it obtained) stay in
    # `errors`.
    advisories: list[str] = field(default_factory=list)

    @property
    def ok(self) -> bool:
        return not self.errors


@dataclass
class _LeadEvidence:
    progress: list[tuple[str, str, str]] = field(default_factory=list)  # (ts, phase-id, line)
    sidecar_reads: dict[str, list[str]] = field(default_factory=dict)  # basename -> [ts]
    activities: list[LeadEvent] = field(default_factory=list)
    scanned_files: list[Path] = field(default_factory=list)
    window: tuple[str | None, str | None] = (None, None)


def _ensure_token_usage_importable() -> None:
    """okstra_token_usage 패키지를 레이아웃별(repo/scripts, runtime/python,
    OKSTRA_PYTHONPATH)로 해소 — validate-run.py `_import_token_usage` 와 동일 후보."""
    here = Path(__file__).resolve().parent
    candidates = [here.parent / "scripts", here.parent / "python"]
    env_pp = os.environ.get("OKSTRA_PYTHONPATH", "").strip()
    if env_pp:
        candidates.append(Path(env_pp))
    for candidate in candidates:
        if candidate.is_dir() and (candidate / "okstra_token_usage").is_dir():
            if str(candidate) not in sys.path:
                sys.path.insert(0, str(candidate))
            break


def _norm(value: str) -> str:
    return re.sub(r"[^a-z0-9]", "", (value or "").lower())


def _is_report_writer(worker: dict) -> bool:
    return "reportwriter" in _norm(str(worker.get("role", ""))) or "reportwriter" in _norm(
        str(worker.get("workerId", ""))
    )


def _worker_role(value: str) -> str | None:
    """역할 필드 전체에서 Unicode 대소문자 차이만 제거한다."""
    return value.casefold() if value else None


def _worker_roles(worker: dict) -> set[str]:
    """명단 `role` 과 `workerId` 를 활동 `agent` 가 쓸 수 있는 별칭으로 펼친다.

    활동은 `codex` / `codex-worker` 를 쓰고 명단은 `Codex worker` 를 쓴다.
    역할 문자열만 보면 둘은 다른 사람이 된다.
    """
    roles: set[str] = set()
    for raw in (worker.get("role"), worker.get("workerId")):
        role = _worker_role(str(raw or ""))
        if not role:
            continue
        compact = role.replace(" ", "-")
        roles.add(role)
        roles.add(compact)
        if role.endswith(" worker"):
            roles.add(role[: -len(" worker")])
        if compact.endswith("-worker"):
            roles.add(compact[: -len("-worker")])
    return roles


def _analysis_workers(team_state: Mapping[str, Any]) -> list[dict]:
    workers = [
        worker
        for worker in (team_state.get("workers") or [])
        if isinstance(worker, dict)
    ]
    return [worker for worker in workers if not _is_report_writer(worker)]


def _scan_one_jsonl(
    path: Path, since: str | None, until: str | None
) -> tuple[list[tuple[str, str, str]], dict[str, list[str]], str | None]:
    """jsonl 한 파일에서 (progress, sidecar reads, agentName) 을 추출한다."""
    from okstra_token_usage.paths import ts_in_window

    progress: list[tuple[str, str, str]] = []
    reads: dict[str, list[str]] = {}
    agent_name: str | None = None
    try:
        fh = path.open(encoding="utf-8")
    except OSError:
        return progress, reads, agent_name
    with fh:
        for raw in fh:
            try:
                rec = json.loads(raw)
            except (json.JSONDecodeError, UnicodeDecodeError):
                continue
            if agent_name is None and rec.get("agentName"):
                agent_name = rec["agentName"]
            if rec.get("type") != "assistant" or rec.get("isSidechain"):
                continue
            ts = rec.get("timestamp") or ""
            if ts and not ts_in_window(ts, since, until):
                continue
            msg = rec.get("message") or {}
            for block in msg.get("content") or []:
                if not isinstance(block, dict):
                    continue
                if block.get("type") == "text":
                    for m in _PROGRESS_LINE_RE.finditer(block.get("text") or ""):
                        line = f"PROGRESS: {m.group('phase')}{m.group('rest')}".rstrip()
                        progress.append((ts, m.group("phase"), line))
                elif block.get("type") == "tool_use":
                    for base in _read_evidence_basenames(block):
                        reads.setdefault(base, []).append(ts)
    return progress, reads, agent_name


def _collect_lead_evidence(
    team_state: dict,
    team_state_path: Path,
    run_manifest: Mapping[str, Any],
    project_root: Path,
    task_type: str,
    suffix: str | None,
    projects_root: Path | None,
) -> tuple[_LeadEvidence | None, str | None]:
    """lead 후보 jsonl 을 스캔해 증거를 모은다.

    후보 = {기록된 lead.sessionId} ∪ {team 태그는 있으나 agentName 이 없는 세션}.
    후자는 `claude --resume` 으로 lead 세션이 fork 된 경우(새 sessionId,
    agentName 없음)를 흡수한다 — worker 세션은 agentName 이 있어 자연 배제된다.
    """
    from okstra_token_usage.claude import find_claude_team_sessions
    from okstra_token_usage.collect import (
        resolve_run_window,
        resolve_team_needles_with_source,
    )
    from okstra_token_usage.paths import claude_project_dir

    since, until = resolve_run_window(team_state_path, team_state, relax_start=False)
    lead_sid = (team_state.get("lead") or {}).get("sessionId") or ""
    team_needles, _needle_source = resolve_team_needles_with_source(
        team_state, project_root, since, until, projects_root=projects_root
    )
    sessions = find_claude_team_sessions(
        project_root, team_needles, lead_sid, projects_root=projects_root
    )
    # `claude --resume` 로 fork 되거나 leadSessionIds[] 에 기록된 lead 세대는
    # team 태그 needle 로 안 잡힐 수 있어 명시적으로 후보에 추가한다.
    proj_dir = claude_project_dir(project_root, projects_root)
    for sid in (team_state.get("leadSessionIds") or []):
        if not isinstance(sid, str) or not sid:
            continue
        candidate = proj_dir / f"{sid}.jsonl"
        if candidate.is_file():
            sessions.setdefault(sid, candidate)
    evidence = _LeadEvidence(window=(since, until))
    for sid, path in sorted(sessions.items()):
        progress, reads, agent_name = _scan_one_jsonl(path, since, until)
        if agent_name and sid != lead_sid:
            continue  # agentName 이 찍힌 세션은 worker — lead 후보에서 제외
        evidence.scanned_files.append(path)
        evidence.progress.extend(progress)
        for base, ts_list in reads.items():
            evidence.sidecar_reads.setdefault(base, []).extend(ts_list)
    if not evidence.scanned_files:
        return None, (
            f"lead session jsonl not found under {proj_dir} "
            f"(lead.sessionId={lead_sid or '<empty>'}) — selected adapter evidence "
            "source `claude-jsonl` cannot verify the PROGRESS checkpoint / "
            "implementation entry-guard conformance cannot be verified, which "
            "fails the run (same principle as the token-usage accuracy contract)."
        )
    evidence.progress.sort()
    for ts_list in evidence.sidecar_reads.values():
        ts_list.sort()
    if _is_activity_contract_v1_planning(run_manifest):
        activities, events_path, error = _read_scoped_lead_activities(
            team_state, run_manifest, project_root, task_type, suffix
        )
        if error:
            return None, error
        evidence.activities.extend(activities)
        if events_path is not None:
            evidence.scanned_files.append(events_path)
    return evidence, None


def _resolve_lead_events_path(
    team_state: Mapping[str, Any],
    run_manifest: Mapping[str, Any],
    project_root: Path,
) -> tuple[Path | None, str | None]:
    raw = run_manifest.get("leadEventsPath") or team_state.get("leadEventsPath") or (
        (team_state.get("artifacts") or {}).get("leadEventsPath")
        if isinstance(team_state.get("artifacts"), dict)
        else ""
    )
    if not raw:
        return None, (
            "artifact lead event log path missing from team-state "
            "(`leadEventsPath`) — selected adapter evidence source `artifact-only` "
            "cannot verify conformance."
        )
    path = Path(str(raw))
    if not path.is_absolute():
        path = project_root / path
    if not path.is_file():
        return None, (
            f"artifact lead event log not found: {path} — selected adapter evidence "
            "source `artifact-only` cannot verify conformance."
        )
    return path, None


def _event_matches_run(
    event: LeadEvent,
    team_state: Mapping[str, Any],
    run_manifest: Mapping[str, Any],
    task_type: str,
    run_seq: str,
) -> bool:
    task_key = str(run_manifest.get("taskKey") or team_state.get("taskKey") or "")
    expected_runtime = str(
        run_manifest.get("leadRuntime")
        or team_state.get("leadRuntime")
        or "claude-code"
    )
    if event.lead_runtime != expected_runtime:
        return False
    if task_key and event.task_key != task_key:
        return False
    return event.task_type == task_type and event.run_seq == run_seq


def _run_sequence(run_manifest: Mapping[str, Any], suffix: str | None) -> str:
    sequences = run_manifest.get("runSequencesByCategory")
    if isinstance(sequences, Mapping):
        value = sequences.get("manifests")
        if isinstance(value, str) and value:
            return value
    if suffix and "-" in suffix:
        return suffix.rsplit("-", 1)[1]
    return ""


def _read_scoped_lead_activities(
    team_state: Mapping[str, Any],
    run_manifest: Mapping[str, Any],
    project_root: Path,
    task_type: str,
    suffix: str | None,
) -> tuple[list[LeadEvent], Path | None, str | None]:
    run_seq = _run_sequence(run_manifest, suffix)
    if not run_seq:
        return [], None, (
            "activity contract cannot scope lead events because the run sequence "
            "is missing from the run manifest and team-state filename."
        )
    events_path, error = _resolve_lead_events_path(
        team_state, run_manifest, project_root
    )
    if error:
        return [], None, error
    try:
        events = read_lead_events(events_path)
    except LeadEventParseError as exc:
        return [], events_path, (
            "artifact lead event log is malformed — selected adapter evidence "
            f"source cannot verify conformance: {exc}"
        )
    activities = [
        event
        for event in events
        if event.event_type == "activity"
        and _event_matches_run(
            event, team_state, run_manifest, task_type, run_seq
        )
    ]
    return activities, events_path, None


def _progress_line_from_event(event) -> tuple[str, str, str] | None:
    details = event.details
    phase = details.get("phase")
    if not isinstance(phase, str) or not phase:
        line_candidate = details.get("line")
        if isinstance(line_candidate, str):
            match = _PROGRESS_LINE_RE.search(line_candidate)
            if match:
                phase = match.group("phase")
        if not isinstance(phase, str) or not phase:
            return None
    line = details.get("line")
    if not isinstance(line, str) or not line:
        message = details.get("message")
        tail = f" {message}" if isinstance(message, str) and message else ""
        line = f"PROGRESS: {phase}{tail}"
    return (event.timestamp, phase, line)


def _sidecar_read_from_event(event) -> tuple[str, str] | None:
    details = event.details
    if event.event_type != "sidecar-read" and details.get("kind") != "implementation-sidecar":
        return None
    basename = details.get("basename")
    if not isinstance(basename, str) or not basename:
        raw_path = details.get("path") or details.get("filePath")
        basename = Path(str(raw_path or "")).name
    if basename not in _TRACKED_READ_BASENAMES:
        return None
    return (basename, event.timestamp)


def _collect_artifact_lead_evidence(
    team_state: dict,
    run_manifest: Mapping[str, Any],
    project_root: Path,
    task_type: str,
    suffix: str | None,
) -> tuple[_LeadEvidence | None, str | None]:
    if not suffix or "-" not in suffix:
        return None, (
            "artifact lead event log cannot be scoped because team-state filename "
            "does not expose a run artifact suffix."
        )
    events_path, error = _resolve_lead_events_path(
        team_state, run_manifest, project_root
    )
    if error:
        return None, error

    try:
        events = read_lead_events(events_path)
    except LeadEventParseError as exc:
        return None, (
            "artifact lead event log is malformed — selected adapter evidence "
            f"source `artifact-only` cannot verify conformance: {exc}"
        )

    run_seq = _run_sequence(run_manifest, suffix)
    evidence = _LeadEvidence(scanned_files=[events_path])
    for event in events:
        if not _event_matches_run(
            event, team_state, run_manifest, task_type, run_seq
        ):
            continue
        if event.event_type == "activity":
            evidence.activities.append(event)
        elif event.event_type in ("progress", "progress-checkpoint"):
            progress = _progress_line_from_event(event)
            if progress is not None:
                evidence.progress.append(progress)
        elif event.event_type in ("artifact-read", "sidecar-read"):
            read = _sidecar_read_from_event(event)
            if read is not None:
                basename, timestamp = read
                evidence.sidecar_reads.setdefault(basename, []).append(timestamp)

    evidence.progress.sort()
    for ts_list in evidence.sidecar_reads.values():
        ts_list.sort()
    return evidence, None


def _conformance_evidence_source(
    team_state: dict,
) -> tuple[str | None, str | None]:
    dispatch_mode = str(team_state.get("dispatchMode", "")).strip()
    if dispatch_mode in _WORKER_DISPATCH_MODES:
        return "artifact-only", None

    adapter = team_state.get("leadAdapter")
    if isinstance(adapter, dict):
        accounting = str(adapter.get("sessionAccounting", "")).strip()
        if accounting in {"claude-jsonl", "artifact-only"}:
            return accounting, None
        if accounting:
            return None, f"unsupported sessionAccounting: `{accounting}`"

    legacy_runtime = str(team_state.get("leadRuntime", "") or "claude-code")
    try:
        accounting = default_host_registry().resolve(legacy_runtime).descriptor.session_accounting
    except HostNotRegistered:
        accounting = "claude-jsonl"
    if accounting == "artifact-only":
        return "artifact-only", None
    return "claude-jsonl", None


def _convergence_rounds_ran(run_dir: Path, suffix: str | None) -> bool:
    """이 run 의 convergence state artifact 가 실제 round 를 1회 이상 돌았는지.
    auto-disable(`totalRounds: 0`)·artifact 부재는 phase-5.5 라인을 요구하지 않는다."""
    if not suffix:
        return False
    path = run_dir / "state" / f"convergence-{suffix}.json"
    try:
        doc = json.loads(path.read_text())
    except (OSError, json.JSONDecodeError):
        return False
    return isinstance(doc, dict) and (doc.get("totalRounds") or 0) >= 1


def _plan_body_rounds_ran(run_dir: Path, suffix: str | None) -> int:
    """이 run 의 plan-body 검증이 실제로 돈 라운드 수.

    각 라운드는 새 워커 배치를 띄우므로 라운드마다 배치 경계가 하나씩 생긴다.
    Phase 6 이후에 도는 구간이라 기존 배치-정리 강제(수렴 라운드 1 직전 /
    report-writer 디스패치 직전)의 바깥이었고, 그래서 self-fix 를 여러 라운드
    돈 run 은 매 라운드의 완료 워커가 그대로 남았다."""
    if not suffix:
        return 0
    try:
        doc = json.loads(_plan_body_state_path(run_dir, suffix).read_text())
    except (OSError, json.JSONDecodeError):
        return 0
    if not isinstance(doc, dict):
        return 0
    # 계약은 이 값을 중첩 `planBodyVerification`(자가수정 루프가 끝난 뒤의 최종
    # 상태)에 둔다. 최상위는 라운드별 이력이라 라운드 수의 자리가 아니다.
    projection = doc.get("planBodyVerification")
    rounds = projection.get("roundCount") if isinstance(projection, dict) else None
    return rounds if isinstance(rounds, int) and rounds > 0 else 0


def _plan_body_state_path(run_dir: Path, suffix: str) -> Path:
    """이 run 의 plan-body 상태 파일.

    `suffix` 는 team-state 이름에서 나온 **state** seq 다. `okstra_ctl/paths.py`
    가 이 파일을 같은 seq 로 정의하므로 둘이 같은 곳을 가리킨다 — 리드가 리포트
    seq 로 쓰면 여기서 못 찾고, 라운드 수가 0 으로 읽힌다.
    """
    return run_dir / "state" / f"plan-body-verification-{suffix}.json"


def _ids_reported_as_asked(report_path: Path) -> list[str]:
    """리포트가 "사용자에게 물었다"고 기록한 열린 승인 차단 행의 id."""
    doc = _read_report_data(report_path)
    asked: list[str] = []
    for row in doc.get("clarificationItems") or []:
        if not isinstance(row, dict):
            continue
        if row.get("blocks") != "approval" or row.get("status") != "open":
            continue
        if not str(row.get("userConfirmation") or "").startswith("asked-"):
            continue
        row_id = str(row.get("id") or "").strip()
        if row_id:
            asked.append(row_id)
    return asked


def _is_activity_contract_v1_planning(
    run_manifest: Mapping[str, Any],
) -> bool:
    return (
        run_manifest.get("activityContractVersion") == 1
        and run_manifest.get("taskType") == "implementation-planning"
    )


def _activity_index(events: list[LeadEvent]) -> dict[str, list[LeadEvent]]:
    indexed: dict[str, list[LeadEvent]] = {}
    for event in events:
        kind = str(event.details.get("kind") or "")
        indexed.setdefault(kind, []).append(event)
    return indexed


def _read_report_data(report_path: Path) -> Mapping[str, Any]:
    """`--report` 가 data.json · markdown · html 이어도 같은 레코드를 연다.

    Phase 7 는 data.json 을 넘긴다. `.md` 만 받던 동안 투영과 self-fix 횟수가
    빈 객체에서 나와 `projected=<missing>` / `selfFixRoundsApplied=0` 이 됐다.
    """
    try:
        data = json.loads(final_report_data_path(report_path).read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    return data if isinstance(data, Mapping) else {}


def _progress_workers(
    evidence: _LeadEvidence,
    phase: str,
    *,
    completed_only: bool = False,
) -> set[str]:
    workers: set[str] = set()
    for _timestamp, event_phase, line in evidence.progress:
        if event_phase != phase:
            continue
        if completed_only and _COMPLETED_STATUS_RE.search(line) is None:
            continue
        worker = _progress_worker(line)
        if worker is not None:
            workers.add(worker)
    return workers


def _progress_worker(line: str) -> str | None:
    match = _WORKER_FIELD_RE.search(line)
    return match.group("worker") if match else None


def _activity_has_agent(events: list[LeadEvent], roles: set[str]) -> bool:
    return any(
        _worker_role(str(event.details.get("agent") or "")) in roles
        for event in events
    )


def _roster_roles_for_value(value: str, workers: list[dict]) -> set[str]:
    candidate = _worker_role(value)
    for worker in workers:
        roles = _worker_roles(worker)
        if candidate in roles:
            return roles
    return {candidate} if candidate else set()


def _check_activity_id_order(events: list[LeadEvent], errors: list[str]) -> None:
    activity_ids = [str(event.details.get("activityId") or "") for event in events]
    invalid = [
        value or "<empty>"
        for value in activity_ids
        if not _ACTIVITY_ID_RE.fullmatch(value)
    ]
    if invalid:
        errors.append(
            "activity contract: activityId values must use A-NNN format; "
            f"invalid={invalid}."
        )
    duplicates = sorted(
        {value for value in activity_ids if activity_ids.count(value) > 1}
    )
    if duplicates:
        errors.append(
            "activity contract: activityId values must be unique; "
            f"duplicates={duplicates}."
        )
    if invalid:
        return
    numbers = [
        int(_ACTIVITY_ID_RE.fullmatch(value).group(1))
        for value in activity_ids
    ]
    if any(current <= previous for previous, current in zip(numbers, numbers[1:])):
        errors.append(
            "activity contract: activityId values must be in strict A-NNN order; "
            f"recorded={activity_ids}."
        )


def _check_projected_agent_activity(
    events: list[LeadEvent],
    report_data: Mapping[str, Any],
    errors: list[str],
) -> None:
    expected = [
        {field: event.details.get(field) for field in ACTIVITY_FIELDS}
        for event in events
    ]
    raw_projected = report_data.get("agentActivity")
    projected = (
        [
            {field: row.get(field) for field in ACTIVITY_FIELDS}
            for row in raw_projected
            if isinstance(row, Mapping)
        ]
        if isinstance(raw_projected, list)
        else raw_projected
    )
    if projected == expected:
        return
    mismatch = "length"
    if isinstance(projected, list) and len(projected) == len(expected):
        mismatch = next(
            (
                f"index {index}"
                for index, (actual, wanted) in enumerate(zip(projected, expected))
                if actual != wanted
            ),
            "content",
        )
    errors.append(
        "activity contract: final-report data.json agentActivity must exactly "
        "match canonical activity events in order and core fields; "
        f"mismatch={mismatch}, expected={len(expected)}, "
        f"projected={len(projected) if isinstance(projected, list) else '<missing>'}."
    )


def _check_activity_worker_pairs(
    evidence: _LeadEvidence,
    indexed: dict[str, list[LeadEvent]],
    analysis_workers: list[dict],
    errors: list[str],
) -> None:
    expected = (
        ("phase-4-dispatch", "worker-dispatched", False),
        ("phase-5-collect", "worker-completed", True),
    )
    for phase, kind, completed_only in expected:
        workers = _progress_workers(
            evidence, phase, completed_only=completed_only
        )
        for worker in sorted(workers):
            roles = _roster_roles_for_value(worker, analysis_workers)
            if _activity_has_agent(indexed.get(kind, []), roles):
                continue
            errors.append(
                f"activity contract: missing `{kind}` for worker `{worker}` "
                f"recorded by `PROGRESS: {phase}`."
            )


def _check_activity_worker_agents(
    indexed: dict[str, list[LeadEvent]],
    analysis_workers: list[dict],
    errors: list[str],
) -> None:
    accepted = {
        role
        for worker in analysis_workers
        for role in _worker_roles(worker)
    }
    roster = [str(worker.get("role") or "") for worker in analysis_workers]
    for kind in ("worker-dispatched", "worker-completed"):
        for event in indexed.get(kind, []):
            agent = str(event.details.get("agent") or "")
            if _worker_role(agent) in accepted:
                continue
            errors.append(
                f"activity contract: `{kind}` agent `{agent or '<empty>'}` must "
                f"match an actual manifest roster role; roster={roster}."
            )


def _allowed_automatic_rounds(self_fix_rounds: int, *, gating: bool = True) -> int:
    """이 run 이 돌아도 되는 자동 plan-body 라운드 수.

    상한이 2 로 고정돼 있던 동안 다른 규칙과 정면으로 충돌했다.
    `validators/validate-run.py` `_validate_verdict_rounds_outlive_self_fix` 는
    자가수정 뒤 **모든** 항목이 그 rewrite 이후 라운드의 판정을 들고 있어야
    한다고 요구한다. 그런데 자가수정 직후의 표적 재검증은 고친 항목만 덮으므로,
    나머지 항목을 덮는 배치가 하나 더 필요하다 — 실측 run 에서 114항목 중 18개가
    표적 라운드, 남은 96개가 그다음 라운드였다. 상한 2 는 그 마지막 배치를
    금지하고, 금지에 따르면 96개가 고쳐지기 전 본문에 대한 판정을 게이트에 싣는다.

    기준선은 2다 — 초기 검증과 재검증 한 번. 자가수정이 없어도 두 번째 배치가
    필요할 수 있다: 동수이거나 상대 워커가 판정을 못 낸 항목은 `needs-reverify`
    로 그 라운드에 실린다. 자가수정은 여기에 배치를 하나씩 더한다. 표적 재검증은
    기준선의 두 번째 배치가 받고, 늘어나는 것은 그 라운드가 덮지 못한 잔여 항목을
    일소하는 배치다.

    ``gating=false`` 자문 경로는 추출과 1라운드만 남긴다. self-fix 와 일소
    배치는 돌리지 않는다.
    """
    if not gating:
        return 1
    return 2 + max(self_fix_rounds, 0)


def _self_fix_rounds_from_state(run_dir: Path, suffix: str | None) -> int | None:
    """이번 런 상태 파일의 자가수정 횟수. 없으면 None — 호출자가 리포트로 폴백.

    리포트 칸은 이어진 seq 의 누적이라, 이번 창의 `self-fix-applied` 건수와
    비교하면 어긋난다.
    """
    if not suffix:
        return None
    try:
        doc = json.loads(_plan_body_state_path(run_dir, suffix).read_text())
    except (OSError, json.JSONDecodeError):
        return None
    if not isinstance(doc, dict):
        return None
    projection = doc.get("planBodyVerification")
    value = (
        projection.get("selfFixRoundsApplied")
        if isinstance(projection, dict)
        else None
    )
    if isinstance(value, int) and value >= 0:
        return value
    value = doc.get("selfFixRoundsApplied")
    return value if isinstance(value, int) and value >= 0 else None


def _is_plan_body_verification_activity(event: LeadEvent) -> bool:
    """`verification-round-completed` 가 계획 본문 배치인지.

    같은 kind 로 적대 재검증 라운드도 남는다. 그 건을 본문 라운드에 넣으면
    recorded 가 roundCount 보다 커진다. 요약이 `adversarial reverify` 이면
    본문이 아니다. 픽스처 요약(`verification-round-completed for …`)은 본문이다.
    """
    summary = str(event.details.get("summary") or "").lower()
    return "adversarial reverify" not in summary and "adversarial re-verify" not in summary


def _plan_body_verification(report_data: Mapping[str, Any]) -> Mapping[str, Any] | None:
    planning = report_data.get("implementationPlanning")
    verification = (
        planning.get("planBodyVerification")
        if isinstance(planning, Mapping)
        else None
    )
    return verification if isinstance(verification, Mapping) else None


def _self_fix_rounds_applied(report_data: Mapping[str, Any]) -> int:
    plan_verification = _plan_body_verification(report_data)
    value = (
        plan_verification.get("selfFixRoundsApplied")
        if plan_verification is not None
        else 0
    )
    return value if isinstance(value, int) and value >= 0 else 0


def _plan_body_gating(report_data: Mapping[str, Any]) -> bool:
    plan_verification = _plan_body_verification(report_data)
    if plan_verification is None:
        return True
    return plan_verification.get("gating") is not False


def _matching_user_reverification_round(
    event: LeadEvent | None,
    clarification_id: str,
    plan_item_ids: set[str],
    verification_rounds: int,
) -> int | None:
    if event is None or event.details.get("outcome") != "resolved":
        return None
    evidence_refs = {
        ref
        for ref in (event.details.get("evidenceRefs") or [])
        if isinstance(ref, str)
    }
    clarification_refs = {
        ref for ref in evidence_refs if _CLARIFICATION_ID_RE.fullmatch(ref)
    }
    event_item_ids = {
        item_id
        for item_id in (event.details.get("planItemIds") or [])
        if isinstance(item_id, str)
    }
    round_numbers = {
        int(match.group(1))
        for ref in evidence_refs
        for match in [_TARGETED_REVERIFICATION_REF_RE.fullmatch(ref)]
        if match is not None
    }
    if clarification_refs != {clarification_id} or event_item_ids != plan_item_ids:
        return None
    if len(round_numbers) != 1:
        return None
    round_number = next(iter(round_numbers))
    return round_number if round_number <= verification_rounds else None


def _resolved_correctness_reverification_rounds(
    indexed: dict[str, list[LeadEvent]],
    report_data: Mapping[str, Any],
    verification_rounds: int,
) -> set[int]:
    evaluated_by_id = {
        str(event.details.get("activityId") or ""): event
        for event in indexed.get("user-decision-evaluated", [])
    }
    rounds: set[int] = set()
    for row in report_data.get("clarificationItems") or []:
        if not isinstance(row, Mapping) or row.get("status") != "resolved":
            continue
        context = row.get("approvalContext")
        if not isinstance(context, Mapping):
            continue
        if context.get("classification") != "correctness-critical":
            continue
        resolution = context.get("resolution")
        if not isinstance(resolution, Mapping):
            continue
        clarification_id = str(row.get("id") or "")
        plan_item_ids = {
            item_id
            for item_id in (context.get("planItemIds") or [])
            if isinstance(item_id, str)
        }
        if not plan_item_ids:
            continue
        check_refs = [
            ref for ref in (resolution.get("checkRefs") or []) if isinstance(ref, str)
        ]
        matched_rounds = [
            _matching_user_reverification_round(
                evaluated_by_id.get(check_ref),
                clarification_id,
                plan_item_ids,
                verification_rounds,
            )
            for check_ref in check_refs
        ]
        if matched_rounds and all(
            round_number is not None for round_number in matched_rounds
        ):
            unique_rounds = set(matched_rounds)
            if len(unique_rounds) == 1:
                rounds.update(unique_rounds)
    return rounds


def _check_activity_round_counts(
    indexed: dict[str, list[LeadEvent]],
    report_data: Mapping[str, Any],
    run_dir: Path,
    suffix: str | None,
    errors: list[str],
) -> None:
    verification_rounds = _plan_body_rounds_ran(run_dir, suffix)
    recorded_verifications = len([
        event
        for event in indexed.get("verification-round-completed", [])
        if _is_plan_body_verification_activity(event)
    ])
    user_reverification_rounds = _resolved_correctness_reverification_rounds(
        indexed,
        report_data,
        verification_rounds,
    )
    automatic_rounds = verification_rounds - len(user_reverification_rounds)
    state_self_fix = _self_fix_rounds_from_state(run_dir, suffix)
    self_fix_rounds = (
        state_self_fix
        if state_self_fix is not None
        else _self_fix_rounds_applied(report_data)
    )
    allowed_rounds = _allowed_automatic_rounds(
        self_fix_rounds, gating=_plan_body_gating(report_data),
    )
    if automatic_rounds > allowed_rounds:
        errors.append(
            f"activity contract: at most {allowed_rounds} plan verification "
            f"batch(es) are allowed for {self_fix_rounds} self-fix round(s); "
            f"roundCount={verification_rounds}, "
            f"humanReverificationRounds={sorted(user_reverification_rounds)}."
        )
    if recorded_verifications != automatic_rounds:
        errors.append(
            "activity contract: `verification-round-completed` count must match "
            f"automatic plan-body rounds={automatic_rounds} from "
            f"roundCount={verification_rounds} in "
            f"{_plan_body_state_path(run_dir, suffix or '')}; "
            f"recorded={recorded_verifications}."
        )
    recorded_self_fixes = len(indexed.get("self-fix-applied", []))
    if recorded_self_fixes != self_fix_rounds:
        errors.append(
            "activity contract: `self-fix-applied` count must match "
            f"selfFixRoundsApplied={self_fix_rounds}; recorded={recorded_self_fixes}."
        )


def _activity_references(event: LeadEvent, reference: str) -> bool:
    refs = event.details.get("evidenceRefs")
    return isinstance(refs, list) and reference in refs


def _check_activity_user_decisions(
    indexed: dict[str, list[LeadEvent]],
    report_data: Mapping[str, Any],
    errors: list[str],
) -> None:
    rows = report_data.get("clarificationItems")
    if not isinstance(rows, list):
        return
    required = indexed.get("user-decision-required", [])
    evaluated = indexed.get("user-decision-evaluated", [])
    for row in rows:
        if not isinstance(row, Mapping) or row.get("blocks") != "approval":
            continue
        clarification_id = str(row.get("id") or "<unknown>")
        confirmation = str(row.get("userConfirmation") or "")
        if confirmation.startswith("asked-") and not any(
            _activity_references(event, clarification_id) for event in required
        ):
            errors.append(
                "activity contract: missing `user-decision-required` referencing "
                f"approval clarification `{clarification_id}`."
            )
        if row.get("status") == "resolved" and not any(
            _activity_references(event, clarification_id) for event in evaluated
        ):
            errors.append(
                "activity contract: missing `user-decision-evaluated` referencing "
                f"resolved approval clarification `{clarification_id}`."
            )


def _check_activity_contract(
    evidence: _LeadEvidence,
    team_state: Mapping[str, Any],
    run_manifest: Mapping[str, Any],
    report_path: Path,
    run_dir: Path,
    suffix: str | None,
    errors: list[str],
) -> None:
    if not _is_activity_contract_v1_planning(run_manifest):
        return
    indexed = _activity_index(evidence.activities)
    workers = [
        worker
        for worker in (team_state.get("workers") or [])
        if isinstance(worker, dict)
    ]
    analysis_workers = _analysis_workers(team_state)
    report_data = _read_report_data(report_path)
    _check_activity_id_order(evidence.activities, errors)
    _check_projected_agent_activity(evidence.activities, report_data, errors)
    _check_activity_worker_pairs(evidence, indexed, analysis_workers, errors)
    _check_activity_worker_agents(indexed, workers, errors)
    _check_activity_round_counts(indexed, report_data, run_dir, suffix, errors)
    _check_activity_user_decisions(indexed, report_data, errors)


def _check_user_confirm_checkpoints(
    by_phase: dict[str, list[tuple[str, str]]],
    asked_ids: list[str],
    errors: list[str],
) -> None:
    """"물어봤다"는 리포트의 기록에는 물어본 흔적이 세션에 남아 있어야 한다.

    승인을 막는 행은 run 하나를 통째로 대기 상태로 만든다. 그런 행을 세우기 전에
    사용자에게 그 자리에서 묻는 것이 계약인데, 그 이행 여부를 리포트의 자기 신고
    말고는 확인할 길이 없었다. 한 번은 리드가 차단이 생길 것을 미리 예측해 말해
    놓고도 묻지 않은 채 행으로 만들었고, run 은 사용자가 이미 답해 둔 질문 앞에서
    self-fix 예산을 전부 썼다. 체크포인트가 그 격차를 좁힌다 — 세션에 질문이 남지
    않았다면 리드는 묻지 않은 것이다."""
    if not asked_ids:
        return
    detail = "prompts/lead/okstra-lead-contract.md 'User confirmation before an approval blocker'"
    asked_lines = " ".join(line for _ts, line in by_phase.get("user-confirm", []))
    for row_id in asked_ids:
        if row_id in asked_lines:
            continue
        errors.append(
            f"PROGRESS checkpoint missing: `user-confirm {row_id}` — the final "
            f"report records `{row_id}` as an approval blocker the user was "
            "asked about, but no such line exists in this run's evidence. Either "
            "ask before writing the row, or record what actually happened "
            f"(`deferred-no-interactive-session`) ({detail})."
        )


def _phase_mentions_worker(
    lines: list[tuple[str, str]], roles: set[str]
) -> bool:
    return any(
        _worker_role(worker) in roles
        for _ts, line in lines
        if (worker := _progress_worker(line)) is not None
    )


def _check_worker_checkpoint_lines(
    by_phase: dict[str, list[tuple[str, str]]],
    analysis_workers: list[dict],
    errors: list[str],
) -> None:
    """phase-4-dispatch / phase-5-collect 의 per-worker 라인 (SKILL.md: once per worker)."""
    for worker in analysis_workers:
        role = str(worker.get("role", "")).strip() or "<unknown role>"
        status = str(worker.get("status", "")).strip()
        roles = _worker_roles(worker)
        if status in _ATTEMPTED_STATUSES and not _phase_mentions_worker(
            by_phase.get("phase-4-dispatch", []), roles
        ):
            errors.append(
                f"PROGRESS checkpoint missing: no `phase-4-dispatch worker=<role>` "
                f"line names worker `{role}` — one line per dispatched worker, "
                "prompts/lead/okstra-lead-contract.md 'Progress reporting (BLOCKING)'."
            )
        if status == "completed" and not _phase_mentions_worker(
            by_phase.get("phase-5-collect", []), roles
        ):
            errors.append(
                f"PROGRESS checkpoint missing: no `phase-5-collect worker=<role>` "
                f"line names completed worker `{role}` — one line per collected "
                "result, prompts/lead/okstra-lead-contract.md 'Progress reporting (BLOCKING)'."
            )


def _check_batch_cleanup_checkpoints(
    by_phase: dict[str, list[tuple[str, str]]],
    convergence_ran: bool,
    report_writer_dispatched: bool,
    errors: list[str],
) -> None:
    """phase-batch-cleanup: 각 배치 경계 직전에 이전 배치 pane/teammate 정리가
    실제로 일어났는지 (prompts/profiles/_common-contract.md 'Phase-start cleanup').
    R1=convergence round 1 직전, R2=report-writer dispatch 직전(수렴이 있었으면
    마지막 라운드 이후 별도 1회). ISO-8601 ts 는 lexicographic 비교가 곧 시간순."""
    detail = "prompts/lead/okstra-lead-contract.md 'Run-scoped worker-resource lifecycle'"
    cleanup_ts = sorted(ts for ts, _line in by_phase.get("phase-batch-cleanup", []))
    conv_ts = sorted(ts for ts, _line in by_phase.get("phase-5.5-convergence", []))
    synth_ts = sorted(ts for ts, _line in by_phase.get("phase-6-synthesis", []))

    if convergence_ran and conv_ts and not any(ts <= conv_ts[0] for ts in cleanup_ts):
        errors.append(
            "PROGRESS checkpoint missing: no `phase-batch-cleanup` line before the "
            f"first `phase-5.5-convergence` round ({conv_ts[0]}) — the prior analysis "
            f"batch's panes/teammates must be cleared first ({detail})."
        )

    if report_writer_dispatched and synth_ts:
        lower = conv_ts[-1] if (convergence_ran and conv_ts) else ""
        if not any(lower <= ts <= synth_ts[0] for ts in cleanup_ts):
            errors.append(
                "PROGRESS checkpoint missing: no `phase-batch-cleanup` line before "
                f"`phase-6-synthesis` ({synth_ts[0]}) — the prior batch must be cleared "
                f"before dispatching report-writer ({detail})."
            )


def _check_plan_verify_cleanup_checkpoints(
    by_phase: dict[str, list[tuple[str, str]]],
    plan_body_rounds: int,
    errors: list[str],
) -> None:
    """plan-body 라운드도 배치 경계다 — 라운드마다 새 검증 워커를 띄운다.

    Phase 6 뒤에 도는 구간이라 위 두 지점의 바깥이었고, self-fix 를 다섯 라운드
    돈 run 은 라운드마다 완료 워커를 남겨 유휴 세션이 쌓였다. 라운드 2 이상은
    직전 라운드의 워커가 반드시 존재하므로, 각 라운드 직전에 정리가 있어야
    한다(라운드 1의 앞 경계는 report-writer 디스패치 정리가 이미 덮는다)."""
    if plan_body_rounds < 2:
        return
    detail = "prompts/lead/plan-body-verification.md §\"Round protocol\" step 7"
    round_ts = sorted(ts for ts, _line in by_phase.get("phase-5.5.9-plan-verify", []))
    if not round_ts:
        errors.append(
            "PROGRESS checkpoint missing: `phase-5.5.9-plan-verify` — the state "
            f"file records {plan_body_rounds} plan-body rounds, each of which "
            f"dispatches a worker batch and must announce itself ({detail})."
        )
        return
    cleanup_ts = sorted(ts for ts, _line in by_phase.get("phase-batch-cleanup", []))
    for index, dispatched_at in enumerate(round_ts[1:], start=1):
        previous = round_ts[index - 1]
        if not any(previous <= ts <= dispatched_at for ts in cleanup_ts):
            errors.append(
                "PROGRESS checkpoint missing: no `phase-batch-cleanup` line "
                f"between plan-body rounds ({previous} → {dispatched_at}) — the "
                f"previous round's completed verifiers must be cleared before "
                f"the next round dispatches ({detail})."
            )


def _check_progress_checkpoints(
    evidence: _LeadEvidence,
    team_state: dict,
    run_dir: Path,
    suffix: str | None,
    report_path: Path,
    errors: list[str],
    advisories: list[str] | None = None,
) -> None:
    # The narration lines are advisory (see `SessionConformanceResult`); the
    # user-confirmation cross-check below stays in `errors`.
    narration = advisories if advisories is not None else errors
    by_phase: dict[str, list[tuple[str, str]]] = {}
    for ts, phase, line in evidence.progress:
        by_phase.setdefault(phase, []).append((ts, line))

    def require(phase: str, condition: bool, detail: str) -> None:
        if condition and phase not in by_phase:
            narration.append(
                f"PROGRESS checkpoint missing: `{phase}` ({detail}) — "
                "prompts/lead/okstra-lead-contract.md 'Progress reporting (BLOCKING)'."
            )

    intake = by_phase.get("phase-1-intake", [])
    if not any("complete" not in line.lower() for _ts, line in intake):
        narration.append(
            "PROGRESS checkpoint missing: `phase-1-intake reading task bundle` "
            "(start-of-Phase-1 line) — prompts/lead/okstra-lead-contract.md 'Progress reporting (BLOCKING)'."
        )
    if not any("complete" in line.lower() for _ts, line in intake):
        narration.append(
            "PROGRESS checkpoint missing: `phase-1-intake complete` "
            "(after all intake reads) — prompts/lead/okstra-lead-contract.md 'Progress reporting (BLOCKING)'."
        )

    workers = [w for w in (team_state.get("workers") or []) if isinstance(w, dict)]
    analysis_workers = _analysis_workers(team_state)
    any_dispatched = any(
        str(w.get("status", "")).strip() in _DISPATCHED_STATUSES for w in workers
    )
    require("phase-2-prompts", bool(workers), "before any Write to assigned prompt paths")
    require(
        "phase-3-team-create",
        any_dispatched,
        "in Phase 3 after recording teamCreate in team-state — the "
        "`using implicit team (split-pane=<on|off>)` line, or the "
        "`skipped (concurrent-run)` variant in the concurrent-run path",
    )
    _check_worker_checkpoint_lines(by_phase, analysis_workers, narration)
    convergence_ran = _convergence_rounds_ran(run_dir, suffix)
    require(
        "phase-5.5-convergence",
        convergence_ran,
        "at the start of each convergence round (state artifact records totalRounds >= 1)",
    )
    report_writer = next((w for w in workers if _is_report_writer(w)), None)
    report_writer_dispatched = (
        report_writer is not None
        and str(report_writer.get("status", "")).strip() in _DISPATCHED_STATUSES
    )
    require(
        "phase-6-synthesis",
        report_writer_dispatched,
        "at the start of Phase 6 (report-writer dispatch)",
    )
    require("phase-7-persist", True, "at the start of Phase 7")
    _check_batch_cleanup_checkpoints(
        by_phase, convergence_ran, report_writer_dispatched, narration
    )
    _check_plan_verify_cleanup_checkpoints(
        by_phase, _plan_body_rounds_ran(run_dir, suffix), narration
    )
    _check_user_confirm_checkpoints(
        by_phase, _ids_reported_as_asked(report_path), errors
    )


def _parse_iso(ts: str) -> datetime | None:
    try:
        return datetime.fromisoformat(ts.replace("Z", "+00:00"))
    except ValueError:
        return None


def _check_heartbeat_sidecar(path: Path, worker_id: str, errors: list[str]) -> None:
    rel = path.name
    if not path.is_file():
        errors.append(f"registered audit sidecar missing: {path}")
        return
    try:
        content = path.read_text(encoding="utf-8")
    except OSError as exc:
        errors.append(f"worker audit sidecar unreadable: {rel} ({exc})")
        return
    entries = [(m.group("stage"), m.group("ts")) for m in _HEARTBEAT_LINE_RE.finditer(content)]
    if not entries:
        errors.append(
            f"`{rel}` has no `- PROGRESS: <stage> <ISO-8601-UTC>` heartbeat lines — "
            "the in-process worker MUST write `started` immediately and append one "
            "line per stage at <= 5-minute cadence (agents/workers/claude-worker.md "
            "'Heartbeat', prompts/lead/okstra-lead-contract.md Common Mistakes)."
        )
        return
    if entries[0][0] != "started":
        errors.append(
            f"`{rel}`: first heartbeat stage must be `started` "
            f"(found `{entries[0][0]}`) — the sidecar is written BEFORE the "
            "per-file reads, with a `- PROGRESS: started <ISO>` line."
        )
    if worker_id == "report-writer":
        for stage, _raw_ts in entries:
            # `in-stage:<stage>` 는 계약이 장기 단계에 요구하는 진행 라인이다 —
            # allowlist 가 이를 배제하면 워커는 cadence 를 지킬 방법이 없다.
            base = stage[len(_IN_STAGE_PREFIX):] if stage.startswith(_IN_STAGE_PREFIX) else stage
            if base not in _REPORT_WRITER_HEARTBEAT_STAGES:
                errors.append(
                    f"`{rel}`: heartbeat stage `{stage}` is not allowed for report-writer."
                )
    # result 파일이 존재하면 마지막 단계 마커도 있어야 한다. timeout 으로 중단된
    # worker(result 없음)에는 요구하지 않는다 — hang 이전 구간의 cadence 만 본다.
    result_file = path.with_name(rel.replace("-audit-", "-"))
    if result_file.exists() and not any(s == "write-result-start" for s, _ in entries):
        errors.append(
            f"`{rel}`: heartbeat is missing the `write-result-start` stage line "
            f"although the worker result `{result_file.name}` exists — every "
            "stage must append its own line (agents/workers/claude-worker.md 'Heartbeat')."
        )
    prev: datetime | None = None
    prev_stage = ""
    for stage, raw_ts in entries:
        ts = _parse_iso(raw_ts)
        if ts is None:
            errors.append(
                f"`{rel}`: heartbeat line for stage `{stage}` has an unparseable "
                f"ISO-8601 timestamp `{raw_ts}`."
            )
            continue
        if prev is not None:
            # 예산은 구간을 여는 단계에서 고른다 — 이 공백이 재는 것은 직전에
            # 선언된 단계의 작업 시간이다.
            budget = max_gap_seconds_after(prev_stage)
            gap = (ts - prev).total_seconds()
            if gap < 0:
                errors.append(
                    f"`{rel}`: heartbeat timestamps regress at stage `{stage}` ({raw_ts})."
                )
            elif gap > budget:
                errors.append(
                    f"`{rel}`: heartbeat gap after stage `{prev_stage}` is {int(gap)}s "
                    f"(budget {budget}s) — emit "
                    "`- PROGRESS: in-stage:<stage> <ISO>` during long stages."
                )
        prev = ts
        prev_stage = stage


def _check_worker_liveness(
    project_root: Path, team_state: Mapping[str, Any], errors: list[str]
) -> None:
    """Validate the liveness artifact selected by each dispatch record."""
    dispatches = team_state.get("workerDispatches")
    if not isinstance(dispatches, list):
        return
    for record in dispatches:
        if not isinstance(record, Mapping):
            continue
        mode = record.get("livenessMode")
        worker_id = record.get("workerId")
        if not isinstance(worker_id, str) or not worker_id:
            continue
        if mode not in (_LIVENESS_AUDIT_HEARTBEAT, _LIVENESS_WRAPPER_STATUS):
            errors.append(
                f"worker `{worker_id}` dispatch has missing or unknown livenessMode: {mode!r}."
            )
            continue
        if mode == _LIVENESS_AUDIT_HEARTBEAT:
            raw_path = record.get("auditSidecarPath")
            if not isinstance(raw_path, str) or not raw_path:
                errors.append(
                    f"worker `{worker_id}` audit-heartbeat dispatch has no registered audit sidecar path."
                )
                continue
            path = Path(raw_path)
            if not path.is_absolute():
                path = project_root / path
            _check_heartbeat_sidecar(path, worker_id, errors)
        elif mode == _LIVENESS_WRAPPER_STATUS:
            raw_path = record.get("statusSidecarPath")
            if not isinstance(raw_path, str) or not raw_path:
                errors.append(
                    f"worker `{worker_id}` wrapper-status dispatch has no registered status sidecar path."
                )
                continue
            path = Path(raw_path)
            if not path.is_absolute():
                path = project_root / path
            status = read_wrapper_status(path)
            if status is None:
                errors.append(f"registered wrapper status sidecar unreadable: {path}")


def _instruction_set_dir(run_dir: Path, suffix: str | None, project_root: Path) -> Path | None:
    """이 run 의 instruction-set 디렉터리. 해소 불가면 None (경로를 모르면 스테이징
    여부도 판정할 수 없고, 모르는 것을 근거로 run 을 떨어뜨리지 않는다)."""
    if not suffix:
        return None
    manifest = run_dir / "manifests" / f"run-manifest-{suffix}.json"
    try:
        data = json.loads(manifest.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError, UnicodeDecodeError):
        return None
    raw = data.get("instructionSetPath") if isinstance(data, dict) else None
    if not isinstance(raw, str) or not raw:
        return None
    path = Path(raw)
    if not path.is_absolute():
        path = project_root / path
    return path if path.is_dir() else None


def _check_cmux_adapter_read(
    evidence: _LeadEvidence, team_state: dict, errors: list[str]
) -> None:
    """검사 4 — cmux 어댑터 읽음. 모든 task-type 에 적용된다.

    다른 어댑터는 lead 의 런타임이 고르지만 이것은 환경이 고른다. 그래서 읽지
    않은 lead 에게는 이 경로가 존재한다는 사실 자체가 닿지 않고, 자기 런타임이
    아는 host 네이티브 디스패치로 되돌아간다 — cmux 경로에서 okstra 가 소유한
    바로 그 일이다."""
    adapter = team_state.get("leadAdapter")
    name = str(adapter.get("name", "")).strip() if isinstance(adapter, dict) else ""
    if name != CMUX_ADAPTER_NAME:
        return
    if evidence.sidecar_reads.get(CMUX_ADAPTER_BASENAME):
        return
    dispatch_mode = str(team_state.get("dispatchMode", "")).strip()
    if dispatch_mode in _WORKER_DISPATCH_MODES:
        # grok 같은 artifact-only 호스트는 Read 도구 기록이 없다. 워커가
        # okstra 디스패치로 나갔으면 어댑터가 가리키는 경로를 탄 것이다.
        return
    errors.append(
        f"cmux adapter: no read of `{CMUX_ADAPTER_BASENAME}` (a `Read` call or a "
        f"shell command naming it) found in the "
        "selected adapter evidence source within this run's window — the cmux "
        "adapter is selected by environment, not by lead runtime, so it MUST be "
        f"read before dispatch ({_CMUX_ADAPTER_CITE})."
    )


def _check_implementation_entry_guard(
    evidence: _LeadEvidence, errors: list[str], instruction_set: Path | None
) -> None:
    """검사 3 — entry guard. fresh-read 규칙(이전 run 기억으로 갈음 불가)은 run
    윈도우 스코핑이 보장한다: 이번 윈도우 안의 Read 만 증거로 인정된다.

    anchor 가 있는 행은 존재 + 순서를, anchor 가 `None` 인 행은 존재만 본다.
    `staged_per_run` 행은 이 run 의 instruction-set 에 실제로 있을 때만 요구한다."""
    anchors: dict[str, str] = {}
    for ts, phase, _line in evidence.progress:  # progress 는 ts 정렬 — 첫 항목이 최초 발생
        if phase in ("phase-6-synthesis", "phase-7-persist") and ts:
            anchors.setdefault(phase, ts)
    for row in _ENTRY_GUARD_READS:
        if row.staged_per_run and not (
            instruction_set and (instruction_set / row.basename).is_file()
        ):
            continue
        ts_list = evidence.sidecar_reads.get(row.basename) or []
        if not ts_list:
            demand = (
                f"it MUST be read fresh {row.anchor[1]} of every implementation run"
                if row.anchor
                else "the staged copy MUST be read at least once inside that window"
            )
            errors.append(
                f"implementation entry guard: no read of `{row.basename}` (a `Read` call "
                f"or a shell command naming it) found in "
                f"the selected adapter evidence source within this run's window — "
                f"{demand} ({row.cite})."
            )
            continue
        if not row.anchor:
            continue
        anchor_phase, when = row.anchor
        anchor_ts = anchors.get(anchor_phase)
        if anchor_ts and min(ts_list) >= anchor_ts:
            errors.append(
                f"implementation entry guard: `{row.basename}` was first read at "
                f"{min(ts_list)}, not before the first `PROGRESS: {anchor_phase}` "
                f"line ({anchor_ts}) — it must be loaded {when}, before "
                f"that checkpoint ({row.cite})."
            )


def validate_session_conformance(
    *,
    team_state: dict,
    team_state_path: Path,
    run_manifest: Mapping[str, Any],
    project_root: Path,
    report_path: Path,
    task_type: str,
    claude_projects_dir: Path | None = None,
) -> SessionConformanceResult:
    """Run post-hoc checks for the neutral lead and activity contracts.

    `claude_projects_dir` injects the Claude project root for tests and diagnostics.
    Heartbeat validation runs before selecting the adapter evidence source.
    """
    result = SessionConformanceResult()
    _ensure_token_usage_importable()
    try:
        from okstra_token_usage.collect import run_artifact_suffix
    except ImportError as exc:  # pragma: no cover — 설치본은 항상 패키지를 동반
        result.errors.append(f"okstra_token_usage import failed — {exc}")
        return result

    run_dir = report_path.parent.parent
    suffix = run_artifact_suffix(team_state_path)
    _check_worker_liveness(project_root, team_state, result.errors)

    evidence_source, source_error = _conformance_evidence_source(team_state)
    if source_error:
        result.errors.append(source_error)
        return result
    if evidence_source == "artifact-only":
        evidence, error = _collect_artifact_lead_evidence(
            team_state, run_manifest, project_root, task_type, suffix
        )
    else:
        evidence, error = _collect_lead_evidence(
            team_state,
            team_state_path,
            run_manifest,
            project_root,
            task_type,
            suffix,
            claude_projects_dir,
        )
    if error:
        result.errors.append(error)
        return result
    _check_progress_checkpoints(
        evidence,
        team_state,
        run_dir,
        suffix,
        report_path,
        result.errors,
        result.advisories,
    )
    _check_activity_contract(
        evidence,
        team_state,
        run_manifest,
        report_path,
        run_dir,
        suffix,
        result.errors,
    )
    _check_cmux_adapter_read(evidence, team_state, result.errors)
    if task_type == "implementation":
        _check_implementation_entry_guard(
            evidence,
            result.errors,
            _instruction_set_dir(run_dir, suffix, project_root),
        )
    return result