iop/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_specs.py

#!/usr/bin/env python3
"""Convert persisted execution-target decisions into dispatcher agent specs."""

from __future__ import annotations

from dataclasses import dataclass
import json
from pathlib import Path
from typing import Any


@dataclass(frozen=True)
class AgentSpec:
    cli: str
    model: str
    display: str
    local_pi: bool = False
    reasoning_effort: str | None = None
    thinking_level: str | None = None
    command_model: str | None = None


def effective_reasoning_effort(spec: AgentSpec) -> str | None:
    if spec.cli in {"codex", "claude", "claude-glm"}:
        return spec.reasoning_effort or "xhigh"
    if spec.cli == "opencode":
        return spec.reasoning_effort or "max"
    return None


def effective_pi_thinking_level(spec: AgentSpec) -> str | None:
    return spec.thinking_level or "high" if spec.cli == "pi" else None


def pi_display(model: str, thinking_level: str | None) -> str:
    suffix = f" {thinking_level}" if thinking_level is not None else ""
    return f"pi/iop/{model}{suffix}"


def agent_spec_from_record(
    record: dict[str, Any],
    target_resolver=None,
) -> AgentSpec | None:
    cli = str(record.get("cli") or "")
    model = str(record.get("model") or "")
    if not cli or not model:
        return None
    reasoning_effort = record.get("reasoning_effort")
    thinking_level = record.get("thinking_level")
    selected = record.get("selected")
    selected = selected if isinstance(selected, dict) else {}
    command_model = record.get("command_model") or selected.get("command_model")
    reasoning_effort = (
        str(reasoning_effort) if reasoning_effort is not None else None
    )
    thinking_level = str(thinking_level) if thinking_level is not None else None
    command_model = str(command_model) if command_model is not None else None
    canonical = None
    if target_resolver is not None:
        resolver_reasoning = reasoning_effort or (
            "max" if cli == "opencode" else None
        )
        try:
            canonical = target_resolver(
                cli,
                model,
                thinking_level,
                resolver_reasoning,
            )
        except (AttributeError, TypeError, ValueError):
            canonical = None
    if canonical is not None:
        if command_model is None:
            command_model = canonical.command_model
        if reasoning_effort is None:
            reasoning_effort = canonical.reasoning_effort
    if cli in {"codex", "claude", "claude-glm", "opencode"}:
        effort = reasoning_effort or ("max" if cli == "opencode" else "xhigh")
        display = f"{cli}/{model} {effort}"
    elif cli == "pi":
        display = pi_display(model, thinking_level)
    else:
        display = f"{cli}/{model}"
    return AgentSpec(
        cli,
        model,
        display,
        local_pi=cli == "pi",
        reasoning_effort=reasoning_effort,
        thinking_level=thinking_level,
        command_model=command_model,
    )


