#!/usr/bin/env python3
"""Dispatch every independently-ready agent-task pair until the task group completes."""

from __future__ import annotations

import argparse
import asyncio
import fcntl
import importlib.util
import hashlib
import json
import os
import re
import signal
import shutil
import subprocess
import sys
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo


_OBSERVATION_MODULE_NAME = "agent_task_dispatcher_observation"


def load_sibling_observation_module():
    loaded = sys.modules.get(_OBSERVATION_MODULE_NAME)
    if loaded is not None:
        return loaded
    spec = importlib.util.spec_from_file_location(
        _OBSERVATION_MODULE_NAME,
        Path(__file__).with_name("dispatcher_observation.py"),
    )
    if spec is None or spec.loader is None:
        raise RuntimeError("failed to load dispatcher observation module")
    module = importlib.util.module_from_spec(spec)
    sys.modules[_OBSERVATION_MODULE_NAME] = module
    try:
        spec.loader.exec_module(module)
    except BaseException:
        sys.modules.pop(_OBSERVATION_MODULE_NAME, None)
        raise
    return module


observation = load_sibling_observation_module()
SEP = observation.SEP
banner = observation.banner
attempt_event = observation.attempt_event
validation_claim = observation.validation_claim

PLAN_RE = re.compile(r"^PLAN-(local|cloud)-G(0[1-9]|10)\.md$")
REVIEW_RE = re.compile(r"^CODE_REVIEW-(local|cloud)-G(0[1-9]|10)\.md$")
PLAN_LOG_RE = re.compile(
    r"^plan_(local|cloud)_G(0[1-9]|10)_(0|[1-9][0-9]*)\.log$"
)
REVIEW_LOG_RE = re.compile(
    r"^code_review_(local|cloud)_G(0[1-9]|10)_(0|[1-9][0-9]*)\.log$"
)
SUBTASK_RE = re.compile(r"^(?P<index>\d{2})(?:\+(?P<deps>\d{2}(?:,\d{2})*))?_[a-z0-9_]+$")
MODIFIED_FILES_HEADINGS = ("Modified Files Summary", "수정 파일 요약")
MODIFIED_FILES_HEADER_CELLS = frozenset({"file", "files", "path", "paths", "파일", "경로"})
PLACEHOLDER_PATH_RE = re.compile(
    r"(?:[<>{}]|\.\.\.|(?:^|[/_.-])(?:tbd|todo|placeholder)(?:$|[/_.-]))",
    re.IGNORECASE,
)
IMPLEMENTATION_CHECKLIST_HEADINGS = ("Implementation Checklist", "구현 체크리스트")
# The canonical English and legacy Korean verdict contracts are paired: a
# heading only accepts the verdict label of its own schema. Mixed pairs are not
# a documented schema and must fail closed.
CODE_REVIEW_RESULT_SCHEMAS = (
    ("Code Review Result", "Overall Verdict"),
    ("코드리뷰 결과", "종합 판정"),
)
USER_REVIEW_SCHEMAS = (
    {
        "status_heading": "Status",
        "reason_heading": "Reason",
        "type_label": "Type",
        "target_label": "Target",
        "evidence_heading": "Blocking Evidence",
        "evidence_label": "Blocking rationale",
        "decision_headings": ("Required User Action",),
        "resume_heading": "Resume Condition",
    },
    {
        "status_heading": "상태",
        "reason_heading": "사유",
        "type_label": "유형",
        "target_label": "연결 대상",
        "evidence_heading": "차단 근거",
        "evidence_label": "차단 판단 근거",
        "decision_headings": ("사용자 조치 또는 결정", "연결 결정 필요"),
        "resume_heading": "재개 조건",
    },
)
VERDICT_SCHEMA_MATCHERS = tuple(
    (
        re.compile(rf"^##\s*{re.escape(heading)}[ \t]*$", re.MULTILINE),
        re.compile(
            rf"^(?:-\s*)?(?:\*\*)?{re.escape(label)}(?:\*\*)?\s*:\s*"
            rf"(?:(PASS|WARN|FAIL)|`(PASS|WARN|FAIL)`)[ \t]*$",
            re.MULTILINE,
        ),
        re.compile(
            rf"^###\s+{re.escape(label)}[ \t]*$\s*^(?:\*\*)?"
            rf"(?:(PASS|WARN|FAIL)|`(PASS|WARN|FAIL)`)(?:\*\*)?[ \t]*$",
            re.MULTILINE,
        ),
    )
    for heading, label in CODE_REVIEW_RESULT_SCHEMAS
)
MILESTONE_TASK_ID_PATTERN = r"[A-Za-z0-9]+(?:[-_+=][A-Za-z0-9]+){0,3}"
MILESTONE_TASK_ID_RE = re.compile(rf"\A{MILESTONE_TASK_ID_PATTERN}\Z")
PLAN_IDENTITY_RE = re.compile(
    r"\A<!--\s+task=(?P<task>\S+)\s+plan=(?P<plan>\d+)\s+tag=(?P<tag>\S+)"
    r"(?:\s+milestone-task=(?P<milestone_task>[^,\s]+(?:,[^,\s]+)*))?"
    r"\s+-->[ \t]*(?:\r?\n|\Z)"
)
MILESTONE_ITEM_RE = re.compile(
    rf"^-\s+\[[ xX]\]\s+\[({MILESTONE_TASK_ID_PATTERN})\]", re.MULTILINE
)
MILESTONE_FEATURE_SECTION_RE = re.compile(
    r"^##[ \t]+기능[ \t]*\r?\n(?P<body>.*?)(?=^##[ \t]+|\Z)",
    re.MULTILINE | re.DOTALL,
)
IMPLEMENTATION_CHECKBOX_RE = re.compile(
    r"^-\s+\[([^\]\r\n]*)\]", re.MULTILINE
)
WORK_LOG_NAME = "WORK_LOG.md"
WORK_LOG_ARCHIVE_RE = re.compile(r"^work_log_(\d+)\.log$")
WORK_LOG_HEADER = (
    "| seq | time | event | task | loop | role | attempt | model | result | locator |"
)
WORK_LOG_SEPARATOR = "|---:|---|---|---|---:|---|---:|---|---|---|"
LEGACY_WORK_LOG_HEADER = (
    "| seq | time | event | task | role | attempt | model | result | locator |"
)
LEGACY_WORK_LOG_SEPARATOR = "|---:|---|---|---|---|---:|---|---|---|"
WORK_LOG_EXECUTION_LOOP_RE = re.compile(
    r"__p(?P<loop>\d+)__(?:worker|selfcheck|review)__a\d+(?=$|[/\\])"
)
AGENT_PROCESS_MARKER_ENV = "AGENT_TASK_EXECUTION_ID"
EXECUTION_CATALOG_PATH: Path | None = None
DISPATCHER_CHILD_BOUNDARY_PROMPT = (
    "You are a child agent already launched by the dispatcher, not the "
    "orchestration caller. Execute only the assigned role directly. Do not "
    "start, monitor, or wait for orchestration through dispatch.py or "
    "orchestrate-agent-task-loop. You may run dispatch.py --validate-plan only "
    "when required by plan or code-review finalization because that mode "
    "validates one candidate PLAN without starting or monitoring orchestration."
)
REPOSITORY_LANGUAGE_PROMPT = "Follow the repository's language and output rules."
SELF_CHECK_PROMPT_PREFIX = REPOSITORY_LANGUAGE_PROMPT
DISPATCHER_TIMEZONE_NAME = "Asia/Seoul"
KST = ZoneInfo(DISPATCHER_TIMEZONE_NAME)
DEFAULT_MAX_PARALLEL = 3


def validated_max_parallel(value: int) -> int:
    """Validate and return a non-negative integer for --max-parallel.

    Rejects negative values and non-integer types. Used both for CLI
    argument parsing and for programmatic callers that may pass arbitrary
    namespaces.
    """
    if not isinstance(value, int) or isinstance(value, bool):
        raise ValueError(
            f"--max-parallel must be an integer >= 0, got {value!r}"
        )
    if value < 0:
        raise ValueError(
            f"--max-parallel must be >= 0, got {value}"
        )
    return value


STREAM_HEARTBEAT_SECONDS = 30
MODEL_RESPONSE_STALL_SECONDS = 3 * 60
RECOVERY_FAILURE_LIMIT = 10
GENERIC_FAILURE_LIMIT_PER_TARGET = 3
SELF_CHECK_UNCHECKED_RETRY_LIMIT = 10
REVIEW_NO_PROGRESS_LIMIT = 10
PROVIDER_TRANSPORT_FAILURES = frozenset(
    {"provider-connection", "provider-stream-disconnect"}
)
FAILURE_EVIDENCE_LIMIT = 2000
# Used to reject stale locators and to bound a live model response that stops
# producing both stream and native-session progress outside tool execution.
RUNTIME_FAILURE_PATTERNS = {
    "context-limit": [
        r"context (?:length|window)", r"maximum context", r"prompt is too long",
        r"too many tokens", r"token limit", r"exceeded.{0,40}token",
        r"output (?:token )?limit", r"maximum output", r"\bmax_tokens\b",
        r"response (?:is )?too long",
    ],
    "provider-quota": [
        r"rate.?limit", r"\bquota\b", r"resource[_ ]?exhausted", r"\b429\b",
        r"too many requests", r"usage limit", r"capacity limit",
        r"\bsession limit\b",
    ],
    "model-unavailable": [
        r"model.{0,40}(?:not found|unavailable)", r"overloaded",
        r"temporarily unavailable",
        r"reasoning[_ ]effort must be one of",
    ],
    "provider-connection": [
        r"\bprovider[_ -]?tunnel[_ -]?error\b",
        (
            r"(?:provider|backend|inference (?:server|endpoint))"
            r".{0,160}(?:connection refused|dial tcp)"
        ),
    ],
    "provider-stream-disconnect": [
        r"backend connection failed during streaming request",
        r"sse stream before done",
        r"(?:model|inference) server was unresponsive",
        r"backend watchdog",
        r"model will be reloaded automatically on retry",
        (
            r"(?:provider|backend|sse).{0,160}"
            r"curl error: failure when receiving data from the peer"
        ),
    ],
}
TARGET_FAILOVER_FAILURES = frozenset(
    {"context-limit", "provider-quota", "model-unavailable"}
)
RECOVERABLE_RUNTIME_FAILURES = TARGET_FAILOVER_FAILURES | PROVIDER_TRANSPORT_FAILURES
QUALIFIED_FAILOVER_FAILURES = RECOVERABLE_RUNTIME_FAILURES
BOUNDED_TARGET_FAILURES = frozenset({"generic-error", "session-stall"})
SELECTOR_FAILOVER_FAILURES = QUALIFIED_FAILOVER_FAILURES | BOUNDED_TARGET_FAILURES


class DispatcherAlreadyRunning(RuntimeError):
    """A live dispatcher owns the workspace; this is non-terminal tracking state."""


class DispatcherTerminalStateError(RuntimeError):
    """Persistent workspace state prevents safe dispatch before work can start."""


class DispatcherInterruptedWithActiveWork(RuntimeError):
    """A control-plane error occurred after one or more agent tasks had started."""


class ExecutionDecisionError(RuntimeError):
    """A selector decision is invalid for this task and must fail closed."""


def now_iso() -> str:
    return datetime.now(KST).isoformat()


def work_log_now_kst() -> str:
    return datetime.now(KST).strftime("%y-%m-%d %H:%M:%S KST")


def sha256_file(path: Path | None) -> str:
    if path is None or not path.exists():
        return "none"
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(65536), b""):
            digest.update(chunk)
    return digest.hexdigest()


DISPATCHER_SOURCE_PATH = Path(__file__).resolve()
DISPATCHER_SOURCE_SHA256 = sha256_file(DISPATCHER_SOURCE_PATH)
DISPATCHER_PROCESS_STARTED_AT = now_iso()


def dispatcher_source_provenance() -> dict[str, Any]:
    current_sha256 = sha256_file(DISPATCHER_SOURCE_PATH)
    return {
        "dispatcher_pid": os.getpid(),
        "dispatcher_process_start_token": process_start_token(os.getpid()),
        "dispatcher_process_started_at": DISPATCHER_PROCESS_STARTED_AT,
        "dispatcher_source_path": str(DISPATCHER_SOURCE_PATH),
        "dispatcher_source_sha256": DISPATCHER_SOURCE_SHA256,
        "dispatcher_source_current_sha256": current_sha256,
        "dispatcher_source_matches_loaded": current_sha256 == DISPATCHER_SOURCE_SHA256,
    }


def plan_identity(path: Path | None) -> str:
    if path is None or not path.exists():
        return "none"
    text = path.read_text(encoding="utf-8", errors="replace")[:1024]
    match = PLAN_IDENTITY_RE.search(text)
    if not match:
        return sha256_file(path)
    fields = [match.group(name) for name in ("task", "plan", "tag")]
    if match.group("milestone_task"):
        fields.append(match.group("milestone_task"))
    identity = "\0".join(fields)
    return "meta:" + hashlib.sha256(identity.encode()).hexdigest()


def milestone_task_ids(metadata: re.Match[str]) -> tuple[str, ...]:
    value = metadata.group("milestone_task")
    return tuple(value.split(",")) if value else ()


def milestone_feature_task_ids(text: str) -> set[str]:
    feature_section = MILESTONE_FEATURE_SECTION_RE.search(text)
    if feature_section is None:
        return set()
    return set(MILESTONE_ITEM_RE.findall(feature_section.group("body")))


def metadata_work_unit_id(metadata: re.Match[str]) -> str:
    work_unit_id = (
        f"{metadata.group('task')}::plan-{metadata.group('plan')}::"
        f"tag-{metadata.group('tag')}"
    )
    if metadata.group("milestone_task"):
        work_unit_id += f"::milestone-task-{metadata.group('milestone_task')}"
    return work_unit_id


def validate_plan_metadata(path: Path, workspace: Path) -> list[str]:
    try:
        head = path.read_text(encoding="utf-8", errors="replace")[:1024]
    except OSError as exc:
        return [f"PLAN metadata를 읽을 수 없다: {exc}"]
    metadata = PLAN_IDENTITY_RE.search(head)
    if metadata is None:
        return [
            "첫 줄 generation header를 판별할 수 없다: "
            "<!-- task=... plan=N tag=... [milestone-task=id[,id...]] -->"
        ]

    task_group = metadata.group("task").split("/", 1)[0]
    task_ids = milestone_task_ids(metadata)
    invalid_ids = [
        task_id
        for task_id in task_ids
        if MILESTONE_TASK_ID_RE.fullmatch(task_id) is None
    ]
    if invalid_ids:
        return [
            "milestone-task id 문법이 Milestone item-id 계약과 다르다: "
            + ", ".join(invalid_ids)
        ]
    if len(task_ids) != len(set(task_ids)):
        return ["milestone-task에 중복 Task id가 있다"]
    if not task_group.startswith("m-"):
        return ["비마일스톤 task에는 milestone-task를 둘 수 없다"] if task_ids else []
    if not task_ids:
        return ["m-* PLAN 첫 줄에는 milestone-task=<id[,id...]>가 필요하다"]

    slug = task_group[2:]
    candidates = sorted(
        path
        for path in (workspace / "agent-roadmap" / "phase").glob(
            f"*/milestones/{slug}.md"
        )
        if path.is_file()
    )
    if len(candidates) != 1:
        return [
            f"milestone-task target은 활성 Milestone과 정확히 하나 매칭되어야 한다: "
            f"slug={slug!r}, matches={len(candidates)}"
        ]
    milestone_text = candidates[0].read_text(encoding="utf-8", errors="replace")
    known_ids = milestone_feature_task_ids(milestone_text)
    unknown_ids = [task_id for task_id in task_ids if task_id not in known_ids]
    if unknown_ids:
        return [
            "milestone-task가 활성 Milestone 기능 Task id와 일치하지 않는다: "
            + ", ".join(unknown_ids)
        ]
    return []


