#!/usr/bin/env python3
"""Dispatch active agent-task PLAN/CODE_REVIEW pairs until completion or a hard blocker."""

from __future__ import annotations

import argparse
import asyncio
import fcntl
import hashlib
import json
import os
import re
import sys
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


SEP = "-" * 42
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$")
SUBTASK_RE = re.compile(r"^(?P<index>\d{2})(?:\+(?P<deps>\d{2}(?:,\d{2})*))?_[a-z0-9_]+$")
VERDICT_RE = re.compile(r"종합 판정\s*:\s*(PASS|WARN|FAIL)")
PLAN_IDENTITY_RE = re.compile(
    r"<!--\s+task=(?P<task>\S+)\s+plan=(?P<plan>\d+)\s+tag=(?P<tag>\S+)\s+-->"
)
FINAL_CHECK_RE = re.compile(r"^-\s+\[[xX]\]\s+CODE_REVIEW-\*-G\?\?\.md", re.MULTILINE)
PROMOTABLE_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"usage limit", r"capacity limit",
    ],
    "model-unavailable": [
        r"model.{0,40}(?:not found|unavailable)", r"overloaded",
        r"temporarily unavailable",
    ],
}


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


def banner(event: str, task: str, lines: list[str] | None = None) -> None:
    display_task = task.rsplit("/", 1)[-1]
    print(SEP, flush=True)
    print(f"{event}: {display_task}", flush=True)
    print(SEP, flush=True)
    if display_task != task:
        print(f"task={task}", flush=True)
    for line in lines or []:
        print(line, flush=True)


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()


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)
    identity = "\0".join(match.group(name) for name in ("task", "plan", "tag"))
    return "meta:" + hashlib.sha256(identity.encode()).hexdigest()


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 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
    local_pi: bool = False


@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


class StateStore:
    def __init__(self, workspace: Path):
        workspace_id = hashlib.sha256(str(workspace).encode()).hexdigest()[:16]
        git_marker = 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 (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" / 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 RuntimeError(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:
            raise RuntimeError(f"같은 workspace의 dispatcher가 이미 실행 중이다: {self.root}") from exc
        if self.path.exists():
            try:
                self.data = json.loads(self.path.read_text(encoding="utf-8"))
            except (json.JSONDecodeError, OSError):
                self.data = {"tasks": {}, "attempt_counters": {}}
        else:
            self.data = {"tasks": {}, "attempt_counters": {}}

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

    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,
                "review_no_progress": 0,
            }
            tasks[task.name] = current
            self.save()
        return current

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

    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) -> None:
        for value in self.data.get("tasks", {}).values():
            value["blocked"] = None
        self.save()


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 extract_write_set(plan: Path | None) -> tuple[set[str], bool]:
    if plan is None or not plan.exists():
        return set(), False
    text = plan.read_text(encoding="utf-8", errors="replace")
    match = re.search(r"^## 수정 파일 요약\s*$([\s\S]*?)(?=^##\s|\Z)", text, re.MULTILINE)
    if not match:
        return set(), False
    result: set[str] = set()
    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
        for value in re.findall(r"`([^`]+)`", cells[0]):
            normalized = re.sub(r":\d+(?::\d+)?$", "", value.strip())
            if normalized and not normalized.startswith(("http://", "https://")):
                result.add(normalized)
    return result, bool(result)


def latest_verdict_log(directory: Path) -> Path | None:
    logs = sorted(directory.glob("code_review_*.log"), key=lambda p: p.stat().st_mtime)
    if not logs:
        return None
    newest = logs[-1]
    text = newest.read_text(encoding="utf-8", errors="replace")
    return newest if VERDICT_RE.search(text) else None