def agent_spec_from_locator(locator: Path | None, target_resolver=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
    return (
        agent_spec_from_record(record, target_resolver)
        if isinstance(record, dict)
        else None
    )


def _selected_schema(decision: dict[str, Any], error_type):
    selected = decision.get("selected")
    if not isinstance(selected, dict):
        raise error_type("selector selected가 object가 아니다")
    adapter = selected.get("adapter")
    target = selected.get("target")
    execution_class = selected.get("execution_class")
    selfcheck = selected.get("selfcheck_required")
    if (
        not isinstance(adapter, str)
        or not isinstance(target, str)
        or not target
        or not isinstance(selfcheck, bool)
        or execution_class not in {"local_model", "cloud_model"}
    ):
        raise error_type("selector selected schema가 유효하지 않다")
    return selected, adapter, target, execution_class, selfcheck


def _validate_promotion_path(
    decision: dict[str, Any], canonical, initial_keys: set[tuple], selector, error_type
) -> None:
    promotion_path = decision.get("promotion_path")
    if not isinstance(promotion_path, list) or len(promotion_path) < 2:
        raise error_type("selector promotion path가 없다")
    resolved_path = []
    for index, entry in enumerate(promotion_path):
        if not isinstance(entry, dict):
            raise error_type(f"selector promotion path[{index}]가 object가 아니다")
        resolved = selector.policy.canonical_target(
            entry.get("adapter"),
            entry.get("target"),
            entry.get("thinking_level"),
            entry.get("reasoning_effort"),
        )
        if resolved is None:
            raise error_type(
                f"selector promotion path[{index}] target이 canonical이 아니다"
            )
        resolved_path.append(resolved)
    if _target_key(resolved_path[0]) not in initial_keys:
        raise error_type("selector promotion path 시작 target이 잘못됐다")
    if any(
        selector.policy.promotion_target(previous) != current
        for previous, current in zip(resolved_path, resolved_path[1:])
    ):
        raise error_type("selector promotion path 순서가 잘못됐다")
    if resolved_path[-1] != canonical:
        raise error_type("selector promotion path tail이 selected와 다르다")


def spec_from_route_target(canonical, error_type) -> AgentSpec:
    adapter = canonical.adapter
    target = canonical.target
    if adapter == "pi":
        if not target.startswith("iop/"):
            raise error_type("Pi selector target/schema가 유효하지 않다")
        model = target.removeprefix("iop/")
        return AgentSpec(
            "pi",
            model,
            pi_display(model, canonical.thinking_level),
            local_pi=True,
            thinking_level=canonical.thinking_level,
        )
    if adapter not in {"agy", "claude", "claude-glm", "codex", "opencode"}:
        raise error_type(f"selector adapter/schema가 유효하지 않다: {adapter!r}")
    effort = canonical.reasoning_effort
    suffix = f" {effort}" if effort is not None else ""
    return AgentSpec(
        adapter,
        target,
        f"{adapter}/{target}{suffix}",
        reasoning_effort=effort,
        command_model=canonical.command_model,
    )


def _target_key(target) -> tuple:
    return (
        target.adapter,
        target.target,
        target.thinking_level,
        target.reasoning_effort,
    )


def agent_spec_from_decision(
    decision: dict[str, Any], selector, error_type
) -> AgentSpec:
    selected, adapter, target, execution_class, selfcheck = _selected_schema(
        decision, error_type
    )
    thinking = selected.get("thinking_level")
    reasoning = selected.get("reasoning_effort")
    try:
        selector._validate_prior_decision(decision)
        selector._validate_prior_candidate_identity(
            decision,
            stage=decision["stage"],
            lane=decision["lane"],
            grade=decision["grade"],
        )
        catalog = decision.get("catalog")
        if (
            isinstance(catalog, dict)
            and catalog.get("revision") != selector.policy.CATALOG.revision
        ):
            return spec_from_snapshot(
                decision,
                error_type,
                selector.policy.canonical_target,
            )
        evaluated_at = selector.datetime.fromisoformat(
            decision["decision"]["evaluated_at"]
        )
        policy_targets = selector.policy.select_policy(
            stage=decision["stage"],
            lane=decision["lane"],
            grade=decision["grade"],
            evaluated_at=evaluated_at,
        ).candidates
        canonical = selector.policy.canonical_target(
            adapter, target, thinking, reasoning
        )
    except Exception as exc:
        raise error_type(f"selector policy validation 실패: {exc}") from exc
    if canonical is None or (
        canonical.execution_class != execution_class
        or canonical.selfcheck_required != selfcheck
    ):
        raise error_type("selector selected가 canonical policy target이 아니다")
    initial_keys = {_target_key(item) for item in policy_targets}
    if _target_key(canonical) not in initial_keys:
        _validate_promotion_path(
            decision, canonical, initial_keys, selector, error_type
        )
    return spec_from_route_target(canonical, error_type)


def _validate_snapshot_contract(
    adapter: str,
    target: str,
    execution_class: str,
    selfcheck: bool,
    error_type,
) -> None:
    if adapter == "pi":
        if not target.startswith("iop/"):
            raise error_type(
                f"Pi completing decision target이 iop/ prefix가 아니다: {target}"
            )
        if (execution_class, selfcheck) != ("local_model", True):
            raise error_type(
                "Pi completing decision execution/selfcheck 계약이 유효하지 않다: "
                f"target={target} execution_class={execution_class} "
                f"selfcheck_required={selfcheck}"
            )
        return
    if adapter not in {"agy", "claude", "claude-glm", "codex", "opencode"}:
        raise error_type(
            f"completing decision adapter가 유효하지 않다: {adapter!r}"
        )
    if execution_class != "cloud_model" or selfcheck:
        raise error_type(
            "cloud completing decision execution/selfcheck 계약이 유효하지 않다: "
            f"{adapter}/{execution_class}/{selfcheck}"
        )


def spec_from_snapshot(
    decision: dict[str, Any],
    error_type,
    target_resolver=None,
) -> AgentSpec:
    """Build a spec from the target snapshot pinned in a persisted decision."""
    selected, adapter, target, execution_class, selfcheck = _selected_schema(
        decision, error_type
    )
    thinking = selected.get("thinking_level")
    reasoning = selected.get("reasoning_effort")
    command_model = selected.get("command_model")
    canonical = None
    if target_resolver is not None:
        resolver_reasoning = reasoning or (
            "max" if adapter == "opencode" else None
        )
        try:
            canonical = target_resolver(
                adapter,
                target,
                thinking,
                resolver_reasoning,
            )
        except (AttributeError, TypeError, ValueError):
            canonical = None
    if canonical is not None:
        if command_model is None:
            command_model = canonical.command_model
        if reasoning is None:
            reasoning = canonical.reasoning_effort
    _validate_snapshot_contract(
        adapter, target, execution_class, selfcheck, error_type
    )
    if adapter == "pi":
        if thinking is not None and thinking not in {"low", "medium", "high"}:
            raise error_type(
                f"Pi completing decision thinking_level이 유효하지 않다: {thinking!r}"
            )
        model = target.removeprefix("iop/")
        return AgentSpec(
            adapter,
            model,
            pi_display(model, thinking),
            local_pi=True,
            thinking_level=thinking,
        )
    if adapter == "claude-glm":
        if command_model is None:
            raise error_type(
                "claude-glm completing decision에 command_model이 없다"
            )
        return AgentSpec(
            adapter,
            target,
            f"{adapter}/{target} xhigh",
            command_model=str(command_model),
        )
    if adapter == "opencode":
        if reasoning is not None and reasoning not in {"medium", "high", "max"}:
            raise error_type(
                "opencode completing decision reasoning_effort가 유효하지 않다: "
                f"{reasoning!r}"
            )
        if command_model is None:
            raise error_type(
                "opencode completing decision에 command_model이 없다"
            )
        effort = str(reasoning or "max")
        return AgentSpec(
            adapter,
            target,
            f"{adapter}/{target} {effort}",
            reasoning_effort=effort,
            command_model=str(command_model),
        )
    effort = reasoning or ("xhigh" if adapter in {"claude", "codex"} else None)
    suffix = f" {effort}" if effort else ""
    return AgentSpec(
        adapter,
        target,
        f"{adapter}/{target}{suffix}",
        reasoning_effort=effort,
        command_model=str(command_model) if command_model is not None else None,
    )