308 lines
11 KiB
Python
308 lines
11 KiB
Python
#!/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]) -> 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")
|
|
command_model = record.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
|
|
if cli == "claude-glm" and model == "glm-5.2" and not command_model:
|
|
command_model = "sonnet"
|
|
if cli == "opencode" and model == "glm-5.2" and not command_model:
|
|
command_model = "iop-glm/glm-5.2"
|
|
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) -> 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) 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_canonical(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)
|
|
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_canonical(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}"
|
|
)
|
|
model = target.removeprefix("iop/")
|
|
glm_cloud = model == "glm-5.2"
|
|
legacy_glm = glm_cloud and execution_class == "local_model" and selfcheck
|
|
expected = ("cloud_model", False) if glm_cloud else ("local_model", True)
|
|
if not legacy_glm and (execution_class, selfcheck) != expected:
|
|
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) -> 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")
|
|
_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":
|
|
return AgentSpec(
|
|
adapter,
|
|
target,
|
|
f"{adapter}/{target} xhigh",
|
|
command_model=str(command_model or "sonnet"),
|
|
)
|
|
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}"
|
|
)
|
|
effort = str(reasoning or "max")
|
|
return AgentSpec(
|
|
adapter,
|
|
target,
|
|
f"{adapter}/{target} {effort}",
|
|
reasoning_effort=effort,
|
|
command_model=str(command_model or target),
|
|
)
|
|
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,
|
|
)
|