def scan_tasks(workspace: Path, task_group: str | None) -> list[Task]:
    task_root = workspace / "agent-task"
    if not task_root.is_dir():
        raise RuntimeError(f"agent-task 디렉터리가 없다: {task_root}")
    directories: list[Path] = []
    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"
    )
    for group in groups:
        if not group.is_dir():
            continue
        directories.append(group)
        directories.extend(sorted(p for p in group.iterdir() if p.is_dir()))

    tasks: list[Task] = []
    for directory in directories:
        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:
            continue
        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)}개다")
        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)
        write_set, write_set_known = extract_write_set(plan)
        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 task/plan/tag metadata를 판별할 수 없다")
            elif metadata.group("task") != name:
                errors.append(
                    f"PLAN task metadata가 디렉터리와 다르다: {metadata.group('task')}"
                )
            if not write_set_known:
                errors.append("수정 파일 요약 write-set을 판별할 수 없다")
        task = 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),
            lane=lane,
            grade=grade,
        )
        tasks.append(task)
    return sorted(tasks, key=lambda t: (t.index, t.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():
        for year in archive.iterdir():
            if not year.is_dir():
                continue
            for month in year.iterdir():
                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 route_agent(task: Task) -> AgentSpec:
    if task.lane is None or task.grade is None:
        raise RuntimeError("PLAN route를 판별할 수 없다")
    grade = task.grade
    if task.lane == "local":
        if grade <= 5:
            return AgentSpec("pi", "ornith:35b", "pi/iop/ornith:35b", local_pi=True)
        if grade <= 8:
            return AgentSpec("pi", "laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True)
        return AgentSpec("claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh")
    if grade <= 2:
        return AgentSpec("agy", "Gemini 3.5 Flash (Low)", "agy/Gemini 3.5 Flash (Low)")
    if grade <= 4:
        return AgentSpec("agy", "Gemini 3.5 Flash (Medium)", "agy/Gemini 3.5 Flash (Medium)")
    if grade <= 6:
        return AgentSpec("agy", "Gemini 3.5 Flash (High)", "agy/Gemini 3.5 Flash (High)")
    if grade <= 8:
        return AgentSpec("claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh")
    return AgentSpec("codex", "gpt-5.6-sol", "codex/gpt-5.6-sol xhigh")


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 task_requires_selfcheck(task: Task) -> bool:
    return task.lane == "local" and task.grade is not None and task.grade <= 8


def task_stage(task: Task, state: dict[str, Any]) -> str:
    if task.user_review:
        return "user-review"
    if task.errors:
        return "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_RE.search(text):
            return "review"
        if re.search(r"^-\s*상태:(?![ \t]*없음[ \t]*$).+", text, re.MULTILINE):
            return "review"
    if state.get("worker_done"):
        if task_requires_selfcheck(task) and not state.get("selfcheck_done"):
            return "selfcheck"
        return "review"
    if task.review and task.review.exists() and FINAL_CHECK_RE.search(text):
        return "review"
    return "worker"


def classify_failure(output: str) -> str:
    lowered = output.lower()
    for category, patterns in PROMOTABLE_PATTERNS.items():
        if any(re.search(pattern, lowered, re.DOTALL) for pattern in patterns):
            return category
    return "generic-error"


def terminal_diagnostic(cli: str, channel: str, line: str) -> str | None:
    if channel == "stderr":
        return line
    try:
        value = json.loads(line)
    except json.JSONDecodeError:
        if cli == "agy" and re.match(
            r"^\s*(?:error|fatal|provider error|model error)\b", line, re.IGNORECASE
        ):
            return line
        return None
    event_type = str(value.get("type", ""))
    if cli == "codex" and event_type in {"turn.failed", "error"}:
        return json.dumps(value.get("error", value), ensure_ascii=False)
    if cli == "claude":
        subtype = str(value.get("subtype", ""))
        if event_type == "result" and (value.get("is_error") or subtype.startswith("error")):
            return str(value.get("result") or value)
        if event_type == "system" and subtype.startswith("error"):
            return json.dumps(value, ensure_ascii=False)
    return None


def agy_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:]:
        if re.search(r"\b(?:error|fatal)\b", line, re.IGNORECASE):
            diagnostics.append(line)
        elif re.search(r"RESOURCE_EXHAUSTED|rate.?limit|model.{0,40}unavailable", line, re.IGNORECASE):
            diagnostics.append(line)
    return diagnostics


def promoted_spec(spec: AgentSpec, recovery_count: int) -> AgentSpec | None:
    if spec.cli == "agy":
        return AgentSpec("claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh")
    if spec.cli == "claude":
        return AgentSpec("codex", "gpt-5.6-sol", "codex/gpt-5.6-sol xhigh")
    if spec.cli == "codex" and recovery_count < 1:
        return AgentSpec("codex", "gpt-5.6-sol", "codex/gpt-5.6-sol xhigh")
    return None


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
    session_id = value.get("thread_id") or value.get("session_id")
    rendered: list[str] = []
    if cli == "codex":
        if value.get("type") == "thread.started" and session_id:
            rendered.append(f"session={session_id}")
        item = value.get("item") or {}
        item_type = item.get("type")
        if item_type == "agent_message" and item.get("text"):
            rendered.extend(str(item["text"]).splitlines())
        elif item_type == "command_execution":
            rendered.append(f"$ {item.get('command', '')} (exit={item.get('exit_code', '?')})")
        elif item_type in {"mcp_tool_call", "web_search"}:
            rendered.append(f"{item_type}: {item.get('server', '')} {item.get('tool', item.get('query', ''))}")
        elif value.get("type") == "turn.failed":
            rendered.append(str(value.get("error", value)))
    elif cli == "claude":
        message = value.get("message") or {}
        for block in message.get("content") or []:
            if block.get("type") == "text":
                rendered.extend(str(block.get("text", "")).splitlines())
            elif block.get("type") == "tool_use":
                rendered.append(f"tool={block.get('name', '')}")
        if value.get("type") == "result" and value.get("result"):
            rendered.extend(str(value["result"]).splitlines())
        session_id = session_id or value.get("session_id")
    return rendered, str(session_id) if session_id else None


def native_session_path(cli: str, workspace: Path, session_id: str | None, attempt_dir: Path) -> str | None:
    if cli == "claude" and session_id:
        encoded = str(workspace).replace("/", "-")
        return str(Path.home() / ".claude" / "projects" / encoded / f"{session_id}.jsonl")
    if cli == "pi" and session_id:
        matches = list((attempt_dir / "pi-sessions").glob(f"*{session_id}*.jsonl"))
        return str(matches[0]) if matches else str(attempt_dir / "pi-sessions")
    if cli == "codex" and session_id:
        matches = list((Path.home() / ".codex" / "sessions").glob(f"**/*{session_id}*.jsonl"))
        return str(matches[0]) if matches else str(Path.home() / ".codex" / "sessions")
    return None


def agy_conversations() -> dict[Path, int]:
    root = Path.home() / ".gemini" / "antigravity-cli" / "conversations"
    if not root.is_dir():
        return {}
    return {path: path.stat().st_mtime_ns for path in root.glob("*.db")}


def build_command(spec: AgentSpec, prompt: str, workspace: Path, session_id: str, attempt_dir: Path) -> list[str]:
    if spec.cli == "codex":
        return [
            "codex", "exec", "--json", "-C", str(workspace), "-m", spec.model,
            "-c", 'model_reasoning_effort="xhigh"',
            "--dangerously-bypass-approvals-and-sandbox", prompt,
        ]
    if spec.cli == "claude":
        return [
            "claude", "-p", "--output-format", "stream-json", "--verbose",
            "--session-id", session_id, "--model", spec.model, "--effort", "xhigh",
            "--dangerously-skip-permissions", prompt,
        ]
    if spec.cli == "agy":
        return [
            "agy", "--print", "--print-timeout", "8h", "--model", spec.model,
            "--dangerously-skip-permissions", "--log-file", str(attempt_dir / "agy-cli.log"), prompt,
        ]
    if spec.cli == "pi":
        return [
            "pi", "-p", "--approve", "--provider", "iop", "--model", spec.model,
            "--thinking", "high", "--session-id", session_id,
            "--session-dir", str(attempt_dir / "pi-sessions"), prompt,
        ]
    raise RuntimeError(f"지원하지 않는 CLI: {spec.cli}")


async def invoke(
    workspace: Path,
    store: StateStore,
    task: Task,
    role: str,
    spec: AgentSpec,
    prompt: str,
) -> tuple[int, str | None, Path]:
    attempt = store.next_attempt(task, role)
    identity = f"{safe_name(task.name)}__p{plan_number(task)}__{role}__a{attempt:02d}"
    attempt_dir = store.runs / f"{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}__{identity}"
    attempt_dir.mkdir(parents=True, exist_ok=False)
    locator_path = attempt_dir / "locator.json"
    output_path = attempt_dir / "output.log"
    session_id = str(uuid.uuid4())
    record: dict[str, Any] = {
        "execution_id": identity,
        "task": task.name,
        "plan_number": plan_number(task),
        "role": role,
        "attempt": attempt,
        "workspace": str(workspace),
        "cli": spec.cli,
        "model": spec.model,
        "plan_path": str(task.plan) if task.plan else None,
        "review_path": str(task.review) if task.review else None,
        "session_id": session_id if spec.cli in {"claude", "pi"} else None,
        "native_session_path": native_session_path(spec.cli, workspace, session_id, attempt_dir),
        "output_log": str(output_path),
        "cli_log": str(attempt_dir / "agy-cli.log") if spec.cli == "agy" else None,
        "started_at": now_iso(),
        "status": "running",
    }
    write_json(locator_path, record)
    command = build_command(spec, prompt, workspace, session_id, attempt_dir)
    before_agy = agy_conversations() if spec.cli == "agy" else {}
    prefix = f"[{task.directory.name}][{role}][a{attempt:02d}]"
    diagnostics: list[str] = []
    try:
        process = await asyncio.create_subprocess_exec(
            *command,
            cwd=workspace,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            limit=10 * 1024 * 1024,
        )
    except FileNotFoundError:
        line = f"command not found: {command[0]}"
        output_path.write_text(line + "\n", encoding="utf-8")
        record.update(status="failed", finished_at=now_iso(), exit_code=127, failure_class="generic-error")
        write_json(locator_path, record)
        print(f"{prefix} {line}", flush=True)
        return 127, "generic-error", 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
        with output_path.open("w", encoding="utf-8") as output:
            while finished_streams < len(readers):
                try:
                    channel, raw = await asyncio.wait_for(queue.get(), timeout=30)
                except asyncio.TimeoutError:
                    print(f"{prefix} 실행 중... locator={locator_path}", flush=True)
                    continue
                if raw is None:
                    finished_streams += 1
                    continue
                line = raw.decode("utf-8", errors="replace").rstrip("\n")
                output.write(f"[{channel}] {line}\n")
                output.flush()
                diagnostic = terminal_diagnostic(spec.cli, channel, line)
                if diagnostic:
                    diagnostics.append(diagnostic)
                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
                    record["native_session_path"] = native_session_path(
                        spec.cli, workspace, discovered, attempt_dir
                    )
                    write_json(locator_path, record)
                for display_line in rendered:
                    if display_line:
                        print(f"{prefix} {display_line}", flush=True)
        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)
        if process.returncode is None:
            process.terminate()
            await process.wait()
        raise

    if spec.cli == "agy":
        after_agy = agy_conversations()
        changed = [
            path for path, mtime in after_agy.items()
            if path not in before_agy or before_agy[path] != mtime
        ]
        if changed:
            selected = max(changed, key=lambda path: after_agy[path])
            record["session_id"] = selected.stem
            record["native_session_path"] = str(selected)
        diagnostics.extend(agy_log_diagnostics(attempt_dir / "agy-cli.log"))
    if not record.get("native_session_path"):
        record["native_session_path"] = native_session_path(
            spec.cli, workspace, record.get("session_id"), attempt_dir
        )
    failure_class = None if return_code == 0 else classify_failure("\n".join(diagnostics[-50:]))
    record.update(
        status="succeeded" if return_code == 0 else "failed",
        finished_at=now_iso(),
        exit_code=return_code,
        failure_class=failure_class,
    )
    write_json(locator_path, record)
    return return_code, failure_class, locator_path


def base_prompt(task: Task, role: str, spec: AgentSpec) -> str:
    if role == "review":
        target = task.review or task.directory
        if task.review:
            return f"Read {target.resolve()} and start the review. Final in Korean."
        return f"Continue the review for {target.resolve()}. Final in Korean."
    if task.plan is None:
        raise RuntimeError("worker PLAN이 없다")
    target = task.plan.resolve()
    if role == "selfcheck":
        if task.review is None:
            raise RuntimeError("selfcheck CODE_REVIEW 파일이 없다")
        return (
            "Think in English. Final in Korean. This is a self-check of completed work, "
            f"not a review. Read {target} and finish any missing work. Recheck and fix "
            f"your work. Read {task.review.resolve()} and fill any missing implementation "
            "fields."
        )
    if spec.local_pi:
        return f"Think in English. Final in Korean. Read {target} and complete the task."
    return f"Read {target} and complete the task. Final in Korean."


def continuation_prompt(task: Task, role: str, locator: Path) -> str:
    return (
        f"Continue from {locator.resolve()}. Check the saved context and current "
        "workspace. Final in Korean."
    )


async def run_escalating(
    workspace: Path,
    store: StateStore,
    task: Task,
    role: str,
    initial: AgentSpec,
) -> tuple[bool, Path | None]:
    spec = initial
    previous_locator: Path | None = None
    codex_recovery_count = 0
    while True:
        prompt = base_prompt(task, role, spec) if previous_locator is None else continuation_prompt(
            task, role, previous_locator
        )
        rc, failure, locator = await invoke(workspace, store, task, role, spec, prompt)
        if rc == 0:
            return True, locator
        failure = failure or "generic-error"
        if spec.local_pi or failure == "generic-error":
            banner("작업차단", task.name, [f"reason={failure}", f"locator={locator}"])
            return False, locator
        next_spec = promoted_spec(spec, codex_recovery_count)
        if next_spec is None:
            banner("작업차단", task.name, [f"reason={failure}", f"locator={locator}"])
            return False, locator
        if spec.cli == "codex":
            codex_recovery_count += 1
        banner(
            "모델승격",
            task.name,
            [f"from={spec.display}", f"to={next_spec.display}", f"reason={failure}", f"locator={locator}"],
        )
        spec = next_spec
        previous_locator = locator


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
    match = VERDICT_RE.search(path.read_text(encoding="utf-8", errors="replace"))
    return match.group(1) if match else None


def matching_archive_directories(workspace: Path, task: Task) -> 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] = []
    for year in archive.iterdir():
        if not year.is_dir():
            continue
        for month in year.iterdir():
            if not month.is_dir():
                continue
            parent = month if len(parts) == 1 else month / group
            if not parent.is_dir():
                continue
            for candidate in parent.iterdir():
                if candidate.is_dir() and suffix_re.match(candidate.name) and (candidate / "complete.log").is_file():
                    matches.append(candidate)
    return matches


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,
    semaphores: dict[str, asyncio.Semaphore],
) -> None:
    spec = route_agent(task)
    resource = f"pi:{spec.model}" if spec.cli == "pi" else spec.cli
    async with semaphores[resource]:
        banner("작업시작", task.name, [f"model={spec.display}", f"plan={task.plan.resolve()}"])
        success, locator = await run_escalating(workspace, store, task, "worker", spec)
        if not success:
            store.update_task(task, blocked=f"worker failure locator={locator}")
            return
        store.update_task(
            task,
            worker_done=True,
            worker_cli=spec.cli,
            worker_model=spec.model,
            selfcheck_done=not spec.local_pi,
            blocked=None,
        )