def write_json(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    temporary.replace(path)


def milestone_work_log_path(task: Task) -> Path:
    return (
        task.directory.parent / WORK_LOG_NAME
        if "/" in task.name
        else task.directory / WORK_LOG_NAME
    )


def work_log_task_name(task: Task, role: str) -> str:
    """Return the role-specific active artifact shown in the task column."""
    artifact = task.review if role == "review" else task.plan
    if artifact is None:
        return task.name
    return f"{task.name}/{artifact.name}"


def work_log_loop_number(task: Task, execution_id: str) -> int:
    """Keep one loop identity even when a reviewer archives the active PLAN."""
    match = WORK_LOG_EXECUTION_LOOP_RE.search(execution_id)
    return int(match.group("loop")) if match else plan_number(task)


def append_work_log_event(
    path: Path,
    *,
    task_name: str,
    loop: int,
    event: str,
    execution_id: str,
    role: str,
    attempt: int,
    model: str,
    result: str,
    locator: Path,
) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a+", encoding="utf-8") as stream:
        fcntl.flock(stream.fileno(), fcntl.LOCK_EX)
        try:
            stream.seek(0)
            text = stream.read()
            if not text:
                stream.write(
                    "# Milestone Work Log\n\n"
                    "> Dispatcher-owned execution timeline. Workers and reviewers do not edit this file.\n\n"
                    f"{WORK_LOG_HEADER}\n"
                    f"{WORK_LOG_SEPARATOR}\n"
                )
                sequence = 1
            else:
                stream.seek(0, os.SEEK_END)
                if WORK_LOG_HEADER not in text:
                    if not text.endswith("\n"):
                        stream.write("\n")
                    stream.write(
                        "\n## Dispatcher Timeline\n\n"
                        "> Dispatcher-owned. Workers and reviewers do not edit this section.\n\n"
                        f"{WORK_LOG_HEADER}\n"
                        f"{WORK_LOG_SEPARATOR}\n"
                    )
                sequence = 1 + max(
                    (
                        int(match.group(1))
                        for match in re.finditer(r"^\|\s*(\d+)\s*\|", text, re.MULTILINE)
                    ),
                    default=0,
                )
                if not text.endswith("\n"):
                    stream.write("\n")

            def cell(value: Any) -> str:
                return str(value).replace("|", r"\|").replace("\n", " ")

            stream.write(
                f"| {sequence} | {work_log_now_kst()} | {cell(event)} | "
                f"{cell(task_name)} | "
                f"{loop} | {cell(role)} | {attempt} | {cell(model)} | {cell(result)} | "
                f"{cell(locator.resolve())} |\n"
            )
            stream.flush()
        finally:
            fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
    return path


def append_milestone_event(
    task: Task,
    *,
    event: str,
    execution_id: str,
    role: str,
    attempt: int,
    model: str,
    result: str,
    locator: Path,
) -> Path:
    return append_work_log_event(
        milestone_work_log_path(task),
        task_name=work_log_task_name(task, role),
        loop=work_log_loop_number(task, execution_id),
        event=event,
        execution_id=execution_id,
        role=role,
        attempt=attempt,
        model=model,
        result=result,
        locator=locator,
    )


def safe_name(value: str) -> str:
    return re.sub(r"[^A-Za-z0-9_.-]+", "__", value).strip("_") or "task"


@dataclass(frozen=True)
class AgentSpec:
    cli: str
    model: str
    display: str
    native_resume: bool = False
    target_id: str | None = None
    execution_class: str = "cloud_model"
    selfcheck_required: bool = False
    reasoning_effort: str | None = None
    runtime: dict[str, Any] = field(default_factory=dict)

def agent_spec_from_record(record: dict[str, Any]) -> AgentSpec | None:
    cli = str(record.get("cli") or "")
    model = str(record.get("model") or "")
    if not cli or not model:
        return None
    runtime = record.get("runtime")
    if not isinstance(runtime, dict):
        runtime = {}
    target_id = record.get("target_id")
    if target_id is not None and (not isinstance(target_id, str) or not target_id):
        return None
    execution_class = record.get("execution_class", "cloud_model")
    if execution_class not in {"local_model", "cloud_model"}:
        return None
    selfcheck_required = record.get("selfcheck_required", False)
    if not isinstance(selfcheck_required, bool):
        return None
    reasoning_effort = record.get("reasoning_effort")
    if reasoning_effort is not None and (
        not isinstance(reasoning_effort, str) or not reasoning_effort
    ):
        return None
    native_resume = bool(runtime.get("native_session_monitor"))
    display = f"{cli}/{model}"
    if reasoning_effort is not None:
        display += f" {reasoning_effort}"
    return AgentSpec(
        cli,
        model,
        display,
        native_resume=native_resume,
        target_id=target_id,
        execution_class=execution_class,
        selfcheck_required=selfcheck_required,
        reasoning_effort=reasoning_effort,
        runtime=dict(runtime),
    )


def agent_spec_from_locator(locator: Path | None) -> AgentSpec | None:
    if locator is None:
        return None
    try:
        record = json.loads(locator.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    if not isinstance(record, dict):
        return None
    return agent_spec_from_record(record)


@dataclass(frozen=True)
class NativeSessionState:
    phase: str
    expected_tool_call_ids: tuple[str, ...] = ()
    completed_tool_call_ids: tuple[str, ...] = ()
    pending_tool_call_ids: tuple[str, ...] = ()
    reason: str = ""


@dataclass
class Task:
    name: str
    directory: Path
    plan: Path | None
    review: Path | None
    user_review: Path | None
    recovery: bool
    errors: list[str] = field(default_factory=list)
    index: int = 0
    deps: tuple[str, ...] = ()
    write_set: set[str] = field(default_factory=set)
    write_set_known: bool = False
    plan_hash: str = "none"
    lane: str | None = None
    grade: int | None = None


def task_target_files(task: Task) -> list[str]:
    """Return the canonical plan-declared file targets for dispatcher output."""
    return sorted(str(Path(path).resolve()) for path in task.write_set)


def task_observation_lines(task: Task) -> list[str]:
    """Render the task directory and declared file targets for operator logs."""
    lines = [f"task_dir={task.directory.resolve()}"]
    targets = task_target_files(task)
    if targets:
        lines.extend(f"target_file={path}" for path in targets)
    else:
        lines.append(
            "target_file=unavailable (Modified Files Summary has no valid file claim)"
        )
    return lines


def next_execution_identity(
    store: StateStore,
    task: Task,
    role: str,
) -> tuple[int, str]:
    attempt = store.next_attempt(task, role)
    identity = (
        f"{safe_name(task.name)}__p{plan_number(task)}__{role}__a{attempt:02d}"
    )
    return attempt, identity


class StateStore:
    def __init__(self, workspace: Path):
        self.workspace = workspace.resolve()
        self.workspace_id = hashlib.sha256(
            str(self.workspace).encode()
        ).hexdigest()[:16]
        git_marker = self.workspace / ".git"
        git_directory: Path | None = None
        if git_marker.is_dir():
            git_directory = git_marker
        elif git_marker.is_file():
            marker = git_marker.read_text(encoding="utf-8", errors="replace").strip()
            if marker.startswith("gitdir:"):
                candidate = Path(marker.split(":", 1)[1].strip())
                git_directory = (
                    candidate
                    if candidate.is_absolute()
                    else (self.workspace / candidate).resolve()
                )
        candidates = []
        if git_directory is not None:
            candidates.append(git_directory / "agent-task-dispatcher")
        state_base = Path(os.environ.get("XDG_STATE_HOME", str(Path.home() / ".local" / "state")))
        candidates.append(state_base / "agent-task-dispatcher" / self.workspace_id)
        self.root = candidates[-1]
        last_error: OSError | None = None
        for candidate in candidates:
            try:
                candidate.mkdir(parents=True, exist_ok=True)
                self.root = candidate
                last_error = None
                break
            except OSError as exc:
                last_error = exc
        if last_error is not None:
            raise DispatcherTerminalStateError(
                f"dispatcher state 디렉터리를 만들 수 없다: {candidates}"
            ) from last_error
        self.path = self.root / "state.json"
        self.runs = self.root / "runs"
        self.runs.mkdir(exist_ok=True)
        self.lock_stream = (self.root / "dispatcher.lock").open("a+", encoding="utf-8")
        try:
            fcntl.flock(self.lock_stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError as exc:
            self.lock_stream.seek(0)
            owner = self.lock_stream.read().strip() or "owner metadata unavailable"
            self.lock_stream.close()
            raise DispatcherAlreadyRunning(
                f"같은 workspace의 dispatcher가 이미 실행 중이다: "
                f"{self.root}; owner={owner}"
            ) from exc
        try:
            self.lock_stream.seek(0)
            self.lock_stream.truncate()
            self.lock_stream.write(
                json.dumps(dispatcher_source_provenance(), ensure_ascii=False) + "\n"
            )
            self.lock_stream.flush()
        except OSError as exc:
            self.lock_stream.close()
            raise DispatcherTerminalStateError(
                f"dispatcher lock owner metadata를 기록할 수 없다: {self.root}"
            ) from exc
        if self.path.exists():
            try:
                self.data = json.loads(self.path.read_text(encoding="utf-8"))
            except (json.JSONDecodeError, OSError) as exc:
                self.lock_stream.close()
                raise DispatcherTerminalStateError(
                    f"dispatcher state를 읽을 수 없다: {self.path}"
                ) from exc
            if not isinstance(self.data, dict):
                self.lock_stream.close()
                raise DispatcherTerminalStateError(
                    f"dispatcher state가 object가 아니다: {self.path}"
                )
        else:
            self.data = {"tasks": {}, "attempt_counters": {}}
        try:
            self._bind_workspace_identity()
            self.write_claim_snapshot()
        except DispatcherTerminalStateError:
            self.lock_stream.close()
            raise

    def _bind_workspace_identity(self) -> None:
        expected = {
            "id": self.workspace_id,
            "root": str(self.workspace),
        }
        current = self.data.get("workspace_identity")
        if current is None:
            self.data["workspace_identity"] = expected
            return
        if not isinstance(current, dict):
            raise DispatcherTerminalStateError(
                f"dispatcher workspace identity가 object가 아니다: {self.path}"
            )
        if (
            current.get("id") != expected["id"]
            or current.get("root") != expected["root"]
        ):
            raise DispatcherTerminalStateError(
                "dispatcher state의 workspace identity가 현재 checkout과 다르다: "
                f"state={current} current={expected}"
            )

    def write_claim_snapshot(self) -> dict[str, dict[str, Any]]:
        raw = self.data.setdefault("write_claims", {})
        if not isinstance(raw, dict):
            raise DispatcherTerminalStateError(
                f"dispatcher write_claims가 object가 아니다: {self.path}"
            )
        snapshot: dict[str, dict[str, Any]] = {}
        for owner, value in raw.items():
            if not isinstance(owner, str) or not owner or not isinstance(value, dict):
                raise DispatcherTerminalStateError(
                    f"dispatcher write claim 형식이 유효하지 않다: owner={owner!r}"
                )
            paths = value.get("paths")
            exclusive = value.get("exclusive", False)
            if not isinstance(paths, list) or not isinstance(exclusive, bool):
                raise DispatcherTerminalStateError(
                    f"dispatcher write claim 경로 형식이 유효하지 않다: owner={owner}"
                )
            if value.get("workspace_id") != self.workspace_id:
                raise DispatcherTerminalStateError(
                    "dispatcher write claim의 workspace identity가 다르다: "
                    f"owner={owner}"
                )
            canonical: list[str] = []
            for raw_path in paths:
                if not isinstance(raw_path, str) or not raw_path:
                    raise DispatcherTerminalStateError(
                        f"dispatcher write claim 경로가 유효하지 않다: owner={owner}"
                    )
                path = Path(raw_path)
                resolved = path.resolve()
                try:
                    resolved.relative_to(self.workspace)
                except ValueError as exc:
                    raise DispatcherTerminalStateError(
                        "dispatcher write claim이 workspace 밖을 가리킨다: "
                        f"owner={owner} path={raw_path}"
                    ) from exc
                if not path.is_absolute() or str(resolved) != raw_path or resolved == self.workspace:
                    raise DispatcherTerminalStateError(
                        "dispatcher write claim 경로가 canonical file이 아니다: "
                        f"owner={owner} path={raw_path}"
                    )
                canonical.append(raw_path)
            if (not canonical and not exclusive) or len(canonical) != len(set(canonical)):
                raise DispatcherTerminalStateError(
                    f"dispatcher write claim 경로 집합이 유효하지 않다: owner={owner}"
                )
            record = dict(value)
            record["paths"] = sorted(canonical)
            snapshot[owner] = record
        return snapshot

    def replace_write_claims(
        self,
        claims: dict[str, dict[str, Any]],
        *,
        persist: bool,
    ) -> None:
        previous = self.data.get("write_claims", {})
        self.data["write_claims"] = claims
        try:
            self.write_claim_snapshot()
            if persist and previous != claims:
                self.save()
        except BaseException:
            self.data["write_claims"] = previous
            raise

    def adopt_active_write_claim(self, task: Task) -> None:
        claims = self.write_claim_snapshot()
        if task.name in claims:
            return
        timestamp = now_iso()
        claims[task.name] = {
            "task": task.name,
            "plan_hash": task.plan_hash,
            "paths": sorted(task.write_set) if task.write_set_known else [],
            "exclusive": not task.write_set_known,
            "workspace_id": self.workspace_id,
            "acquired_at": timestamp,
            "updated_at": timestamp,
            "source": "active-recovery",
        }
        self.replace_write_claims(claims, persist=True)

    def release_write_claim(self, task_name: str, *, persist: bool = True) -> bool:
        claims = self.write_claim_snapshot()
        if task_name not in claims:
            return False
        del claims[task_name]
        self.replace_write_claims(claims, persist=persist)
        return True

    def save(self) -> None:
        write_json(self.path, self.data)

    def close(self) -> None:
        if not self.lock_stream.closed:
            self.lock_stream.close()

    def task_state(self, task: Task) -> dict[str, Any]:
        tasks = self.data.setdefault("tasks", {})
        current = tasks.get(task.name)
        if not current or current.get("plan_hash") != task.plan_hash:
            current = {
                "plan_hash": task.plan_hash,
                "worker_done": False,
                "worker_cli": None,
                "worker_model": None,
                "selfcheck_done": False,
                "blocked": None,
                "active_stage": None,
                "active_locator": None,
                "review_no_progress": 0,
                "selfcheck_incomplete": 0,
                "selfcheck_context_locator": None,
                "recovery_failures": {},
                "execution_decisions": {},
                "route_transition_history": [],
                "stage_failure_budgets": {},
                "generic_failure_budgets": {},
                "retry_failover_pending": False,
                "retry_failover_context": None,
                "blocker_evidence": None,
            }
            tasks[task.name] = current
            self.save()
        return current

    def peek_task_state(self, task: Task) -> dict[str, Any]:
        current = self.data.get("tasks", {}).get(task.name)
        if current and current.get("plan_hash") == task.plan_hash:
            return dict(current)
        return {
            "plan_hash": task.plan_hash,
            "worker_done": False,
            "worker_cli": None,
            "worker_model": None,
            "selfcheck_done": False,
            "blocked": None,
            "active_stage": None,
            "active_locator": None,
            "review_no_progress": 0,
            "selfcheck_incomplete": 0,
            "selfcheck_context_locator": None,
            "recovery_failures": {},
            "execution_decisions": {},
            "route_transition_history": [],
            "generic_failure_budgets": {},
            "retry_failover_pending": False,
            "retry_failover_context": None,
            "blocker_evidence": None,
        }

    def update_task(self, task: Task, **values: Any) -> None:
        state = self.task_state(task)
        state.update(values)
        self.save()

    def mark_active(self, task: Task, stage: str, locator: Path | None = None) -> None:
        self.update_task(
            task,
            active_stage=stage,
            active_locator=str(locator) if locator else None,
            active_started_at=now_iso(),
        )

    def clear_active(self, task: Task) -> None:
        self.update_task(
            task,
            active_stage=None,
            active_locator=None,
            active_started_at=None,
        )

    def consume_matching_retry_handoff(self, task: Task, locator_path: str) -> bool:
        """Atomically consume a pending retry handoff when a matching locator exists.

        When a worker writes its locator and sets active_locator, the pending
        retry-failover state must be cleared in the same transaction.
        This prevents a crash window where a restart sees the pending handoff
        and creates a duplicate invocation.

        Returns True if the pending handoff was consumed, False if no matching
        locator was found (active_locator is None or differs from locator_path).
        """
        state = self.task_state(task)
        active = state.get("active_locator")
        if active != locator_path:
            return False
        pending = state.get("retry_failover_pending")
        if not pending:
            return False
        context = state.get("retry_failover_context")
        if not isinstance(context, dict):
            return False
        context_locator = context.get("locator")
        if context_locator != locator_path:
            return False
        # Snapshot current state to restore on save failure. This ensures the
        # crash window is not widened by a partial consume: if the save fails,
        # the pending handoff remains intact both in-memory and on-disk.
        pre_state = dict(state)
        pre_keys = set(state.keys())
        pre_values = {k: state.get(k) for k in ["retry_failover_pending", "retry_failover_context"]}
        try:
            self.update_task(
                task,
                retry_failover_pending=False,
                retry_failover_context=None,
            )
        except Exception:
            # Restore the pre-consume state on any failure.
            for k, v in pre_values.items():
                state[k] = v
            # Restore key existence: if a key existed before, restore its value;
            # if a key did not exist before, ensure it is not present.
            for k in list(state.keys()):
                if k not in pre_keys:
                    del state[k]
            for k, v in pre_values.items():
                if k not in state:
                    state[k] = v
            raise
        return True

    def commit_retry_handoff_locator(
        self, task: Task, handoff_id: str, locator_path: str,
    ) -> bool:
        """Atomically commit a new locator and consume a matching pending retry handoff.

        This is the durable one-save transition for retry handoff. It matches
        the pending handoff by stable handoff_id (not by locator path, which
        changes on each attempt) and atomically updates active_locator, clears
        the pending flag, and clears the context in a single save.

        On save failure the pre-state is fully restored both in-memory and on
        disk so the crash window is not widened.

        Returns True if a matching pending handoff was consumed, False if no
        pending handoff with the given handoff_id was found.
        """
        state = self.task_state(task)
        pending = state.get("retry_failover_pending")
        if not pending:
            return False
        context = state.get("retry_failover_context")
        if not isinstance(context, dict):
            return False
        if context.get("handoff_id") != handoff_id:
            return False
        # Snapshot current state to restore on save failure.
        pre_state = dict(state)
        pre_keys = set(state.keys())
        pre_values = {
            k: state.get(k)
            for k in [
                "retry_failover_pending",
                "retry_failover_context",
                "active_locator",
            ]
        }
        try:
            self.update_task(
                task,
                active_locator=locator_path,
                retry_failover_pending=False,
                retry_failover_context=None,
            )
        except Exception:
            for k, v in pre_values.items():
                state[k] = v
            for k in list(state.keys()):
                if k not in pre_keys:
                    del state[k]
            for k, v in pre_values.items():
                if k not in state:
                    state[k] = v
            raise
        return True

    def next_attempt(self, task: Task, role: str) -> int:
        key = f"{task.name}|{task.plan_hash}|{role}"
        counters = self.data.setdefault("attempt_counters", {})
        number = int(counters.get(key, 0))
        counters[key] = number + 1
        self.save()
        return number

    def clear_blocked(self, task_group: str | None = None) -> None:
        prefix = f"{task_group}/" if task_group else None
        for task_name, value in self.data.get("tasks", {}).items():
            if (
                task_group is not None
                and task_name != task_group
                and not task_name.startswith(prefix)
            ):
                continue
            value["blocked"] = None
            value["review_no_progress"] = 0
            value["selfcheck_incomplete"] = 0
            value["selfcheck_context_locator"] = None
            value["recovery_failures"] = {}
            value["stage_failure_budgets"] = {}
            value["generic_failure_budgets"] = {}
            value["retry_failover_pending"] = False
        self.save()

    def accept_catalog_revision(
        self,
        tasks: list[Task],
        *,
        catalog_revision: str,
        catalog_source: str,
        persist: bool,
    ) -> list[str]:
        """Reset stale worker selection only after an explicit operator action."""
        accepted: list[str] = []
        staged: list[tuple[Task, dict[str, Any], dict[str, Any], str]] = []
        for task in tasks:
            state = self.data.get("tasks", {}).get(task.name)
            if not isinstance(state, dict) or state.get("plan_hash") != task.plan_hash:
                continue
            decisions = state.get("execution_decisions")
            if not isinstance(decisions, dict):
                continue
            mismatched: list[tuple[str, dict[str, Any]]] = []
            for stage, decision in decisions.items():
                if not isinstance(decision, dict):
                    continue
                evidence = decision.get("catalog")
                prior_revision = (
                    evidence.get("revision") if isinstance(evidence, dict) else None
                )
                if prior_revision and prior_revision != catalog_revision:
                    mismatched.append((stage, decision))
            if not mismatched:
                continue
            if [stage for stage, _ in mismatched] != ["worker"]:
                raise DispatcherTerminalStateError(
                    "catalog revision 수락은 미완료 worker 선택에만 허용된다: "
                    f"task={task.name} stages={[stage for stage, _ in mismatched]}"
                )
            if state.get("worker_done"):
                raise DispatcherTerminalStateError(
                    "완료된 worker의 catalog revision은 재선택할 수 없다: "
                    f"task={task.name}"
                )
            active_live, active_detail = external_active_is_live(
                state,
                expected_workspace=self.workspace,
                expected_workspace_id=self.workspace_id,
                expected_runs_root=self.runs,
            )
            if active_live:
                raise DispatcherTerminalStateError(
                    "실행 중 worker가 있어 catalog revision을 재선택할 수 없다: "
                    f"task={task.name}; {active_detail}"
                )
            staged.append((task, state, mismatched[0][1], active_detail))

        for task, state, prior_decision, active_detail in staged:
            decisions = dict(state.get("execution_decisions", {}))
            decisions.pop("worker", None)
            recovery_failures = dict(state.get("recovery_failures", {}))
            recovery_failures.pop("worker", None)
            stage_budgets = dict(state.get("stage_failure_budgets", {}))
            work_unit_id = prior_decision.get("work_unit_id")
            if isinstance(work_unit_id, str):
                stage_budgets.pop(f"{work_unit_id}|worker", None)
            generic_budgets = dict(state.get("generic_failure_budgets", {}))
            if isinstance(work_unit_id, str):
                prefix = f"{work_unit_id}|worker|"
                generic_budgets = {
                    key: value
                    for key, value in generic_budgets.items()
                    if not key.startswith(prefix)
                }
            history = list(state.get("catalog_revision_acceptance_history", []))
            history.append(
                {
                    "accepted_at": now_iso(),
                    "stage": "worker",
                    "work_unit_id": work_unit_id,
                    "previous_catalog": prior_decision.get("catalog"),
                    "previous_selected": prior_decision.get("selected"),
                    "accepted_catalog": {
                        "revision": catalog_revision,
                        "source": catalog_source,
                    },
                    "stale_active_detail": active_detail,
                }
            )
            state.update(
                execution_decisions=decisions,
                catalog_revision_acceptance_history=history,
                recovery_failures=recovery_failures,
                stage_failure_budgets=stage_budgets,
                generic_failure_budgets=generic_budgets,
                retry_failover_pending=False,
                retry_failover_context=None,
                blocker_evidence=None,
                blocked=None,
                active_stage=None,
                active_locator=None,
                active_started_at=None,
            )
            accepted.append(task.name)
        if persist and accepted:
            self.save()
        return accepted

    def mark_retry_failover(self, task_group: str | None = None, workspace: Path | None = None) -> None:
        prefix = f"{task_group}/" if task_group else None
        for task_name, value in self.data.get("tasks", {}).items():
            if (
                task_group is not None
                and task_name != task_group
                and not task_name.startswith(prefix)
            ):
                continue
            if not value.get("blocked"):
                continue
            blocker_evidence = value.get("blocker_evidence") if isinstance(value.get("blocker_evidence"), dict) else {}
            decisions = value.get("execution_decisions", {})
            worker_decision = decisions.get("worker") if isinstance(decisions, dict) else None
            role = blocker_evidence.get("role")
            failure_class = blocker_evidence.get("failure_class")
            locator = blocker_evidence.get("locator")
            selected = blocker_evidence.get("selected")
            work_unit_id = blocker_evidence.get("work_unit_id")
            qualified = (
                role == "worker"
                and failure_class in QUALIFIED_FAILOVER_FAILURES
                and isinstance(locator, str)
                and locator.strip()
                and isinstance(selected, dict)
                and isinstance(work_unit_id, str)
                and isinstance(worker_decision, dict)
                and worker_decision.get("work_unit_id") == work_unit_id
            )
            handoff_id = str(uuid.uuid4())
            retry_context = ({
                "role": role,
                "failure_class": failure_class,
                "locator": locator,
                "selected": selected,
                "work_unit_id": work_unit_id,
                "handoff_id": handoff_id,
            } if qualified else None)

            value["blocked"] = None
            value["review_no_progress"] = 0
            value["selfcheck_incomplete"] = 0
            value["selfcheck_context_locator"] = None
            value["recovery_failures"] = {}
            value["stage_failure_budgets"] = {}
            value["retry_failover_pending"] = qualified
            value["retry_failover_context"] = retry_context
            value["blocker_evidence"] = None
        self.save()


    def prepare_orchestration(
        self,
        scope: str,
        tasks: list[Task],
        workspace: Path,
    ) -> None:
        orchestrations = self.data.setdefault("orchestrations", {})
        current = orchestrations.get(scope)
        if current is None or (current.get("status") == "complete" and tasks):
            current = {"status": "running", "tasks": {}}
            orchestrations[scope] = current
        changed = False
        tracked = current.setdefault("tasks", {})
        for task in tasks:
            record = tracked.get(task.name)
            if record is None:
                tracked[task.name] = {
                    "status": "active",
                    "archive": None,
                    "archive_baseline": [
                        str(path.resolve())
                        for path in matching_archive_directories_by_name(
                            workspace,
                            task.name,
                            require_complete=False,
                        )
                    ],
                }
                changed = True
                continue
            if record.get("status") != "complete" and (
                record.get("status") != "active" or "reason" in record
            ):
                record["status"] = "active"
                record.pop("reason", None)
                changed = True
        if changed or current.get("status") != "running":
            current["status"] = "running"
            self.save()

    def mark_orchestration_task_complete(
        self,
        scope: str,
        task_name: str,
        archive: str | Path,
    ) -> None:
        archive_path = Path(archive).resolve()
        if not archive_path.is_dir() or not (archive_path / "complete.log").is_file():
            raise RuntimeError(
                f"완료 archive에 complete.log가 없다: task={task_name} archive={archive_path}"
            )
        current = self.data.setdefault("orchestrations", {}).setdefault(
            scope, {"status": "running", "tasks": {}}
        )
        tracked = current.setdefault("tasks", {})
        record = tracked.setdefault(
            task_name,
            {"status": "active", "archive": None, "archive_baseline": []},
        )
        record.update(status="complete", archive=str(archive_path))
        record.pop("reason", None)
        self.release_write_claim(task_name, persist=False)
        self.save()
        cleanup_completed_task_attempt_logs(self.runs, task_name)

    def mark_orchestration_blocked(
        self,
        scope: str,
        outcomes: dict[str, tuple[str, str]],
    ) -> None:
        current = self.data.setdefault("orchestrations", {}).setdefault(
            scope, {"status": "running", "tasks": {}}
        )
        current["status"] = "blocked"
        tracked = current.setdefault("tasks", {})
        for task_name, (status, reason) in outcomes.items():
            record = tracked.setdefault(
                task_name,
                {
                    "status": "active",
                    "archive": None,
                    "archive_baseline": [],
                },
            )
            if record.get("status") == "complete":
                continue
            record.update(status=status, reason=reason)
        self.save()

    def reconcile_orchestration(
        self,
        scope: str,
        workspace: Path,
        active_or_running: set[str],
    ) -> tuple[dict[str, str], dict[str, str]]:
        current = self.data.setdefault("orchestrations", {}).setdefault(
            scope, {"status": "running", "tasks": {}}
        )
        completed: dict[str, str] = {}
        errors: dict[str, str] = {}
        changed = False
        for task_name, record in current.setdefault("tasks", {}).items():
            if record.get("status") == "complete":
                archive = str(record.get("archive") or "")
                if archive and (Path(archive) / "complete.log").is_file():
                    completed[task_name] = archive
                    if task_name not in active_or_running:
                        changed = (
                            self.release_write_claim(task_name, persist=False)
                            or changed
                        )
                else:
                    errors[task_name] = "persisted complete archive가 유효하지 않다"
                continue
            if task_name in active_or_running:
                continue
            baseline = set(str(path) for path in record.get("archive_baseline", []))
            candidates = [
                path
                for path in matching_archive_directories_by_name(workspace, task_name)
                if str(path.resolve()) not in baseline
            ]
            if len(candidates) == 1:
                archive = str(candidates[0].resolve())
                record.update(status="complete", archive=archive)
                completed[task_name] = archive
                changed = self.release_write_claim(task_name, persist=False) or changed
                changed = True
            elif not candidates:
                errors[task_name] = (
                    "관찰된 task가 active와 새 complete.log archive 모두에서 사라졌다"
                )
            else:
                errors[task_name] = (
                    "새 complete.log archive가 여러 개라 완료 경로를 확정할 수 없다: "
                    + ",".join(str(path) for path in candidates)
                )
        if changed:
            self.save()
        for task_name in completed:
            if task_name not in active_or_running:
                cleanup_completed_task_attempt_logs(self.runs, task_name)
        return completed, errors

    def orchestration_tasks(self, scope: str) -> set[str]:
        current = self.data.get("orchestrations", {}).get(scope, {})
        return set(current.get("tasks", {}))

    def mark_orchestration_complete(self, scope: str) -> None:
        current = self.data.setdefault("orchestrations", {}).setdefault(
            scope, {"status": "running", "tasks": {}}
        )
        current["status"] = "complete"
        self.save()


def orchestration_live_agent_processes(
    store: StateStore,
    scope: str,
) -> dict[str, str]:
    """Return observed tasks with live or conservatively active evidence."""
    task_states = store.data.get("tasks", {})
    live: dict[str, str] = {}
    for task_name in store.orchestration_tasks(scope):
        state = task_states.get(task_name)
        if not isinstance(state, dict):
            continue
        is_live, detail = external_active_is_live(
            state,
            expected_workspace=store.workspace,
            expected_workspace_id=store.workspace_id,
            expected_runs_root=store.runs,
        )
        if is_live:
            live[task_name] = detail
    return live


def workspace_live_agent_processes(
    store: StateStore,
) -> dict[str, str]:
    """Return observed tasks across the entire physical workspace with live or conservatively active evidence."""
    task_states = store.data.get("tasks", {})
    live: dict[str, str] = {}
    if isinstance(task_states, dict):
        for task_name, state in task_states.items():
            if not isinstance(state, dict):
                continue
            is_live, detail = external_active_is_live(
                state,
                expected_workspace=store.workspace,
                expected_workspace_id=store.workspace_id,
                expected_runs_root=store.runs,
            )
            if is_live:
                live[task_name] = detail
    return live


def parse_route(plan: Path | None) -> tuple[str | None, int | None]:
    if plan is None:
        return None, None
    match = PLAN_RE.match(plan.name)
    if not match:
        return None, None
    return match.group(1), int(match.group(2))


def parse_task_name(task_root: Path, directory: Path) -> str:
    return directory.relative_to(task_root).as_posix()


def inspect_write_set(
    plan: Path | None,
    workspace: Path,
) -> tuple[set[str], list[str]]:
    if plan is None:
        return set(), ["PLAN 경로가 없다"]
    if not plan.is_file():
        return set(), [f"PLAN 파일이 없다: {plan}"]
    workspace = workspace.resolve()
    try:
        text = plan.read_text(encoding="utf-8", errors="replace")
    except OSError as exc:
        return set(), [f"PLAN 파일을 읽을 수 없다: {plan}: {exc}"]
    matches = []
    for heading in MODIFIED_FILES_HEADINGS:
        pattern = rf"^##\s*{re.escape(heading)}[ \t]*$([\s\S]*?)(?=^##\s|\Z)"
        for m in re.finditer(pattern, text, re.MULTILINE):
            matches.append(m)
    if not matches:
        return set(), ["Modified Files Summary 섹션이 없다"]
    if len(matches) != 1:
        return set(), [
            f"Modified Files Summary 섹션은 정확히 1개여야 한다: count={len(matches)}"
        ]
    match = matches[0]
    result: set[str] = set()
    diagnostics: list[str] = []
    for line in match.group(1).splitlines():
        if not line.lstrip().startswith("|"):
            continue
        cells = [cell.strip() for cell in line.strip().strip("|").split("|")]
        if not cells:
            continue
        if all(set(cell) <= {":", "-"} for cell in cells):
            continue
        if cells[0].casefold() in MODIFIED_FILES_HEADER_CELLS:
            continue
        claims = re.findall(r"`([^`]+)`", cells[0])
        if not claims:
            diagnostics.append(
                f"정확한 backtick workspace 파일 경로가 없는 claim 행: {cells[0]}"
            )
            continue
        for value in claims:
            normalized = re.sub(r":\d+(?::\d+)?$", "", value.strip())
            if not normalized:
                diagnostics.append("빈 경로 claim은 허용되지 않는다")
                continue
            if PLACEHOLDER_PATH_RE.search(normalized):
                diagnostics.append(
                    f"placeholder 또는 malformed path claim은 허용되지 않는다: {normalized}"
                )
                continue
            if normalized.startswith(("http://", "https://")):
                diagnostics.append(f"URL claim은 허용되지 않는다: {normalized}")
                continue
            if "\\" in normalized:
                diagnostics.append(
                    f"malformed path claim은 허용되지 않는다: {normalized}"
                )
                continue
            if any(character in normalized for character in "*?[]"):
                diagnostics.append(
                    f"glob 또는 broad path claim은 허용되지 않는다: {normalized}"
                )
                continue
            if normalized.endswith(("/", "\\")):
                diagnostics.append(
                    f"디렉터리 claim은 허용되지 않는다: {normalized}"
                )
                continue
            candidate = Path(normalized)
            try:
                resolved = (
                    candidate.resolve()
                    if candidate.is_absolute()
                    else (workspace / candidate).resolve()
                )
            except (OSError, RuntimeError) as exc:
                diagnostics.append(
                    f"경로를 canonicalize할 수 없다: {normalized}: {exc}"
                )
                continue
            try:
                resolved.relative_to(workspace)
            except ValueError:
                diagnostics.append(
                    f"workspace 밖 claim은 허용되지 않는다: {normalized}"
                )
                continue
            if resolved == workspace:
                diagnostics.append("workspace root claim은 허용되지 않는다")
                continue
            if resolved.is_dir():
                diagnostics.append(
                    f"디렉터리 claim은 허용되지 않는다: {normalized}"
                )
                continue
            result.add(str(resolved))
    if not result:
        diagnostics.append("정확한 workspace 파일 claim이 하나 이상 필요하다")
    return result, diagnostics


def extract_write_set(plan: Path | None, workspace: Path) -> tuple[set[str], bool]:
    write_set, diagnostics = inspect_write_set(plan, workspace)
    if diagnostics:
        return set(), False
    return write_set, True


def latest_verdict_log(directory: Path) -> Path | None:
    candidates: list[tuple[int, Path]] = []
    for path in directory.glob("code_review_*.log"):
        match = REVIEW_LOG_RE.fullmatch(path.name)
        if match is None or not path.is_file() or read_verdict(path) is None:
            continue
        candidates.append((int(match.group(3)), path))
    if not candidates:
        return None
    return max(
        candidates,
        key=lambda candidate: (candidate[0], candidate[1].name),
    )[1]


def matching_plan_log(directory: Path, review_log: Path | None) -> Path | None:
    if review_log is None:
        return None
    review_identity = plan_identity(review_log)
    matches = [
        path
        for path in directory.glob("plan_*.log")
        if PLAN_LOG_RE.fullmatch(path.name) is not None
        and path.is_file()
        and plan_identity(path) == review_identity
    ]
    return max(matches, key=lambda path: path.stat().st_mtime_ns) if matches else None


def read_task_directory(workspace: Path, directory: Path) -> Task | None:
    """Read one already-known task directory without scanning the task group."""
    task_root = workspace / "agent-task"
    if not directory.is_dir():
        return None
    plans = sorted(p for p in directory.iterdir() if p.is_file() and PLAN_RE.match(p.name))
    reviews = sorted(p for p in directory.iterdir() if p.is_file() and REVIEW_RE.match(p.name))
    users = sorted(directory.glob("USER_REVIEW.md"))
    complete = directory / "complete.log"
    recovery_log = latest_verdict_log(directory)
    if not plans and not reviews and not users and not complete.exists() and recovery_log is None:
        return None
    name = parse_task_name(task_root, directory)
    errors: list[str] = []
    if len(plans) > 1:
        errors.append(f"active PLAN이 {len(plans)}개다")
    if len(reviews) > 1:
        errors.append(f"active CODE_REVIEW가 {len(reviews)}개다")
    if len(users) > 1:
        errors.append(f"USER_REVIEW가 {len(users)}개다")
    if users and (plans or reviews):
        errors.append("USER_REVIEW stop state와 active PLAN/CODE_REVIEW가 공존한다")
    plan = plans[0] if len(plans) == 1 else None
    review = reviews[0] if len(reviews) == 1 else None
    recovery = complete.exists() or recovery_log is not None
    if bool(plan) != bool(review) and not recovery:
        errors.append("active PLAN/CODE_REVIEW pair가 불완전하다")
    relative = directory.relative_to(task_root)
    subtask = relative.parts[1] if len(relative.parts) == 2 else None
    index = 0
    deps: tuple[str, ...] = ()
    if subtask:
        match = SUBTASK_RE.match(subtask)
        if match:
            index = int(match.group("index"))
            deps = tuple((match.group("deps") or "").split(",")) if match.group("deps") else ()
        else:
            errors.append(f"split subtask 이름이 계약과 다르다: {subtask}")
    lane, grade = parse_route(plan)
    recovery_plan = matching_plan_log(directory, recovery_log)
    write_set_source = plan or recovery_plan
    write_set: set[str] = set()
    write_set_known = False
    if recovery_log is not None and recovery_plan is None:
        errors.append(
            "PLAN Modified Files Summary를 복구할 matching PLAN log가 없다"
        )
    elif write_set_source is not None:
        write_set, write_set_diagnostics = inspect_write_set(
            write_set_source,
            workspace,
        )
        write_set_known = bool(write_set) and not write_set_diagnostics
        errors.extend(
            f"PLAN Modified Files Summary가 유효하지 않다: {diagnostic}"
            for diagnostic in write_set_diagnostics
        )
    if plan is not None:
        metadata = PLAN_IDENTITY_RE.search(
            plan.read_text(encoding="utf-8", errors="replace")[:1024]
        )
        if metadata is None:
            errors.append("PLAN 첫 줄 generation metadata를 판별할 수 없다")
        elif metadata.group("task") != name:
            errors.append(
                f"PLAN task metadata가 디렉터리와 다르다: {metadata.group('task')}"
            )
        errors.extend(validate_plan_metadata(plan, workspace))
        if review is not None and plan_identity(plan) != plan_identity(review):
            errors.append("PLAN/CODE_REVIEW generation metadata가 다르다")
    return Task(
        name=name,
        directory=directory,
        plan=plan,
        review=review,
        user_review=users[0] if len(users) == 1 else None,
        recovery=recovery,
        errors=errors,
        index=index,
        deps=deps,
        write_set=write_set,
        write_set_known=write_set_known,
        plan_hash=(
            plan_identity(plan)
            if plan
            else sha256_file(
                recovery_log
                or (users[0] if len(users) == 1 else complete)
            )
        ),
        lane=lane,
        grade=grade,
    )




@dataclass
class StageFailureBudget:
    """Persistent failure counter shared by one work unit and stage."""

    store: StateStore
    task: Task
    work_unit_id: str
    stage: str


    @classmethod
    def from_decision(cls, store: StateStore, task: Task, decision: dict[str, Any]) -> "StageFailureBudget":
        work_unit_id = decision.get("work_unit_id")
        stage = decision.get("stage")
        if not isinstance(work_unit_id, str) or not work_unit_id or not isinstance(stage, str) or not stage:
            raise ExecutionDecisionError("stage failure budget identity가 유효하지 않다")
        return cls(store, task, work_unit_id, stage)


    @property
    def key(self) -> str:
        return f"{self.work_unit_id}|{self.stage}"

    def _budgets(self) -> dict[str, Any]:
        state = self.store.task_state(self.task)
        budgets = state.get("stage_failure_budgets", {})
        if not isinstance(budgets, dict):
            raise ExecutionDecisionError("persisted stage failure budgets schema가 유효하지 않다")
        return dict(budgets)

    def count(self) -> int:
        entry = self._budgets().get(self.key, {})
        if not isinstance(entry, dict):
            raise ExecutionDecisionError("persisted stage failure budget entry가 유효하지 않다")
        return int(entry.get("count", 0))

    def record_failure(self, *, target: dict[str, Any], transition: str) -> int:
        budgets = self._budgets()
        entry = dict(budgets.get(self.key, {}))
        count = int(entry.get("count", 0)) + 1
        entry.update(
            work_unit_id=self.work_unit_id, stage=self.stage, count=count,
            last_target={
                "target_id": target.get("target_id"),
                "agent": target.get("agent"),
                "model": target.get("model"),
            },
            last_transition=transition,
        )
        budgets[self.key] = entry
        self.store.update_task(self.task, stage_failure_budgets=budgets)
        return count

    def reset_on_success(self) -> None:
        budgets = self._budgets()
        budgets.pop(self.key, None)
        self.store.update_task(self.task, stage_failure_budgets=budgets)


@dataclass
class GenericFailureBudget:
    """Persistent generic-error counter scoped to one selected target."""

    store: StateStore
    task: Task
    work_unit_id: str
    stage: str
    target_id: str

    @classmethod
    def from_decision(
        cls,
        store: StateStore,
        task: Task,
        decision: dict[str, Any],
    ) -> "GenericFailureBudget":
        work_unit_id = decision.get("work_unit_id")
        stage = decision.get("stage")
        selected = decision.get("selected")
        target_id = selected.get("target_id") if isinstance(selected, dict) else None
        if not all(
            isinstance(value, str) and value
            for value in (work_unit_id, stage, target_id)
        ):
            raise ExecutionDecisionError(
                "generic failure budget identity가 유효하지 않다"
            )
        return cls(store, task, work_unit_id, stage, target_id)

    @property
    def key(self) -> str:
        return f"{self.work_unit_id}|{self.stage}|{self.target_id}"

    def _budgets(self) -> dict[str, Any]:
        state = self.store.task_state(self.task)
        budgets = state.get("generic_failure_budgets", {})
        if not isinstance(budgets, dict):
            raise ExecutionDecisionError(
                "persisted generic failure budgets schema가 유효하지 않다"
            )
        return dict(budgets)

    def count(self) -> int:
        entry = self._budgets().get(self.key, {})
        if not isinstance(entry, dict):
            raise ExecutionDecisionError(
                "persisted generic failure budget entry가 유효하지 않다"
            )
        return int(entry.get("count", 0))

    def record_failure(self) -> int:
        budgets = self._budgets()
        entry = dict(budgets.get(self.key, {}))
        count = int(entry.get("count", 0)) + 1
        entry.update(
            work_unit_id=self.work_unit_id,
            stage=self.stage,
            target_id=self.target_id,
            count=count,
        )
        budgets[self.key] = entry
        self.store.update_task(self.task, generic_failure_budgets=budgets)
        return count

    def reset_stage_on_success(self) -> None:
        budgets = self._budgets()
        prefix = f"{self.work_unit_id}|{self.stage}|"
        budgets = {
            key: value for key, value in budgets.items() if not key.startswith(prefix)
        }
        self.store.update_task(self.task, generic_failure_budgets=budgets)


def scan_tasks(
    workspace: Path,
    task_group: str | None,
    *,
    exclude_names: set[str] | None = None,
) -> list[Task]:
    task_root = workspace / "agent-task"
    if not task_root.is_dir():
        raise DispatcherTerminalStateError(
            f"agent-task 디렉터리가 없다: {task_root}"
        )
    directories: list[Path] = []
    try:
        groups = [task_root / task_group] if task_group else sorted(
            p for p in task_root.iterdir() if p.is_dir() and p.name != "archive"
        )
    except FileNotFoundError:
        return []
    for group in groups:
        if not group.is_dir():
            continue
        directories.append(group)
        try:
            directories.extend(sorted(p for p in group.iterdir() if p.is_dir()))
        except FileNotFoundError:
            continue
    tasks = [
        task
        for directory in directories
        if (
            exclude_names is None
            or parse_task_name(task_root, directory) not in exclude_names
        )
        if (task := read_task_directory(workspace, directory)) is not None
    ]
    return sorted(tasks, key=lambda task: (task.index, task.name))


def dependency_candidates(workspace: Path, task: Task, predecessor: str) -> list[Path]:
    parts = task.name.split("/")
    if len(parts) != 2:
        return []
    group = parts[0]
    task_root = workspace / "agent-task"
    found: list[Path] = []
    active_group = task_root / group
    for pattern in (f"{predecessor}_*/complete.log", f"{predecessor}+*/complete.log"):
        found.extend(active_group.glob(pattern))
    archive = task_root / "archive"
    if archive.is_dir():
        try:
            years = list(archive.iterdir())
        except FileNotFoundError:
            years = []
        for year in years:
            if not year.is_dir():
                continue
            try:
                months = list(year.iterdir())
            except FileNotFoundError:
                continue
            for month in months:
                archived_group = month / group
                if not archived_group.is_dir():
                    continue
                for pattern in (f"{predecessor}_*/complete.log", f"{predecessor}+*/complete.log"):
                    found.extend(archived_group.glob(pattern))
    return sorted(set(path.resolve() for path in found))


def dependency_state(workspace: Path, task: Task) -> tuple[bool, str]:
    missing: list[str] = []
    ambiguous: list[str] = []
    for predecessor in task.deps:
        candidates = dependency_candidates(workspace, task, predecessor)
        if not candidates:
            missing.append(predecessor)
        elif len(candidates) > 1:
            ambiguous.append(f"{predecessor}={','.join(str(p) for p in candidates)}")
    if ambiguous:
        return False, "dependency ambiguity: " + "; ".join(ambiguous)
    if missing:
        return False, "predecessor complete.log 대기: " + ",".join(missing)
    return True, "ready"


def live_predecessors(
    task: Task,
    active_task_names: set[str],
) -> list[str]:
    parts = task.name.split("/")
    if len(parts) != 2 or not task.deps:
        return []
    group = parts[0]
    live: list[str] = []
    for predecessor in task.deps:
        prefix = re.compile(rf"^{re.escape(predecessor)}(?:[+_])")
        if any(
            name.startswith(f"{group}/")
            and prefix.match(name.split("/", 1)[1])
            for name in active_task_names
        ):
            live.append(predecessor)
    return live


def _selector_module():
    if "agent_task_execution_target_selector" in sys.modules:
        return sys.modules["agent_task_execution_target_selector"]
    path = Path(__file__).resolve().parent / "select_execution_target.py"
    spec = importlib.util.spec_from_file_location("agent_task_execution_target_selector", path)
    if spec is None or spec.loader is None:
        raise ExecutionDecisionError(f"selector load 실패: {path}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def _decision_file(task: Task, stage: str) -> Path:
    path = task.plan if stage == "worker" else task.review
    if path is None or not path.is_file():
        raise ExecutionDecisionError(f"{stage} selector 입력 파일이 없다")
    return path


def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
    try:
        selector = _selector_module()
        selector._validate_prior_decision(decision)
        selected = decision["selected"]
        catalog_evidence = decision["catalog"]
        catalog = selector.load_runtime_catalog(catalog_evidence["source"])
        if catalog.revision != catalog_evidence["revision"]:
            raise ExecutionDecisionError(
                "실행 카탈로그가 target 선택 이후 변경됐다"
            )
        target = selector.policy.canonical_target(
            catalog, selected["target_id"]
        )
        if target is None or selector._target_snapshot(target) != selected:
            raise ExecutionDecisionError(
                "selector selected가 주입된 카탈로그 target과 일치하지 않는다"
            )
    except ExecutionDecisionError:
        raise
    except Exception as exc:
        raise ExecutionDecisionError(
            f"selector catalog validation 실패: {exc}"
        ) from exc
    runtime = dict(target.runtime)
    display = f"{target.agent}/{target.model}"
    if target.reasoning_effort is not None:
        display += f" {target.reasoning_effort}"
    return AgentSpec(
        target.agent,
        target.model,
        display,
        native_resume=bool(runtime.get("native_session_monitor")),
        target_id=target.catalog_id,
        execution_class=target.execution_class,
        selfcheck_required=target.selfcheck_required,
        reasoning_effort=target.reasoning_effort,
        runtime=runtime,
    )


def _spec_from_completing_decision(decision: dict[str, Any]) -> AgentSpec:
    """Lightweight AgentSpec extraction from a persisted completing decision.

    Unlike `agent_spec_from_decision`, this does not re-validate against the
    selector policy. The completing decision is already authoritative evidence
    of the target that succeeded, so re-running policy is unnecessary and would
    defeat the purpose of pinning the selfcheck target.
    """
    return agent_spec_from_decision(decision)


def select_execution_decision(
    task: Task, *, stage: str, prior_decision: dict[str, Any] | None = None,
    evaluated_at: datetime | None = None,
    transition: str | None = None,
    failure_class: str | None = None,
) -> dict[str, Any]:
    try:
        selector = _selector_module()
    except Exception as exc:
        code = getattr(exc, "code", exc.__class__.__name__)
        raise ExecutionDecisionError(
            f"{stage} selector load 실패 [{code}]: {exc}"
        ) from exc
    try:
        if transition is None:
            if prior_decision is not None and stage == "worker" and task.plan and task.plan.is_file():
                current_id = work_unit_id_from_file(task.plan)
                prior_id = prior_decision.get("work_unit_id") if isinstance(prior_decision, dict) else None
                if current_id and isinstance(prior_id, str) and prior_id and prior_id != current_id:
                    prior_decision = None
            transition = "resume" if prior_decision is not None else "initial"
        return selector.select_execution_target(
            _decision_file(task, stage), stage=stage,
            evaluated_at=evaluated_at or datetime.now(KST),
            catalog_path=EXECUTION_CATALOG_PATH,
            transition=transition,
            prior_decision=prior_decision,
            failure_class=failure_class,
        )
    except (OSError, ValueError, selector.SelectorInputError) as exc:
        code = getattr(exc, "code", exc.__class__.__name__)
        raise ExecutionDecisionError(
            f"{stage} selector decision 실패 [{code}]: {exc}"
        ) from exc


def work_unit_id_from_file(path: Path) -> str | None:
    if not path.is_file():
        return None
    try:
        head = path.read_text(encoding="utf-8", errors="replace")[:1024]
        match = PLAN_IDENTITY_RE.search(head)
        if match:
            return metadata_work_unit_id(match)
    except Exception:
        pass
    return None


def official_review_plan_source(task: Task) -> Path:
    """Resolve the authoritative PLAN generation for an official review."""
    if task.plan is not None or task.review is not None:
        if task.plan is None or task.review is None:
            raise ExecutionDecisionError(
                "official review active PLAN/CODE_REVIEW pair가 불완전하다"
            )
        return task.plan
    recovery_log = latest_verdict_log(task.directory)
    recovery_plan = matching_plan_log(task.directory, recovery_log)
    if recovery_log is None or recovery_plan is None:
        raise ExecutionDecisionError(
            "official review recovery의 matching archived PLAN identity를 복구할 수 없다"
        )
    return recovery_plan


def official_review_source_identity(task: Task) -> tuple[str, int, str]:
    source = official_review_plan_source(task)
    route_match = PLAN_RE.match(source.name) or PLAN_LOG_RE.match(source.name)
    if route_match is None:
        raise ExecutionDecisionError(
            f"official review PLAN route를 복구할 수 없다: {source.name}"
        )
    try:
        head = source.read_text(encoding="utf-8", errors="replace")[:1024]
    except OSError as exc:
        raise ExecutionDecisionError(
            f"official review PLAN source를 읽을 수 없다: {source}"
        ) from exc
    metadata = PLAN_IDENTITY_RE.search(head)
    if metadata is None or metadata.group("task") != task.name:
        raise ExecutionDecisionError(
            f"official review PLAN work-unit identity를 복구할 수 없다: {source}"
        )
    work_unit_id = metadata_work_unit_id(metadata)
    return route_match.group(1), int(route_match.group(2)), work_unit_id


def synthesized_official_review_decision(
    task: Task, *, evaluated_at: datetime | None = None
) -> dict[str, Any]:
    lane, grade, work_unit_id = official_review_source_identity(task)
    evaluated = evaluated_at or datetime.now(KST)
    if evaluated.tzinfo is None or evaluated.utcoffset() is None:
        raise ExecutionDecisionError(
            "official review evaluated_at이 timezone-aware가 아니다"
        )
    selector = _selector_module()
    initial = selector.select_execution_target_for_route(
        work_unit_id=work_unit_id,
        stage="review",
        lane=lane,
        grade=grade,
        evaluated_at=evaluated,
        catalog_path=EXECUTION_CATALOG_PATH,
    )
    if task.plan is None and task.review is None:
        return selector.select_execution_target_for_route(
            work_unit_id=work_unit_id,
            stage="review",
            lane=lane,
            grade=grade,
            evaluated_at=evaluated,
            catalog_path=EXECUTION_CATALOG_PATH,
            transition="resume",
            prior_decision=initial,
        )
    return initial


def read_or_preview_stage_decision(
    task: Task,
    state: dict[str, Any],
    *,
    stage: str,
    dry_run: bool = False,
    evaluated_at: datetime | None = None,
) -> dict[str, Any]:
    decisions = state.get("execution_decisions", {}) if isinstance(state, dict) else {}
    prior = decisions.get(stage) if isinstance(decisions, dict) else None

    if stage == "review":
        lane, grade, work_unit_id = official_review_source_identity(task)
        if (
            isinstance(prior, dict)
            and isinstance(prior.get("decision"), dict)
        ):
            if (
                prior.get("work_unit_id") != work_unit_id
                or prior.get("stage") != "review"
                or prior.get("lane") != lane
                or prior.get("grade") != grade
            ):
                raise ExecutionDecisionError(
                    "persisted official review decision이 recovery source identity/route와 다르다"
                )
            agent_spec_from_decision(prior)
            return prior
        return synthesized_official_review_decision(
            task, evaluated_at=evaluated_at
        )

    if isinstance(prior, dict) and task.plan and task.plan.is_file():
        work_unit_id = work_unit_id_from_file(task.plan)
        if work_unit_id and prior.get("work_unit_id") == work_unit_id:
            return prior

    return select_execution_decision(
        task,
        stage=stage,
        prior_decision=prior,
        evaluated_at=evaluated_at,
    )


def selector_evidence_lines(decision: dict[str, Any] | None) -> list[str]:
    if not isinstance(decision, dict):
        return []
    selected = decision.get("selected", {})
    if not isinstance(selected, dict):
        return []
    work_unit = decision.get("work_unit_id", "none")
    decision_info = decision.get("decision", {})
    if not isinstance(decision_info, dict):
        decision_info = {}
    rule_id = decision_info.get("rule_id", decision.get("rule_id", "none"))
    priority = decision_info.get(
        "policy_priority", decision.get("priority", "none")
    )
    transition = decision.get("transition", {})
    trigger = transition.get("trigger", "none") if isinstance(transition, dict) else "none"
    candidates = decision.get("candidates", [])
    cand_strs = []
    if isinstance(candidates, list):
        for c in candidates:
            if isinstance(c, dict):
                rank = c.get("candidate_rank", "?")
                agent = c.get("agent", "?")
                model = c.get("model", "?")
                cand_strs.append(f"#{rank}:{agent}/{model}")

    reasons = decision_info.get(
        "reason_codes", selected.get("reason_codes", [])
    )
    reason_str = ",".join(reasons) if isinstance(reasons, list) else str(reasons)

    lines = [
        f"work_unit_id={work_unit}",
        f"rule_id={rule_id}",
        f"priority={priority}",
        f"transition={trigger}",
    ]
    if cand_strs:
        lines.append(f"candidates={';'.join(cand_strs)}")
    if reason_str:
        lines.append(f"reason_codes={reason_str}")
    return lines


def selector_runtime_evidence(decision: dict[str, Any]) -> dict[str, Any]:
    """Return canonical selector fields persisted in runtime audit records."""
    return {
        "work_unit_id": decision.get("work_unit_id"),
        "candidates": decision.get("candidates"),
        "selected": decision.get("selected"),
        "decision": decision.get("decision"),
        "catalog": decision.get("catalog"),
        "transition": decision.get("transition"),
    }


def commit_execution_decision(
    store: StateStore, task: Task, stage: str, decision: dict[str, Any],
) -> None:
    state = store.task_state(task)
    decisions, history = state.get("execution_decisions", {}), state.get("route_transition_history", [])
    if not isinstance(decisions, dict) or not isinstance(history, list):
        raise ExecutionDecisionError("persisted selector state schema가 유효하지 않다")
    decisions = dict(decisions)
    decisions[stage] = decision

    stage_budget_count = 0
    try:
        stage_budget_count = StageFailureBudget.from_decision(store, task, decision).count()
    except Exception:
        pass

    selected = decision.get("selected", {})
    history_entry = {
        "stage": stage,
        "transition": decision.get("transition", {}).get("trigger") if isinstance(decision.get("transition"), dict) else None,
        "work_unit_id": decision.get("work_unit_id"),
        "candidates": decision.get("candidates"),
        "selected": selected,
        "decision": decision.get("decision"),
        "reason_codes": decision.get("decision", {}).get("reason_codes", [])
        if isinstance(decision.get("decision"), dict)
        else [],
        "catalog": decision.get("catalog"),
        "stage_budget": stage_budget_count,
    }
    history = [*history, history_entry]
    # Preserve retry handoff state whenever a pending retry is in flight.
    # The retry handoff (stable handoff_id, pending flag, context) must survive
    # the decision commit so that the subsequent production invoke() can read
    # it, embed the handoff_id in the new locator record, and atomically
    # consume the pending handoff via commit_retry_handoff_locator().
    # Clearing it here would force invoke() to fall back to a generic
    # active-locator update and lose the crash-safe handoff identity.
    # invoke() handles consumption regardless of failover or resume transition.
    is_retry_in_flight = bool(state.get("retry_failover_pending"))
    update_kwargs = {
        "execution_decisions": decisions,
        "route_transition_history": history,
        "blocked": None,
        "blocker_evidence": None,
    }
    if not is_retry_in_flight:
        update_kwargs["retry_failover_pending"] = False
        update_kwargs["retry_failover_context"] = None
    store.update_task(task, **update_kwargs)


def persisted_execution_decision(
    store: StateStore, task: Task, *, stage: str,
    transition: str | None = None,
    failure_class: str | None = None,
    evaluated_at: datetime | None = None,
) -> tuple[dict[str, Any], AgentSpec]:
    state = store.task_state(task)
    decisions = state.get("execution_decisions", {})
    if not isinstance(decisions, dict):
        raise ExecutionDecisionError("persisted selector state schema가 유효하지 않다")
    is_retry = retry_failover_pending(state) and stage == "worker"
    prior_decision = decisions.get(stage)

    retry_ctx = state.get("retry_failover_context") if isinstance(state.get("retry_failover_context"), dict) else {}

    if stage == "review":
        decision = read_or_preview_stage_decision(
            task, state, stage=stage, evaluated_at=evaluated_at
        )
    else:
        if transition is None:
            if is_retry:
                transition = "failover"
                failure_class = failure_class or retry_ctx.get("failure_class")
                if failure_class not in QUALIFIED_FAILOVER_FAILURES:
                    raise ExecutionDecisionError(
                        "retry failover requires persisted qualified runtime failure evidence"
                    )
            elif prior_decision is not None and stage == "worker" and task.plan and task.plan.is_file():
                current_id = work_unit_id_from_file(task.plan)
                prior_id = prior_decision.get("work_unit_id") if isinstance(prior_decision, dict) else None
                if current_id and isinstance(prior_id, str) and prior_id and prior_id != current_id:
                    prior_decision = None
                transition = "resume" if prior_decision is not None else "initial"
            else:
                transition = "resume" if prior_decision is not None else "initial"

        try:
            decision = select_execution_decision(
                task, stage=stage, prior_decision=prior_decision,
                transition=transition,
                failure_class=failure_class,
                evaluated_at=evaluated_at,
            )
        except ExecutionDecisionError as exc:
            # A route with no next target resumes the selected runtime so the
            # retry budget can make the terminal decision deterministically.
            if is_retry and transition == "failover" and "no_failover_candidate" in str(exc):
                decision = select_execution_decision(
                    task, stage=stage, prior_decision=prior_decision,
                    transition="resume",
                    evaluated_at=evaluated_at,
                )
            else:
                raise

    spec = agent_spec_from_decision(decision)
    commit_execution_decision(store, task, stage, decision)
    return decision, spec


def has_persisted_worker_decision(state: dict[str, Any], task: Task | None = None) -> bool:
    decisions = state.get("execution_decisions", {})
    if not isinstance(decisions, dict):
        return False
    prior = decisions.get("worker")
    if prior is None:
        return False
    if task is not None and task.plan and task.plan.is_file():
        current_id = work_unit_id_from_file(task.plan)
        if current_id and prior.get("work_unit_id") != current_id:
            return False
    return True


def retry_failover_pending(state: dict[str, Any]) -> bool:
    return bool(state.get("retry_failover_pending"))


def plan_number(task: Task) -> int:
    if task.plan and task.plan.exists():
        match = PLAN_IDENTITY_RE.search(
            task.plan.read_text(encoding="utf-8", errors="replace")[:1024]
        )
        if match:
            return int(match.group("plan"))
    return 0


def completing_decision_requires_selfcheck(state: dict[str, Any]) -> bool:
    completing = state.get("completing_decision")
    if not isinstance(completing, dict):
        return False
    selected = completing.get("selected")
    if not isinstance(selected, dict):
        return False
    return selected.get("selfcheck_required") is True


def _validated_completing_decision(
    task: Task, decision: dict[str, Any]
) -> tuple[dict[str, Any], AgentSpec]:
    """Strictly validate a completing decision against the task contract.

    Enforces that the decision's stage is "worker", its work_unit_id matches
    the task's PLAN identity, and its selected fields pass the canonical
    agent/execution-class/selfcheck normalization through `_spec_from_completing_decision`.

    Returns the validated decision and its normalized AgentSpec.
    Raises ExecutionDecisionError on any contract violation so that callers
    can fail closed rather than advancing to an inconsistent stage.
    """
    if not isinstance(decision, dict):
        raise ExecutionDecisionError(
            "completing decision이 dict가 아니다"
        )
    if decision.get("stage") != "worker":
        raise ExecutionDecisionError(
            f"completing decision stage가 worker가 아니다: {decision.get('stage')!r}"
        )
    expected_work_unit_id = work_unit_id_from_file(task.plan)
    if decision.get("work_unit_id") != expected_work_unit_id:
        raise ExecutionDecisionError(
            f"completing decision work_unit_id 불일치: "
            f"persisted={decision.get('work_unit_id')!r} "
            f"plan={expected_work_unit_id!r}"
        )
    spec = _spec_from_completing_decision(decision)
    return decision, spec


def _completing_decision_is_valid(
    task: Task, state: dict[str, Any]
) -> bool:
    """Check whether the persisted completing decision satisfies the task contract.

    Validates stage, work_unit_id, and selected agent/execution-class/selfcheck
    combination. Used by task_stage to prevent a worker_done state with no
    authoritative completing decision from advancing to review.
    """
    completing = state.get("completing_decision")
    if not isinstance(completing, dict):
        return False
    try:
        _validated_completing_decision(task, completing)
    except ExecutionDecisionError:
        return False
    return True


def concrete_user_review_value(value: str) -> bool:
    normalized = value.strip().strip("`").strip()
    normalized = re.sub(r"^-\s*", "", normalized).strip()
    if not normalized or re.search(r"\{[^}]+\}|<[^>]+>", normalized):
        return False
    return normalized.casefold() not in {
        "-",
        "n/a",
        "na",
        "none",
        "unknown",
        "미정",
        "없음",
        "해당 없음",
    }


def user_review_blocker_state(path: Path) -> tuple[bool, str]:
    if not path.is_file():
        return False, "파일이 없다"
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError as exc:
        return False, f"파일을 읽을 수 없다: {exc}"
    matched_schemas = [
        schema
        for schema in USER_REVIEW_SCHEMAS
        if re.search(
            rf"^##\s*{re.escape(schema['status_heading'])}[ \t]*$",
            text,
            re.MULTILINE,
        )
    ]
    if len(matched_schemas) != 1:
        return False, "지원하는 USER_REVIEW schema가 정확히 하나가 아니다"
    schema = matched_schemas[0]
    status = markdown_section(text, schema["status_heading"]).strip().strip("`")
    if status != "USER_REVIEW":
        return False, "상태가 USER_REVIEW가 아니다"
    reason = markdown_section(text, schema["reason_heading"])
    gate_type_matches = re.findall(
        rf"(?m)^-\s*{re.escape(schema['type_label'])}:\s*"
        r"(milestone-lock|external-execution)\s*$",
        reason,
    )
    if len(gate_type_matches) != 1:
        return False, "지원하는 user-review 유형이 정확히 하나가 아니다"
    gate_type = gate_type_matches[0]
    target = re.search(
        rf"(?m)^-\s*{re.escape(schema['target_label'])}:\s*(.+?)\s*$",
        reason,
    )
    target_value = target.group(1) if target else ""
    if not concrete_user_review_value(target_value):
        return False, "구체적인 연결 대상이 없다"
    if gate_type == "milestone-lock" and (
        "agent-roadmap/" not in target_value
        or "/milestones/" not in target_value
        or ".md" not in target_value
    ):
        return False, "구체적인 Milestone 연결 대상이 없다"
    evidence = markdown_section(text, schema["evidence_heading"])
    evidence_line = re.search(
        rf"(?m)^-\s*{re.escape(schema['evidence_label'])}:\s*(.+?)\s*$",
        evidence,
    )
    if evidence_line is None or not concrete_user_review_value(
        evidence_line.group(1)
    ):
        return False, "구체적인 차단 판단 근거가 없다"
    decision = markdown_section(text, schema["decision_headings"])
    unresolved = [
        value
        for value in re.findall(r"(?m)^-\s*\[\s\]\s+(.+?)\s*$", decision)
        if concrete_user_review_value(value)
    ]
    if not unresolved:
        return False, "미해결 사용자 조치 또는 결정 항목이 없다"
    resume = markdown_section(text, schema["resume_heading"])
    resume_conditions = [
        line
        for line in resume.splitlines()
        if concrete_user_review_value(line)
    ]
    if not resume_conditions:
        return False, "구체적인 재개 조건이 없다"
    return True, f"unresolved {gate_type} user action or decision"


def task_stage(task: Task, state: dict[str, Any]) -> str:
    if task.errors:
        return "blocked"
    if task.user_review:
        if task.plan is not None or task.review is not None:
            return "blocked"
        blocking, _ = user_review_blocker_state(task.user_review)
        return "user-review" if blocking else "blocked"
    if task.recovery and (task.plan is None or task.review is None):
        return "review"
    if task.review and task.review.exists():
        text = task.review.read_text(encoding="utf-8", errors="replace")
        if verdict_from_text(text):
            return "review"
    if state.get("worker_done"):
        # A provider exit code is not implementation completion.  Re-enter the
        # worker stage until the implementation-owned review contract is
        # actually materialized (or contains complete blocker evidence).
        if implementation_review_errors(task):
            return "worker"
        if not _completing_decision_is_valid(task, state):
            return "blocked"
        if completing_decision_requires_selfcheck(state) and not state.get("selfcheck_done"):
            return "selfcheck"
        return "review"
    return "worker"


def markdown_section(text: str, heading: str | tuple[str, ...]) -> str:
    headings = (heading,) if isinstance(heading, str) else heading
    matches = []
    for h in headings:
        for m in re.finditer(rf"^##\s*{re.escape(h)}[ \t]*$", text, re.MULTILINE):
            matches.append(m)
    if len(matches) != 1:
        return ""
    match = matches[0]
    next_heading = re.search(r"^##\s+", text[match.end():], re.MULTILINE)
    end = match.end() + next_heading.start() if next_heading else len(text)
    return text[match.end():end].strip()


def implementation_review_errors(task: Task) -> list[str]:
    if task.review is None or not task.review.is_file():
        return ["CODE_REVIEW 파일 없음"]
    text = task.review.read_text(encoding="utf-8", errors="replace")
    checklist = markdown_section(text, IMPLEMENTATION_CHECKLIST_HEADINGS)
    checkbox_values = IMPLEMENTATION_CHECKBOX_RE.findall(checklist)
    if not checkbox_values:
        return ["구현 체크리스트 미완료"]
    if (
        any(not value.strip() for value in checkbox_values)
        and not implementation_blocker_evidence_complete(text)
    ):
        return ["구현 체크리스트 미완료"]
    return []


def implementation_blocker_evidence_complete(text: str) -> bool:
    """Allow official review to classify an externally blocked implementation.

    Workers must not invent a verdict merely to bypass the checklist gate.  The
    exception therefore requires both an explicit blocked runtime/status field
    and a concrete, non-placeholder resume condition.  The official reviewer
    still owns USER_REVIEW, follow-up, and terminal-state classification.
    """
    blocked_status = re.search(
        r"(?im)^\s*(?:status\s*=\s*|authorization/runtime state:\s*`?)"
        r"(?:blocked|차단)\b",
        text,
    )
    if blocked_status is None:
        return False
    for match in re.finditer(
        r"(?im)^\s*(?:resume condition|재개 조건)\s*:\s*(.+?)\s*$",
        text,
    ):
        value = match.group(1).strip().strip("`").strip()
        lowered = value.lower()
        if (
            len(value) < 8
            or "<" in value
            or ">" in value
            or lowered in {"none", "n/a", "na", "없음"}
            or any(
                token in lowered
                for token in (
                    "actual state",
                    "exact condition",
                    "none or",
                    "placeholder",
                    "tbd",
                    "todo",
                )
            )
        ):
            continue
        return True
    return False


def classify_failure_with_evidence(output: str) -> tuple[str, str | None]:
    lines = output.splitlines()
    for category, patterns in RUNTIME_FAILURE_PATTERNS.items():
        for line in reversed(lines):
            lowered = line.lower()
            if any(re.search(pattern, lowered, re.DOTALL) for pattern in patterns):
                return category, line
    return "generic-error", None


def classify_failure(output: str) -> str:
    return classify_failure_with_evidence(output)[0]


def termination_signal(return_code: int) -> tuple[str, bool] | None:
    signal_number: int | None = None
    inferred = False
    if return_code < 0:
        signal_number = -return_code
    elif return_code > 128:
        signal_number = return_code - 128
        inferred = True
    if signal_number is None:
        return None
    try:
        return signal.Signals(signal_number).name, inferred
    except ValueError:
        return None


def failure_report_lines(failure: str, locator: Path) -> list[str]:
    record: dict[str, Any] = {}
    try:
        record = json.loads(locator.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        pass
    failure_class = str(record.get("failure_class") or failure)
    source = str(record.get("failure_source") or "unverified")
    provider_confirmed = bool(
        record.get("provider_transport_failure_confirmed", False)
    )
    lines = [
        f"failure_class={failure_class}",
        f"failure_source={source}",
        "provider_transport_failure_confirmed="
        f"{str(provider_confirmed).lower()}",
    ]
    if record.get("dispatcher_pid") is not None:
        lines.append(f"dispatcher_pid={record['dispatcher_pid']}")
    if record.get("agent_pid") is not None:
        lines.append(f"agent_pid={record['agent_pid']}")
    if record.get("dispatcher_source_sha256"):
        lines.append(
            f"dispatcher_source_sha256={record['dispatcher_source_sha256']}"
        )
    source_matches_loaded = record.get("dispatcher_source_matches_loaded")
    if source_matches_loaded is not None:
        lines.append(
            "dispatcher_source_matches_loaded="
            f"{str(bool(source_matches_loaded)).lower()}"
        )
    if (
        source_matches_loaded is False
        and record.get("dispatcher_source_current_sha256")
    ):
        lines.append(
            "dispatcher_source_current_sha256="
            f"{record['dispatcher_source_current_sha256']}"
        )
    if provider_confirmed:
        evidence_source = record.get("failure_evidence_source")
        evidence = record.get("failure_evidence_excerpt")
        if evidence_source:
            lines.append(f"provider_evidence_source={evidence_source}")
        if evidence:
            rendered = str(evidence).replace("\r", r"\r").replace("\n", r"\n")
            lines.append(f"provider_evidence={rendered}")
    if failure_class == "session-stall":
        lines.extend(
            [
                f"timeout_phase={record.get('native_session_phase') or 'unknown'}",
                f"timeout_seconds={record.get('session_stall_seconds') or 'unknown'}",
                "termination_initiator="
                f"{record.get('termination_initiator') or 'dispatcher'}",
            ]
        )
    elif failure_class == "process-terminated":
        lines.extend(
            [
                f"termination_signal={record.get('termination_signal') or 'unknown'}",
                "termination_initiator="
                f"{record.get('termination_initiator') or 'unknown'}",
            ]
        )
    lines.append(f"locator={locator}")
    return lines


def json_agent_terminal_outcome(
    value: object,
    terminal_success_contract: str | None = None,
) -> str | None:
    """Return the authoritative final outcome encoded by a JSONL event."""
    if not isinstance(value, dict):
        return None
    event_type = str(value.get("type", "")).lower()
    if terminal_success_contract in {None, "turn_completed"}:
        if event_type == "turn.completed":
            return "succeeded"
        if event_type == "turn.failed":
            return "failed"
    if terminal_success_contract not in {None, "agent_end"}:
        return None
    if event_type != "agent_end":
        return None
    if value.get("willRetry") is True:
        return None
    messages = value.get("messages")
    if not isinstance(messages, list):
        return None
    for message in reversed(messages):
        if (
            not isinstance(message, dict)
            or str(message.get("role", "")).lower() != "assistant"
        ):
            continue
        stop_reason = message.get("stopReason") or message.get("stop_reason")
        normalized = str(stop_reason or "").lower()
        if normalized in {"error", "aborted"}:
            return "failed"
        if normalized == "stop":
            return "succeeded"
        return None
    return None


def json_agent_terminal_outcome_from_line(
    line: str,
    terminal_success_contract: str | None = None,
) -> str | None:
    try:
        return json_agent_terminal_outcome(
            json.loads(line), terminal_success_contract
        )
    except json.JSONDecodeError:
        return None


def terminal_diagnostic(cli: str, channel: str, line: str) -> str | None:
    if channel == "stderr":
        return line
    try:
        value = json.loads(line)
    except json.JSONDecodeError:
        return line if re.match(r"^\s*(?:error|fatal)\b", line, re.IGNORECASE) else None
    if not isinstance(value, dict):
        return None
    event_type = str(value.get("type", "")).lower()
    severity = str(value.get("severity") or value.get("level") or "").lower()
    status = str(value.get("status") or "").lower()
    subtype = str(value.get("subtype") or "").lower()
    if (
        json_agent_terminal_outcome(value) == "failed"
        or event_type in {"error", "fatal", "request.failed", "turn.failed", "rate_limit_event"}
        or severity in {"error", "fatal"}
        or subtype.startswith("error")
        or bool(value.get("is_error"))
        or (
            status in {"failed", "rejected"}
            and any(field in value for field in ("code", "error", "error_code", "status_code"))
        )
    ):
        return json.dumps(value, ensure_ascii=False)
    return None


def collaboration_tool(line: str) -> str | None:
    try:
        value = json.loads(line)
    except json.JSONDecodeError:
        return None
    item = value.get("item") or {}
    if (
        value.get("type") == "item.started"
        and item.get("type") == "collab_tool_call"
        and item.get("tool")
    ):
        return str(item["tool"])
    return None


def json_agent_command_transition(line: str) -> tuple[str, str] | None:
    """Return the lifecycle transition for one Codex command execution item."""
    try:
        value = json.loads(line)
    except json.JSONDecodeError:
        return None
    if not isinstance(value, dict):
        return None
    event_type = value.get("type")
    item = value.get("item")
    if (
        event_type not in {"item.started", "item.completed"}
        or not isinstance(item, dict)
        or item.get("type") != "command_execution"
        or not isinstance(item.get("id"), str)
        or not item["id"]
    ):
        return None
    return ("started" if event_type == "item.started" else "completed", item["id"])


async def terminate_process_group(
    process: asyncio.subprocess.Process,
    grace_seconds: float = 5,
) -> None:
    """Terminate the exact subprocess group and escalate if descendants remain."""
    try:
        os.killpg(process.pid, signal.SIGTERM)
    except ProcessLookupError:
        if process.returncode is None:
            await process.wait()
        return

    if process.returncode is None:
        try:
            await asyncio.wait_for(process.wait(), timeout=grace_seconds)
        except TimeoutError:
            try:
                os.killpg(process.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
            await process.wait()
            return

    try:
        os.killpg(process.pid, 0)
    except ProcessLookupError:
        return
    try:
        os.killpg(process.pid, signal.SIGKILL)
    except ProcessLookupError:
        pass


def auxiliary_log_diagnostics(path: Path) -> list[str]:
    if not path.exists():
        return []
    diagnostics: list[str] = []
    for line in path.read_text(encoding="utf-8", errors="replace").splitlines()[-200:]:
        failure_class, evidence = classify_failure_with_evidence(line)
        if (
            failure_class not in RECOVERABLE_RUNTIME_FAILURES
            or evidence is None
        ):
            continue
        if failure_class == "provider-quota" and not re.search(
            (
                r"RESOURCE[_ ]?EXHAUSTED"
                r"|\b(?:HTTP|status(?: code)?)\s*[:=]?\s*429\b"
                r"|\btoo many requests\b"
                r"|(?:rate.?limit|quota|capacity).{0,40}"
                r"(?:exceed|exhaust|reached|reject)"
                r"|(?:exceed|exhaust|reached|reject).{0,40}"
                r"(?:rate.?limit|quota|capacity)"
                r"|(?:rate.?limit|quota).{0,40}retry after"
            ),
            line,
            re.IGNORECASE,
        ):
            continue
        diagnostics.append(line)
    return diagnostics


def attempt_terminal_diagnostics(
    attempt_directory: Path,
    record: dict[str, Any],
) -> list[tuple[str, str]]:
    spec = agent_spec_from_record(record)
    if spec is None:
        return []
    try:
        stream_lines = (attempt_directory / "stream.log").read_text(
            encoding="utf-8",
            errors="replace",
        ).splitlines()
    except OSError:
        stream_lines = []
    diagnostics: list[tuple[str, str]] = []
    for stream_line in stream_lines:
        match = re.match(r"^\[(stdout|stderr)\]\s?(.*)$", stream_line)
        if match is None:
            continue
        channel, payload = match.groups()
        diagnostic = terminal_diagnostic(spec.cli, channel, payload)
        if diagnostic:
            diagnostics.append((f"{spec.cli}:{channel}", diagnostic))
    for raw_path in record.get("auxiliary_logs", []):
        path = Path(str(raw_path))
        diagnostics.extend(
            (f"{spec.cli}:auxiliary-log", diagnostic)
            for diagnostic in auxiliary_log_diagnostics(path)
        )
    return diagnostics


def render_json_line(cli: str, line: str) -> tuple[list[str], str | None]:
    try:
        value = json.loads(line)
    except json.JSONDecodeError:
        return [line.rstrip()], None
    if not isinstance(value, dict):
        return [line.rstrip()], None
    session_id = value.get("thread_id") or value.get("session_id")
    rendered: list[str] = []
    for field in ("text", "result", "message", "output"):
        item = value.get(field)
        if isinstance(item, str) and item:
            rendered.extend(item.splitlines())
    nested = value.get("item")
    if isinstance(nested, dict):
        for field in ("text", "message", "output"):
            item = nested.get(field)
            if isinstance(item, str) and item:
                rendered.extend(item.splitlines())
    if not rendered and terminal_diagnostic(cli, "stdout", line):
        rendered.append(json.dumps(value, ensure_ascii=False))
    return rendered, str(session_id) if session_id else None


def native_session_path(
    spec: AgentSpec,
    workspace: Path,
    session_id: str | None,
    attempt_dir: Path,
) -> str | None:
    template = spec.runtime.get("session_path")
    if not isinstance(template, str) or not template or not session_id:
        return None
    values = {
        "agent": spec.cli,
        "attempt_dir": str(attempt_dir),
        "model": spec.model,
        "prompt": "",
        "reasoning_effort": str(spec.reasoning_effort or ""),
        "resume_session": "",
        "resume_session_dir": "",
        "session_id": session_id,
        "target_id": str(spec.target_id or ""),
        "workspace": str(workspace),
    }
    rendered = str(template).format_map(values)
    candidate = Path(rendered).expanduser()
    if not candidate.is_absolute():
        candidate = attempt_dir / candidate
    if any(character in str(candidate) for character in "*?["):
        matches = sorted(
            candidate.parent.glob(candidate.name),
            key=lambda path: path.stat().st_mtime_ns,
        )
        return str(matches[-1]) if matches else str(candidate.parent)
    return str(candidate)


def native_session_mtime_ns(path: str | None) -> int | None:
    if not path:
        return None
    candidate = Path(path)
    return candidate.stat().st_mtime_ns if candidate.is_file() else None


def native_session_state(path: str | None) -> NativeSessionState:
    if not path:
        return NativeSessionState("starting", reason="native-session-path-missing")
    candidate = Path(path)
    if not candidate.is_file():
        return NativeSessionState("starting", reason="native-session-file-missing")
    return NativeSessionState("active", reason="native-session-file-present")


def native_session_phase(path: str | None) -> str:
    return native_session_state(path).phase


def log_tail_excerpt(path: Path, *, byte_limit: int = 8192, char_limit: int = 2000) -> str:
    """Return a bounded recent log excerpt without loading a long reasoning stream."""
    try:
        with path.open("rb") as stream:
            stream.seek(max(0, path.stat().st_size - byte_limit))
            text = stream.read().decode("utf-8", errors="replace")
    except OSError as exc:
        return f"<stream log unavailable: {exc}>"
    return text[-char_limit:]


def process_start_token(value: Any) -> str | None:
    """Read Linux process start ticks so PID reuse is not treated as liveness."""
    try:
        pid = int(value)
        text = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
        close = text.rfind(")")
        fields = text[close + 2 :].split()
        return fields[19] if close >= 0 and len(fields) > 19 else None
    except (TypeError, ValueError, OSError):
        return None


def process_is_alive(value: Any, expected_start_token: Any = None) -> bool:
    """Return whether the same attempt/dispatcher process still exists."""
    try:
        pid = int(value)
        if pid <= 0:
            return False
        os.kill(pid, 0)
    except (TypeError, ValueError, OSError):
        return False
    current_token = process_start_token(pid)
    if (
        expected_start_token is not None
        and current_token is not None
        and str(expected_start_token) != current_token
    ):
        return False
    return True


def marked_agent_process_pids(marker: str) -> list[int]:
    """Find live processes carrying the per-attempt environment marker."""
    expected = f"{AGENT_PROCESS_MARKER_ENV}={marker}".encode()
    matches: list[int] = []
    for environ in Path("/proc").glob("[0-9]*/environ"):
        try:
            values = environ.read_bytes().split(b"\0")
            pid = int(environ.parent.name)
        except (OSError, ValueError):
            continue
        if expected in values:
            matches.append(pid)
    return sorted(matches)


def locator_workspace_ownership(
    locator_path: Path,
    locator: dict[str, Any],
    *,
    expected_workspace: Path | None = None,
    expected_workspace_id: str | None = None,
    expected_runs_root: Path | None = None,
) -> tuple[bool, str]:
    if expected_workspace is None and expected_workspace_id is None:
        return True, ""
    try:
        expected_root = (
            expected_workspace.resolve() if expected_workspace is not None else None
        )
        expected_id = expected_workspace_id
        if expected_id is None and expected_root is not None:
            expected_id = hashlib.sha256(str(expected_root).encode()).hexdigest()[:16]
        if expected_runs_root is None:
            return False, "현재 workspace의 locator runs root가 없다"
        resolved_runs = expected_runs_root.resolve()
        resolved_locator = locator_path.resolve()
        resolved_locator.relative_to(resolved_runs)
    except (OSError, RuntimeError, ValueError):
        return (
            False,
            "foreign workspace locator path: "
            f"locator={locator_path} expected_runs={expected_runs_root}",
        )

    recorded_workspace = locator.get("workspace")
    recorded_workspace_id = locator.get("workspace_id")
    if recorded_workspace_id not in (None, "") and (
        str(recorded_workspace_id) != str(expected_id)
    ):
        return (
            False,
            "foreign workspace locator id: "
            f"recorded={recorded_workspace_id} expected={expected_id}",
        )
    if recorded_workspace not in (None, ""):
        try:
            recorded_root = Path(str(recorded_workspace)).resolve()
        except (OSError, RuntimeError):
            return False, "locator workspace 경로를 canonicalize할 수 없다"
        if expected_root is not None and recorded_root != expected_root:
            return (
                False,
                "foreign workspace locator root: "
                f"recorded={recorded_root} expected={expected_root}",
            )
    evidence_fields = ["stream_log"]
    runtime = locator.get("runtime")
    if isinstance(runtime, dict) and runtime.get("native_session_monitor"):
        evidence_fields.append("native_session_path")
    for field in evidence_fields:
        raw_evidence = locator.get(field)
        if raw_evidence in (None, ""):
            continue
        try:
            Path(str(raw_evidence)).resolve().relative_to(resolved_runs)
        except (OSError, RuntimeError, ValueError):
            return (
                False,
                "foreign workspace locator evidence: "
                f"field={field} path={raw_evidence}",
            )
    # An identity-less legacy locator is accepted only because physical
    # containment under the current store's runs root was already proved.
    return True, ""


def external_active_is_live(
    state: dict[str, Any],
    *,
    expected_workspace: Path | None = None,
    expected_workspace_id: str | None = None,
    expected_runs_root: Path | None = None,
) -> tuple[bool, str]:
    raw_locator = state.get("active_locator")
    if not raw_locator:
        return False, "active locator 없음"
    target = Path(str(raw_locator))
    locator_path = target if target.name == "locator.json" else target / "locator.json"
    path_owned, ownership_detail = locator_workspace_ownership(
        locator_path,
        {},
        expected_workspace=expected_workspace,
        expected_workspace_id=expected_workspace_id,
        expected_runs_root=expected_runs_root,
    )
    if not path_owned:
        return False, ownership_detail
    locator: dict[str, Any] = {}
    if locator_path.is_file():
        try:
            locator = json.loads(locator_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            return False, f"locator 판독 실패: {locator_path}"
        if not isinstance(locator, dict):
            return False, f"locator object 형식이 아니다: {locator_path}"

    owned, ownership_detail = locator_workspace_ownership(
        locator_path,
        locator,
        expected_workspace=expected_workspace,
        expected_workspace_id=expected_workspace_id,
        expected_runs_root=expected_runs_root,
    )
    if not owned:
        return False, ownership_detail

    if locator:
        status = str(locator.get("status") or "")
        if status and status != "running":
            return False, f"locator status={status}"

    # The stream may legitimately remain quiet during long reasoning.  A live
    # process is stronger evidence than a locator or dispatcher heartbeat, and
    # prevents a second dispatcher from duplicating an active attempt.
    agent_pid_recorded = locator.get("agent_pid") not in (None, "")
    for field, token_field in (
        ("agent_pid", "agent_process_start_token"),
        ("dispatcher_pid", "dispatcher_process_start_token"),
    ):
        if process_is_alive(locator.get(field), locator.get(token_field)):
            return True, f"{field}={locator[field]} alive; output stream is monitored"
    process_marker = str(locator.get("agent_process_marker") or "")
    if process_marker:
        marker_pids = marked_agent_process_pids(process_marker)
        if marker_pids:
            return (
                True,
                "agent process marker alive: "
                + ",".join(str(pid) for pid in marker_pids),
            )
        return (
            False,
            "agent process marker is absent from the process table",
        )
    if agent_pid_recorded:
        return (
            False,
            "recorded agent process identity is no longer alive",
        )

    native_raw = locator.get("native_session_path")
    native = Path(str(native_raw)) if native_raw else None
    if native and native.is_dir():
        sessions = list(native.glob("*.jsonl"))
        native = max(sessions, key=lambda path: path.stat().st_mtime_ns) if sessions else None
    if native is None or not native.is_file():
        roots = [target] if target.is_dir() else [target.parent]
        sessions = [
            path
            for root in roots
            for path in root.glob("**/*.jsonl")
        ]
        native = max(sessions, key=lambda path: path.stat().st_mtime_ns) if sessions else None
    now = datetime.now(KST).timestamp()
    runtime = locator.get("runtime")
    monitor_native_session = bool(
        isinstance(runtime, dict) and runtime.get("native_session_monitor")
    )
    stream_progress_at: float | None = None
    stream_raw = locator.get("stream_log")
    stream = Path(str(stream_raw)) if stream_raw else None
    if stream and stream.is_file():
        stream_progress_at = stream.stat().st_mtime
    if native and native.is_file():
        native_progress_at = native.stat().st_mtime
        progress_at = max(native_progress_at, stream_progress_at or 0.0)
        inactive = max(0.0, now - progress_at)
        if monitor_native_session:
            phase = native_session_phase(str(native))
            # Only an exact incomplete toolCall -> toolResult batch is a tool
            # execution interval.  Unknown/starting/model-reasoning states
            # must never be treated as a stalled tool merely because their
            # native event file is quiet.
            if phase == "tool-running":
                return (
                    True,
                    "phase=tool-running with no agent PID evidence; "
                    "time-based duplicate recovery is disabled",
                )
            return (
                True,
                f"phase={phase} native+stream inactive={inactive:.1f}s "
                "with no agent PID evidence; time-based duplicate recovery is disabled",
            )
        return (
            True,
            "native+stream inactive="
            f"{inactive:.1f}s with no agent PID evidence; "
            "time-based duplicate recovery is disabled",
        )

    if stream_progress_at is not None:
        inactive = max(0.0, now - stream_progress_at)
        return (
            True,
            f"stream inactive={inactive:.1f}s with no agent PID evidence; "
            "time-based duplicate recovery is disabled",
        )

    return False, f"active 증거 없음: {raw_locator}"


def native_resume_locator(
    state: dict[str, Any],
    *,
    expected_workspace: Path | None = None,
    expected_workspace_id: str | None = None,
    expected_runs_root: Path | None = None,
) -> Path | None:
    raw_locator = state.get("active_locator")
    if not raw_locator:
        return None
    target = Path(str(raw_locator))
    locator = target if target.name == "locator.json" else target / "locator.json"
    path_owned, _ = locator_workspace_ownership(
        locator,
        {},
        expected_workspace=expected_workspace,
        expected_workspace_id=expected_workspace_id,
        expected_runs_root=expected_runs_root,
    )
    if not path_owned:
        return None
    try:
        record = json.loads(locator.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    if not isinstance(record, dict):
        return None
    owned, _ = locator_workspace_ownership(
        locator,
        record,
        expected_workspace=expected_workspace,
        expected_workspace_id=expected_workspace_id,
        expected_runs_root=expected_runs_root,
    )
    if not owned:
        return None
    if (
        not isinstance(record.get("runtime"), dict)
        or not record["runtime"].get("native_session_monitor")
        or record.get("failure_class") not in {"context-limit", "session-stall"}
        or record.get("status") != "failed"
    ):
        return None
    native_raw = record.get("native_session_path")
    native = Path(str(native_raw)) if native_raw else None
    if native is None or not native.exists():
        return None
    if expected_runs_root is not None:
        try:
            native.resolve().relative_to(expected_runs_root.resolve())
        except (OSError, RuntimeError, ValueError):
            return None
    return locator


def selfcheck_context_resume_locator(
    state: dict[str, Any],
    task: Task,
    *,
    expected_workspace: Path,
    expected_workspace_id: str,
    expected_runs_root: Path,
) -> tuple[Path | None, str]:
    raw_locator = state.get("selfcheck_context_locator")
    if not isinstance(raw_locator, str) or not raw_locator:
        return None, "persisted selfcheck context locator가 없다"
    target = Path(raw_locator)
    locator = target if target.name == "locator.json" else target / "locator.json"
    path_owned, detail = locator_workspace_ownership(
        locator,
        {},
        expected_workspace=expected_workspace,
        expected_workspace_id=expected_workspace_id,
        expected_runs_root=expected_runs_root,
    )
    if not path_owned:
        return None, detail
    try:
        record = json.loads(locator.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None, "persisted selfcheck context locator를 읽을 수 없다"
    if not isinstance(record, dict):
        return None, "persisted selfcheck context locator 형식이 잘못됐다"
    owned, detail = locator_workspace_ownership(
        locator,
        record,
        expected_workspace=expected_workspace,
        expected_workspace_id=expected_workspace_id,
        expected_runs_root=expected_runs_root,
    )
    if not owned:
        return None, detail
    if (
        record.get("task") != task.name
        or record.get("role") != "selfcheck"
        or not isinstance(record.get("runtime"), dict)
        or not record["runtime"].get("native_session_monitor")
        or record.get("status") != "succeeded"
    ):
        return None, "persisted selfcheck context locator identity가 일치하지 않는다"
    native_raw = record.get("native_session_path")
    native = Path(str(native_raw)) if native_raw else None
    if native is not None and native.is_dir():
        sessions = list(native.glob("*.jsonl"))
        native = max(sessions, key=lambda path: path.stat().st_mtime_ns) if sessions else None
    if native is None or not native.is_file():
        return None, "persisted selfcheck native session이 없다"
    try:
        native.resolve().relative_to(expected_runs_root.resolve())
    except (OSError, RuntimeError, ValueError):
        return None, "persisted selfcheck native session이 workspace runs 밖에 있다"
    return locator, ""


def build_command(
    spec: AgentSpec,
    prompt: str,
    workspace: Path,
    session_id: str,
    attempt_dir: Path,
    native_resume_session: Path | None = None,
) -> list[str]:
    template_name = (
        "resume_command"
        if native_resume_session is not None and spec.runtime.get("resume_command")
        else "command"
    )
    template = spec.runtime.get(template_name)
    if not isinstance(template, list) or not template:
        raise RuntimeError(
            f"runtime catalog target {spec.target_id!r} has no {template_name}"
        )
    values = {
        "agent": spec.cli,
        "attempt_dir": str(attempt_dir),
        "model": spec.model,
        "prompt": prompt,
        "reasoning_effort": str(spec.reasoning_effort or ""),
        "resume_session": str(native_resume_session or ""),
        "resume_session_dir": (
            str(native_resume_session.parent)
            if native_resume_session is not None
            else ""
        ),
        "session_id": session_id,
        "target_id": str(spec.target_id or ""),
        "workspace": str(workspace),
    }
    try:
        return [str(part).format_map(values) for part in template]
    except (KeyError, ValueError) as exc:
        raise RuntimeError(
            f"runtime command template expansion failed for {spec.target_id!r}: {exc}"
        ) from exc


def preflight_execution_catalog(
    catalog_path: Path,
    *,
    workspace: Path | None = None,
    run_commands: bool = True,
) -> None:
    selector = _selector_module()
    catalog = selector.load_runtime_catalog(catalog_path)
    checked_workspace = (workspace or Path.cwd()).resolve()
    checked_commands: set[str] = set()
    for target_id, target in catalog.targets.items():
        values = {
            "agent": target.agent,
            "attempt_dir": str(catalog_path.parent),
            "model": target.model,
            "prompt": "",
            "reasoning_effort": str(target.reasoning_effort or ""),
            "resume_session": "",
            "resume_session_dir": "",
            "session_id": "preflight-session",
            "target_id": target_id,
            "workspace": str(checked_workspace),
        }
        command = [str(part).format_map(values) for part in target.runtime["command"]]
        executable = command[0]
        if executable not in checked_commands and shutil.which(executable) is None:
            raise ExecutionDecisionError(
                f"execution catalog target {target_id!r} command not found: {executable}"
            )
        checked_commands.add(executable)
        probe_template = target.runtime.get("preflight_command")
        if not probe_template or not run_commands:
            continue
        probe = [str(part).format_map(values) for part in probe_template]
        probe_environment = {
            str(key): str(item).format_map(values)
            for key, item in target.runtime.get("environment", {}).items()
        }
        completed = subprocess.run(
            probe,
            cwd=checked_workspace,
            env={**os.environ, **probe_environment},
            capture_output=True,
            text=True,
            timeout=15,
            check=False,
        )
        if completed.returncode != 0:
            diagnostic = (completed.stderr or completed.stdout).strip()
            raise ExecutionDecisionError(
                f"execution catalog target {target_id!r} preflight failed: "
                f"{diagnostic or completed.returncode}"
            )


async def invoke(
    workspace: Path,
    store: StateStore,
    task: Task,
    role: str,
    spec: AgentSpec,
    prompt: str,
    resume_locator: Path | None = None,
) -> tuple[int, str | None, Path]:
    worker_signature_before = (
        task_signature(workspace, task) if role == "worker" else None
    )
    attempt, identity = next_execution_identity(store, task, role)
    attempt_dir = store.runs / f"{datetime.now(KST).strftime('%Y%m%dT%H%M%S%z')}__{identity}"
    attempt_dir.mkdir(parents=True, exist_ok=False)
    locator_path = attempt_dir / "locator.json"
    stream_path = attempt_dir / "stream.log"
    normalized_output_path = attempt_dir / "normalized-output.log"
    heartbeat_path = attempt_dir / "heartbeat.log"
    stream_path.touch()
    normalized_output_path.touch()
    heartbeat_path.touch()
    session_id = str(uuid.uuid4())
    process_marker = f"w{store.workspace_id}__{identity}__{uuid.uuid4()}"
    native_resume_session: Path | None = None
    if spec.native_resume and resume_locator and resume_locator.is_file():
        resume_locator_path = (
            resume_locator
            if resume_locator.name == "locator.json"
            else resume_locator / "locator.json"
        )
        path_owned, _ = locator_workspace_ownership(
            resume_locator_path,
            {},
            expected_workspace=store.workspace,
            expected_workspace_id=store.workspace_id,
            expected_runs_root=store.runs,
        )
        if path_owned:
            try:
                prior = json.loads(resume_locator_path.read_text(encoding="utf-8"))
            except (OSError, json.JSONDecodeError):
                prior = {}
        else:
            prior = {}
        owned, _ = locator_workspace_ownership(
            resume_locator_path,
            prior if isinstance(prior, dict) else {},
            expected_workspace=store.workspace,
            expected_workspace_id=store.workspace_id,
            expected_runs_root=store.runs,
        )
        if owned and isinstance(prior, dict):
            prior_native = prior.get("native_session_path")
            candidate = Path(str(prior_native)) if prior_native else None
            if candidate and candidate.is_dir():
                sessions = list(candidate.glob("*.jsonl"))
                candidate = (
                    max(sessions, key=lambda path: path.stat().st_mtime_ns)
                    if sessions
                    else None
                )
            if candidate and candidate.is_file():
                try:
                    candidate.resolve().relative_to(store.runs.resolve())
                except (OSError, RuntimeError, ValueError):
                    candidate = None
            if candidate and candidate.is_file():
                native_resume_session = candidate
                resume_locator = resume_locator_path
                session_id = str(prior.get("session_id") or candidate.stem)
    started_at = now_iso()
    work_log_path = milestone_work_log_path(task)
    record: dict[str, Any] = {
        "execution_id": identity,
        "task": task.name,
        "task_directory": str(task.directory.resolve()),
        "target_files": task_target_files(task),
        "target_files_known": task.write_set_known,
        "plan_number": plan_number(task),
        "role": role,
        "attempt": attempt,
        "workspace": str(store.workspace),
        "workspace_id": store.workspace_id,
        **dispatcher_source_provenance(),
        "cli": spec.cli,
        "model": spec.model,
        "target_id": spec.target_id,
        "execution_class": spec.execution_class,
        "selfcheck_required": spec.selfcheck_required,
        "reasoning_effort": spec.reasoning_effort,
        "runtime": spec.runtime,
        "agent_process_marker": process_marker,
        "plan_path": str(task.plan) if task.plan else None,
        "review_path": str(task.review) if task.review else None,
        "session_id": session_id,
        "native_session_path": (
            str(native_resume_session)
            if native_resume_session is not None
            else native_session_path(spec, workspace, session_id, attempt_dir)
        ),
        "output_log": str(stream_path),
        "stream_log": str(stream_path),
        "normalized_output_log": str(normalized_output_path),
        "heartbeat_log": str(heartbeat_path),
        "auxiliary_logs": [
            str(item).format_map(
                {
                    "agent": spec.cli,
                    "attempt_dir": str(attempt_dir),
                    "model": spec.model,
                    "prompt": "",
                    "reasoning_effort": str(spec.reasoning_effort or ""),
                    "resume_session": str(native_resume_session or ""),
                    "resume_session_dir": (
                        str(native_resume_session.parent)
                        if native_resume_session is not None
                        else ""
                    ),
                    "session_id": session_id,
                    "target_id": str(spec.target_id or ""),
                    "workspace": str(workspace),
                }
            )
            for item in spec.runtime.get("auxiliary_logs", [])
        ],
        "work_log": str(work_log_path.resolve()),
        "started_at": started_at,
        "status": "running",
        "resumed_from_locator": str(resume_locator) if native_resume_session else None,
    }
    stage_decision = None
    if isinstance(store, StateStore):
        decisions = store.task_state(task).get("execution_decisions", {})
        if isinstance(decisions, dict):
            stage_decision = decisions.get(role)
    if isinstance(stage_decision, dict):
        record.update(selector_runtime_evidence(stage_decision))
        try:
            record["stage_budget"] = StageFailureBudget.from_decision(store, task, stage_decision).count()
        except Exception:
            record["stage_budget"] = 0
    # Resolve the retry handoff identity assigned when a pending target
    # failover was created before the first durable locator write, so
    # the first record on disk already carries the stable handoff ID a
    # crash/restart can match against (the locator path changes on every
    # attempt).
    retry_handoff_id: str | None = None
    if isinstance(store, StateStore):
        retry_ctx = store.task_state(task).get("retry_failover_context")
        if isinstance(retry_ctx, dict):
            retry_handoff_id = retry_ctx.get("handoff_id")
    if retry_handoff_id:
        record["retry_handoff_id"] = retry_handoff_id
    write_json(locator_path, record)
    if isinstance(store, StateStore):
        if retry_handoff_id:
            # One-save transition: update active_locator, clear pending flag,
            # and clear context together. Restore pre-state on save failure.
            # A mismatch (False) or a save fault (raises) must stop before
            # the provider process seam so we never launch a duplicate
            # invocation against a retry intent we failed to commit.
            if not store.commit_retry_handoff_locator(task, retry_handoff_id, str(locator_path)):
                raise ExecutionDecisionError("retry handoff commit mismatch")
        else:
            store.update_task(task, active_locator=str(locator_path))
    prefix = f"[{task.directory.name}][{role}][a{attempt:02d}]"

    def persist_locator_record() -> None:
        """Do not abort a live model solely because a locator refresh failed."""
        try:
            write_json(locator_path, record)
        except OSError as exc:
            record["locator_write_error"] = str(exc)
            attempt_event(
                prefix,
                f"locator 기록 경고: locator={locator_path} error={exc}",
            )

    for line in task_observation_lines(task):
        attempt_event(prefix, line)
    attempt_event(prefix, f"locator={locator_path}")
    try:
        append_milestone_event(
            task,
            event="START",
            execution_id=identity,
            role=role,
            attempt=attempt,
            model=spec.display,
            result="running",
            locator=locator_path,
        )
    except OSError as exc:
        line = f"milestone work log setup failed: {exc}"
        heartbeat_path.write_text(line + "\n", encoding="utf-8")
        record.update(
            status="failed",
            finished_at=now_iso(),
            exit_code=1,
            failure_class="work-log-setup",
            failure_source="work-log",
            provider_transport_failure_confirmed=False,
            work_log_error=str(exc),
        )
        persist_locator_record()
        attempt_event(prefix, line)
        return 1, "work-log-setup", locator_path
    command = build_command(
        spec,
        prompt,
        workspace,
        session_id,
        attempt_dir,
        native_resume_session=native_resume_session,
    )
    diagnostics: list[str] = []
    diagnostic_origins: list[str] = []
    control_violation: str | None = None
    session_stall_seconds: float | None = None
    terminal_success_contract = spec.runtime.get("terminal_success")
    terminal_outcome_seen: str | None = None
    try:
        runtime_values = {
            "agent": spec.cli,
            "attempt_dir": str(attempt_dir),
            "model": spec.model,
            "prompt": prompt,
            "reasoning_effort": str(spec.reasoning_effort or ""),
            "resume_session": str(native_resume_session or ""),
            "resume_session_dir": (
                str(native_resume_session.parent)
                if native_resume_session is not None
                else ""
            ),
            "session_id": session_id,
            "target_id": str(spec.target_id or ""),
            "workspace": str(workspace),
        }
        runtime_environment = {
            str(key): str(value).format_map(runtime_values)
            for key, value in spec.runtime.get("environment", {}).items()
        }
        process = await asyncio.create_subprocess_exec(
            *command,
            cwd=workspace,
            env={
                **os.environ,
                AGENT_PROCESS_MARKER_ENV: process_marker,
                **runtime_environment,
            },
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            limit=10 * 1024 * 1024,
            start_new_session=True,
        )
        # Keep the child PID in the locator before monitoring output.  If this
        # dispatcher is interrupted, a later dispatcher can distinguish a
        # genuinely live, silent model from a stale locator and must not launch
        # a duplicate continuation.
        record["agent_pid"] = process.pid
        record["agent_process_start_token"] = process_start_token(process.pid)
        persist_locator_record()
    except FileNotFoundError:
        line = f"command not found: {command[0]}"
        heartbeat_path.write_text(line + "\n", encoding="utf-8")
        failure_class = "generic-error"
        try:
            append_milestone_event(
                task,
                event="FINISH",
                execution_id=identity,
                role=role,
                attempt=attempt,
                model=spec.display,
                result=f"failed:{failure_class}:127",
                locator=locator_path,
            )
        except OSError as exc:
            record["work_log_runtime_error"] = str(exc)
            failure_class = "work-log-runtime-write"
        record.update(
            status="failed",
            finished_at=now_iso(),
            exit_code=127,
            failure_class=failure_class,
            failure_source=(
                "work-log" if failure_class == "work-log-runtime-write" else "cli-launch"
            ),
            provider_transport_failure_confirmed=False,
        )
        persist_locator_record()
        attempt_event(prefix, line)
        return 127, failure_class, locator_path

    readers: list[asyncio.Task[None]] = []
    try:
        assert process.stdout is not None and process.stderr is not None
        queue: asyncio.Queue[tuple[str, bytes | None]] = asyncio.Queue()

        async def pump(channel: str, stream: asyncio.StreamReader) -> None:
            try:
                while True:
                    raw = await stream.readline()
                    if not raw:
                        break
                    await queue.put((channel, raw))
            finally:
                await queue.put((channel, None))

        readers = [
            asyncio.create_task(pump("stdout", process.stdout)),
            asyncio.create_task(pump("stderr", process.stderr)),
        ]
        finished_streams = 0
        loop = asyncio.get_running_loop()
        last_native_mtime: int | None = None
        last_stream_mtime: int | None = None
        last_native_progress_at = loop.time()
        last_stream_progress_at = loop.time()
        active_command_execution_ids: set[str] = set()
        with (
            stream_path.open("w", encoding="utf-8") as stream_log,
            normalized_output_path.open("w", encoding="utf-8") as normalized_output_log,
            heartbeat_path.open("a", encoding="utf-8") as heartbeat_log,
        ):
            while finished_streams < len(readers):
                try:
                    channel, raw = await asyncio.wait_for(
                        queue.get(), timeout=STREAM_HEARTBEAT_SECONDS
                    )
                except asyncio.TimeoutError:
                    try:
                        stream_mtime = stream_path.stat().st_mtime_ns
                    except OSError:
                        stream_mtime = None
                    if stream_mtime is not None:
                        record["stream_log_mtime_ns"] = stream_mtime
                        if stream_mtime != last_stream_mtime:
                            last_stream_mtime = stream_mtime
                            last_stream_progress_at = loop.time()
                            record.pop("native_silence_inspection", None)
                    native_path = (
                        str(native_resume_session)
                        if native_resume_session is not None
                        else native_session_path(
                            spec,
                            workspace,
                            record.get("session_id"),
                            attempt_dir,
                        )
                    )
                    if native_path:
                        record["native_session_path"] = native_path
                    native_mtime = native_session_mtime_ns(
                        record.get("native_session_path")
                    )
                    if native_mtime is not None:
                        record["native_session_mtime_ns"] = native_mtime
                        if native_mtime != last_native_mtime:
                            last_native_mtime = native_mtime
                            last_native_progress_at = loop.time()
                            # Native events and the separately flushed stream log
                            # are peer progress signals. A trailing toolResult only
                            # selects the timeout budget; it never overrides later
                            # reasoning/text output.
                            record["native_activity_state"] = "working"
                    native_state = native_session_state(
                        record.get("native_session_path")
                    )
                    native_phase = native_state.phase
                    is_native_tool_execution = native_phase == "tool-running"
                    is_command_execution = bool(active_command_execution_ids)
                    # Outside a toolCall->toolResult interval, model stdout/stderr
                    # is the liveness signal. A completed tool result changes phase
                    # but must not reset the model-response silence clock.
                    native_inactive_seconds = loop.time() - (
                        max(last_native_progress_at, last_stream_progress_at)
                        if is_native_tool_execution
                        else last_stream_progress_at
                    )
                    if spec.native_resume:
                        record["native_session_phase"] = native_phase
                        record["native_session_phase_reason"] = (
                            native_state.reason
                        )
                        record["native_expected_tool_call_ids"] = list(
                            native_state.expected_tool_call_ids
                        )
                        record["native_completed_tool_call_ids"] = list(
                            native_state.completed_tool_call_ids
                        )
                        record["native_pending_tool_call_ids"] = list(
                            native_state.pending_tool_call_ids
                        )
                        record["native_stall_timeout_seconds"] = None
                        record.setdefault("native_activity_state", "starting")
                    if (
                        spec.native_resume
                        and not is_native_tool_execution
                        and not is_command_execution
                        and native_inactive_seconds >= MODEL_RESPONSE_STALL_SECONDS
                        and session_stall_seconds is None
                    ):
                        session_stall_seconds = native_inactive_seconds
                        inspection = {
                            "at": now_iso(),
                            "silence_seconds": round(native_inactive_seconds, 3),
                            "stream_tail": log_tail_excerpt(stream_path),
                        }
                        record["native_silence_inspection"] = inspection
                        diagnostic = (
                            f"native-session {native_phase} stream produced no update for "
                            f"{native_inactive_seconds:.1f}s; terminating the stalled model "
                            "process for native-session recovery"
                        )
                        heartbeat_log.write(f"[silence-inspection] {diagnostic}\n")
                        heartbeat_log.flush()
                        persist_locator_record()
                        attempt_event(prefix, f"모델응답정지: {diagnostic}")
                        await terminate_process_group(process)
                    non_native_inactive_seconds = loop.time() - max(
                        last_native_progress_at, last_stream_progress_at
                    )
                    if (
                        not spec.native_resume
                        and not is_command_execution
                        and non_native_inactive_seconds
                        >= MODEL_RESPONSE_STALL_SECONDS
                        and session_stall_seconds is None
                    ):
                        session_stall_seconds = non_native_inactive_seconds
                        inspection = {
                            "at": now_iso(),
                            "silence_seconds": round(non_native_inactive_seconds, 3),
                            "stream_tail": log_tail_excerpt(stream_path),
                        }
                        record["stream_silence_inspection"] = inspection
                        diagnostic = (
                            f"{spec.cli} emitted no stream output or native-session event for "
                            f"{non_native_inactive_seconds:.1f}s; terminating the stalled model "
                            "process for retry recovery"
                        )
                        heartbeat_log.write(f"[silence-inspection] {diagnostic}\n")
                        heartbeat_log.flush()
                        persist_locator_record()
                        attempt_event(prefix, f"모델응답정지: {diagnostic}")
                        await terminate_process_group(process)
                    heartbeat = (
                        f"작업중... locator={locator_path} "
                        f"native_session={record.get('native_session_path') or 'none'} "
                        f"native_mtime_ns={record.get('native_session_mtime_ns', 'none')}"
                    )
                    if spec.native_resume:
                        heartbeat += (
                            f" native_activity={record.get('native_activity_state')}"
                            f" native_phase={native_phase}"
                        )
                    if active_command_execution_ids:
                        heartbeat += (
                            " command_execution_ids="
                            + ",".join(sorted(active_command_execution_ids))
                        )
                    heartbeat_log.write(f"[heartbeat] {heartbeat}\n")
                    heartbeat_log.flush()
                    persist_locator_record()
                    # Heartbeat is recovery state, not a user-visible lifecycle
                    # event. Keep it out of the caller-facing event stream.
                    continue
                if raw is None:
                    finished_streams += 1
                    continue
                record.pop("native_silence_inspection", None)
                record.pop("stream_silence_inspection", None)
                if spec.native_resume and channel == "stdout":
                    record["native_activity_state"] = "streaming"
                line = raw.decode("utf-8", errors="replace").rstrip("\n")
                stream_log.write(f"[{channel}] {line}\n")
                stream_log.flush()
                if channel == "stdout":
                    command_transition = json_agent_command_transition(line)
                    if command_transition is not None:
                        transition, command_id = command_transition
                        if transition == "started":
                            active_command_execution_ids.add(command_id)
                        else:
                            active_command_execution_ids.discard(command_id)
                        record["active_command_execution_ids"] = sorted(
                            active_command_execution_ids
                        )
                if channel == "stdout" and terminal_success_contract:
                    terminal_outcome = json_agent_terminal_outcome_from_line(
                        line, str(terminal_success_contract)
                    )
                    if terminal_outcome is not None:
                        terminal_outcome_seen = terminal_outcome
                diagnostic = terminal_diagnostic(spec.cli, channel, line)
                if diagnostic:
                    diagnostics.append(diagnostic)
                    diagnostic_origins.append(f"{spec.cli}:{channel}")
                if role == "review" and channel == "stdout":
                    invoked_tool = collaboration_tool(line)
                    if invoked_tool and control_violation is None:
                        control_violation = invoked_tool
                        diagnostics.append(
                            f"official review invoked forbidden collaboration tool: "
                            f"{invoked_tool}"
                        )
                        diagnostic_origins.append("dispatcher:review-control")
                        attempt_event(
                            prefix,
                            f"리뷰 제어 계약 위반: collaboration-tool="
                            f"{invoked_tool}",
                        )
                        await terminate_process_group(process)
                rendered, discovered = (
                    render_json_line(spec.cli, line) if channel == "stdout" else ([line], None)
                )
                if discovered and record.get("session_id") != discovered:
                    record["session_id"] = discovered
                    if native_resume_session is None:
                        record["native_session_path"] = native_session_path(
                            spec, workspace, discovered, attempt_dir
                        )
                    persist_locator_record()
                for display_line in rendered:
                    if display_line:
                        normalized_output_log.write(display_line + "\n")
                        normalized_output_log.flush()
                        # Child output is retained for recovery and review but is
                        # not itself a dispatcher lifecycle event.
        await asyncio.gather(*readers)
        return_code = await process.wait()
    except asyncio.CancelledError:
        for reader in readers:
            reader.cancel()
        if readers:
            await asyncio.gather(*readers, return_exceptions=True)
        await terminate_process_group(process)
        runtime_error: OSError | None = None
        try:
            append_milestone_event(
                task,
                event="FINISH",
                execution_id=identity,
                role=role,
                attempt=attempt,
                model=spec.display,
                result="failed:cancelled",
                locator=locator_path,
            )
        except OSError as exc:
            runtime_error = exc
        record.update(
            status="failed",
            finished_at=now_iso(),
            exit_code="cancelled",
            failure_class="cancelled",
            failure_source="caller-cancel",
            provider_transport_failure_confirmed=False,
        )
        if runtime_error is not None:
            record["work_log_runtime_error"] = str(runtime_error)
        persist_locator_record()
        raise

    for raw_path in record.get("auxiliary_logs", []):
        aux_diagnostics = auxiliary_log_diagnostics(Path(str(raw_path)))
        diagnostics.extend(aux_diagnostics)
        diagnostic_origins.extend(
            f"{spec.cli}:auxiliary-log" for _ in aux_diagnostics
        )
    native_path = (
        str(native_resume_session)
        if native_resume_session is not None
        else native_session_path(
            spec, workspace, record.get("session_id"), attempt_dir
        )
    )
    if native_path:
        record["native_session_path"] = native_path
    native_mtime = native_session_mtime_ns(record.get("native_session_path"))
    if native_mtime is not None:
        record["native_session_mtime_ns"] = native_mtime
    failure_source: str | None = None
    failure_evidence: str | None = None
    failure_evidence_source: str | None = None
    provider_transport_failure_confirmed = False
    termination = termination_signal(return_code)
    if termination is not None:
        record["termination_signal"] = termination[0]
        record["termination_signal_inferred"] = termination[1]
    if control_violation:
        failure_class = "review-control-violation"
        failure_source = "dispatcher-control"
        failure_evidence_source = "dispatcher:review-control"
        for index in range(len(diagnostics) - 1, -1, -1):
            if diagnostic_origins[index] == failure_evidence_source:
                failure_evidence = diagnostics[index]
                break
    elif session_stall_seconds is not None:
        failure_class = "session-stall"
        failure_source = "dispatcher-stall-timeout"
        record["session_stall_seconds"] = round(session_stall_seconds, 3)
        record["termination_initiator"] = "dispatcher"
    elif return_code != 0 and termination is not None:
        failure_class = "process-terminated"
        failure_source = "process-termination"
        record["termination_initiator"] = "unknown"
    elif (
        return_code == 0
        and terminal_success_contract is not None
        and terminal_outcome_seen == "succeeded"
    ):
        # A declared terminal-success event is authoritative. Earlier JSONL
        # error events and stderr diagnostics may describe a recovered
        # transport (for example Codex WebSocket -> HTTPS fallback), not the
        # final attempt outcome.
        failure_class = None
    else:
        classified_failure, classified_evidence = classify_failure_with_evidence(
            "\n".join(diagnostics[-50:])
        )
        stdout_diagnostic_index = next(
            (
                index
                for index in range(len(diagnostic_origins) - 1, -1, -1)
                if diagnostic_origins[index].endswith(":stdout")
            ),
            None,
        )
        if return_code != 0 or classified_evidence is not None:
            failure_class = classified_failure
            failure_evidence = classified_evidence
        elif stdout_diagnostic_index is not None:
            failure_class = "generic-error"
            failure_evidence = diagnostics[stdout_diagnostic_index]
        elif terminal_success_contract is not None and terminal_outcome_seen != "succeeded":
            failure_class = "generic-error"
            failure_source = "dispatcher-terminal-contract"
            failure_evidence = (
                f"required {terminal_success_contract} terminal success event "
                "was not observed"
            )
            failure_evidence_source = "dispatcher:terminal-contract"
        else:
            failure_class = None
        if (
            failure_class is not None
            and failure_evidence is not None
            and failure_evidence_source is None
        ):
            for index in range(len(diagnostics) - 1, -1, -1):
                if diagnostics[index] == failure_evidence:
                    failure_evidence_source = diagnostic_origins[index]
                    break
        if failure_source is None:
            if failure_class in PROVIDER_TRANSPORT_FAILURES:
                failure_source = "provider-terminal-diagnostic"
                provider_transport_failure_confirmed = failure_evidence is not None
            elif failure_evidence is not None:
                failure_source = "cli-terminal-diagnostic"
            elif return_code != 0:
                failure_source = "cli-exit"
    if (
        role == "worker"
        and task.review is not None
        and return_code == 0
        and failure_class is None
    ):
        worker_errors = implementation_review_errors(task)
        worker_signature_after = task_signature(workspace, task)
        if worker_signature_before == worker_signature_after or worker_errors:
            failure_class = "generic-error"
            failure_source = "dispatcher-worker-completion-contract"
            details = []
            if worker_signature_before == worker_signature_after:
                details.append("no claimed file or implementation evidence changed")
            details.extend(worker_errors)
            failure_evidence = "; ".join(details)
            failure_evidence_source = "dispatcher:worker-completion-contract"
    try:
        append_milestone_event(
            task,
            event="FINISH",
            execution_id=identity,
            role=role,
            attempt=attempt,
            model=spec.display,
            result=(
                f"succeeded:0"
                if return_code == 0 and failure_class is None
                else f"failed:{failure_class or 'generic-error'}:{return_code}"
            ),
            locator=locator_path,
        )
    except OSError as exc:
        if failure_class is not None:
            record["prior_failure_class"] = failure_class
        record["work_log_runtime_error"] = str(exc)
        failure_class = "work-log-runtime-write"
        failure_source = "work-log"
        failure_evidence = None
        failure_evidence_source = None
        provider_transport_failure_confirmed = False
    if failure_evidence is not None:
        record["failure_evidence_excerpt"] = failure_evidence[:FAILURE_EVIDENCE_LIMIT]
        record["failure_evidence_truncated"] = (
            len(failure_evidence) > FAILURE_EVIDENCE_LIMIT
        )
    if failure_evidence_source is not None:
        record["failure_evidence_source"] = failure_evidence_source
    record.update(
        status="succeeded" if return_code == 0 and failure_class is None else "failed",
        finished_at=now_iso(),
        exit_code=return_code,
        failure_class=failure_class,
        failure_source=failure_source,
        provider_transport_failure_confirmed=provider_transport_failure_confirmed,
    )
    persist_locator_record()
    return return_code, failure_class, locator_path


def dispatcher_child_prompt(body: str) -> str:
    return f"{DISPATCHER_CHILD_BOUNDARY_PROMPT} {body}"


def selfcheck_prompt(task: Task, *, unchecked_items: bool = False) -> str:
    if task.plan is None:
        raise RuntimeError("selfcheck PLAN이 없다")
    if task.review is None:
        raise RuntimeError("selfcheck CODE_REVIEW 파일이 없다")
    if unchecked_items:
        body = (
            f"Read {task.plan.resolve()}; complete every unchecked implementation "
            f"item and update {task.review.resolve()}. {REPOSITORY_LANGUAGE_PROMPT}"
        )
    else:
        body = (
            f"Read {task.plan.resolve()}; review all work once, fix omissions, "
            f"and update {task.review.resolve()}. {REPOSITORY_LANGUAGE_PROMPT}"
        )
    return f"{SELF_CHECK_PROMPT_PREFIX} {body}"


def base_prompt(
    task: Task,
    role: str,
    spec: AgentSpec,
    *,
    unchecked_items: bool = False,
) -> str:
    if role == "review":
        target = task.review or task.directory
        if task.review:
            return dispatcher_child_prompt(
                f"Read {target.resolve()} and start the review. "
                f"{REPOSITORY_LANGUAGE_PROMPT}"
            )
        return dispatcher_child_prompt(
            f"Continue the review for {target.resolve()}. "
            f"{REPOSITORY_LANGUAGE_PROMPT}"
        )
    if task.plan is None:
        raise RuntimeError("worker PLAN이 없다")
    target = task.plan.resolve()
    if role == "selfcheck":
        return selfcheck_prompt(task, unchecked_items=unchecked_items)
    if spec.native_resume:
        return dispatcher_child_prompt(
            f"Read {target} and complete the task. {REPOSITORY_LANGUAGE_PROMPT}"
        )
    return dispatcher_child_prompt(
        f"Read {target} and complete the task. {REPOSITORY_LANGUAGE_PROMPT}"
    )



def build_context_package(
    workspace: Path, task: Task, locator: Path, *, previous_spec: AgentSpec, next_spec: AgentSpec
) -> dict[str, Any]:
    """Build the fail-closed continuation context for a target transition."""

    if not locator.is_file():
        raise ExecutionDecisionError("logical context locator가 없다")
    try:
        record = json.loads(locator.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ExecutionDecisionError("logical context locator를 읽을 수 없다") from exc
    if not isinstance(record, dict):
        raise ExecutionDecisionError("logical context locator 형식이 잘못됐다")
    workspace_value = record.get("workspace")
    if not isinstance(workspace_value, str) or not workspace_value:
        raise ExecutionDecisionError("logical context workspace가 없다")
    recorded_workspace = Path(workspace_value)
    if not recorded_workspace.is_absolute() or recorded_workspace.resolve() != workspace.resolve():
        raise ExecutionDecisionError("logical context workspace가 일치하지 않는다")
    if record.get("task") != task.name:
        raise ExecutionDecisionError("logical context task가 일치하지 않는다")
    if task.plan is None or not task.plan.is_file():
        raise ExecutionDecisionError("logical context PLAN이 없다")
    plan_value = record.get("plan_path")
    if not isinstance(plan_value, str) or not plan_value:
        raise ExecutionDecisionError("logical context PLAN 경로가 없다")
    plan_path = Path(plan_value)
    if not plan_path.is_absolute() or plan_path.resolve() != task.plan.resolve():
        raise ExecutionDecisionError("logical context PLAN 경로가 일치하지 않는다")
    required = {
        "normalized_output": ("normalized_output_log", "normalized-output.log"),
        "raw_log": ("stream_log", "stream.log"),
    }
    paths: dict[str, str] = {}
    attempt_dir = locator.resolve().parent
    for field, (record_field, filename) in required.items():
        value = record.get(record_field)
        if not isinstance(value, str) or not value:
            raise ExecutionDecisionError(f"logical context {field} artifact가 없다")
        path = Path(value)
        expected = attempt_dir / filename
        if not path.is_absolute() or path.resolve() != expected or not expected.is_file():
            raise ExecutionDecisionError(f"logical context {field} artifact가 locator attempt와 일치하지 않는다")
        paths[field] = str(expected)
    same_native = previous_spec.native_resume and next_spec.native_resume
    package = {
        "plan": str(task.plan.resolve()), "locator": str(locator.resolve()),
        "workspace": str(workspace.resolve()), **paths,
        "resume_mode": "native" if same_native else "logical",
    }
    if same_native:
        native = Path(str(record.get("native_session_path", "")))
        if not native.is_file():
            raise ExecutionDecisionError("same-native-session logical context native session이 없다")
        package["native_session_path"] = str(native.resolve())
    return package


def canonical_selector_failover_route(decision: dict[str, Any] | None) -> bool:
    if not isinstance(decision, dict):
        return False
    candidates = decision.get("candidates")
    return isinstance(candidates, list) and len(candidates) > 1


def logical_context_prompt(context: dict[str, Any]) -> str:
    plan = context["plan"]
    locator = context["locator"]
    workspace = context["workspace"]
    raw_log = context["raw_log"]
    normalized_output = context["normalized_output"]
    return dispatcher_child_prompt(
        f"{REPOSITORY_LANGUAGE_PROMPT} "
        f"Read plan={plan}, locator={locator}, workspace={workspace}, "
        f"raw_log={raw_log}, normalized_output={normalized_output} and complete the task."
    )


def continuation_prompt_from_package(
    context_package: dict[str, Any],
    *,
    target: dict[str, Any] | None = None,
    native_resume: bool = False,
) -> str:
    if native_resume or context_package.get("resume_mode") == "native":
        return dispatcher_child_prompt(
            f"{REPOSITORY_LANGUAGE_PROMPT} Continue this session and complete "
            "the current task."
        )
    plan = context_package["plan"]
    locator = context_package["locator"]
    workspace = context_package["workspace"]
    raw_log = context_package["raw_log"]
    normalized_output = context_package["normalized_output"]
    return dispatcher_child_prompt(
        f"{REPOSITORY_LANGUAGE_PROMPT} "
        f"Read plan={plan}, locator={locator}, workspace={workspace}, "
        f"raw_log={raw_log}, normalized_output={normalized_output} and complete the task."
    )


def continuation_prompt(
    task: Task,
    role: str,
    locator: Path | None = None,
    *,
    native_resume: bool = False,
    resume_same_native_session: bool = False,
    context: dict[str, Any] | None = None,
    unchecked_items: bool = False,
) -> str:
    if native_resume and role == "selfcheck":
        if resume_same_native_session:
            if unchecked_items:
                return selfcheck_prompt(task, unchecked_items=True)
            return (
                f"{SELF_CHECK_PROMPT_PREFIX} Continue."
            )
        return selfcheck_prompt(task, unchecked_items=unchecked_items)
    if context is not None:
        return continuation_prompt_from_package(
            context,
            native_resume=resume_same_native_session or context.get("resume_mode") == "native",
        )
    if native_resume:
        if resume_same_native_session:
            return dispatcher_child_prompt(
                f"{REPOSITORY_LANGUAGE_PROMPT} Continue this session and complete "
                "the current task."
            )
        target = task.plan or task.directory
        return dispatcher_child_prompt(
            f"Read {target.resolve()} and complete the task. "
            f"{REPOSITORY_LANGUAGE_PROMPT}"
        )
    if role == "review":
        return dispatcher_child_prompt(
            f"Continue the review for {task.directory.resolve()}. "
            f"{REPOSITORY_LANGUAGE_PROMPT}"
        )
    return dispatcher_child_prompt(
        f"Continue from {locator.resolve() if locator else task.directory.resolve()}. Check the saved context and current "
        f"workspace. {REPOSITORY_LANGUAGE_PROMPT}"
    )


async def run_escalating(
    workspace: Path,
    store: StateStore,
    task: Task,
    role: str,
    initial: AgentSpec,
    initial_resume_locator: Path | None = None,
    *,
    unchecked_items: bool = False,
) -> tuple[bool, Path | None]:
    spec = initial
    previous_locator = initial_resume_locator
    review_control_retries = 0
    native_recovery_retries = 0
    generic_retries = 0
    terminal_recovery_retries = 0
    native_resume_locator = initial_resume_locator
    recovery_failures = 0
    stage_budget: StageFailureBudget | None = None
    if isinstance(store, StateStore):
        state = store.task_state(task)
        persisted = state.get("recovery_failures", {})
        if isinstance(persisted, dict):
            recovery_failures = int(persisted.get(role, 0))
        decisions = state.get("execution_decisions", {})
        decision = decisions.get(role) if isinstance(decisions, dict) and role in {"worker", "review"} else None
        if isinstance(decision, dict):
            stage_budget = StageFailureBudget.from_decision(store, task, decision)
            recovery_failures = stage_budget.count()
    if recovery_failures >= RECOVERY_FAILURE_LIMIT:
        locator = initial_resume_locator
        reason = (
            f"{role} recovery failure limit already exhausted: "
            f"{recovery_failures}/{RECOVERY_FAILURE_LIMIT}"
        )
        if isinstance(store, StateStore):
            decision = store.task_state(task).get("execution_decisions", {}).get(role, {})
            store.update_task(
                task,
                blocked=f"{reason} locator={locator}",
                blocker_evidence={
                    "role": role,
                    "failure_class": None,
                    "locator": str(locator) if locator else None,
                    "selected": decision.get("selected") if isinstance(decision, dict) else None,
                    "work_unit_id": decision.get("work_unit_id") if isinstance(decision, dict) else None,
                },
            )
        banner(
            "작업차단",
            task.name,
            [
                "reason=recovery-failure-limit",
                f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}",
                f"locator={locator}",
            ],
        )
        return False, locator
    context: dict[str, Any] | None = None
    while True:
        prompt = (
            base_prompt(task, role, spec, unchecked_items=unchecked_items)
            if previous_locator is None
            else continuation_prompt(
                task,
                role,
                previous_locator,
                native_resume=spec.native_resume,
                resume_same_native_session=native_resume_locator is not None,
                context=context,
                unchecked_items=unchecked_items,
            )
        )
        context = None
        rc, failure, locator = await invoke(
            workspace,
            store,
            task,
            role,
            spec,
            prompt,
            resume_locator=native_resume_locator,
        )
        native_resume_locator = None
        if rc == 0 and failure is None:
            if isinstance(store, StateStore):
                state = store.task_state(task)
                persisted = dict(state.get("recovery_failures", {}))
                persisted.pop(role, None)
                store.update_task(task, recovery_failures=persisted)
                if stage_budget is not None:
                    stage_budget.reset_on_success()
                decisions = state.get("execution_decisions", {})
                completing = (
                    decisions.get(role) if isinstance(decisions, dict) else None
                )
                if isinstance(completing, dict):
                    GenericFailureBudget.from_decision(
                        store, task, completing
                    ).reset_stage_on_success()
            return True, locator
        failure = failure or "generic-error"
        if failure in {
            "work-log-blocked",
            "work-log-incomplete",
            "work-log-setup",
            "work-log-runtime-write",
        }:
            banner("작업차단", task.name, failure_report_lines(failure, locator))
            return False, locator
        recovery_failures += 1
        if stage_budget is not None:
            selected = stage_budget.store.task_state(task)["execution_decisions"][role]["selected"]
            transition = stage_budget.store.task_state(task)["execution_decisions"][role]["transition"]["trigger"]
            recovery_failures = stage_budget.record_failure(target=selected, transition=transition)
        if isinstance(store, StateStore):
            state = store.task_state(task)
            persisted = dict(state.get("recovery_failures", {}))
            persisted[role] = recovery_failures
            store.update_task(task, recovery_failures=persisted)
        if recovery_failures >= RECOVERY_FAILURE_LIMIT:
            reason = (
                f"{role} recovery failure limit exhausted: "
                f"{recovery_failures}/{RECOVERY_FAILURE_LIMIT}"
            )
            if isinstance(store, StateStore):
                decision = store.task_state(task).get("execution_decisions", {}).get(role, {})
                store.update_task(
                    task,
                    blocked=f"{reason} locator={locator}",
                    blocker_evidence={
                        "role": role,
                        "failure_class": failure,
                        "locator": str(locator) if locator else None,
                        "selected": decision.get("selected") if isinstance(decision, dict) else None,
                        "work_unit_id": decision.get("work_unit_id") if isinstance(decision, dict) else None,
                    },
                )
            banner(
                "작업차단",
                task.name,
                [
                    "reason=recovery-failure-limit",
                    *failure_report_lines(failure, locator),
                    f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}",
                ],
            )
            return False, locator
        if role == "review" and failure == "review-control-violation":
            review_control_retries += 1
            banner(
                "리뷰재시도",
                task.name,
                [
                    *failure_report_lines(failure, locator),
                    f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}",
                ],
            )
            previous_locator = None
            await asyncio.sleep(min(30, 2 ** min(review_control_retries, 5)))
            continue
        current_decision = None
        if isinstance(store, StateStore):
            task_state = store.task_state(task)
            decisions = task_state.get("execution_decisions", {})
            if isinstance(decisions, dict):
                current_decision = decisions.get(role)

        generic_failure_count = 0
        if failure in BOUNDED_TARGET_FAILURES and isinstance(current_decision, dict):
            generic_failure_count = GenericFailureBudget.from_decision(
                store, task, current_decision
            ).record_failure()

        generic_failover_ready = (
            failure in BOUNDED_TARGET_FAILURES
            and generic_failure_count >= GENERIC_FAILURE_LIMIT_PER_TARGET
        )
        if canonical_selector_failover_route(current_decision) and (
            failure in QUALIFIED_FAILOVER_FAILURES or generic_failover_ready
        ):
            try:
                next_decision = select_execution_decision(
                    task,
                    stage=role,
                    prior_decision=current_decision,
                    transition="failover",
                    failure_class=failure,
                )
                next_spec = agent_spec_from_decision(next_decision)
                if next_spec != spec:
                    if locator is None:
                        raise ExecutionDecisionError("logical context locator가 없다")
                    context = build_context_package(
                        workspace, task, locator, previous_spec=spec, next_spec=next_spec
                    )
                    commit_execution_decision(store, task, role, next_decision)
                    banner(
                        "실행대상전환" if role == "worker" else "리뷰실행대상전환",
                        task.name,
                        [
                            f"from={spec.display}",
                            f"to={next_spec.display}",
                            *failure_report_lines(failure, locator),
                        ],
                    )
                    spec = next_spec
                    previous_locator = locator
                    continue
                else:
                    commit_execution_decision(store, task, role, next_decision)
            except (ExecutionDecisionError, OSError, ValueError) as exc:
                code = getattr(exc, "code", "")
                if not code:
                    if "no_failover_candidate" in str(exc):
                        code = "no_failover_candidate"
                    else:
                        code = exc.__class__.__name__
                if generic_failover_ready and code == "no_failover_candidate":
                    reason = (
                        f"{role} generic failure limit exhausted for target: "
                        f"{generic_failure_count}/"
                        f"{GENERIC_FAILURE_LIMIT_PER_TARGET}"
                    )
                    selected = (
                        current_decision.get("selected")
                        if isinstance(current_decision, dict)
                        else None
                    )
                    store.update_task(
                        task,
                        blocked=f"{reason} locator={locator}",
                        blocker_evidence={
                            "role": role,
                            "failure_class": failure,
                            "locator": str(locator) if locator else None,
                            "selected": selected,
                            "work_unit_id": current_decision.get("work_unit_id")
                            if isinstance(current_decision, dict)
                            else None,
                        },
                    )
                    banner(
                        "작업차단",
                        task.name,
                        [
                            "reason=generic-failure-limit",
                            *failure_report_lines(failure, locator),
                            f"retry={generic_failure_count}/"
                            f"{GENERIC_FAILURE_LIMIT_PER_TARGET}",
                        ],
                    )
                    return False, locator
                store.update_task(
                    task, blocked=f"{role} selector decision 실패 [{code}]: {exc}"
                )
                banner(
                    "작업차단",
                    task.name,
                    [f"reason={code}", *failure_report_lines(failure, locator)],
                )
                return False, locator
        if generic_failover_ready:
            reason = (
                f"{role} generic failure limit exhausted for target: "
                f"{generic_failure_count}/{GENERIC_FAILURE_LIMIT_PER_TARGET}"
            )
            selected = (
                current_decision.get("selected")
                if isinstance(current_decision, dict)
                else None
            )
            store.update_task(
                task,
                blocked=f"{reason} locator={locator}",
                blocker_evidence={
                    "role": role,
                    "failure_class": failure,
                    "locator": str(locator) if locator else None,
                    "selected": selected,
                    "work_unit_id": current_decision.get("work_unit_id")
                    if isinstance(current_decision, dict)
                    else None,
                },
            )
            banner(
                "작업차단",
                task.name,
                [
                    "reason=generic-failure-limit",
                    *failure_report_lines(failure, locator),
                    f"retry={generic_failure_count}/"
                    f"{GENERIC_FAILURE_LIMIT_PER_TARGET}",
                ],
            )
            return False, locator
        if spec.native_resume:
            if failure in {"context-limit", "session-stall"}:
                native_recovery_retries += 1
                banner(
                    "native-session세션연속재시작",
                    task.name,
                    [
                        f"model={spec.display}",
                        *failure_report_lines(failure, locator),
                        f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}",
                    ],
                )
                previous_locator = locator
                native_resume_locator = locator
                await asyncio.sleep(min(30, 2 ** min(native_recovery_retries, 5)))
                continue
            native_recovery_retries += 1
            if failure == "session-stall":
                event = "세션응답복구재시도"
            elif failure in {
                "provider-connection",
                "provider-stream-disconnect",
            }:
                event = "세션연결재시도"
            else:
                event = "native-session복구재시도"
            banner(
                event,
                task.name,
                [
                    f"model={spec.display}",
                    *failure_report_lines(failure, locator),
                    f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}",
                ],
            )
            previous_locator = locator
            await asyncio.sleep(min(30, 2 ** min(native_recovery_retries, 5)))
            continue
        if failure in BOUNDED_TARGET_FAILURES:
            generic_retries += 1
            banner(
                "작업복구재시도",
                task.name,
                [
                    f"model={spec.display}",
                    *failure_report_lines(failure, locator),
                    f"retry={generic_failure_count}/"
                    f"{GENERIC_FAILURE_LIMIT_PER_TARGET}",
                ],
            )
            previous_locator = locator
            await asyncio.sleep(min(30, 2 ** min(generic_retries, 5)))
            continue
        if failure not in RECOVERABLE_RUNTIME_FAILURES:
            terminal_recovery_retries += 1
            banner(
                "모델복구재시도",
                task.name,
                [
                    f"model={spec.display}",
                    *failure_report_lines(failure, locator),
                    f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}",
                ],
            )
            previous_locator = locator
            await asyncio.sleep(
                min(30, 2 ** min(terminal_recovery_retries, 5))
            )
            continue
        if role == "review":
            terminal_recovery_retries += 1
            banner(
                "리뷰재시도",
                task.name,
                [
                    f"model={spec.display}",
                    "reason=review-route-has-no-next-target",
                    *failure_report_lines(failure, locator),
                    f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}",
                ],
            )
            previous_locator = locator
            await asyncio.sleep(
                min(30, 2 ** min(terminal_recovery_retries, 5))
            )
            continue
        terminal_recovery_retries += 1
        banner(
            "모델복구재시도",
            task.name,
            [
                f"model={spec.display}",
                *failure_report_lines(failure, locator),
                f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}",
            ],
        )
        previous_locator = locator
        await asyncio.sleep(min(30, 2 ** min(terminal_recovery_retries, 5)))


def task_signature(workspace: Path, task: Task) -> str:
    digest = hashlib.sha256()
    if not task.directory.exists():
        return "moved"
    for path in sorted(p for p in task.directory.iterdir() if p.is_file()):
        if (
            PLAN_RE.match(path.name)
            or REVIEW_RE.match(path.name)
            or path.name.endswith(".log")
        ):
            digest.update(path.name.encode())
            digest.update(sha256_file(path).encode())
    for raw_path in sorted(task.write_set):
        path = Path(raw_path)
        path = path if path.is_absolute() else workspace / path
        digest.update(raw_path.encode())
        if path.is_file():
            digest.update(str(path.stat().st_mode).encode())
            digest.update(sha256_file(path).encode())
        elif path.exists():
            digest.update(b"non-file")
        else:
            digest.update(b"missing")
    return digest.hexdigest()


def read_verdict(path: Path) -> str | None:
    if not path.exists():
        return None
    return verdict_from_text(path.read_text(encoding="utf-8", errors="replace"))


def verdict_from_text(text: str) -> str | None:
    selected: tuple[re.Match[str], re.Pattern[str], re.Pattern[str]] | None = None
    for heading_re, line_re, block_re in VERDICT_SCHEMA_MATCHERS:
        headings = list(heading_re.finditer(text))
        if not headings:
            continue
        # A duplicated heading, or headings from both schemas, is ambiguous.
        if len(headings) != 1 or selected is not None:
            return None
        selected = (headings[0], line_re, block_re)
    if selected is None:
        return None
    heading, line_re, block_re = selected
    next_heading = re.search(r"^##\s+", text[heading.end():], re.MULTILINE)
    end = heading.end() + next_heading.start() if next_heading else len(text)
    section = text[heading.end():end]
    inline_matches = list(line_re.finditer(section))
    block_matches = list(block_re.finditer(section))
    matches = inline_matches + block_matches
    if len(matches) != 1:
        return None
    return next(
        (value for value in matches[0].groups() if value in {"PASS", "WARN", "FAIL"}),
        None,
    )


def matching_archive_directories_by_name(
    workspace: Path,
    task_name: str,
    *,
    require_complete: bool = True,
) -> list[Path]:
    archive = workspace / "agent-task" / "archive"
    parts = task_name.split("/")
    if not archive.is_dir() or len(parts) not in {1, 2}:
        return []
    group = parts[0]
    final_name = parts[-1]
    suffix_re = re.compile(rf"^{re.escape(final_name)}(?:_\d+)?$")
    matches: list[Path] = []
    try:
        years = list(archive.iterdir())
    except FileNotFoundError:
        return []
    for year in years:
        if not year.is_dir():
            continue
        try:
            months = list(year.iterdir())
        except FileNotFoundError:
            continue
        for month in months:
            if not month.is_dir():
                continue
            parent = month if len(parts) == 1 else month / group
            if not parent.is_dir():
                continue
            try:
                candidates = list(parent.iterdir())
            except FileNotFoundError:
                continue
            for candidate in candidates:
                if (
                    candidate.is_dir()
                    and suffix_re.match(candidate.name)
                    and (
                        not require_complete
                        or (candidate / "complete.log").is_file()
                    )
                ):
                    matches.append(candidate)
    return sorted(matches)


def matching_archive_directories(workspace: Path, task: Task) -> list[Path]:
    return matching_archive_directories_by_name(workspace, task.name)


def task_group_name(task_name: str) -> str:
    return task_name.split("/", 1)[0]


def work_log_event_cells(line: str) -> list[str] | None:
    stripped = line.strip()
    if not stripped.startswith("|") or not stripped.endswith("|"):
        return None
    cells = [
        cell.strip().replace(r"\|", "|")
        for cell in re.split(r"(?<!\\)\|", stripped[1:-1])
    ]
    if len(cells) == 10:
        return cells
    if len(cells) != 9:
        return None
    if cells[0] == "seq":
        legacy_loop = "loop"
    elif cells[0].startswith("---"):
        legacy_loop = "---:"
    else:
        match = WORK_LOG_EXECUTION_LOOP_RE.search(cells[8])
        legacy_loop = match.group("loop") if match else "0"
    return [*cells[:4], legacy_loop, *cells[4:]]


def merge_work_log_sources(sources: set[Path]) -> str:
    """Merge duplicate dispatcher timelines without losing conflicting rows."""
    allowed_metadata = {
        "# Milestone Work Log",
        "## Dispatcher Timeline",
        "> Dispatcher-owned execution timeline. Workers and reviewers do not edit this file.",
        "> Dispatcher-owned. Workers and reviewers do not edit this section.",
        WORK_LOG_HEADER,
        WORK_LOG_SEPARATOR,
        LEGACY_WORK_LOG_HEADER,
        LEGACY_WORK_LOG_SEPARATOR,
    }
    unique_rows: dict[tuple[str, ...], tuple[str, ...]] = {}
    ordered_rows: list[tuple[str, int, int, list[str]]] = []

    for source_index, source in enumerate(sorted(sources)):
        try:
            lines = source.read_text(
                encoding="utf-8",
                errors="replace",
            ).splitlines()
        except OSError as exc:
            raise ValueError(
                f"WORK_LOG source를 읽을 수 없다: source={source} error={exc}"
            ) from exc
        for line_number, line in enumerate(lines, start=1):
            cells = work_log_event_cells(line)
            if cells is None:
                if not line.strip() or line.strip() in allowed_metadata:
                    continue
                raise ValueError(
                    "WORK_LOG 병합 대상에 안전하게 보존할 수 없는 내용이 있다: "
                    f"source={source} line={line_number}"
                )
            if cells[0] == "seq" or cells[0].startswith("---"):
                continue
            if cells[2] not in {"START", "FINISH"}:
                raise ValueError(
                    "WORK_LOG 병합 대상에 지원하지 않는 event가 있다: "
                    f"source={source} line={line_number} event={cells[2]}"
                )
            try:
                sequence = int(cells[0])
                int(cells[4])
                int(cells[6])
            except ValueError as exc:
                raise ValueError(
                    "WORK_LOG 병합 대상의 sequence, loop 또는 attempt가 유효하지 않다: "
                    f"source={source} line={line_number}"
                ) from exc
            key = (cells[2], cells[3], cells[4], cells[5], cells[6], cells[9])
            fingerprint = tuple(cells[1:])
            previous = unique_rows.get(key)
            if previous is not None:
                if previous != fingerprint:
                    raise ValueError(
                        "WORK_LOG 병합 충돌: "
                        f"event={cells[2]} task={cells[3]} loop={cells[4]} "
                        f"role={cells[5]} attempt={cells[6]} locator={cells[9]}"
                    )
                continue
            unique_rows[key] = fingerprint
            ordered_rows.append((cells[1], source_index, sequence, cells))

    if not ordered_rows:
        raise ValueError("WORK_LOG 병합 대상에 timeline row가 없다")

    header = (
        "# Milestone Work Log\n\n"
        "> Dispatcher-owned execution timeline. Workers and reviewers do not edit this file.\n\n"
        f"{WORK_LOG_HEADER}\n"
        f"{WORK_LOG_SEPARATOR}\n"
    )
    rendered_rows: list[str] = []
    for sequence, (_, _, _, cells) in enumerate(sorted(ordered_rows), start=1):
        escaped = [str(value).replace("|", r"\|").replace("\n", " ") for value in cells]
        escaped[0] = str(sequence)
        rendered_rows.append("| " + " | ".join(escaped) + " |\n")
    return header + "".join(rendered_rows)


def unfinished_work_log_attempts(path: Path) -> list[dict[str, Any]]:
    """Return START rows that have no matching FINISH row."""
    try:
        lines = path.read_text(
            encoding="utf-8",
            errors="replace",
        ).splitlines()
    except OSError:
        raise
    open_attempts: dict[str, dict[str, Any]] = {}
    for line in lines:
        cells = work_log_event_cells(line)
        if cells is None or cells[2] not in {"START", "FINISH"}:
            continue
        try:
            sequence = int(cells[0])
            loop = int(cells[4])
            attempt = int(cells[6])
        except ValueError:
            continue
        locator = cells[9]
        key = locator or "\0".join(
            (cells[3], cells[4], cells[5], cells[6], cells[7])
        )
        if cells[2] == "START":
            open_attempts[key] = {
                "sequence": sequence,
                "task_name": cells[3],
                "loop": loop,
                "role": cells[5],
                "attempt": attempt,
                "model": cells[7],
                "locator": locator,
            }
        else:
            open_attempts.pop(key, None)
    return sorted(
        open_attempts.values(),
        key=lambda record: int(record["sequence"]),
    )


def close_unfinished_work_log_attempts(path: Path) -> int:
    """Close orphaned START rows after verified group completion."""
    unfinished = unfinished_work_log_attempts(path)
    for record in unfinished:
        locator = Path(str(record["locator"]))
        append_work_log_event(
            path,
            task_name=str(record["task_name"]),
            loop=int(record["loop"]),
            event="FINISH",
            execution_id=f"reconciled-{record['sequence']}",
            role=str(record["role"]),
            attempt=int(record["attempt"]),
            model=str(record["model"]),
            result="reconciled:verified-complete-archive",
            locator=locator,
        )
    return len(unfinished)


def archived_task_group_directories(
    workspace: Path,
    task_group: str,
) -> list[Path]:
    """Return month-local archive directories for one logical task group."""
    archive_root = workspace / "agent-task" / "archive"
    if not archive_root.is_dir():
        return []
    suffix_re = re.compile(rf"^{re.escape(task_group)}(?:_\d+)?$")
    matches: list[Path] = []
    try:
        years = list(archive_root.iterdir())
    except FileNotFoundError:
        return []
    for year in years:
        if not year.is_dir():
            continue
        try:
            months = list(year.iterdir())
        except FileNotFoundError:
            continue
        for month in months:
            if not month.is_dir():
                continue
            try:
                candidates = list(month.iterdir())
            except FileNotFoundError:
                continue
            matches.extend(
                candidate
                for candidate in candidates
                if candidate.is_dir() and suffix_re.fullmatch(candidate.name)
            )
    return sorted(matches)


def next_work_log_archive_number(
    workspace: Path,
    task_group: str,
) -> int:
    numbers = [
        int(match.group(1))
        for directory in archived_task_group_directories(workspace, task_group)
        for path in directory.glob("work_log_*.log")
        if (match := WORK_LOG_ARCHIVE_RE.fullmatch(path.name))
    ]
    return max(numbers, default=-1) + 1


def completed_group_archive_directory(
    task_group: str,
    task_names: set[str],
    completed_tasks: dict[str, str],
) -> Path | None:
    candidates: list[tuple[int, str, Path]] = []
    for task_name in task_names:
        archive_raw = completed_tasks.get(task_name)
        if not archive_raw:
            continue
        archive = Path(archive_raw)
        if not archive.is_dir():
            continue
        complete_log = archive / "complete.log"
        target = archive if task_name == task_group else archive.parent
        try:
            completed_at = (
                complete_log.stat().st_mtime_ns
                if complete_log.is_file()
                else archive.stat().st_mtime_ns
            )
        except OSError:
            continue
        candidates.append((completed_at, str(target), target))
    return (
        max(candidates, key=lambda item: (item[0], item[1]))[2]
        if candidates
        else None
    )


def archive_completed_group_work_logs(
    workspace: Path,
    observed_tasks: set[str],
    completed_tasks: dict[str, str],
    active_or_running: set[str],
) -> tuple[dict[str, str], dict[str, str]]:
    """Archive each completed task-group timeline after its last writer exits."""
    observed_by_group: dict[str, set[str]] = {}
    for task_name in observed_tasks:
        observed_by_group.setdefault(task_group_name(task_name), set()).add(
            task_name
        )
    active_groups = {
        task_group_name(task_name)
        for task_name in active_or_running
    }
    archived: dict[str, str] = {}
    errors: dict[str, str] = {}
    for task_group, task_names in sorted(observed_by_group.items()):
        if task_group in active_groups or not task_names <= set(completed_tasks):
            continue
        active_source = (
            workspace / "agent-task" / task_group / WORK_LOG_NAME
        )
        legacy_sources = {
            Path(completed_tasks[task_name]) / WORK_LOG_NAME
            for task_name in task_names
            if (Path(completed_tasks[task_name]) / WORK_LOG_NAME).is_file()
        }
        sources = {
            *legacy_sources,
            *([active_source] if active_source.is_file() else []),
        }
        if not sources:
            continue
        target_directory = completed_group_archive_directory(
            task_group,
            task_names,
            completed_tasks,
        )
        if target_directory is None:
            errors[task_group] = (
                "검증된 task archive에서 WORK_LOG 대상 디렉터리를 정할 수 없다"
            )
            continue
        archive_number = next_work_log_archive_number(
            workspace,
            task_group,
        )
        destination = target_directory / f"work_log_{archive_number}.log"
        source = next(iter(sources))
        if destination.exists():
            errors[task_group] = (
                "WORK_LOG archive destination이 이미 존재한다: "
                "sources="
                + ",".join(str(path) for path in sorted(sources))
                + f" destination={destination}"
            )
            continue
        source = next(iter(sources))
        temporary = destination.with_name(destination.name + ".tmp")
        try:
            if len(sources) == 1:
                close_unfinished_work_log_attempts(source)
                source.replace(destination)
            else:
                if temporary.exists():
                    raise OSError(
                        "WORK_LOG archive temporary destination이 이미 존재한다: "
                        f"temporary={temporary}"
                    )
                temporary.write_text(
                    merge_work_log_sources(sources),
                    encoding="utf-8",
                )
                close_unfinished_work_log_attempts(temporary)
                temporary.replace(destination)
                for merged_source in sources:
                    merged_source.unlink()
        except (OSError, ValueError) as exc:
            if temporary.exists():
                try:
                    temporary.unlink()
                except OSError:
                    pass
            errors[task_group] = (
                "WORK_LOG archive 실패: sources="
                + ",".join(str(path) for path in sorted(sources))
                + " "
                f"destination={destination} error={exc}"
            )
            continue
        if active_source in sources:
            for task_name in sorted(task_names, reverse=True):
                if "/" not in task_name:
                    continue
                try:
                    (workspace / "agent-task" / task_name).rmdir()
                except OSError:
                    pass
            try:
                active_source.parent.rmdir()
            except OSError:
                pass
        archived[task_group] = str(destination.resolve())
    return archived, errors


def task_attempt_log_directories(runs: Path, task_name: str) -> list[Path]:
    """Return dispatcher-owned attempt directories whose locator names the task."""
    matches: list[Path] = []
    if not runs.is_dir():
        return matches
    for attempt_dir in runs.iterdir():
        if not attempt_dir.is_dir():
            continue
        locator = attempt_dir / "locator.json"
        try:
            record = json.loads(locator.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            continue
        if record.get("task") != task_name:
            continue
        matches.append(attempt_dir)
    return matches


def cleanup_completed_task_attempt_logs(runs: Path, task_name: str) -> int:
    """Remove only dispatcher-owned logs for a task after its complete archive exists."""
    removed = 0
    for attempt_dir in task_attempt_log_directories(runs, task_name):
        try:
            shutil.rmtree(attempt_dir)
        except OSError as exc:
            attempt_event(
                "[attempt-log-cleanup-warning]",
                f"task={task_name} path={attempt_dir} error={exc}",
            )
            continue
        removed += 1
    return removed


def review_fingerprints(workspace: Path, task: Task) -> set[tuple[str, str]]:
    directories = [task.directory] if task.directory.is_dir() else []
    directories.extend(matching_archive_directories(workspace, task))
    fingerprints: set[tuple[str, str]] = set()
    for directory in directories:
        for path in directory.iterdir():
            if path.is_file() and (path.name.startswith("code_review_") or REVIEW_RE.match(path.name)):
                fingerprints.add((str(path.resolve()), sha256_file(path)))
    return fingerprints


def review_outcome(
    workspace: Path, task: Task, prior_fingerprints: set[tuple[str, str]]
) -> dict[str, str]:
    if task.directory.is_dir():
        directories = [task.directory]
        archives: list[Path] = []
    else:
        archives = matching_archive_directories(workspace, task)
        directories = list(archives)
    newest_directory: Path | None = None
    newest_log: Path | None = None
    newest_mtime = -1
    for directory in directories:
        logs = list(directory.glob("code_review_*.log"))
        logs.extend(path for path in directory.iterdir() if path.is_file() and REVIEW_RE.match(path.name))
        for log in logs:
            mtime = log.stat().st_mtime_ns
            fingerprint = (str(log.resolve()), sha256_file(log))
            if fingerprint not in prior_fingerprints and mtime > newest_mtime and read_verdict(log):
                newest_directory = directory
                newest_log = log
                newest_mtime = mtime
    verdict = read_verdict(newest_log) if newest_log else "UNKNOWN"
    if newest_directory in archives:
        state = "archived"
    elif newest_directory and (newest_directory / "USER_REVIEW.md").exists():
        state = "user-review"
    elif newest_directory and (newest_directory / "complete.log").exists():
        state = "complete-finalization"
    elif newest_log and REVIEW_RE.match(newest_log.name):
        state = "finalization-pending"
    elif newest_directory and any(REVIEW_RE.match(path.name) for path in newest_directory.iterdir() if path.is_file()):
        state = "follow-up"
    else:
        state = "changed"
    return {
        "verdict": verdict or "UNKNOWN",
        "state": state,
        "path": str(newest_directory or task.directory),
        "review_log": str(newest_log) if newest_log else "unknown",
    }


async def run_worker(
    workspace: Path,
    store: StateStore,
    task: Task,
    resume_locator: Path | None = None,
) -> None:
    retry_context = store.task_state(task).get("retry_failover_context")
    if resume_locator is None and isinstance(retry_context, dict):
        locator_value = retry_context.get("locator")
        if isinstance(locator_value, str) and locator_value:
            resume_locator = Path(locator_value)

    # If an active_locator already exists from a prior attempt that wrote its
    # locator but crashed before consuming the pending handoff, consume it now
    # to prevent a duplicate invocation. The locator write is the durable
    # commitment; the pending handoff is the logical intent. Consume the intent
    # when the commitment is already present.
    if isinstance(store, StateStore):
        prior_state = store.task_state(task)
        prior_active = prior_state.get("active_locator")
        prior_pending = prior_state.get("retry_failover_pending")
        if prior_active and prior_pending:
            consumed = False
            # Prefer handoff_id matching: read the stable identity from the
            # active locator file so we can match across crash boundaries
            # where the locator path changes.
            try:
                prior_locator_data = json.loads(Path(prior_active).read_text(encoding="utf-8"))
                if isinstance(prior_locator_data, dict):
                    handoff_id = prior_locator_data.get("retry_handoff_id")
                    if handoff_id:
                        consumed = store.commit_retry_handoff_locator(
                            task, handoff_id, prior_active,
                        )
            except (OSError, json.JSONDecodeError):
                pass
            if not consumed:
                # Fallback: match by locator path when handoff_id is
                # unavailable (e.g. crash between state update and locator
                # write, or pre-existing state from a prior dispatcher
                # version).
                consumed = store.consume_matching_retry_handoff(task, prior_active)

    try:
        decision, spec = persisted_execution_decision(
            store, task, stage="worker"
        )
    except ExecutionDecisionError as exc:
        store.update_task(task, blocked=str(exc))
        banner("작업차단", task.name, [f"reason={exc}"])
        return
    work_log = milestone_work_log_path(task)
    banner(
        "작업시작",
        task.name,
        [
            f"model={spec.display}",
            f"plan={task.plan.resolve()}",
            f"work_log={work_log.resolve()}",
            *task_observation_lines(task),
        ],
    )
    success, locator = await run_escalating(
        workspace,
        store,
        task,
        "worker",
        spec,
        initial_resume_locator=resume_locator,
    )
    if not success:
        current = store.task_state(task).get("blocked")
        store.update_task(
            task, blocked=current or f"worker failure locator={locator}"
        )
        return
    completed_spec = agent_spec_from_locator(locator) or spec
    try:
        _mark_worker_done(
            store,
            task,
            initial_decision=decision,
            worker_cli=completed_spec.cli,
            worker_model=completed_spec.model,
        )
    except ExecutionDecisionError as exc:
        store.update_task(task, worker_done=False, blocked=f"worker completion validation failed: {exc}")
        banner("작업차단", task.name, [f"reason={exc}"])
        return


def _require_same_runtime_identity(
    expected_spec: AgentSpec,
    worker_cli: str,
    worker_model: str,
) -> None:
    """Verify the worker CLI/model matches the validated completing decision spec.

    Ensures the actual worker that ran is the same runtime identity that the
    completing decision authorizes. Prevents a cloud-completed worker from
    being recorded as a native-session selfcheck target or vice versa.
    """
    if expected_spec.cli != worker_cli:
        raise ExecutionDecisionError(
            f"worker runtime CLI 불일치: expected={expected_spec.cli} actual={worker_cli}"
        )
    if expected_spec.model != worker_model:
        raise ExecutionDecisionError(
            f"worker runtime model 불일치: expected={expected_spec.model} actual={worker_model}"
        )


def _mark_worker_done(
    store: StateStore,
    task: Task,
    *,
    initial_decision: dict[str, Any],
    worker_cli: str,
    worker_model: str,
) -> None:
    """Persist worker completion with the authoritative completing decision.

    Uses the persisted execution_decisions worker entry as the sole authoritative
    source. Does not fall back to initial_decision even when the persisted
    decision is malformed—malformed persisted state blocks completion rather
    than silently reverting to a speculative initial decision.

    Validates the completing decision through the strict contract validator
    and verifies the worker CLI/model identity matches the normalized spec.
    On any validation failure, raises ExecutionDecisionError to prevent
    worker_done from being recorded.
    """
    decisions = store.task_state(task).get("execution_decisions", {})
    if not isinstance(decisions, dict) or "worker" not in decisions:
        raise ExecutionDecisionError(
            "persisted worker decision이 execution_decisions에 없다"
        )
    decision = decisions["worker"]
    if not isinstance(decision, dict):
        raise ExecutionDecisionError(
            "persisted worker decision이 dict가 아니다"
        )
    validated_decision, expected_spec = _validated_completing_decision(
        task, decision
    )
    _require_same_runtime_identity(expected_spec, worker_cli, worker_model)
    selected = validated_decision["selected"]
    execution_class = selected["execution_class"]
    store.update_task(
        task,
        worker_done=True,
        worker_cli=worker_cli,
        worker_model=worker_model,
        completing_decision=validated_decision,
        execution_class=execution_class,
        selfcheck_done=not selected["selfcheck_required"],
        review_no_progress=0,
        blocked=None,
    )


async def run_selfcheck(
    workspace: Path,
    store: StateStore,
    task: Task,
    resume_locator: Path | None = None,
) -> None:
    completing = store.task_state(task).get("completing_decision")
    if not isinstance(completing, dict):
        store.update_task(
            task, blocked="completing decision이 없어 selfcheck를 실행할 수 없다"
        )
        banner(
            "작업차단", task.name,
            ["reason=missing-completing-decision"],
        )
        return
    try:
        _completed_decision, spec = _validated_completing_decision(task, completing)
    except ExecutionDecisionError as exc:
        store.update_task(task, blocked=str(exc))
        banner("작업차단", task.name, [f"reason={exc}"])
        return
    if not spec.selfcheck_required:
        raise RuntimeError("selfcheck_required가 아닌 route에 selfcheck stage가 배정됐다")
    work_log = milestone_work_log_path(task)
    banner(
        "자가검증시작",
        task.name,
        [
            f"model={spec.display}",
            f"plan={task.plan.resolve()}",
            f"work_log={work_log.resolve()}",
            *task_observation_lines(task),
        ],
    )
    # 0 means the full pass is pending. After it fails the checklist gate,
    # each additional count represents one completed unchecked-item retry.
    incomplete_results = 0
    if isinstance(store, StateStore):
        incomplete_results = int(
            store.task_state(task).get("selfcheck_incomplete", 0)
        )
    incomplete_retries = max(0, incomplete_results - 1)
    if incomplete_retries >= SELF_CHECK_UNCHECKED_RETRY_LIMIT:
        locator = resume_locator
        reason = (
            "selfcheck unchecked-item retry limit already exhausted: "
            f"{incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}"
        )
        store.update_task(task, blocked=f"{reason} locator={locator}")
        banner(
            "작업차단",
            task.name,
            [
                "reason=selfcheck-incomplete-limit",
                "mode=unchecked-items",
                f"retry={incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}",
                f"locator={locator}",
            ],
        )
        return
    if incomplete_results > 0 and resume_locator is None:
        if not spec.native_resume:
            reason = "selfcheck retry에 필요한 native resume 계약이 target에 없다"
            store.update_task(task, blocked=reason)
            banner(
                "작업차단",
                task.name,
                ["reason=selfcheck-context-unavailable", reason],
            )
            return
        resume_locator, context_error = selfcheck_context_resume_locator(
            store.task_state(task),
            task,
            expected_workspace=store.workspace,
            expected_workspace_id=store.workspace_id,
            expected_runs_root=store.runs,
        )
        if resume_locator is None:
            reason = f"selfcheck context resume 실패: {context_error}"
            store.update_task(task, blocked=reason)
            banner(
                "작업차단",
                task.name,
                ["reason=selfcheck-context-unavailable", context_error],
            )
            return
    while True:
        unchecked_items = incomplete_results > 0
        success, locator = await run_escalating(
            workspace,
            store,
            task,
            "selfcheck",
            spec,
            initial_resume_locator=resume_locator,
            unchecked_items=unchecked_items,
        )
        if not success:
            current = store.task_state(task).get("blocked")
            store.update_task(
                task, blocked=current or f"selfcheck failure locator={locator}"
            )
            return
        errors = implementation_review_errors(task)
        if not errors:
            break
        if not spec.native_resume:
            reason = "selfcheck checklist가 미완료지만 target에 native resume 계약이 없다"
            store.update_task(task, blocked=reason)
            banner(
                "작업차단",
                task.name,
                ["reason=selfcheck-context-unavailable", reason],
            )
            return
        if locator is None:
            reason = "selfcheck 성공 locator가 없어 context를 이어갈 수 없다"
            store.update_task(task, blocked=reason)
            banner(
                "작업차단",
                task.name,
                ["reason=selfcheck-context-unavailable", reason],
            )
            return
        incomplete_results += 1
        incomplete_retries = max(0, incomplete_results - 1)
        if incomplete_retries >= SELF_CHECK_UNCHECKED_RETRY_LIMIT:
            reason = (
                "selfcheck checklist remains incomplete after unchecked-item retry: "
                f"{incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}"
            )
            store.update_task(
                task,
                blocked=f"{reason} locator={locator}",
                selfcheck_incomplete=incomplete_results,
                selfcheck_context_locator=str(locator),
            )
            banner(
                "작업차단",
                task.name,
                [
                    "reason=selfcheck-incomplete-limit",
                    f"detail={'; '.join(errors)}",
                    f"retry={incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}",
                    f"locator={locator}",
                ],
            )
            return
        store.update_task(
            task,
            selfcheck_incomplete=incomplete_results,
            selfcheck_context_locator=str(locator),
        )
        resume_locator = locator
        banner(
            "자가검증재시도",
            task.name,
            [
                f"reason={'; '.join(errors)}",
                "mode=unchecked-items",
                f"retry={incomplete_retries + 1}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}",
                f"locator={locator}",
            ],
        )
    store.update_task(
        task,
        selfcheck_done=True,
        selfcheck_incomplete=0,
        selfcheck_context_locator=None,
        blocked=None,
    )


async def run_review(
    workspace: Path,
    store: StateStore,
    task: Task,
    resume_locator: Path | None = None,
) -> str | None:
    try:
        _, spec = persisted_execution_decision(
            store, task, stage="review"
        )
    except ExecutionDecisionError as exc:
        store.update_task(task, blocked=str(exc))
        banner("작업차단", task.name, [f"reason={exc}"])
        return None
    state = store.task_state(task)
    prior_no_progress = int(state.get("review_no_progress", 0))
    if prior_no_progress >= REVIEW_NO_PROGRESS_LIMIT:
        locator = state.get("active_locator")
        reason = (
            "review no-progress limit already exhausted: "
            f"{prior_no_progress}/{REVIEW_NO_PROGRESS_LIMIT}"
        )
        store.update_task(task, blocked=f"{reason} locator={locator}")
        banner(
            "작업차단",
            task.name,
            [
                "reason=review-no-progress-limit",
                f"unchanged_review_attempts={prior_no_progress}/{REVIEW_NO_PROGRESS_LIMIT}",
                f"locator={locator}",
            ],
        )
        return None
    before = task_signature(workspace, task)
    prior_review_fingerprints = review_fingerprints(workspace, task)
    target = task.review.resolve() if task.review else task.directory.resolve()
    banner(
        "리뷰시작",
        task.name,
        [
            f"model={spec.display}",
            f"review={target}",
            *task_observation_lines(task),
        ],
    )
    success, locator = await run_escalating(
        workspace,
        store,
        task,
        "review",
        spec,
        initial_resume_locator=resume_locator,
    )
    if not success:
        current = store.task_state(task).get("blocked")
        store.update_task(
            task, blocked=current or f"review failure locator={locator}"
        )
        return None
    after = task_signature(workspace, task)
    if before == after:
        state = store.task_state(task)
        count = int(state.get("review_no_progress", 0)) + 1
        if count >= REVIEW_NO_PROGRESS_LIMIT:
            reason = (
                "review made no progress: "
                f"{count}/{REVIEW_NO_PROGRESS_LIMIT} locator={locator}"
            )
            store.update_task(
                task,
                review_no_progress=count,
                blocked=reason,
            )
            banner(
                "작업차단",
                task.name,
                [
                    "reason=review-no-progress-limit",
                    f"unchanged_review_attempts={count}/{REVIEW_NO_PROGRESS_LIMIT}",
                    f"locator={locator}",
                ],
            )
            return None
        store.update_task(task, review_no_progress=count)
        banner(
            "루프정체경고",
            task.name,
            [
                f"unchanged_review_attempts={count}/{REVIEW_NO_PROGRESS_LIMIT}",
                f"locator={locator}",
            ],
        )
        await asyncio.sleep(min(30, count * 5))
    else:
        store.update_task(task, review_no_progress=0, blocked=None)
        outcome = review_outcome(workspace, task, prior_review_fingerprints)
        banner(
            "리뷰결과",
            task.name,
            [
                f"verdict={outcome['verdict']}",
                f"state={outcome['state']}",
                f"path={outcome['path']}",
                f"review_log={outcome['review_log']}",
                f"locator={locator}",
            ],
        )
        if outcome["verdict"] == "PASS" and outcome["state"] == "archived":
            banner("작업완료", task.name, [f"archive={outcome['path']}", f"locator={locator}"])
            return outcome["path"]
        if outcome["verdict"] == "UNKNOWN" or outcome["state"] == "changed":
            raise RuntimeError(
                "official review가 판정과 다음 파일 상태를 materialize하지 않았다; "
                f"locator={locator}"
            )
        if outcome["state"] == "archived":
            raise RuntimeError(
                f"PASS가 아닌 review가 완료 archive로 이동했다: "
                f"verdict={outcome['verdict']} locator={locator}"
            )
    return None


def status_lines(
    task: Task,
    stage: str,
    dependency: str,
    decision: dict[str, Any] | None = None,
) -> list[str]:
    route = f"{task.lane}-G{task.grade:02d}" if task.lane and task.grade else "recovery"
    base = [f"stage={stage}", f"route={route}", f"dependency={dependency}"]
    if decision is not None:
        return base + selector_evidence_lines(decision)
    return base


def select_dispatch_candidates(
    store: StateStore,
    ready: list[tuple[Task, str]],
    *,
    persist: bool,
    available_slots: int | None = None,
) -> tuple[
    list[tuple[Task, str]],
    list[tuple[Task, str, str]],
    str,
]:
    ready_reviews = [(task, stage) for task, stage in ready if stage == "review"]
    ready_workers = [(task, stage) for task, stage in ready if stage in {"worker", "selfcheck"}]
    ordered = ready_reviews + ready_workers
    claims = store.write_claim_snapshot()
    selected: list[tuple[Task, str]] = []
    deferred: list[tuple[Task, str, str]] = []
    timestamp = now_iso()
    for task, stage in ordered:
        if not task.write_set_known or not task.write_set:
            deferred.append(
                (
                    task,
                    stage,
                    "valid non-empty Modified Files Summary write claim이 필요하다",
                )
            )
            continue
        requested = sorted(task.write_set)
        invalid_path: str | None = None
        for raw_path in requested:
            path = Path(raw_path)
            resolved = path.resolve()
            try:
                resolved.relative_to(store.workspace)
            except ValueError:
                invalid_path = raw_path
                break
            if (
                not path.is_absolute()
                or str(resolved) != raw_path
                or resolved == store.workspace
            ):
                invalid_path = raw_path
                break
        if invalid_path is not None:
            deferred.append(
                (
                    task,
                    stage,
                    f"write claim 경로가 canonical workspace file이 아니다: {invalid_path}",
                )
            )
            continue

        conflict: tuple[str, str] | None = None
        requested_set = set(requested)
        for owner in sorted(claims):
            if owner == task.name:
                continue
            other = claims[owner]
            if other.get("exclusive"):
                conflict = (owner, "<exclusive-workspace-claim>")
                break
            intersection = sorted(requested_set & set(other.get("paths", [])))
            if intersection:
                conflict = (owner, intersection[0])
                break
        if conflict is not None:
            owner, path = conflict
            deferred.append(
                (
                    task,
                    stage,
                    f"write claim 충돌 대기: owner={owner}; path={path}",
                )
            )
            continue

        # Capacity-only admission: admit and acquire/replace a claim only
        # while a slot remains.  A newly capacity-deferred task gets a stable
        # wait reason and no new claim; a task that already owns its lifecycle
        # claim keeps it unchanged while waiting.
        if available_slots is not None and len(selected) >= available_slots:
            if task.name in claims:
                deferred.append(
                    (
                        task,
                        stage,
                        f"capacity waiting: limit reached (selected={len(selected)}/{available_slots})",
                    )
                )
            else:
                deferred.append(
                    (
                        task,
                        stage,
                        f"capacity waiting: limit reached (selected={len(selected)}/{available_slots})",
                    )
                )
            continue

        previous = claims.get(task.name, {})
        claims[task.name] = {
            "task": task.name,
            "plan_hash": task.plan_hash,
            "paths": requested,
            "exclusive": False,
            "workspace_id": store.workspace_id,
            "acquired_at": previous.get("acquired_at") or timestamp,
            "updated_at": timestamp,
            "source": "plan",
        }
        selected.append((task, stage))

    if persist:
        store.replace_write_claims(claims, persist=True)
    return selected, deferred, ""


def ensure_review_shared_state(workspace: Path) -> None:
    helper = workspace / "agent-ops" / "bin" / "ai-ignore.sh"
    if not helper.is_file():
        raise RuntimeError(f"review shared-state helper가 없다: {helper}")
    command = [
        "bash",
        "-c",
        'source "$1" && agent_ops_ensure_gitignore_task_artifact_block "$2"',
        "agent-task-review-preflight",
        str(helper),
        str(workspace / ".gitignore"),
    ]
    completed = subprocess.run(
        command,
        cwd=workspace,
        capture_output=True,
        text=True,
        check=False,
    )
    if completed.returncode != 0:
        diagnostic = (completed.stderr or completed.stdout).strip()
        raise RuntimeError(
            f"review shared-state preflight 실패: {diagnostic or completed.returncode}"
        )


async def dispatch(args: argparse.Namespace) -> int:
    workspace = Path(args.workspace).resolve()
    store = StateStore(workspace)
    try:
        try:
            return await dispatch_with_store(args, workspace, store)
        except Exception as exc:
            # A scheduler/control-plane exception must not make asyncio.run()
            # cancel already-running agent attempts.  Keep this loop alive until
            # every owned background task finishes naturally; the next
            # dispatcher run reconciles their file/state results.
            current = asyncio.current_task()
            active = [
                task
                for task in asyncio.all_tasks()
                if task is not current and not task.done()
            ]
            if active:
                banner(
                    "디스패처복구대기",
                    args.task_group or "agent-task",
                    [
                        f"running_async_tasks={len(active)}",
                        "scheduler 예외와 무관하게 실행 중 agent를 자연 종료까지 추적",
                    ],
                )
                await asyncio.gather(*active, return_exceptions=True)
                raise DispatcherInterruptedWithActiveWork(
                    f"running agent가 있던 scheduler 예외: {exc}"
                ) from exc
            raise
    finally:
        store.close()


async def dispatch_with_store(
    args: argparse.Namespace,
    workspace: Path,
    store: StateStore,
) -> int:
    orchestration_scope = args.task_group or "__all__"
    if args.retry_blocked and not args.dry_run:
        store.mark_retry_failover(args.task_group)
    running: dict[str, asyncio.Task[str | None]] = {}
    last_wait: dict[str, str] = {}
    completed_tasks: dict[str, str] = {}
    fatal_errors: dict[str, str] = {}
    control_plane_errors: dict[str, str] = {}
    work_log_archive_errors: dict[str, str] = {}
    review_shared_state_ready = False
    candidate_scope: set[str] | None = None
    task_cache: dict[str, Task] | None = None
    resume_locators: dict[str, Path] = {}
    live_external_processes: dict[str, str] = {}
    capacity_waiting: set[str] = set()
    max_parallel = validated_max_parallel(
        getattr(args, "max_parallel", DEFAULT_MAX_PARALLEL)
    )
    catalog_revision_accepted = False

    while True:
        if task_cache is None:
            tasks = scan_tasks(workspace, args.task_group)
            task_cache = {task.name: task for task in tasks}
        else:
            tasks = sorted(task_cache.values(), key=lambda task: (task.index, task.name))
        if (
            getattr(args, "accept_catalog_revision", False)
            and not catalog_revision_accepted
        ):
            selector = _selector_module()
            catalog = selector.load_runtime_catalog(EXECUTION_CATALOG_PATH)
            accepted = store.accept_catalog_revision(
                tasks,
                catalog_revision=catalog.revision,
                catalog_source=str(catalog.source),
                persist=not args.dry_run,
            )
            banner(
                "카탈로그재선택",
                args.task_group,
                [
                    f"mode={'preview' if args.dry_run else 'accepted'}",
                    f"catalog_revision={catalog.revision}",
                    f"tasks={','.join(accepted) if accepted else 'none'}",
                ],
            )
            catalog_revision_accepted = True
        if args.dry_run:
            persistent_errors: dict[str, str] = {}
            observed_tasks: set[str] = set()
            live_external_processes = {}
        else:
            store.prepare_orchestration(orchestration_scope, tasks, workspace)
            live_external_processes = orchestration_live_agent_processes(
                store,
                orchestration_scope,
            )
            active_or_running = (
                {task.name for task in tasks}
                | set(running)
                | set(live_external_processes)
            )
            reconciled_completed, persistent_errors = store.reconcile_orchestration(
                orchestration_scope,
                workspace,
                active_or_running,
            )
            completed_tasks.update(reconciled_completed)
            for task_name in persistent_errors:
                completed_tasks.pop(task_name, None)
            observed_tasks = store.orchestration_tasks(orchestration_scope)
            work_log_archives, work_log_archive_errors = (
                archive_completed_group_work_logs(
                    workspace,
                    observed_tasks,
                    completed_tasks,
                    active_or_running,
                )
            )
            for task_group, archive in sorted(work_log_archives.items()):
                banner(
                    "작업로그아카이브",
                    task_group,
                    [f"archive={archive}"],
                )
        if not tasks and not running:
            if live_external_processes:
                for task_name, detail in sorted(
                    live_external_processes.items()
                ):
                    banner(
                        "작업수행중",
                        task_name,
                        [
                            "이전 dispatcher의 model process를 종료시키지 않고 추적",
                            detail,
                        ],
                    )
                await asyncio.sleep(STREAM_HEARTBEAT_SECONDS)
                continue
            if control_plane_errors:
                banner(
                    "디스패치추적대기",
                    args.task_group or "agent-task",
                    [
                        "예상하지 못한 dispatcher 중단 결과를 재조정해야 함",
                        *(
                            f"interrupted[{name}]={reason}"
                            for name, reason in sorted(control_plane_errors.items())
                        ),
                    ],
                )
                return 3
            if work_log_archive_errors:
                banner(
                    "디스패치추적대기",
                    args.task_group or "agent-task",
                    [
                        "완료 task group의 WORK_LOG archive를 재시도해야 함",
                        *(
                            f"work-log-archive[{group}]={reason}"
                            for group, reason in sorted(
                                work_log_archive_errors.items()
                            )
                        ),
                    ],
                )
                return 3
            if args.task_group and not observed_tasks and not completed_tasks:
                reason = (
                    "명시한 task group에서 관찰된 active task나 "
                    "검증된 complete.log 이력이 없다"
                )
                if not args.dry_run:
                    store.mark_orchestration_blocked(orchestration_scope, {})
                banner(
                    "디스패치차단",
                    args.task_group,
                    [f"reason=unobserved-task-group", reason],
                )
                return 2
            pending_attempt_logs = {
                name: paths
                for name in completed_tasks
                if (
                    paths := task_attempt_log_directories(
                        store.runs,
                        name,
                    )
                )
            }
            if pending_attempt_logs:
                banner(
                    "디스패치추적대기",
                    args.task_group or "agent-task",
                    [
                        "완료 task의 attempt 로그 정리가 아직 끝나지 않음",
                        *(
                            f"attempt-log-cleanup-pending[{name}]="
                            + ",".join(str(path) for path in paths)
                            for name, paths in sorted(pending_attempt_logs.items())
                        ),
                    ],
                )
                return 3
            incomplete = sorted(observed_tasks - completed_tasks.keys())
            if incomplete or fatal_errors or persistent_errors:
                details = [
                    *(f"incomplete={name}" for name in incomplete),
                    *(
                        f"persistent[{name}]={reason}"
                        for name, reason in sorted(persistent_errors.items())
                    ),
                    *(f"error[{name}]={reason}" for name, reason in sorted(fatal_errors.items())),
                ]
                if not args.dry_run:
                    store.mark_orchestration_blocked(
                        orchestration_scope,
                        {
                            name: (
                                "blocked",
                                persistent_errors.get(name)
                                or fatal_errors.get(name)
                                or "관찰된 task가 완료되지 않았다",
                            )
                            for name in (
                                set(incomplete)
                                | set(fatal_errors)
                                | set(persistent_errors)
                            )
                        },
                    )
                banner("디스패치차단", args.task_group or "agent-task", details)
                return 2
            if not args.dry_run:
                store.mark_orchestration_complete(orchestration_scope)
            banner(
                "작업완료",
                args.task_group or "agent-task",
                [
                    "active task 없음",
                    f"verified_complete_tasks={len(completed_tasks)}",
                    *(f"complete[{name}]={path}" for name, path in sorted(completed_tasks.items())),
                ],
            )
            return 0

        task_by_name = {task.name: task for task in tasks}
        finished_names: set[str] = set()
        complete_log_created = False
        for name, future in list(running.items()):
            if not future.done():
                continue
            finished_names.add(name)
            completed_archive: str | None = None
            try:
                completed_archive = future.result()
                if completed_archive:
                    complete_log_created = True
                    store.mark_orchestration_task_complete(
                        orchestration_scope, name, completed_archive
                    )
                    completed_tasks[name] = completed_archive
            except Exception as exc:  # keep other independent tasks alive
                banner(
                    "디스패치추적대기",
                    name,
                    [f"agent coroutine exception={exc}"],
                )
                control_plane_errors[name] = str(exc)
            task = task_by_name.get(name)
            if task:
                store.clear_active(task)
                if not completed_archive:
                    refreshed = read_task_directory(workspace, task.directory)
                    if refreshed is None:
                        task_cache.pop(name, None)
                    else:
                        task_cache[name] = refreshed
            if completed_archive:
                task_cache.pop(name, None)
            del running[name]
        if finished_names:
            if complete_log_created:
                # Running reviewers may be archiving their own active directory
                # while this completion-triggered scan runs. Preserve their
                # already-loaded Task snapshots and do not reread those mutable
                # directories until their futures finish.
                running_snapshots = {
                    name: task_cache[name]
                    for name in running
                    if name in task_cache
                }
                tasks = scan_tasks(
                    workspace,
                    args.task_group,
                    exclude_names=set(running),
                )
                task_cache = {
                    **{task.name: task for task in tasks},
                    **running_snapshots,
                }
                tasks = sorted(
                    task_cache.values(),
                    key=lambda task: (task.index, task.name),
                )
                store.prepare_orchestration(orchestration_scope, tasks, workspace)
                live_external_processes = orchestration_live_agent_processes(
                    store,
                    orchestration_scope,
                )
                active_or_running = (
                    {task.name for task in tasks}
                    | set(running)
                    | set(live_external_processes)
                )
                reconciled_completed, persistent_errors = store.reconcile_orchestration(
                    orchestration_scope,
                    workspace,
                    active_or_running,
                )
                completed_tasks.update(reconciled_completed)
                for task_name in persistent_errors:
                    completed_tasks.pop(task_name, None)
                observed_tasks = store.orchestration_tasks(orchestration_scope)
                work_log_archives, work_log_archive_errors = (
                    archive_completed_group_work_logs(
                        workspace,
                        observed_tasks,
                        completed_tasks,
                        active_or_running,
                    )
                )
                for task_group, archive in sorted(
                    work_log_archives.items()
                ):
                    banner(
                        "작업로그아카이브",
                        task_group,
                        [f"archive={archive}"],
                    )
                candidate_scope = None
            else:
                tasks = sorted(task_cache.values(), key=lambda task: (task.index, task.name))
                candidate_scope = finished_names | capacity_waiting
                capacity_waiting = set()

        if not tasks and not running:
            # A completion-triggered full scan may have removed the last active task.
            continue

        # Derive workspace-global capacity.  Count unique task names across
        # current running futures and same-workspace live/conservative evidence,
        # regardless of --task-group.  Do not count pump/heartbeat/selector/
        # selector coroutines as extra slots.
        workspace_live = workspace_live_agent_processes(store)
        workspace_live = {
            name: detail
            for name, detail in workspace_live.items()
            if name not in finished_names
        }
        occupied_names = set(running) | set(workspace_live)
        available_slots: int | None = (
            None if max_parallel == 0 else max(0, max_parallel - len(occupied_names))
        )

        ready: list[tuple[Task, str]] = []
        waiting_tasks: list[str] = []
        externally_active: list[tuple[Task, str]] = []
        blocked_details: dict[str, tuple[str, str, str]] = {}
        for task in tasks:
            # A running future owns this task directory.  Re-reading its review
            # or dependency files can race with review finalization/archive and
            # must never interrupt unrelated tasks.
            if task.name in running:
                continue
            if task.name in control_plane_errors:
                reason = (
                    "예상하지 못한 agent coroutine 중단 결과를 "
                    "다음 dispatcher가 재조정해야 함: "
                    f"{control_plane_errors[task.name]}"
                )
                blocked_details[task.name] = (
                    "디스패치추적대기",
                    "interrupted",
                    reason,
                )
                waiting_tasks.append(task.name)
                continue
            state = store.peek_task_state(task) if args.dry_run else store.task_state(task)
            active_predecessors = live_predecessors(
                task,
                set(running) | set(live_external_processes),
            )
            if active_predecessors:
                dependency_ready = False
                dependency = (
                    "predecessor FINISH 대기: "
                    + ",".join(active_predecessors)
                )
            else:
                dependency_ready, dependency = dependency_state(
                    workspace,
                    task,
                )
            stage = task_stage(task, state)
            if state.get("active_stage"):
                active_stage = str(state["active_stage"])
                active_live, active_detail = external_active_is_live(
                    state,
                    expected_workspace=store.workspace,
                    expected_workspace_id=store.workspace_id,
                    expected_runs_root=store.runs,
                )
                if active_live:
                    reason = f"외부 실행중: stage={active_stage}; {active_detail}"
                    active_key = (
                        f"active|{active_stage}|{state.get('active_locator') or 'unknown'}"
                    )
                    externally_active.append((task, active_stage))
                    blocked_details[task.name] = ("작업중", stage, reason)
                    if not args.dry_run:
                        store.adopt_active_write_claim(task)
                    if not args.dry_run and last_wait.get(task.name) != active_key:
                        banner("작업중", task.name, status_lines(task, stage, reason))
                        last_wait[task.name] = active_key
                    continue
                if not args.dry_run:
                    resume_locator = native_resume_locator(
                        state,
                        expected_workspace=store.workspace,
                        expected_workspace_id=store.workspace_id,
                        expected_runs_root=store.runs,
                    )
                    if resume_locator is not None:
                        resume_locators[task.name] = resume_locator
                    banner(
                        "작업복구",
                        task.name,
                        status_lines(task, stage, f"stale active 제외: {active_detail}"),
                    )
                    store.clear_active(task)
                    state = store.task_state(task)
            reason = ""
            if task.errors:
                reason = "; ".join(task.errors)
            elif task.user_review:
                blocking, detail = user_review_blocker_state(task.user_review)
                if task.plan is not None or task.review is not None:
                    blocking = False
                    detail = "active PLAN/CODE_REVIEW와 공존한다"
                reason = (
                    f"USER_REVIEW 대기: {task.user_review}; {detail}"
                    if blocking
                    else (
                        "USER_REVIEW stop 계약 불충족: "
                        f"{task.user_review}; {detail}"
                    )
                )
            elif state.get("blocked"):
                reason = str(state["blocked"])
            elif not dependency_ready:
                reason = dependency
            if reason:
                wait_key = f"{stage}|{reason}"
                event = (
                    "작업차단"
                    if (
                        task.errors
                        or stage in {"blocked", "user-review"}
                        or state.get("blocked")
                    )
                    else "작업대기"
                )
                blocked_details[task.name] = (event, stage, reason)
                if not args.dry_run and last_wait.get(task.name) != wait_key:
                    banner(event, task.name, status_lines(task, stage, reason))
                    last_wait[task.name] = wait_key
                waiting_tasks.append(task.name)
                continue
            if task.name not in running and (
                candidate_scope is None or task.name in candidate_scope
            ):
                ready.append((task, stage))

        admission_time = datetime.now(KST)
        if args.dry_run:
            candidates, deferred, _ = select_dispatch_candidates(
                store,
                ready,
                persist=False,
                available_slots=available_slots,
            )
            for task, stage, reason in deferred:
                event = (
                    "작업차단"
                    if reason.startswith(
                        (
                            "valid non-empty",
                            "write claim 경로가",
                        )
                    )
                    else "작업대기"
                )
                blocked_details[task.name] = (event, stage, reason)
                waiting_tasks.append(task.name)
            ready_by_name = {task.name: stage for task, stage in candidates}
            for task in tasks:
                if task.name in ready_by_name:
                    stage = ready_by_name[task.name]
                    selector_stage = "review" if stage == "review" else "worker"
                    preview_state = store.peek_task_state(task)
                    decisions = preview_state.get("execution_decisions", {})
                    prior_decision = (
                        decisions.get(selector_stage)
                        if isinstance(decisions, dict)
                        else None
                    )
                    decision = read_or_preview_stage_decision(
                        task,
                        preview_state,
                        stage=selector_stage,
                        dry_run=args.dry_run,
                    )
                    spec = agent_spec_from_decision(decision)
                    lines = status_lines(task, stage, "ready", decision=decision)
                    banner(
                        "작업대기",
                        task.name,
                        lines + [f"model={spec.display}"],
                    )
                else:
                    event, stage, reason = blocked_details[task.name]
                    banner(event, task.name, status_lines(task, stage, reason))
            return 2 if waiting_tasks and not candidates else 0

        candidates, deferred, _ = select_dispatch_candidates(
            store,
            ready,
            persist=True,
            available_slots=available_slots,
        )
        for task, stage, reason in deferred:
            event = (
                "작업차단"
                if reason.startswith(
                    (
                        "valid non-empty",
                        "write claim 경로가",
                    )
                )
                else "작업대기"
            )
            blocked_details[task.name] = (event, stage, reason)
            waiting_tasks.append(task.name)
            wait_key = f"{stage}|{reason}"
            if last_wait.get(task.name) != wait_key:
                banner(event, task.name, status_lines(task, stage, reason))
                last_wait[task.name] = wait_key

        # Rebuild capacity_waiting from current capacity-only deferrals.
        # Dependency, blocker, invalid-write-set, and claim-collision deferrals
        # are not capacity waiters and rely on their existing wake-up event.
        if available_slots is not None:
            capacity_waiting = {
                task.name
                for task, stage, reason in deferred
                if reason.startswith("capacity waiting:")
            }

        if (
            not review_shared_state_ready
            and any(stage == "review" for _, stage in candidates)
        ):
            try:
                ensure_review_shared_state(workspace)
            except (OSError, RuntimeError) as exc:
                # Shared review setup is a blocker only for reviews.  It must
                # not prevent dependency-independent workers/selfchecks from
                # starting and draining in the same scheduler pass.
                review_removed: list[tuple[Task, str]] = []
                remaining_candidates: list[tuple[Task, str]] = []
                for task, stage in candidates:
                    if stage != "review":
                        remaining_candidates.append((task, stage))
                        continue
                    reason = f"review shared-state preflight failed: {exc}"
                    store.update_task(task, blocked=reason)
                    fatal_errors[task.name] = reason
                    if task.name not in waiting_tasks:
                        waiting_tasks.append(task.name)
                    blocked_details[task.name] = (
                        "작업차단",
                        stage,
                        reason,
                    )
                    banner(
                        "작업차단",
                        task.name,
                        status_lines(task, stage, reason),
                    )
                    review_removed.append((task, stage))
                candidates = remaining_candidates

                # Block every ready review that was deferred (e.g. by capacity).
                for task, stage, _ in deferred:
                    if stage == "review":
                        reason = f"review shared-state preflight failed: {exc}"
                        store.update_task(task, blocked=reason)
                        fatal_errors[task.name] = reason
                        if task.name not in waiting_tasks:
                            waiting_tasks.append(task.name)
                        blocked_details[task.name] = (
                            "작업차단",
                            stage,
                            reason,
                        )
                        banner(
                            "작업차단",
                            task.name,
                            status_lines(task, stage, reason),
                        )

                # Refill freed runtime slots from disjoint non-review capacity
                # waiters in stable deferred order using persistent claim admission.
                # Reviews that had received slots retain their existing claims;
                # reviews that never received slots do not synthesize claims.
                if available_slots is not None and review_removed:
                    freed_slots = len(review_removed)
                    refill_inputs = [
                        (task, stage)
                        for task, stage, reason in deferred
                        if stage in {"worker", "selfcheck"}
                        and reason.startswith("capacity waiting:")
                    ]
                    if refill_inputs:
                        refilled, refill_deferred, _ = select_dispatch_candidates(
                            store,
                            refill_inputs,
                            persist=True,
                            available_slots=freed_slots,
                        )
                        candidates.extend(refilled)
                        for task, stage, reason in refill_deferred:
                            event = (
                                "작업차단"
                                if reason.startswith(
                                    (
                                        "valid non-empty",
                                        "write claim 경로가",
                                    )
                                )
                                else "작업대기"
                            )
                            blocked_details[task.name] = (event, stage, reason)
                            wait_key = f"{stage}|{reason}"
                            if last_wait.get(task.name) != wait_key:
                                banner(event, task.name, status_lines(task, stage, reason))
                                last_wait[task.name] = wait_key
                        capacity_waiting = {
                            task.name
                            for task, stage, reason in refill_deferred
                            if reason.startswith("capacity waiting:")
                        }
                    else:
                        capacity_waiting = set()

            else:
                review_shared_state_ready = True

        scheduled = False
        for task, stage in candidates:
            store.mark_active(task, stage)
            resume_locator = resume_locators.pop(task.name, None)
            state = store.task_state(task)
            if stage == "review":
                future = asyncio.create_task(
                    run_review(
                        workspace,
                        store,
                        task,
                        **(
                            {"resume_locator": resume_locator}
                            if resume_locator is not None
                            else {}
                        ),
                    )
                )
            elif stage == "selfcheck":
                future = asyncio.create_task(
                    run_selfcheck(
                        workspace,
                        store,
                        task,
                        **(
                            {"resume_locator": resume_locator}
                            if resume_locator is not None
                            else {}
                        ),
                    )
                )
            else:
                future = asyncio.create_task(
                    run_worker(
                        workspace,
                        store,
                        task,
                        **(
                            {"resume_locator": resume_locator}
                            if resume_locator is not None
                            else {}
                        ),
                    )
                )
            running[task.name] = future
            last_wait.pop(task.name, None)
            scheduled = True

        if running:
            await asyncio.wait(running.values(), return_when=asyncio.FIRST_COMPLETED)
            continue
        if work_log_archive_errors:
            banner(
                "디스패치추적대기",
                args.task_group or "agent-task",
                [
                    "완료 task group의 WORK_LOG archive를 재시도해야 함",
                    *(
                        f"work-log-archive[{group}]={reason}"
                        for group, reason in sorted(
                            work_log_archive_errors.items()
                        )
                    ),
                ],
            )
            return 3
        if externally_active:
            banner(
                "디스패치추적대기",
                args.task_group or "agent-task",
                ["새 실행 후보 없음", "active task는 caller가 계속 추적"],
            )
            return 3
        if control_plane_errors:
            banner(
                "디스패치추적대기",
                args.task_group or "agent-task",
                [
                    "실행 중이던 독립 작업을 모두 소진했고 재조정이 필요함",
                    *(
                        f"interrupted[{name}]={reason}"
                        for name, reason in sorted(control_plane_errors.items())
                    ),
                ],
            )
            return 3
        # Capacity-only wait: external live attempts fill the cap, but we must
        # not mark the orchestration blocked.  Report as non-terminal tracking
        # state and return 3 so a restart with a larger limit or after occupancy
        # drops can resume naturally.
        if capacity_waiting and not scheduled:
            external_fillers = set(occupied_names) - set(running)
            if external_fillers:
                banner(
                    "디스패치추적대기",
                    args.task_group or "agent-task",
                    [
                        f"capacity_waiting={','.join(sorted(capacity_waiting))}",
                        f"occupied_by_external={','.join(sorted(external_fillers))}",
                        f"max_parallel={max_parallel}",
                        "용량 대기: 외부 실행이 용량을 채워 다음 dispatcher가 재조정",
                    ],
                )
                return 3
        if not scheduled:
            store.mark_orchestration_blocked(
                orchestration_scope,
                {
                    name: (
                        "blocked" if event == "작업차단" else "waiting",
                        detail,
                    )
                    for name, (event, _, detail) in blocked_details.items()
                },
            )
            reason = f"waiting={','.join(sorted(waiting_tasks))}"
            banner(
                "디스패치차단",
                args.task_group or "agent-task",
                [
                    "실행 가능한 독립 작업을 모두 소진함",
                    reason,
                    f"verified_complete_tasks={len(completed_tasks)}",
                    *(
                        f"complete[{name}]={path}"
                        for name, path in sorted(completed_tasks.items())
                    ),
                    *(
                        f"{name}: stage={stage}; reason={detail}"
                        for name, (_, stage, detail) in sorted(
                            blocked_details.items()
                        )
                    ),
                ],
            )
            return 2


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--workspace", default=".", help="repository root (default: current directory)")
    parser.add_argument("--task-group", help="run only agent-task/<task_group>")
    parser.add_argument(
        "--execution-catalog",
        help=(
            "runtime agent/model catalog JSON override; alternatively set "
            "AGENT_TASK_EXECUTION_CATALOG; defaults to the bundled catalog"
        ),
    )
    parser.add_argument("--dry-run", action="store_true", help="classify and print without launching CLIs")
    parser.add_argument("--retry-blocked", action="store_true", help="clear dispatcher-local blocked state")
    parser.add_argument(
        "--accept-catalog-revision",
        action="store_true",
        help=(
            "explicitly replace stale, inactive, incomplete worker selection "
            "for the required --task-group after a catalog update"
        ),
    )
    parser.add_argument(
        "--max-parallel",
        type=int,
        default=DEFAULT_MAX_PARALLEL,
        metavar="MAX_PARALLEL",
        help=(
            "physical-workspace global cap on unique active task-stage "
            f"attempts; default is {DEFAULT_MAX_PARALLEL}; 0 is unlimited"
        ),
    )
    parser.add_argument(
        "--validate-plan",
        metavar="PATH",
        help="validate one PLAN Modified Files Summary without starting the dispatcher",
    )
    return parser.parse_args()


def main() -> int:
    global EXECUTION_CATALOG_PATH
    args = parse_args()
    if args.accept_catalog_revision and not args.task_group:
        print(
            "dispatcher error: --accept-catalog-revision requires --task-group",
            file=sys.stderr,
        )
        return 2
    try:
        validated_max_parallel(
            getattr(args, "max_parallel", DEFAULT_MAX_PARALLEL)
        )
    except ValueError as exc:
        print(f"dispatcher error: {exc}", file=sys.stderr)
        return 2
    validate_plan = getattr(args, "validate_plan", None)
    if validate_plan:
        workspace = Path(args.workspace).resolve()
        candidate = Path(validate_plan)
        if not candidate.is_absolute():
            candidate = (Path.cwd() / candidate).resolve()
        metadata_diagnostics = validate_plan_metadata(candidate, workspace)
        if metadata_diagnostics:
            for diagnostic in metadata_diagnostics:
                print(f"plan metadata error: {diagnostic}", file=sys.stderr)
            return 2
        write_set, diagnostics = inspect_write_set(candidate, workspace)
        if diagnostics:
            for diagnostic in diagnostics:
                print(f"plan write-set error: {diagnostic}", file=sys.stderr)
            return 2
        for path in sorted(write_set):
            validation_claim(path)
        return 0
    try:
        selector = _selector_module()
        EXECUTION_CATALOG_PATH = selector.resolve_catalog_path(
            getattr(args, "execution_catalog", None)
        )
        preflight_execution_catalog(
            EXECUTION_CATALOG_PATH,
            workspace=Path(args.workspace).resolve(),
            run_commands=not args.dry_run,
        )
    except Exception as exc:
        code = getattr(exc, "code", exc.__class__.__name__)
        print(f"dispatcher catalog error [{code}]: {exc}", file=sys.stderr)
        return 2
    if os.environ.get(AGENT_PROCESS_MARKER_ENV):
        print(
            "nested dispatcher invocation rejected: this process is already a "
            "dispatcher child; continue the assigned role directly and do not "
            "wait for the parent dispatcher",
            file=sys.stderr,
        )
        return 4
    try:
        return asyncio.run(dispatch(args))
    except KeyboardInterrupt:
        print("\n중단됨", file=sys.stderr)
        return 130
    except DispatcherAlreadyRunning as exc:
        print(f"dispatcher active: {exc}", file=sys.stderr)
        return 3
    except DispatcherTerminalStateError as exc:
        print(f"dispatcher error: {exc}", file=sys.stderr)
        return 2
    except Exception as exc:
        # An unexpected dispatcher failure is not proof that the task group is
        # drained or terminal.  The caller must inspect active PIDs/locators and
        # recover instead of treating it like exit 2.
        print(f"dispatcher interrupted: {exc}", file=sys.stderr)
        return 3


if __name__ == "__main__":
    raise SystemExit(main())