async def run_selfcheck(
    workspace: Path,
    store: StateStore,
    task: Task,
    semaphores: dict[str, asyncio.Semaphore],
) -> None:
    spec = route_agent(task)
    if not spec.local_pi:
        raise RuntimeError("Pi가 아닌 route에 selfcheck stage가 배정됐다")
    resource = f"pi:{spec.model}"
    async with semaphores[resource]:
        banner("자가검증시작", task.name, [f"model={spec.display}", f"plan={task.plan.resolve()}"])
        success, locator = await run_escalating(workspace, store, task, "selfcheck", spec)
        if not success:
            store.update_task(task, blocked=f"selfcheck failure locator={locator}")
            return
        store.update_task(task, selfcheck_done=True, blocked=None)


async def run_review(
    workspace: Path,
    store: StateStore,
    task: Task,
    review_semaphore: asyncio.Semaphore,
) -> None:
    spec = AgentSpec("codex", "gpt-5.6-sol", "codex/gpt-5.6-sol xhigh")
    before = task_signature(workspace, task)
    prior_review_fingerprints = review_fingerprints(workspace, task)
    async with review_semaphore:
        target = task.review.resolve() if task.review else task.directory.resolve()
        banner("리뷰시작", task.name, [f"model={spec.display}", f"review={target}"])
        success, locator = await run_escalating(workspace, store, task, "review", spec)
    if not success:
        store.update_task(task, blocked=f"review failure locator={locator}")
        return
    after = task_signature(workspace, task)
    if before == after:
        state = store.task_state(task)
        count = int(state.get("review_no_progress", 0)) + 1
        store.update_task(task, review_no_progress=count)
        banner("루프정체경고", task.name, [f"unchanged_review_attempts={count}", 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}"])


def status_lines(task: Task, stage: str, dependency: str) -> list[str]:
    route = f"{task.lane}-G{task.grade:02d}" if task.lane and task.grade else "recovery"
    return [f"stage={stage}", f"route={route}", f"dependency={dependency}"]


def select_dispatch_candidates(
    ready: list[tuple[Task, str]], active_stages: set[str]
) -> tuple[list[tuple[Task, str]], list[tuple[Task, 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"}]
    if "review" in active_stages:
        return [], ready, "공식 review 종료 대기"
    if active_stages & {"worker", "selfcheck"}:
        return ready_workers, ready_reviews, "worker/selfcheck 안정화 대기"
    if ready_reviews:
        return ready_reviews[:1], ready_reviews[1:] + ready_workers, "공식 review 우선 처리 대기"
    return ready_workers, [], ""


async def dispatch(args: argparse.Namespace) -> int:
    workspace = Path(args.workspace).resolve()
    store = StateStore(workspace)
    if args.retry_blocked:
        store.clear_blocked()
    semaphores = {
        "pi:ornith:35b": asyncio.Semaphore(3),
        "pi:laguna-s:2.1": asyncio.Semaphore(2),
        "agy": asyncio.Semaphore(1),
        "claude": asyncio.Semaphore(64),
        "codex": asyncio.Semaphore(64),
    }
    review_semaphore = asyncio.Semaphore(1)
    running: dict[str, asyncio.Task[None]] = {}
    running_stages: dict[str, str] = {}
    running_write_sets: dict[str, set[str]] = {}
    last_wait: dict[str, str] = {}

    while True:
        tasks = scan_tasks(workspace, args.task_group)
        if not tasks and not running:
            banner("작업완료", args.task_group or "agent-task", ["active task 없음"])
            return 0

        task_by_name = {task.name: task for task in tasks}
        for name, future in list(running.items()):
            if not future.done():
                continue
            try:
                future.result()
            except Exception as exc:  # keep other independent tasks alive
                banner("작업차단", name, [f"dispatcher exception={exc}"])
                task = task_by_name.get(name)
                if task:
                    store.update_task(task, blocked=f"dispatcher exception={exc}")
            del running[name]
            running_stages.pop(name, None)
            running_write_sets.pop(name, None)

        ready: list[tuple[Task, str]] = []
        hard_blocked: list[str] = []
        blocked_details: dict[str, tuple[str, str, str]] = {}
        for task in tasks:
            state = store.task_state(task)
            dependency_ready, dependency = dependency_state(workspace, task)
            stage = task_stage(task, state)
            reason = ""
            if task.errors:
                reason = "; ".join(task.errors)
            elif stage == "user-review":
                reason = f"USER_REVIEW 대기: {task.user_review}"
            elif state.get("blocked"):
                reason = str(state["blocked"])
            elif not dependency_ready:
                reason = dependency
            if reason:
                wait_key = f"{stage}|{reason}"
                event = (
                    "작업차단"
                    if 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
                hard_blocked.append(task.name)
                continue
            if task.name not in running:
                ready.append((task, stage))

        if args.dry_run:
            ready_by_name = {task.name: stage for task, stage in ready}
            for task in tasks:
                if task.name in ready_by_name:
                    stage = ready_by_name[task.name]
                    spec = (
                        AgentSpec("codex", "gpt-5.6-sol", "codex/gpt-5.6-sol xhigh")
                        if stage == "review"
                        else route_agent(task)
                    )
                    banner("작업대기", task.name, status_lines(task, stage, "ready") + [f"model={spec.display}"])
                else:
                    event, stage, reason = blocked_details[task.name]
                    banner(event, task.name, status_lines(task, stage, reason))
            return 2 if hard_blocked and not ready else 0

        active_stages = {
            running_stages[name] for name, future in running.items() if not future.done()
        }
        candidates, deferred, phase_wait_reason = select_dispatch_candidates(ready, active_stages)
        for task, stage in deferred:
            wait_key = f"{stage}|{phase_wait_reason}"
            if last_wait.get(task.name) != wait_key:
                banner("작업대기", task.name, status_lines(task, stage, phase_wait_reason))
                last_wait[task.name] = wait_key

        reserved_sets = list(running_write_sets.values())
        scheduled = False
        for task, stage in candidates:
            if task.write_set and any(task.write_set & other for other in reserved_sets):
                reason = "write-set 겹침으로 직렬 대기"
                if last_wait.get(task.name) != reason:
                    banner("작업대기", task.name, status_lines(task, stage, reason))
                    last_wait[task.name] = reason
                continue
            if stage == "review":
                future = asyncio.create_task(run_review(workspace, store, task, review_semaphore))
            elif stage == "selfcheck":
                future = asyncio.create_task(run_selfcheck(workspace, store, task, semaphores))
            else:
                future = asyncio.create_task(run_worker(workspace, store, task, semaphores))
            running[task.name] = future
            running_stages[task.name] = stage
            running_write_sets[task.name] = task.write_set
            reserved_sets.append(task.write_set)
            last_wait.pop(task.name, None)
            scheduled = True

        if running:
            await asyncio.wait(running.values(), return_when=asyncio.FIRST_COMPLETED)
            continue
        if not scheduled:
            banner("작업차단", args.task_group or "agent-task", [f"blocked={','.join(sorted(hard_blocked))}"])
            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("--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")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    try:
        return asyncio.run(dispatch(args))
    except KeyboardInterrupt:
        print("\n중단됨", file=sys.stderr)
        return 130
    except RuntimeError as exc:
        print(f"dispatcher error: {exc}", file=sys.stderr)
        return 2


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