toki/iop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge remote-tracking branch 'origin/dev' into HEAD

Browse source
This commit is contained in:
toki 2026-08-05 06:21:17 +09:00
commit abdd4a87ef
12 changed files with 2463 additions and 983 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md
View file

@ -1,6 +1,6 @@
---
name: orchestrate-agent-task-loop
description: Run agent-task work and autonomously execute active PLAN/CODE_REVIEW loops on request. Use when dispatching dependency-ready work in parallel by predecessor completion and workspace write claims, running lane/G-specific Codex, Claude, agy, and Pi workers, adding local-Pi self-checks, converging official Codex reviews, and escalating cloud context until the task loop finishes.
description: Run agent-task work and autonomously execute active PLAN/CODE_REVIEW loops on request. Use when dispatching dependency-ready work in parallel by predecessor completion and workspace write claims, running catalog-selected lane/G workers and reviewers, adding local-Pi self-checks, converging official reviews, and escalating cloud context until the task loop finishes.
---
# Orchestrate Agent Task Loop
@ -66,24 +66,30 @@ Treat Korean text inside code spans or fenced examples as exact runtime or file-
## Routing Contract
| PLAN route | Worker |
|---|---|
| `local-G01`–`local-G06` | Pi `iop/ornith:35b`, thinking high |
| `local-G07`–`local-G08` | KST day/night agy `Gemini 3.6 Flash (High)` → OpenCode `iop-glm/glm-5.2`, variant max → Codex `gpt-5.6-terra`, reasoning high |
| `local-G09`–`local-G10` | Claude `claude-opus-4-8`, effort xhigh |
| `cloud-G01`–`cloud-G02` | Codex `gpt-5.3-codex-spark` → agy `Gemini 3.6 Flash (Low)` → OpenCode `iop-glm/glm-5.2`, variant medium → Codex `gpt-5.6-terra`, reasoning high |
| `cloud-G03`–`cloud-G04` | agy `Gemini 3.6 Flash (Medium)` → OpenCode `iop-glm/glm-5.2`, variant high → Codex `gpt-5.6-terra`, reasoning high |
| `cloud-G05`–`cloud-G06` | agy `Gemini 3.6 Flash (High)` → OpenCode `iop-glm/glm-5.2`, variant max → Codex `gpt-5.6-terra`, reasoning high |
| `cloud-G07`–`cloud-G08` | Claude `claude-opus-4-8`, effort xhigh |
| `cloud-G09`–`cloud-G10` | Codex `gpt-5.6-sol`, reasoning xhigh |
| Every `CODE_REVIEW-*` | Codex `gpt-5.6-sol`, reasoning xhigh |
`scripts/execution_target_catalog.json` is the only assignment source. Do not
reinterpret a target from task prose or environment variables.
- `lanes.worker` and `lanes.review` each define every `local-G01` through
`local-G10` and `cloud-G01` through `cloud-G10` independently.
- A lane's `candidates` array is its complete ordered execution chain. The first
eligible target is the default and qualified terminal failures advance to the
next unused eligible target without returning to an earlier rank.
- `targets` owns adapter, model, command model, execution class, self-check,
thinking, and reasoning options. Reorder or replace existing target ids by
editing only the lane array. Add a model for an existing adapter by adding one
target entry and referencing its id. Only a new CLI/driver requires Python
dispatcher support.
- The dispatcher validates the catalog at process start. New work uses that
revision; already persisted work keeps its ordered candidate snapshot.
- Missing grade lanes, unknown target ids, duplicate candidates or runtime
identities, invalid options, and incomplete time-window metadata fail closed.
Concurrency limits:
- Global physical-workspace limit: omitting `max_parallel` caps execution at `3`; explicit `max_parallel=0` is unlimited. A positive value caps unique active task-stage attempts and is not narrowed by `task_group`. The cap applies across worker, self-check, review, and verified external-active attempts in the same physical workspace.
- Pi `ornith:35b`: 3.
- agy: 1.
- Official Codex review: no separate review-only limit; subject to the global
- Official review: no separate review-only limit; subject to the global
cap.
- Run worker/self-check and official review in parallel only when they belong to different dependency-ready tasks and their canonical PLAN write sets do not collide in the current physical workspace. Prevent duplicate execution of the same task.
- Even with `complete.log`, treat an explicit predecessor as unfinished while live model/review execution evidence for that task remains. Delay only its consumers; do not propagate the delay to dependency-free siblings or other task groups.
@ -111,9 +117,9 @@ Keep control prompts in English, insert absolute paths only, and do not expand t
Never ask a worker, self-check, or review model to create, edit, or summarize `WORK_LOG.md`.
Run Pi self-check only for policy targets marked `local_model` and `selfcheck_required=true`; OpenCode `iop-glm/glm-5.2` is a cloud target and skips self-check. Do not treat a local Pi self-check exit code `0` as success by itself. Set `selfcheck_done=true` only when `## Implementation Checklist` (or legacy `## 구현 체크리스트`) in `CODE_REVIEW_PATH` contains at least one Markdown list checkbox and every `[...]` checkbox value has at least one non-whitespace character. If both canonical and legacy checklist headings are present in the same file, fail closed. Accept any non-empty value, including `x`, `v`, and `✅`. Do not inspect `## Implementation Item Completion`, `Deviations from Plan`, `Key Design Decisions`, `Verification Results`, or final CODE_REVIEW synchronization text. Run the full self-check prompt exactly once. If its checklist condition fails, resume that successful pass's Pi native session and run the unchecked-item retry prompt up to 10 times. Each retry must resume the locator returned by the preceding successful pass so the same conversation context is preserved; never repeat the full review prompt or start a fresh retry session. Persist the latest successful context locator for dispatcher restart, and block instead of starting fresh when that context cannot be resumed. Block that task after the 10th unchecked-item retry remains incomplete, and continue draining independent work.
Run Pi self-check only for catalog targets marked `local_model` and `selfcheck_required=true`; cloud targets skip self-check. Do not treat a local Pi self-check exit code `0` as success by itself. Set `selfcheck_done=true` only when `## Implementation Checklist` (or legacy `## 구현 체크리스트`) in `CODE_REVIEW_PATH` contains at least one Markdown list checkbox and every `[...]` checkbox value has at least one non-whitespace character. If both canonical and legacy checklist headings are present in the same file, fail closed. Accept any non-empty value, including `x`, `v`, and `✅`. Do not inspect `## Implementation Item Completion`, `Deviations from Plan`, `Key Design Decisions`, `Verification Results`, or final CODE_REVIEW synchronization text. Run the full self-check prompt exactly once. If its checklist condition fails, resume that successful pass's Pi native session and run the unchecked-item retry prompt up to 10 times. Each retry must resume the locator returned by the preceding successful pass so the same conversation context is preserved; never repeat the full review prompt or start a fresh retry session. Persist the latest successful context locator for dispatcher restart, and block instead of starting fresh when that context cannot be resumed. Block that task after the 10th unchecked-item retry remains incomplete, and continue draining independent work.
After an AGY/Gemini worker exits `0`, apply the same `CODE_REVIEW_PATH` implementation-checklist regex before accepting worker completion. If it is incomplete, run a fresh quota probe: only an `exhausted` target becomes `provider-quota` and enters the existing selector failover/promotion chain; `available` or `unknown` remains a completion-evidence recovery on Gemini.
After an AGY/Gemini worker exits `0`, apply the same `CODE_REVIEW_PATH` implementation-checklist regex before accepting worker completion. If it is incomplete, run a fresh quota probe: only an `exhausted` target becomes `provider-quota` and enters the ordered lane failover chain; `available` or `unknown` remains a completion-evidence recovery on Gemini.
For Pi worker recovery attempts, pass only `Read {PLAN_PATH}. Continue.` without a locator explanation. Pi self-check recovery must preserve the current full-pass or unchecked-item role and use its concise prompt. For other CLI escalation attempts, pass `Continue from {LOCATOR_PATH}. Check the saved context and current workspace. Keep artifact content in English. Final in Korean.` Preserve the collaboration prohibition and next-state-materialization sentence in official-review escalation and recovery prompts. Do not ask the model to write a separate handoff summary.
@ -229,18 +235,17 @@ When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a termin
- Archive `WORK_LOG.md` as `work_log_N.log` only after the final task review process exits, the dispatcher appends `FINISH`, and a complete scan finds no active/running task in that group. Accept the log at either the active group path or the verified completed single-task archive; do not impose either location contract on common plan/code-review.
3. **Escalate and recover context.**
- For every route that lists Gemini followed by OpenCode GLM and Codex Terra, classify terminal provider errors or stderr evidence of context/output limits, provider quota/rate limits, unavailable models, or confirmed provider transport errors as a qualified failover to the next candidate. For AGY, accept top-level `error`, `fatal`, `request.failed`, or `turn.failed` events; failed/rejected status with a top-level error/code; stderr; or strong `RESOURCE_EXHAUSTED`, HTTP 429, quota, or rate-limit evidence in `agy-cli.log`. For OpenCode, accept stderr or structured error events as terminal diagnostics. Cloud-only escalation remains OpenCode GLM → Codex Terra; never escalate from an assistant message, source text, tool/test output, or a plain quota-configuration string in an AGY log. Legacy `claude-glm` locators retain their typed quota parsing for recovery only.
- Target Codex `gpt-5.6-terra` with reasoning `high` when escalating from Claude to Codex.
- If Codex returns the same error, retry in a fresh Codex session using the locator while preserving the previous Codex model/reasoning and sharing the same stage's 10-consecutive-failure limit. Continue dispatching other tasks during recovery.
- For every multi-candidate catalog lane, classify terminal provider errors or stderr evidence of context/output limits, provider quota/rate limits, unavailable models, or confirmed provider transport errors as a qualified failover to the next catalog candidate. For AGY, accept top-level `error`, `fatal`, `request.failed`, or `turn.failed` events; failed/rejected status with a top-level error/code; stderr; or strong `RESOURCE_EXHAUSTED`, HTTP 429, quota, or rate-limit evidence in `agy-cli.log`. For OpenCode, accept stderr or structured error events as terminal diagnostics. Never fail over from an assistant message, source text, tool/test output, or a plain quota-configuration string in an AGY log. Legacy locators retain their compatibility promotion only for recovery.
- If a selected target has no unused eligible catalog fallback, retry in a fresh session using the locator while preserving that target's model options and sharing the same stage's 10-consecutive-failure limit. Continue dispatching other tasks during recovery.
- When current source reads a locator blocked 10 times as `generic-error` by older dispatcher source, collapse those 10 failures into one terminal error and clear only that task's blocker only if all 10 terminal-evidence records for the same task/plan/role/source/execution target reclassify to the same escalatable error. Include `stream.log` and the attempt's `agy-cli.log` for AGY. Do not adjust automatically when any history is missing or mixed, or when the locator dispatcher source hash equals the current source hash. Dry-run must display this escalation recovery and next model without writing state. Live execution must choose the higher target from the locator's actual failed target, not the initial PLAN route, inherit locator context, and restore the same escalation target and locator from persisted reclassification metadata after immediate restart.
- Recover timeout, crash, process termination, permission, and ordinary implementation errors on the same target within the same stage's 10-consecutive-failure limit, preserving the actual failure class and locator. At exhaustion, block only that task and keep dispatching independent work.
- On success after escalation, record `worker_cli` and `worker_model` from the successful locator's actual target, not the initial PLAN route.
- Never escalate local Pi models to a cloud model. OpenCode GLM may fail over to the policy-owned Codex Terra backup; legacy `claude-glm` recovery keeps the same terminal fallback.
- Never escalate a `local_model` target to a cloud target. Cloud fallback follows only the current lane's ordered candidates; legacy locators may use their persisted compatibility promotion.
- Use attempt identity `<task-name>__p<plan>__<role>__aNN` and namespace the process marker with the physical workspace id. Record canonical workspace root/id, CLI/model/reasoning effort, PLAN/review, `WORK_LOG.md`, session ID, native session path, and raw output log in the locator.
- Store locators under repository `.git/agent-task-dispatcher/runs/`. Fall back to `${XDG_STATE_HOME}/agent-task-dispatcher/<workspace-id>/runs/` only when `.git` state is unwritable.
4. **Converge review.**
- Run every official review in an independent Codex one-shot session with no separate numeric limit. Dispatch all ready reviews with disjoint workspace claims in parallel.
- Run every official review in an independent session on the selected review-lane target with no separate numeric limit. Dispatch all ready reviews with disjoint workspace claims in parallel.
- For finalization recovery without an active PLAN, recover the review target and write claim from the archived plan log for the same first-line generation metadata, including `milestone-task` when present. Keep the claim until the completed archive is verified.
- Forbid collaboration/sub-agent tools in official review and finish inside the current one-shot session. If such a tool call appears, clean up that attempt's independent subprocess group and retry in a fresh review session. Count the failure toward the same stage's 10-consecutive-failure limit.
- Delegate PASS archive, WARN/FAIL follow-up pairs, and review-finalization recovery to the `code-review` file contract.
@ -251,9 +256,9 @@ When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a termin
## Verification Checklist
- [ ] Scan the complete candidate set only on initial entry and immediately after verified `complete.log`; atomically claim and start every non-running, dependency-ready, non-colliding candidate in the same pass.
- [ ] Confirm the actual CLI/model for each route matches the routing table.
- [ ] Confirm the actual CLI/model for each route matches its catalog lane array.
- [ ] Run exactly one full fresh-session self-check only for Pi work, followed by at most 10 unchecked-item retries in that same Pi native session context when its checklist remains incomplete.
- [ ] Run every official review with Codex `gpt-5.6-sol` xhigh and dispatch dependency-ready reviews with disjoint workspace claims in parallel, subject to the global `--max-parallel` cap (no separate review-only limit).
- [ ] Resolve every official review from its explicit `lanes.review` grade entry and dispatch dependency-ready reviews with disjoint workspace claims in parallel, subject to the global `--max-parallel` cap (no separate review-only limit).
- [ ] Locate the native session and output log for every attempt locator.
- [ ] Record every worker/self-check/review attempt `START`/`FINISH` in one task-group `WORK_LOG.md`.
- [ ] For every completed task group that generated `WORK_LOG.md`, archive a `work_log_N.log` containing the final review `FINISH` and leave no active `WORK_LOG.md`.

2
agent-ops/skills/project/orchestrate-agent-task-loop/agents/openai.yaml
View file

@ -1,4 +1,4 @@
interface:
display_name: "Agent Task Loop Orchestrator"
short_description: "Orchestrate PLAN execution and Codex review loops"
short_description: "Orchestrate catalog-routed PLAN and review loops"
default_prompt: "Use $orchestrate-agent-task-loop to execute the active agent-task workflow."

538
agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py
View file

@ -51,6 +51,36 @@ banner = observation.banner
attempt_event = observation.attempt_event
validation_claim = observation.validation_claim
def load_sibling_target_specs_module():
module_name = "agent_task_execution_target_specs"
loaded = sys.modules.get(module_name)
if loaded is not None:
return loaded
spec = importlib.util.spec_from_file_location(
module_name,
Path(__file__).with_name("execution_target_specs.py"),
)
if spec is None or spec.loader is None:
raise RuntimeError("failed to load execution target specs module")
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
try:
spec.loader.exec_module(module)
except BaseException:
sys.modules.pop(spec.name, None)
raise
return module
target_specs = load_sibling_target_specs_module()
AgentSpec = target_specs.AgentSpec
effective_reasoning_effort = target_specs.effective_reasoning_effort
effective_pi_thinking_level = target_specs.effective_pi_thinking_level
pi_display = target_specs.pi_display
agent_spec_from_record = target_specs.agent_spec_from_record
agent_spec_from_locator = target_specs.agent_spec_from_locator
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(
@ -494,85 +524,6 @@ 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
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:
if spec.cli == "pi":
return spec.thinking_level or "high"
return 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")
if reasoning_effort is not None:
reasoning_effort = str(reasoning_effort)
thinking_level = record.get("thinking_level")
if thinking_level is not None:
thinking_level = str(thinking_level)
command_model = record.get("command_model")
if command_model is not None:
command_model = str(command_model)
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"
local_pi = cli == "pi"
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=local_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
if not isinstance(record, dict):
return None
return agent_spec_from_record(record)
@dataclass(frozen=True)
class PiSessionState:
phase: str
@ -1762,229 +1713,15 @@ def _decision_file(task: Task, stage: str) -> Path:
def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
selected = decision.get("selected")
if not isinstance(selected, dict):
raise ExecutionDecisionError("selector selected가 object가 아니다")
adapter, target = selected.get("adapter"), selected.get("target")
thinking_level = selected.get("thinking_level")
reasoning_effort = selected.get("reasoning_effort")
local_pi = selected.get("selfcheck_required")
execution_class = selected.get("execution_class")
if (not isinstance(adapter, str) or not isinstance(target, str) or not target
or not isinstance(local_pi, bool)
or execution_class not in {"local_model", "cloud_model"}):
raise ExecutionDecisionError("selector selected schema가 유효하지 않다")
try:
selector = _selector_module()
selector._validate_prior_decision(decision)
decision_info = decision["decision"]
evaluated_at = selector.datetime.fromisoformat(decision_info["evaluated_at"])
policy_targets = selector.policy.select_policy(
stage=decision["stage"], lane=decision["lane"], grade=decision["grade"],
evaluated_at=evaluated_at,
).candidates
selector._validate_prior_candidate_identity(
decision,
stage=decision["stage"],
lane=decision["lane"],
grade=decision["grade"],
)
canonical = selector.policy.canonical_target(
adapter, target, thinking_level, reasoning_effort
)
except Exception as exc:
raise ExecutionDecisionError(f"selector policy validation 실패: {exc}") from exc
if canonical is None or (
canonical.execution_class != execution_class
or canonical.selfcheck_required != local_pi
):
raise ExecutionDecisionError("selector selected가 canonical policy target이 아니다")
initial_keys = {
(
item.adapter,
item.target,
item.thinking_level,
item.reasoning_effort,
)
for item in policy_targets
}
if (adapter, target, thinking_level, reasoning_effort) not in initial_keys:
promotion_path = decision.get("promotion_path")
if not isinstance(promotion_path, list) or len(promotion_path) < 2:
raise ExecutionDecisionError("selector promotion path가 없다")
resolved_path = []
for index, entry in enumerate(promotion_path):
if not isinstance(entry, dict):
raise ExecutionDecisionError(
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 ExecutionDecisionError(
f"selector promotion path[{index}] target이 canonical이 아니다"
)
resolved_path.append(resolved)
if (
resolved_path[0].adapter,
resolved_path[0].target,
resolved_path[0].thinking_level,
resolved_path[0].reasoning_effort,
) not in initial_keys:
raise ExecutionDecisionError("selector promotion path 시작 target이 잘못됐다")
if any(
selector.policy.promotion_target(previous) != current
for previous, current in zip(resolved_path, resolved_path[1:])
):
raise ExecutionDecisionError("selector promotion path 순서가 잘못됐다")
if resolved_path[-1] != canonical:
raise ExecutionDecisionError("selector promotion path tail이 selected와 다르다")
if adapter == "pi":
if not target.startswith("iop/"):
raise ExecutionDecisionError("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"} or local_pi:
raise ExecutionDecisionError(f"selector adapter/schema가 유효하지 않다: {adapter!r}")
model = target
reasoning_effort = canonical.reasoning_effort or (
"high" if canonical == selector.policy.CODEX_TERRA_HIGH else None
return target_specs.agent_spec_from_decision(
decision, _selector_module(), ExecutionDecisionError
)
suffix = f" {reasoning_effort}" if reasoning_effort is not None else (
" xhigh" if adapter in {"claude", "claude-glm", "codex"} else ""
)
display = f"{adapter}/{target}{suffix}"
if reasoning_effort is not None:
return AgentSpec(
adapter,
model,
display,
reasoning_effort=reasoning_effort,
command_model=canonical.command_model,
)
return AgentSpec(adapter, model, display, command_model=canonical.command_model)
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.
"""
selected = decision.get("selected")
if not isinstance(selected, dict):
raise ExecutionDecisionError(
"completing decision selected schema가 유효하지 않다"
)
adapter = selected.get("adapter")
target = selected.get("target")
execution_class = selected.get("execution_class")
selfcheck_required = selected.get("selfcheck_required")
thinking_level = selected.get("thinking_level")
reasoning_effort = selected.get("reasoning_effort")
if not all(isinstance(value, str) and value for value in (adapter, target, execution_class)):
raise ExecutionDecisionError(
"completing decision selected의 adapter/target/execution_class는 빈 문자열이 아닌 string이어야 한다"
)
if execution_class not in {"local_model", "cloud_model"}:
raise ExecutionDecisionError(
f"completing decision execution_class이 유효하지 않다: {execution_class}"
)
if not isinstance(selfcheck_required, bool):
raise ExecutionDecisionError(
"completing decision selected.selfcheck_required must be a boolean"
)
if adapter == "pi":
if not target.startswith("iop/"):
raise ExecutionDecisionError(
f"Pi completing decision target이 iop/ prefix가 아니다: {target}"
)
model = target.removeprefix("iop/")
glm_cloud = model == "glm-5.2"
expected_class = "cloud_model" if glm_cloud else "local_model"
expected_selfcheck = not glm_cloud
legacy_glm_contract = (
glm_cloud
and execution_class == "local_model"
and selfcheck_required
)
if not legacy_glm_contract and (
execution_class != expected_class
or selfcheck_required != expected_selfcheck
):
raise ExecutionDecisionError(
"Pi completing decision execution/selfcheck 계약이 유효하지 않다: "
f"target={target} execution_class={execution_class} "
f"selfcheck_required={selfcheck_required}"
)
if thinking_level is not None and thinking_level not in {"low", "medium", "high"}:
raise ExecutionDecisionError(
f"Pi completing decision thinking_level이 유효하지 않다: {thinking_level!r}"
)
display = pi_display(model, thinking_level)
return AgentSpec(
adapter,
model,
display,
local_pi=True,
thinking_level=thinking_level,
)
if adapter not in {"agy", "claude", "claude-glm", "codex", "opencode"}:
raise ExecutionDecisionError(
f"completing decision adapter가 유효하지 않다: {adapter!r}"
)
if execution_class != "cloud_model":
raise ExecutionDecisionError(
f"cloud completing decision execution_class이 cloud_model이 아니다: {adapter}/{execution_class}"
)
if selfcheck_required:
raise ExecutionDecisionError(
f"cloud completing decision selfcheck_required가 True이다: {adapter}/{target}"
)
if adapter == "claude-glm":
if target != "glm-5.2":
raise ExecutionDecisionError(
f"claude-glm completing decision target이 유효하지 않다: {target}"
)
return AgentSpec(
adapter,
target,
f"{adapter}/{target} xhigh",
local_pi=False,
command_model="sonnet",
)
if adapter == "opencode":
if target != "glm-5.2":
raise ExecutionDecisionError(
f"opencode completing decision target이 유효하지 않다: {target}"
)
if reasoning_effort not in {"medium", "high", "max"}:
raise ExecutionDecisionError(
"opencode completing decision reasoning_effort가 유효하지 않다: "
f"{reasoning_effort!r}"
)
return AgentSpec(
adapter,
target,
f"{adapter}/{target} {reasoning_effort}",
local_pi=False,
reasoning_effort=reasoning_effort,
command_model="iop-glm/glm-5.2",
)
display = f"{adapter}/{target}"
return AgentSpec(adapter, target, display, local_pi=False)
return target_specs.spec_from_snapshot(
decision, ExecutionDecisionError
)
def select_execution_decision(
@ -2009,9 +1746,23 @@ def select_execution_decision(
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"
evaluated = evaluated_at or datetime.now(KST)
if stage == "review":
lane, grade, work_unit_id = official_review_source_identity(task)
return selector.select_execution_target_for_route(
work_unit_id=work_unit_id,
stage=stage,
lane=lane,
grade=grade,
evaluated_at=evaluated,
transition=transition,
prior_decision=prior_decision,
quota_snapshot=quota_snapshot,
failure_class=failure_class,
)
return selector.select_execution_target(
_decision_file(task, stage), stage=stage,
evaluated_at=evaluated_at or datetime.now(KST),
evaluated_at=evaluated,
transition=transition,
prior_decision=prior_decision,
quota_snapshot=quota_snapshot,
@ -2077,7 +1828,10 @@ def official_review_source_identity(task: Task) -> tuple[str, int, str]:
def synthesized_official_review_decision(
task: Task, *, evaluated_at: datetime | None = None
task: Task,
*,
evaluated_at: datetime | None = None,
quota_snapshot: dict[str, Any] | None = None,
) -> dict[str, Any]:
lane, grade, work_unit_id = official_review_source_identity(task)
evaluated = evaluated_at or datetime.now(KST)
@ -2085,64 +1839,46 @@ def synthesized_official_review_decision(
raise ExecutionDecisionError(
"official review evaluated_at이 timezone-aware가 아니다"
)
recovery_from_archive = task.plan is None and task.review is None
selector = _selector_module()
policy_decision = selector.policy.select_policy(
stage="review", lane=lane, grade=grade, evaluated_at=evaluated
)
selected_target = policy_decision.candidates[0]
target_ref = {
"adapter": selected_target.adapter,
"target": selected_target.target,
}
candidate = {
"candidate_rank": 1,
"adapter": selected_target.adapter,
"target": selected_target.target,
"execution_class": selected_target.execution_class,
"selfcheck_required": selected_target.selfcheck_required,
"quota_mode": "bounded",
"quota_status": "unknown",
"eligibility": "eligible",
"rejection_reason": None,
}
return {
"schema_version": selector.SCHEMA_VERSION,
"work_unit_id": work_unit_id,
"stage": "review",
"lane": lane,
"grade": grade,
"selected": {
"adapter": selected_target.adapter,
"target": selected_target.target,
"execution_class": selected_target.execution_class,
"selfcheck_required": selected_target.selfcheck_required,
},
"candidates": [candidate],
"decision": {
"rule_id": policy_decision.rule_id,
"policy_priority": policy_decision.policy_priority,
"reason_codes": list(policy_decision.reason_codes),
"evaluated_at": evaluated.astimezone(KST).isoformat(),
"timezone": selector.TIMEZONE_NAME,
"time_window": policy_decision.time_window,
"pinned": recovery_from_archive,
},
"quota": {
recovery_from_archive = task.plan is None and task.review is None
effective_quota = (
quota_snapshot
if quota_snapshot is not None
else {
"snapshot_id": None,
"mode": "bounded",
"status": "unknown",
"source": "official_review_fixed_policy",
"source": "official_review_catalog_policy",
"checked_at": None,
"targets": [],
},
"transition": {
"previous_target": dict(target_ref) if recovery_from_archive else None,
"next_target": dict(target_ref) if recovery_from_archive else None,
"trigger": "resume" if recovery_from_archive else "initial",
}
)
try:
decision = selector.select_execution_target_for_route(
work_unit_id=work_unit_id,
stage="review",
lane=lane,
grade=grade,
evaluated_at=evaluated,
quota_snapshot=effective_quota,
quota_probe_command="official_review_catalog_policy",
)
except selector.SelectorInputError as exc:
raise ExecutionDecisionError(
f"official review catalog selector가 실패했다 [{exc.code}]: {exc}"
) from exc
if recovery_from_archive:
selected = decision["selected"]
target_ref = {
"adapter": selected["adapter"],
"target": selected["target"],
}
decision["decision"]["pinned"] = True
decision["transition"] = {
"previous_target": dict(target_ref),
"next_target": dict(target_ref),
"trigger": "resume",
"context_transfer": "none",
},
}
}
return decision
def read_or_preview_stage_decision(
@ -2176,7 +1912,9 @@ def read_or_preview_stage_decision(
agent_spec_from_decision(prior)
return prior
return synthesized_official_review_decision(
task, evaluated_at=evaluated_at
task,
evaluated_at=evaluated_at,
quota_snapshot=quota_snapshot,
)
if isinstance(prior, dict) and task.plan and task.plan.is_file():
@ -2418,6 +2156,26 @@ def derive_work_unit_quota_evidence(
}
def _retry_admission_candidates(prior: dict[str, Any] | None) -> list[Any]:
candidates = prior.get("candidates", []) if isinstance(prior, dict) else []
used = prior.get("used_candidates", []) if isinstance(prior, dict) else []
selected = prior.get("selected") if isinstance(prior, dict) else None
used_keys = {
(entry.get("adapter"), entry.get("target"))
for entry in used
if isinstance(entry, dict)
}
if isinstance(selected, dict):
used_keys.add((selected.get("adapter"), selected.get("target")))
return [
type("Target", (), candidate)()
for candidate in candidates
if isinstance(candidate, dict)
and candidate.get("execution_class") != "local_model"
and (candidate.get("adapter"), candidate.get("target")) not in used_keys
]
def build_admission_batch_snapshot(
store: StateStore,
ready_items: list[tuple[Task, str]],
@ -2431,40 +2189,39 @@ def build_admission_batch_snapshot(
seen_keys = set()
for task, stage in ready_items:
if stage != "worker":
if stage not in {"worker", "review"}:
continue
state = store.peek_task_state(task)
if has_persisted_worker_decision(state, task) and not retry_quota_refresh_pending(state):
selector_stage = "review" if stage == "review" else "worker"
decisions = state.get("execution_decisions", {})
prior = decisions.get(selector_stage) if isinstance(decisions, dict) else None
is_retry = stage == "worker" and retry_quota_refresh_pending(state)
has_persisted = (
has_persisted_worker_decision(state, task)
if stage == "worker"
else isinstance(prior, dict)
)
if has_persisted and not is_retry:
continue
is_retry = retry_quota_refresh_pending(state)
if is_retry:
decisions = state.get("execution_decisions", {})
prior = decisions.get("worker") if isinstance(decisions, dict) else None
candidates = prior.get("candidates", []) if isinstance(prior, dict) else []
used = prior.get("used_candidates", []) if isinstance(prior, dict) else []
selected = prior.get("selected") if isinstance(prior, dict) else None
used_keys = {
(entry.get("adapter"), entry.get("target"))
for entry in used
if isinstance(entry, dict)
}
if isinstance(selected, dict):
used_keys.add((selected.get("adapter"), selected.get("target")))
candidates_to_probe = [
type("Target", (), candidate)()
for candidate in candidates
if isinstance(candidate, dict)
and candidate.get("execution_class") != "local_model"
and (candidate.get("adapter"), candidate.get("target")) not in used_keys
]
candidates_to_probe = _retry_admission_candidates(prior)
else:
lane, grade = task.lane, task.grade
if stage == "review":
try:
lane, grade, _ = official_review_source_identity(task)
except ExecutionDecisionError:
continue
else:
lane, grade = task.lane, task.grade
if not lane or not grade:
continue
try:
pol_dec = policy_mod.select_policy(
stage="worker", lane=lane, grade=grade, evaluated_at=admission_time
stage=selector_stage,
lane=lane,
grade=grade,
evaluated_at=admission_time,
)
except ValueError:
continue
@ -4662,6 +4419,25 @@ def build_context_package(
return package
def failover_context_package(
workspace: Path,
task: Task,
role: str,
locator: Path,
previous_spec: AgentSpec,
next_spec: AgentSpec,
) -> dict[str, Any] | None:
if role == "review":
return None
return build_context_package(
workspace,
task,
locator,
previous_spec=previous_spec,
next_spec=next_spec,
)
def canonical_selector_failover_route(decision: dict[str, Any] | None) -> bool:
if not isinstance(decision, dict):
return False
@ -4993,8 +4769,8 @@ async def run_escalating(
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
context = failover_context_package(
workspace, task, role, locator, spec, next_spec
)
commit_execution_decision(store, task, role, next_decision)
banner(
@ -5119,7 +4895,7 @@ async def run_escalating(
task.name,
[
f"model={spec.display}",
"reason=official-review-fixed-target",
"reason=review-catalog-target-retry",
*failure_report_lines(failure, locator),
f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}",
],

555
agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_catalog.json Normal file
View file

@ -0,0 +1,555 @@
{
"schema_version": 1,
"targets": {
"pi-ornith-high": {
"adapter": "pi",
"target": "iop/ornith:35b",
"execution_class": "local_model",
"selfcheck_required": true,
"thinking_level": "high"
},
"pi-laguna-high": {
"adapter": "pi",
"target": "iop/laguna-s:2.1",
"execution_class": "local_model",
"selfcheck_required": true,
"thinking_level": "high"
},
"agy-gemini-low": {
"adapter": "agy",
"target": "Gemini 3.6 Flash (Low)",
"execution_class": "cloud_model",
"selfcheck_required": false
},
"agy-gemini-medium": {
"adapter": "agy",
"target": "Gemini 3.6 Flash (Medium)",
"execution_class": "cloud_model",
"selfcheck_required": false
},
"agy-gemini-high": {
"adapter": "agy",
"target": "Gemini 3.6 Flash (High)",
"execution_class": "cloud_model",
"selfcheck_required": false
},
"opencode-glm-medium": {
"adapter": "opencode",
"target": "glm-5.2",
"command_model": "iop-glm/glm-5.2",
"execution_class": "cloud_model",
"selfcheck_required": false,
"reasoning_effort": "medium"
},
"opencode-glm-high": {
"adapter": "opencode",
"target": "glm-5.2",
"command_model": "iop-glm/glm-5.2",
"execution_class": "cloud_model",
"selfcheck_required": false,
"reasoning_effort": "high"
},
"opencode-glm-max": {
"adapter": "opencode",
"target": "glm-5.2",
"command_model": "iop-glm/glm-5.2",
"execution_class": "cloud_model",
"selfcheck_required": false,
"reasoning_effort": "max"
},
"legacy-claude-glm": {
"adapter": "claude-glm",
"target": "glm-5.2",
"command_model": "sonnet",
"execution_class": "cloud_model",
"selfcheck_required": false,
"reasoning_effort": "xhigh"
},
"claude-opus-xhigh": {
"adapter": "claude",
"target": "claude-opus-4-8",
"execution_class": "cloud_model",
"selfcheck_required": false,
"reasoning_effort": "xhigh"
},
"claude-haiku-xhigh": {
"adapter": "claude",
"target": "claude-haiku-4-5",
"execution_class": "cloud_model",
"selfcheck_required": false,
"reasoning_effort": "xhigh"
},
"codex-spark-xhigh": {
"adapter": "codex",
"target": "gpt-5.3-codex-spark",
"execution_class": "cloud_model",
"selfcheck_required": false,
"reasoning_effort": "xhigh"
},
"codex-sol-xhigh": {
"adapter": "codex",
"target": "gpt-5.6-sol",
"execution_class": "cloud_model",
"selfcheck_required": false,
"reasoning_effort": "xhigh"
},
"codex-terra-high": {
"adapter": "codex",
"target": "gpt-5.6-terra",
"execution_class": "cloud_model",
"selfcheck_required": false,
"reasoning_effort": "high"
}
},
"lanes": {
"worker": {
"local-G01": {
"candidates": [
"pi-ornith-high"
],
"rule_id": "worker-local-g01-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"local-G02": {
"candidates": [
"pi-ornith-high"
],
"rule_id": "worker-local-g02-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"local-G03": {
"candidates": [
"pi-ornith-high"
],
"rule_id": "worker-local-g03-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"local-G04": {
"candidates": [
"pi-ornith-high"
],
"rule_id": "worker-local-g04-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"local-G05": {
"candidates": [
"pi-ornith-high"
],
"rule_id": "worker-local-g05-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"local-G06": {
"candidates": [
"pi-ornith-high"
],
"rule_id": "worker-local-g06-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"local-G07": {
"candidates": [
"agy-gemini-high",
"opencode-glm-max",
"codex-terra-high"
],
"policy_priority": 20,
"time_windows": {
"kst-day-[07:00,23:00)": {
"rule_id": "worker-local-g07-kst-day-catalog",
"reason_codes": [
"worker_catalog_lane_kst_day"
]
},
"kst-night-[23:00,07:00)": {
"rule_id": "worker-local-g07-kst-night-catalog",
"reason_codes": [
"worker_catalog_lane_kst_night"
]
}
}
},
"local-G08": {
"candidates": [
"agy-gemini-high",
"opencode-glm-max",
"codex-terra-high"
],
"policy_priority": 20,
"time_windows": {
"kst-day-[07:00,23:00)": {
"rule_id": "worker-local-g08-kst-day-catalog",
"reason_codes": [
"worker_catalog_lane_kst_day"
]
},
"kst-night-[23:00,07:00)": {
"rule_id": "worker-local-g08-kst-night-catalog",
"reason_codes": [
"worker_catalog_lane_kst_night"
]
}
}
},
"local-G09": {
"candidates": [
"claude-opus-xhigh",
"codex-terra-high"
],
"rule_id": "worker-local-g09-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"local-G10": {
"candidates": [
"claude-opus-xhigh",
"codex-terra-high"
],
"rule_id": "worker-local-g10-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"cloud-G01": {
"candidates": [
"codex-spark-xhigh",
"agy-gemini-low",
"opencode-glm-medium",
"codex-terra-high"
],
"rule_id": "worker-cloud-g01-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"cloud-G02": {
"candidates": [
"codex-spark-xhigh",
"agy-gemini-low",
"opencode-glm-medium",
"codex-terra-high"
],
"rule_id": "worker-cloud-g02-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"cloud-G03": {
"candidates": [
"agy-gemini-medium",
"opencode-glm-high",
"codex-terra-high"
],
"rule_id": "worker-cloud-g03-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"cloud-G04": {
"candidates": [
"agy-gemini-medium",
"opencode-glm-high",
"codex-terra-high"
],
"rule_id": "worker-cloud-g04-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"cloud-G05": {
"candidates": [
"agy-gemini-high",
"opencode-glm-max",
"codex-terra-high"
],
"rule_id": "worker-cloud-g05-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"cloud-G06": {
"candidates": [
"agy-gemini-high",
"opencode-glm-max",
"codex-terra-high"
],
"rule_id": "worker-cloud-g06-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"cloud-G07": {
"candidates": [
"claude-opus-xhigh",
"codex-terra-high"
],
"rule_id": "worker-cloud-g07-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"cloud-G08": {
"candidates": [
"claude-opus-xhigh",
"codex-terra-high"
],
"rule_id": "worker-cloud-g08-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"cloud-G09": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "worker-cloud-g09-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
},
"cloud-G10": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "worker-cloud-g10-catalog",
"policy_priority": 30,
"reason_codes": [
"worker_catalog_lane"
]
}
},
"review": {
"local-G01": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-local-g01-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"local-G02": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-local-g02-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"local-G03": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-local-g03-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"local-G04": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-local-g04-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"local-G05": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-local-g05-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"local-G06": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-local-g06-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"local-G07": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-local-g07-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"local-G08": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-local-g08-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"local-G09": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-local-g09-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"local-G10": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-local-g10-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"cloud-G01": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-cloud-g01-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"cloud-G02": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-cloud-g02-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"cloud-G03": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-cloud-g03-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"cloud-G04": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-cloud-g04-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"cloud-G05": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-cloud-g05-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"cloud-G06": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-cloud-g06-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"cloud-G07": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-cloud-g07-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"cloud-G08": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-cloud-g08-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"cloud-G09": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-cloud-g09-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
},
"cloud-G10": {
"candidates": [
"codex-sol-xhigh"
],
"rule_id": "review-cloud-g10-catalog",
"policy_priority": 10,
"reason_codes": [
"review_catalog_lane"
]
}
}
},
"promotions": {
"claude-opus-xhigh": "codex-terra-high"
}
}

64
agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_contract.py Normal file
View file

@ -0,0 +1,64 @@
#!/usr/bin/env python3
"""Driver-level validation for operator-defined execution targets."""
from __future__ import annotations
VALID_ADAPTERS = frozenset(
{"pi", "agy", "opencode", "claude-glm", "claude", "codex"}
)
def validate_target_contract(target, path: str, error_type) -> None:
if target.adapter not in VALID_ADAPTERS:
raise error_type(
f"{path}.adapter must be one of {sorted(VALID_ADAPTERS)}; "
"a new adapter requires dispatcher driver support"
)
local = target.adapter == "pi"
expected_class = "local_model" if local else "cloud_model"
if target.execution_class != expected_class:
raise error_type(
f"{path}: {target.adapter} targets must use {expected_class}"
)
if target.selfcheck_required != local:
raise error_type(
f"{path}: selfcheck_required must be {str(local).lower()} "
f"for {target.adapter}"
)
if target.adapter == "pi":
if not target.target.startswith("iop/"):
raise error_type(f"{path}: pi target must start with iop/")
if target.thinking_level is None:
raise error_type(f"{path}: pi target requires thinking_level")
if target.reasoning_effort is not None or target.command_model is not None:
raise error_type(
f"{path}: pi target cannot set reasoning_effort or command_model"
)
elif target.thinking_level is not None:
raise error_type(f"{path}: thinking_level is only valid for pi")
if target.adapter == "agy" and (
target.reasoning_effort is not None or target.command_model is not None
):
raise error_type(
f"{path}: agy target cannot set reasoning_effort or command_model"
)
if target.adapter == "opencode" and (
target.reasoning_effort not in {"medium", "high", "max"}
or target.command_model is None
):
raise error_type(
f"{path}: opencode target requires command_model and "
"medium|high|max reasoning_effort"
)
if target.adapter == "claude-glm" and (
target.command_model is None or target.reasoning_effort != "xhigh"
):
raise error_type(
f"{path}: claude-glm target requires command_model and "
"xhigh reasoning_effort"
)
if target.adapter in {"claude", "codex"} and target.reasoning_effort is None:
raise error_type(
f"{path}: {target.adapter} target requires reasoning_effort"
)

552
agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_policy.py
View file

@ -1,10 +1,16 @@
#!/usr/bin/env python3
"""Pure execution-target policy for Agent Task worker and review stages."""
"""Catalog-backed execution-target policy for Agent Task stages."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
import hashlib
import importlib.util
import json
from pathlib import Path
import sys
from typing import Any
from zoneinfo import ZoneInfo
@ -15,6 +21,12 @@ VALID_STAGES = {"worker", "review"}
VALID_LANES = {"local", "cloud"}
VALID_PI_THINKING_LEVELS = frozenset({"low", "medium", "high"})
VALID_REASONING_EFFORTS = frozenset({"medium", "high", "max", "xhigh"})
VALID_EXECUTION_CLASSES = frozenset({"local_model", "cloud_model"})
CATALOG_SCHEMA_VERSION = 1
CATALOG_PATH = Path(__file__).with_name("execution_target_catalog.json")
TIME_WINDOWS = frozenset(
{"kst-day-[07:00,23:00)", "kst-night-[23:00,07:00)"}
)
@dataclass(frozen=True)
@ -26,6 +38,25 @@ class RouteTarget:
thinking_level: str | None = None
reasoning_effort: str | None = None
command_model: str | None = None
catalog_id: str | None = None
@dataclass(frozen=True)
class LanePolicy:
candidates: tuple[str, ...]
policy_priority: int
rule_id: str | None = None
reason_codes: tuple[str, ...] = ()
time_windows: dict[str, tuple[str, tuple[str, ...]]] | None = None
@dataclass(frozen=True)
class ExecutionTargetCatalog:
schema_version: int
revision: str
targets: dict[str, RouteTarget]
lanes: dict[str, dict[str, LanePolicy]]
promotions: dict[str, str]
@dataclass(frozen=True)
@ -35,84 +66,331 @@ class PolicyDecision:
reason_codes: tuple[str, ...]
time_window: str
candidates: tuple[RouteTarget, ...]
route_id: str
catalog_revision: str
PI_ORNITH = RouteTarget("pi", "iop/ornith:35b", "local_model", True)
AGY_GEMINI_LOW = RouteTarget(
"agy", "Gemini 3.6 Flash (Low)", "cloud_model", False
)
AGY_GEMINI_MEDIUM = RouteTarget(
"agy", "Gemini 3.6 Flash (Medium)", "cloud_model", False
)
AGY_GEMINI_HIGH = RouteTarget(
"agy", "Gemini 3.6 Flash (High)", "cloud_model", False
)
PI_LAGUNA = RouteTarget("pi", "iop/laguna-s:2.1", "local_model", True)
# OpenCode's GLM fallback always runs one reasoning step above the Gemini
# variant it follows: Low -> medium, Medium -> high, High -> max.
OPENCODE_GLM_MEDIUM = RouteTarget(
"opencode",
"glm-5.2",
"cloud_model",
False,
reasoning_effort="medium",
command_model="iop-glm/glm-5.2",
)
OPENCODE_GLM_HIGH = RouteTarget(
"opencode",
"glm-5.2",
"cloud_model",
False,
reasoning_effort="high",
command_model="iop-glm/glm-5.2",
)
OPENCODE_GLM_MAX = RouteTarget(
"opencode",
"glm-5.2",
"cloud_model",
False,
reasoning_effort="max",
command_model="iop-glm/glm-5.2",
)
# Retain the old Claude compatibility target only for persisted locator
# recovery. New policy decisions never select it.
CLAUDE_GLM = RouteTarget(
"claude-glm",
"glm-5.2",
"cloud_model",
False,
# `claude-glm` maps the Claude Code Sonnet alias to the GLM provider.
# Keep the policy/log identity as the actual GLM model while invoking the
# compatibility alias required by Claude Code.
command_model="sonnet",
)
CLAUDE_OPUS = RouteTarget("claude", "claude-opus-4-8", "cloud_model", False)
CLAUDE_HAIKU_XHIGH = RouteTarget(
"claude", "claude-haiku-4-5", "cloud_model", False
)
CODEX_SPARK_XHIGH = RouteTarget(
"codex", "gpt-5.3-codex-spark", "cloud_model", False
)
CODEX_SOL_XHIGH = RouteTarget("codex", "gpt-5.6-sol", "cloud_model", False)
CODEX_TERRA_HIGH = RouteTarget("codex", "gpt-5.6-terra", "cloud_model", False)
class CatalogError(ValueError):
"""Raised when the operator-owned model catalog is malformed."""
CANONICAL_TARGETS = (
PI_ORNITH,
AGY_GEMINI_LOW,
AGY_GEMINI_MEDIUM,
AGY_GEMINI_HIGH,
PI_LAGUNA,
OPENCODE_GLM_MEDIUM,
OPENCODE_GLM_HIGH,
OPENCODE_GLM_MAX,
CLAUDE_GLM,
CLAUDE_OPUS,
CLAUDE_HAIKU_XHIGH,
CODEX_SPARK_XHIGH,
CODEX_SOL_XHIGH,
CODEX_TERRA_HIGH,
)
def _load_target_contract():
module_name = "execution_target_contract"
loaded = sys.modules.get(module_name)
if loaded is not None:
return loaded
path = Path(__file__).with_name("execution_target_contract.py")
spec = importlib.util.spec_from_file_location(module_name, path)
if spec is None or spec.loader is None:
raise CatalogError(f"target contract load failed: {path}")
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
target_contract = _load_target_contract()
VALID_ADAPTERS = target_contract.VALID_ADAPTERS
def _object(value: object, path: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise CatalogError(f"{path} must be an object")
return value
def _non_empty_string(value: object, path: str) -> str:
if not isinstance(value, str) or not value:
raise CatalogError(f"{path} must be a non-empty string")
return value
def _optional_enum(
value: object, allowed: frozenset[str], path: str
) -> str | None:
if value is None:
return None
if not isinstance(value, str) or value not in allowed:
raise CatalogError(f"{path} must be null or one of {sorted(allowed)}")
return value
def _target_from_config(target_id: str, value: object) -> RouteTarget:
path = f"targets.{target_id}"
item = _object(value, path)
allowed = {
"adapter",
"target",
"execution_class",
"selfcheck_required",
"thinking_level",
"reasoning_effort",
"command_model",
}
unknown = sorted(set(item) - allowed)
if unknown:
raise CatalogError(f"{path} has unknown fields: {unknown}")
execution_class = _non_empty_string(
item.get("execution_class"), f"{path}.execution_class"
)
if execution_class not in VALID_EXECUTION_CLASSES:
raise CatalogError(
f"{path}.execution_class must be one of {sorted(VALID_EXECUTION_CLASSES)}"
)
selfcheck = item.get("selfcheck_required")
if not isinstance(selfcheck, bool):
raise CatalogError(f"{path}.selfcheck_required must be a boolean")
command_model = item.get("command_model")
if command_model is not None:
command_model = _non_empty_string(command_model, f"{path}.command_model")
target = RouteTarget(
adapter=_non_empty_string(item.get("adapter"), f"{path}.adapter"),
target=_non_empty_string(item.get("target"), f"{path}.target"),
execution_class=execution_class,
selfcheck_required=selfcheck,
thinking_level=_optional_enum(
item.get("thinking_level"),
VALID_PI_THINKING_LEVELS,
f"{path}.thinking_level",
),
reasoning_effort=_optional_enum(
item.get("reasoning_effort"),
VALID_REASONING_EFFORTS,
f"{path}.reasoning_effort",
),
command_model=command_model,
catalog_id=target_id,
)
target_contract.validate_target_contract(target, path, CatalogError)
return target
def _string_list(value: object, path: str) -> tuple[str, ...]:
if not isinstance(value, list) or not value:
raise CatalogError(f"{path} must be a non-empty array")
values = tuple(
_non_empty_string(entry, f"{path}[{index}]")
for index, entry in enumerate(value)
)
if len(values) != len(set(values)):
raise CatalogError(f"{path} must not contain duplicate ids")
return values
def _lane_from_config(
stage: str,
lane_id: str,
value: object,
targets: dict[str, RouteTarget],
) -> LanePolicy:
path = f"lanes.{stage}.{lane_id}"
item = _object(value, path)
allowed = {
"candidates",
"rule_id",
"policy_priority",
"reason_codes",
"time_windows",
}
unknown = sorted(set(item) - allowed)
if unknown:
raise CatalogError(f"{path} has unknown fields: {unknown}")
candidates = _string_list(item.get("candidates"), f"{path}.candidates")
missing_targets = [target_id for target_id in candidates if target_id not in targets]
if missing_targets:
raise CatalogError(f"{path} references unknown targets: {missing_targets}")
execution_classes = {targets[target_id].execution_class for target_id in candidates}
if len(execution_classes) != 1:
raise CatalogError(
f"{path}.candidates cannot mix local_model and cloud_model targets"
)
priority = item.get("policy_priority")
if isinstance(priority, bool) or not isinstance(priority, int) or priority < 0:
raise CatalogError(f"{path}.policy_priority must be a non-negative integer")
raw_reasons = item.get("reason_codes", [])
if not isinstance(raw_reasons, list) or any(
not isinstance(reason, str) or not reason for reason in raw_reasons
):
raise CatalogError(f"{path}.reason_codes must be an array of strings")
rule_id = item.get("rule_id")
if rule_id is not None:
rule_id = _non_empty_string(rule_id, f"{path}.rule_id")
raw_windows = item.get("time_windows")
windows: dict[str, tuple[str, tuple[str, ...]]] | None = None
if raw_windows is not None:
if rule_id is not None or raw_reasons:
raise CatalogError(
f"{path}: time_windows cannot be combined with base rule metadata"
)
windows_obj = _object(raw_windows, f"{path}.time_windows")
if set(windows_obj) != TIME_WINDOWS:
raise CatalogError(
f"{path}.time_windows must define exactly {sorted(TIME_WINDOWS)}"
)
windows = {}
for window_name, raw_window in windows_obj.items():
window = _object(raw_window, f"{path}.time_windows.{window_name}")
if set(window) != {"rule_id", "reason_codes"}:
raise CatalogError(
f"{path}.time_windows.{window_name} must contain rule_id and reason_codes"
)
reasons = window.get("reason_codes")
if not isinstance(reasons, list) or not reasons or any(
not isinstance(reason, str) or not reason for reason in reasons
):
raise CatalogError(
f"{path}.time_windows.{window_name}.reason_codes must be a non-empty string array"
)
windows[window_name] = (
_non_empty_string(
window.get("rule_id"),
f"{path}.time_windows.{window_name}.rule_id",
),
tuple(reasons),
)
elif rule_id is None:
raise CatalogError(f"{path}.rule_id is required without time_windows")
return LanePolicy(
candidates=candidates,
policy_priority=priority,
rule_id=rule_id,
reason_codes=tuple(raw_reasons),
time_windows=windows,
)
def _read_catalog_root(path: Path) -> dict[str, Any]:
try:
raw = path.read_text(encoding="utf-8")
except OSError as exc:
raise CatalogError(f"cannot read execution target catalog {path}: {exc}") from exc
try:
data = json.loads(raw)
except json.JSONDecodeError as exc:
raise CatalogError(f"invalid JSON in execution target catalog {path}: {exc}") from exc
root = _object(data, "catalog")
if set(root) != {"schema_version", "targets", "lanes", "promotions"}:
raise CatalogError(
"catalog must contain exactly schema_version, targets, lanes, promotions"
)
if root.get("schema_version") != CATALOG_SCHEMA_VERSION:
raise CatalogError(
f"catalog.schema_version must be {CATALOG_SCHEMA_VERSION}"
)
return root
def load_catalog(path: Path = CATALOG_PATH) -> ExecutionTargetCatalog:
root = _read_catalog_root(path)
raw_targets = _object(root.get("targets"), "targets")
if not raw_targets:
raise CatalogError("targets must not be empty")
targets = {
_non_empty_string(target_id, "targets key"): _target_from_config(
target_id, value
)
for target_id, value in raw_targets.items()
}
identities: dict[tuple[object, ...], str] = {}
for target_id, target in targets.items():
identity = (
target.adapter,
target.target,
target.thinking_level,
target.reasoning_effort,
)
if identity in identities:
raise CatalogError(
f"targets {identities[identity]!r} and {target_id!r} have duplicate runtime identity"
)
identities[identity] = target_id
raw_lanes = _object(root.get("lanes"), "lanes")
if set(raw_lanes) != VALID_STAGES:
raise CatalogError(f"lanes must define exactly {sorted(VALID_STAGES)}")
expected_lane_ids = {
f"{lane}-G{grade:02d}" for lane in VALID_LANES for grade in range(1, 11)
}
lanes: dict[str, dict[str, LanePolicy]] = {}
for stage in sorted(VALID_STAGES):
stage_lanes = _object(raw_lanes.get(stage), f"lanes.{stage}")
if set(stage_lanes) != expected_lane_ids:
missing = sorted(expected_lane_ids - set(stage_lanes))
extra = sorted(set(stage_lanes) - expected_lane_ids)
raise CatalogError(
f"lanes.{stage} must define every grade independently; missing={missing}, extra={extra}"
)
lanes[stage] = {
lane_id: _lane_from_config(stage, lane_id, value, targets)
for lane_id, value in stage_lanes.items()
}
raw_promotions = _object(root.get("promotions"), "promotions")
promotions: dict[str, str] = {}
for source, destination in raw_promotions.items():
source_id = _non_empty_string(source, "promotions key")
destination_id = _non_empty_string(
destination, f"promotions.{source_id}"
)
if source_id not in targets or destination_id not in targets:
raise CatalogError(
f"promotions.{source_id} references an unknown target"
)
if source_id == destination_id:
raise CatalogError(f"promotions.{source_id} cannot point to itself")
if (
targets[source_id].execution_class != "cloud_model"
or targets[destination_id].execution_class != "cloud_model"
):
raise CatalogError("promotions may contain only cloud_model targets")
promotions[source_id] = destination_id
for source_id in promotions:
seen: set[str] = set()
current = source_id
while current in promotions:
if current in seen:
raise CatalogError(f"promotions contain a cycle at {current!r}")
seen.add(current)
current = promotions[current]
normalized = json.dumps(root, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
return ExecutionTargetCatalog(
schema_version=CATALOG_SCHEMA_VERSION,
revision=hashlib.sha256(normalized.encode("utf-8")).hexdigest(),
targets=targets,
lanes=lanes,
promotions=promotions,
)
CATALOG = load_catalog()
CATALOG_REVISION = CATALOG.revision
CATALOG_TARGETS_BY_ID = CATALOG.targets
CANONICAL_TARGETS = tuple(CATALOG.targets.values())
def catalog_target(target_id: str) -> RouteTarget:
try:
return CATALOG.targets[target_id]
except KeyError as exc:
raise CatalogError(f"unknown catalog target: {target_id}") from exc
# Compatibility names remain for persisted-state recovery and focused driver tests.
# Lane membership and order live only in execution_target_catalog.json.
PI_ORNITH = catalog_target("pi-ornith-high")
PI_LAGUNA = catalog_target("pi-laguna-high")
AGY_GEMINI_LOW = catalog_target("agy-gemini-low")
AGY_GEMINI_MEDIUM = catalog_target("agy-gemini-medium")
AGY_GEMINI_HIGH = catalog_target("agy-gemini-high")
OPENCODE_GLM_MEDIUM = catalog_target("opencode-glm-medium")
OPENCODE_GLM_HIGH = catalog_target("opencode-glm-high")
OPENCODE_GLM_MAX = catalog_target("opencode-glm-max")
CLAUDE_GLM = catalog_target("legacy-claude-glm")
CLAUDE_OPUS = catalog_target("claude-opus-xhigh")
CLAUDE_HAIKU_XHIGH = catalog_target("claude-haiku-xhigh")
CODEX_SPARK_XHIGH = catalog_target("codex-spark-xhigh")
CODEX_SOL_XHIGH = catalog_target("codex-sol-xhigh")
CODEX_TERRA_HIGH = catalog_target("codex-terra-high")
def canonical_target(
@ -121,7 +399,11 @@ def canonical_target(
thinking_level: str | None = None,
reasoning_effort: str | None = None,
) -> RouteTarget | None:
"""Resolve one policy-owned adapter + target + thinking identity."""
"""Resolve one catalog target, accepting pre-catalog implicit defaults."""
if adapter == "pi" and thinking_level is None:
thinking_level = "high"
if adapter in {"claude", "claude-glm", "codex"} and reasoning_effort is None:
reasoning_effort = "xhigh"
return next(
(
candidate
@ -138,10 +420,11 @@ def canonical_target(
def promotion_target(current: RouteTarget) -> RouteTarget | None:
"""Return the next target in the policy-owned cloud promotion chain."""
if current == CLAUDE_OPUS:
return CODEX_TERRA_HIGH
return None
"""Return a legacy promotion target declared by the catalog."""
if current.catalog_id is None:
return None
destination = CATALOG.promotions.get(current.catalog_id)
return CATALOG.targets.get(destination) if destination else None
@dataclass(frozen=True)
@ -152,7 +435,7 @@ class QuotaProbeSpec:
def quota_probe_spec(target: RouteTarget) -> QuotaProbeSpec | None:
"""Return the policy-owned quota probe spec for a route target."""
"""Return the driver-owned quota probe spec for a route target."""
if target.execution_class == "local_model":
return None
if target.adapter == "agy":
@ -191,98 +474,25 @@ def _kst_time_window(evaluated_at: datetime) -> str:
def select_policy(
*, stage: str, lane: str, grade: int, evaluated_at: datetime
) -> PolicyDecision:
"""Return the ordered target policy for one initial route evaluation."""
"""Return the ordered targets for one explicit stage/lane/grade entry."""
_validate(stage, lane, grade, evaluated_at)
if stage == "review":
return PolicyDecision(
rule_id="official-review-codex",
policy_priority=10,
reason_codes=("official_review_fixed",),
time_window="not_applicable",
candidates=(CODEX_SOL_XHIGH,),
)
if lane == "local":
if grade <= 6:
return PolicyDecision(
rule_id="worker-local-g01-g06",
policy_priority=30,
reason_codes=("local_low_grade",),
time_window="not_applicable",
candidates=(PI_ORNITH,),
)
if grade <= 8:
time_window = _kst_time_window(evaluated_at)
if time_window == "kst-day-[07:00,23:00)":
rule_id = "worker-local-g07-g08-kst-day"
reason_code = "kst_day_gemini_high"
candidates = (
AGY_GEMINI_HIGH,
OPENCODE_GLM_MAX,
CODEX_TERRA_HIGH,
)
else:
rule_id = "worker-local-g07-g08-kst-night"
reason_code = "kst_night_gemini_high"
candidates = (
AGY_GEMINI_HIGH,
OPENCODE_GLM_MAX,
CODEX_TERRA_HIGH,
)
return PolicyDecision(
rule_id=rule_id,
policy_priority=20,
reason_codes=(reason_code,),
time_window=time_window,
candidates=candidates,
)
return PolicyDecision(
rule_id="worker-local-g09-g10",
policy_priority=30,
reason_codes=("local_high_grade_cloud_target",),
time_window="not_applicable",
candidates=(CLAUDE_OPUS,),
)
if grade <= 2:
candidates = (
CODEX_SPARK_XHIGH,
AGY_GEMINI_LOW,
OPENCODE_GLM_MEDIUM,
CODEX_TERRA_HIGH,
)
rule_id = "worker-cloud-g01-g02"
reason_code = "cloud_spark_priority_grade"
elif grade <= 4:
candidates = (
AGY_GEMINI_MEDIUM,
OPENCODE_GLM_HIGH,
CODEX_TERRA_HIGH,
)
rule_id = "worker-cloud-g03-g04"
reason_code = "cloud_gemini_medium_grade"
elif grade <= 6:
candidates = (
AGY_GEMINI_HIGH,
OPENCODE_GLM_MAX,
CODEX_TERRA_HIGH,
)
rule_id = "worker-cloud-g05-g06"
reason_code = "cloud_gemini_high_grade"
elif grade <= 8:
candidates = (CLAUDE_OPUS,)
rule_id = "worker-cloud-g07-g08"
reason_code = "cloud_opus_grade"
lane_id = f"{lane}-G{grade:02d}"
lane_policy = CATALOG.lanes[stage][lane_id]
if lane_policy.time_windows is not None:
time_window = _kst_time_window(evaluated_at)
rule_id, reason_codes = lane_policy.time_windows[time_window]
else:
candidates = (CODEX_SOL_XHIGH,)
rule_id = "worker-cloud-g09-g10"
reason_code = "cloud_codex_grade"
time_window = "not_applicable"
assert lane_policy.rule_id is not None
rule_id, reason_codes = lane_policy.rule_id, lane_policy.reason_codes
return PolicyDecision(
rule_id=rule_id,
policy_priority=30,
reason_codes=(reason_code,),
time_window="not_applicable",
candidates=candidates,
policy_priority=lane_policy.policy_priority,
reason_codes=reason_codes,
time_window=time_window,
candidates=tuple(
CATALOG.targets[target_id] for target_id in lane_policy.candidates
),
route_id=f"{stage}:{lane_id}",
catalog_revision=CATALOG.revision,
)

308
agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_specs.py Normal file
View file

@ -0,0 +1,308 @@
#!/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,
)

347
agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_state.py Normal file
View file

@ -0,0 +1,347 @@
#!/usr/bin/env python3
"""Persisted execution-target snapshot validation helpers."""
from __future__ import annotations
from datetime import datetime
ERROR_CODE = "malformed_prior_decision"
def runtime_key(entry: dict) -> tuple[object, ...]:
return (
entry.get("adapter"),
entry.get("target"),
entry.get("thinking_level"),
entry.get("reasoning_effort"),
)
def route_key(target) -> tuple[object, ...]:
return (
target.adapter,
target.target,
target.thinking_level,
target.reasoning_effort,
)
def _validate_selected_target(prior_decision, canonical_targets, policy, error_type):
selected = prior_decision.get("selected")
if not isinstance(selected, dict):
raise error_type(ERROR_CODE, "prior_decision.selected must be an object")
selected_key = runtime_key(selected)
matching = policy.canonical_target(*selected_key)
if matching is None:
raise error_type(
ERROR_CODE,
f"prior_decision.selected {selected_key} is not a policy-owned target",
)
if (
selected.get("execution_class") != matching.execution_class
or selected.get("selfcheck_required") != matching.selfcheck_required
):
raise error_type(
ERROR_CODE,
f"prior_decision.selected attributes do not match canonical target for {selected_key}",
)
if isinstance(prior_decision.get("catalog"), dict) and (
selected.get("target_id") != matching.catalog_id
or selected.get("command_model") != matching.command_model
):
raise error_type(
ERROR_CODE,
f"prior_decision.selected catalog attributes do not match canonical target for {selected_key}",
)
return selected_key, matching, [route_key(target) for target in canonical_targets]
def _validate_promotion_history(
prior_decision, matching, canonical_keys, policy, error_type
):
promotion_path = prior_decision.get("promotion_path")
if not isinstance(promotion_path, list) or len(promotion_path) < 2:
raise error_type(
ERROR_CODE, "promoted prior_decision requires promotion_path evidence"
)
path_targets = []
for index, entry in enumerate(promotion_path):
if not isinstance(entry, dict):
raise error_type(ERROR_CODE, f"promotion_path[{index}] must be an object")
target = policy.canonical_target(*runtime_key(entry))
if target is None:
raise error_type(ERROR_CODE, f"promotion_path[{index}] is not policy-owned")
path_targets.append(target)
if route_key(path_targets[0]) not in set(canonical_keys):
raise error_type(
ERROR_CODE, "promotion_path must begin at the initial policy target"
)
for previous, current in zip(path_targets, path_targets[1:]):
if policy.promotion_target(previous) != current:
raise error_type(
ERROR_CODE, "promotion_path contains a non-adjacent transition"
)
if path_targets[-1] != matching:
raise error_type(ERROR_CODE, "promotion_path tail does not match selected target")
if "used_candidates" in prior_decision:
raise error_type(
ERROR_CODE, "promotion decision must not carry failover used_candidates"
)
def _validate_used_history(prior_decision, selected_key, canonical_keys, error_type):
used = prior_decision["used_candidates"]
if not isinstance(used, list):
raise error_type(ERROR_CODE, "prior_decision.used_candidates must be a list")
used_keys = []
for index, entry in enumerate(used):
if not isinstance(entry, dict):
raise error_type(
ERROR_CODE,
f"prior_decision.used_candidates[{index}] must be an object",
)
key = runtime_key(entry)
if key not in set(canonical_keys):
raise error_type(
ERROR_CODE,
f"prior_decision.used_candidates[{index}] {key} is not in canonical policy targets {set(canonical_keys)}",
)
used_keys.append(key)
if len(used_keys) != len(set(used_keys)):
raise error_type(
ERROR_CODE, "prior_decision.used_candidates contains duplicate targets"
)
if [canonical_keys.index(key) for key in used_keys] != sorted(
canonical_keys.index(key) for key in used_keys
):
raise error_type(
ERROR_CODE,
"prior_decision.used_candidates order does not match candidate rank order",
)
if used_keys and selected_key != used_keys[-1]:
raise error_type(
ERROR_CODE,
f"prior_decision.selected {selected_key} does not match tail of used_candidates {used_keys[-1]}",
)
def _validate_selected_and_history(
prior_decision, canonical_targets, policy, error_type
):
selected_key, matching, canonical_keys = _validate_selected_target(
prior_decision, canonical_targets, policy, error_type
)
if selected_key not in set(canonical_keys):
_validate_promotion_history(
prior_decision, matching, canonical_keys, policy, error_type
)
return
if "used_candidates" in prior_decision:
_validate_used_history(
prior_decision, selected_key, canonical_keys, error_type
)
return
eligible = [
runtime_key(candidate)
for candidate in prior_decision.get("candidates", [])
if isinstance(candidate, dict) and candidate.get("eligibility") == "eligible"
]
if eligible and selected_key != eligible[0]:
raise error_type(
ERROR_CODE,
f"prior_decision.selected {selected_key} does not match first eligible candidate {eligible[0]} when used_candidates is absent",
)
def _validate_pinned_catalog_snapshot(
prior_decision, error_type, validate_used_candidates
):
candidates = prior_decision.get("candidates")
selected = prior_decision.get("selected")
if not isinstance(candidates, list) or not isinstance(selected, dict):
raise error_type(ERROR_CODE, "pinned catalog snapshot is incomplete")
keys = [runtime_key(candidate) for candidate in candidates]
if len(keys) != len(set(keys)):
raise error_type(ERROR_CODE, "pinned catalog snapshot has duplicate targets")
selected_key = runtime_key(selected)
if selected_key not in keys:
raise error_type(
ERROR_CODE, "pinned selected target is not present in the candidate snapshot"
)
selected_candidate = candidates[keys.index(selected_key)]
identity_fields = (
"target_id", "adapter", "target", "execution_class",
"selfcheck_required", "thinking_level", "reasoning_effort", "command_model",
)
for field in identity_fields:
if selected.get(field) != selected_candidate.get(field):
raise error_type(
ERROR_CODE,
f"pinned selected.{field} does not match its candidate snapshot",
)
if "promotion_path" in prior_decision:
raise error_type(
ERROR_CODE,
"catalog-backed decisions must express fallback in the lane candidate array",
)
if "used_candidates" not in prior_decision:
eligible = [
runtime_key(candidate)
for candidate in candidates
if candidate.get("eligibility") == "eligible"
]
if not eligible or selected_key != eligible[0]:
raise error_type(
ERROR_CODE, "selected target is not the first eligible pinned candidate"
)
return
used = validate_used_candidates(prior_decision.get("used_candidates"))
used_keys = [runtime_key(entry) for entry in used]
if len(used_keys) != len(set(used_keys)):
raise error_type(ERROR_CODE, "used_candidates contains duplicate targets")
if any(key not in keys for key in used_keys):
raise error_type(
ERROR_CODE, "used_candidates contains a target outside the pinned snapshot"
)
if [keys.index(key) for key in used_keys] != sorted(
keys.index(key) for key in used_keys
):
raise error_type(
ERROR_CODE, "used_candidates order does not match the pinned candidate order"
)
if not used_keys or used_keys[-1] != selected_key:
raise error_type(
ERROR_CODE, "selected target does not match used_candidates tail"
)
def _validate_catalog(catalog, expected_route_id, policy, error_type):
if not isinstance(catalog, dict):
return False
if (
catalog.get("schema_version") != policy.CATALOG_SCHEMA_VERSION
or not isinstance(catalog.get("revision"), str)
or not catalog.get("revision")
):
raise error_type(
ERROR_CODE,
"prior_decision.catalog must contain the current schema_version and a non-empty revision",
)
if catalog.get("route_id") != expected_route_id:
raise error_type(
ERROR_CODE,
f"prior_decision.catalog.route_id ({catalog.get('route_id')!r}) does not match {expected_route_id!r}",
)
return catalog.get("revision") != policy.CATALOG.revision
def _canonical_decision(decision_info, stage, lane, grade, policy, error_type):
evaluated_at = decision_info.get("evaluated_at")
if not isinstance(evaluated_at, str):
raise error_type(
ERROR_CODE, "prior_decision.decision.evaluated_at must be a string"
)
try:
parsed = datetime.fromisoformat(evaluated_at)
except (ValueError, TypeError) as exc:
raise error_type(
ERROR_CODE,
f"prior_decision.decision.evaluated_at is not a valid ISO datetime: {evaluated_at!r}",
) from exc
if parsed.tzinfo is None or parsed.utcoffset() is None:
raise error_type(
ERROR_CODE,
f"prior_decision.decision.evaluated_at must be timezone-aware: {evaluated_at!r}",
)
try:
return policy.select_policy(
stage=stage, lane=lane, grade=grade, evaluated_at=parsed
)
except ValueError as exc:
raise error_type(ERROR_CODE, str(exc)) from exc
def _validate_decision_metadata(decision_info, canonical, error_type):
expected = {
"rule_id": canonical.rule_id,
"policy_priority": canonical.policy_priority,
"reason_codes": list(canonical.reason_codes),
"time_window": canonical.time_window,
}
actual = {
"rule_id": decision_info.get("rule_id"),
"policy_priority": decision_info.get("policy_priority"),
"reason_codes": list(decision_info.get("reason_codes", [])),
"time_window": decision_info.get("time_window"),
}
for field, expected_value in expected.items():
if actual[field] != expected_value:
raise error_type(
ERROR_CODE,
f"prior_decision.decision.{field} ({actual[field]!r}) does not match canonical policy ({expected_value!r})",
)
def _validate_candidates(prior_decision, canonical_targets, catalog, error_type):
candidates = prior_decision.get("candidates")
if not isinstance(candidates, list) or len(candidates) != len(canonical_targets):
actual_length = len(candidates) if isinstance(candidates, list) else 0
raise error_type(
ERROR_CODE,
f"prior_decision.candidates length ({actual_length}) does not match canonical policy candidates length ({len(canonical_targets)})",
)
for index, (candidate, target) in enumerate(zip(candidates, canonical_targets)):
if not isinstance(candidate, dict):
raise error_type(
ERROR_CODE, f"prior_decision.candidates[{index}] must be an object"
)
expected = {
"adapter": target.adapter,
"target": target.target,
"execution_class": target.execution_class,
"selfcheck_required": target.selfcheck_required,
"thinking_level": target.thinking_level,
"reasoning_effort": target.reasoning_effort,
}
if isinstance(catalog, dict):
expected.update(
target_id=target.catalog_id, command_model=target.command_model
)
if any(candidate.get(field) != value for field, value in expected.items()):
raise error_type(
ERROR_CODE,
f"prior_decision.candidates[{index}] identity ({candidate.get('adapter')}, {candidate.get('target')}) does not match canonical policy candidate ({target.adapter}, {target.target})",
)
def validate_prior_candidate_identity(
prior_decision,
*,
stage,
lane,
grade,
policy,
error_type,
validate_used_candidates,
):
decision_info = prior_decision.get("decision")
if not isinstance(decision_info, dict):
raise error_type(ERROR_CODE, "prior_decision.decision must be an object")
catalog = prior_decision.get("catalog")
changed = _validate_catalog(
catalog, f"{stage}:{lane}-G{grade:02d}", policy, error_type
)
if changed:
_validate_pinned_catalog_snapshot(
prior_decision, error_type, validate_used_candidates
)
return
canonical = _canonical_decision(
decision_info, stage, lane, grade, policy, error_type
)
_validate_decision_metadata(decision_info, canonical, error_type)
_validate_candidates(prior_decision, canonical.candidates, catalog, error_type)
_validate_selected_and_history(
prior_decision, canonical.candidates, policy, error_type
)

506
agent-ops/skills/project/orchestrate-agent-task-loop/scripts/select_execution_target.py
View file

@ -73,6 +73,19 @@ def _load_policy():
policy = _load_policy()
def _load_state_validation():
path = Path(__file__).resolve().parent / "execution_target_state.py"
spec = importlib.util.spec_from_file_location("execution_target_state", path)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
state_validation = _load_state_validation()
class SelectorInputError(Exception):
"""Input contract violation returned as stderr JSON with a non-zero exit."""
@ -176,6 +189,13 @@ def _validate_prior_selected(selected: object) -> None:
"prior_decision.selected.execution_class must be one of "
f"{sorted(_VALID_EXECUTION_CLASSES)}",
)
for field in ("target_id", "command_model"):
value = selected.get(field)
if value is not None and (not isinstance(value, str) or not value):
raise SelectorInputError(
code,
f"prior_decision.selected.{field} must be null or a non-empty string",
)
if not isinstance(selected.get("selfcheck_required"), bool):
raise SelectorInputError(
code,
@ -235,6 +255,16 @@ def _require_nullable_string(
)
def _require_optional_non_empty_string(
container: dict, field: str, prefix: str, code: str
) -> None:
value = container.get(field)
if value is not None and (not isinstance(value, str) or not value):
raise SelectorInputError(
code, f"{prefix}.{field} must be null or a non-empty string"
)
def _validate_prior_candidates(candidates: object) -> None:
"""Validate every reused candidate against the initial output schema.
@ -265,6 +295,8 @@ def _validate_prior_candidates(candidates: object) -> None:
)
_require_non_empty_string(entry, "adapter", prefix, code)
_require_non_empty_string(entry, "target", prefix, code)
for field in ("target_id", "command_model"):
_require_optional_non_empty_string(entry, field, prefix, code)
_require_string_enum(
entry, "execution_class", _VALID_EXECUTION_CLASSES, prefix, code
)
@ -414,6 +446,21 @@ def _validate_prior_decision(value: object) -> dict:
_validate_prior_candidates(value["candidates"])
_validate_prior_decision_evidence(value["decision"])
_validate_prior_quota(value["quota"])
catalog = value.get("catalog")
if catalog is not None:
if not isinstance(catalog, dict):
raise SelectorInputError(code, "prior_decision.catalog must be an object")
if catalog.get("schema_version") != policy.CATALOG_SCHEMA_VERSION:
raise SelectorInputError(
code,
"prior_decision.catalog.schema_version does not match the supported catalog schema",
)
for field in ("revision", "route_id"):
item = catalog.get(field)
if not isinstance(item, str) or not item:
raise SelectorInputError(
code, f"prior_decision.catalog.{field} must be a non-empty string"
)
return value
@ -839,6 +886,24 @@ def _selected_quota(
}
def _selected_fields(target) -> dict:
def value(field: str):
return target.get(field) if isinstance(target, dict) else getattr(target, field)
selected = {
field: value(field)
for field in ("adapter", "target", "execution_class", "selfcheck_required")
}
for field in ("target_id", "command_model", "thinking_level", "reasoning_effort"):
source_field = (
"catalog_id" if field == "target_id" and not isinstance(target, dict) else field
)
field_value = value(source_field)
if field_value is not None:
selected[field] = field_value
return selected
def _initial(
*,
work_unit_id: str,
@ -864,20 +929,13 @@ def _initial(
)
eligible = status != "exhausted"
candidate = {
**_selected_fields(target),
"candidate_rank": rank,
"adapter": target.adapter,
"target": target.target,
"execution_class": target.execution_class,
"selfcheck_required": target.selfcheck_required,
"quota_mode": mode,
"quota_status": status,
"eligibility": "eligible" if eligible else "ineligible",
"rejection_reason": None if eligible else "quota_exhausted",
}
if target.thinking_level is not None:
candidate["thinking_level"] = target.thinking_level
if target.reasoning_effort is not None:
candidate["reasoning_effort"] = target.reasoning_effort
candidates.append(candidate)
if eligible and selected is None:
selected = target
@ -887,22 +945,18 @@ def _initial(
"no_eligible_target",
"all policy candidates are exhausted according to the quota snapshot",
)
selected_fields = {
"adapter": selected.adapter,
"target": selected.target,
"execution_class": selected.execution_class,
"selfcheck_required": selected.selfcheck_required,
}
if selected.thinking_level is not None:
selected_fields["thinking_level"] = selected.thinking_level
if selected.reasoning_effort is not None:
selected_fields["reasoning_effort"] = selected.reasoning_effort
selected_fields = _selected_fields(selected)
return {
"schema_version": SCHEMA_VERSION,
"work_unit_id": work_unit_id,
"stage": stage,
"lane": lane,
"grade": grade,
"catalog": {
"schema_version": policy.CATALOG_SCHEMA_VERSION,
"revision": decision.catalog_revision,
"route_id": decision.route_id,
},
"selected": selected_fields,
"candidates": candidates,
"decision": {
@ -927,147 +981,8 @@ def _initial(
def _validate_selected_and_used_history(
prior_decision: dict,
canonical_targets: list,
) -> None:
code = "malformed_prior_decision"
selected = prior_decision.get("selected")
if not isinstance(selected, dict):
raise SelectorInputError(code, "prior_decision.selected must be an object")
sel_key = (
selected.get("adapter"),
selected.get("target"),
selected.get("thinking_level"),
selected.get("reasoning_effort"),
)
canon_keys_list = [
(c.adapter, c.target, c.thinking_level, c.reasoning_effort)
for c in canonical_targets
]
canon_keys_set = set(canon_keys_list)
matching_cand = policy.canonical_target(*sel_key)
if matching_cand is None:
raise SelectorInputError(
code,
f"prior_decision.selected {sel_key} is not a policy-owned target",
)
if (
selected.get("execution_class") != matching_cand.execution_class
or selected.get("selfcheck_required") != matching_cand.selfcheck_required
):
raise SelectorInputError(
code,
f"prior_decision.selected attributes do not match canonical target for {sel_key}",
)
if sel_key not in canon_keys_set:
promotion_path = prior_decision.get("promotion_path")
if not isinstance(promotion_path, list) or len(promotion_path) < 2:
raise SelectorInputError(
code,
"promoted prior_decision requires promotion_path evidence",
)
path_targets = []
for index, entry in enumerate(promotion_path):
if not isinstance(entry, dict):
raise SelectorInputError(
code, f"promotion_path[{index}] must be an object"
)
target = policy.canonical_target(
entry.get("adapter"),
entry.get("target"),
entry.get("thinking_level"),
entry.get("reasoning_effort"),
)
if target is None:
raise SelectorInputError(
code, f"promotion_path[{index}] is not policy-owned"
)
path_targets.append(target)
if (
path_targets[0].adapter,
path_targets[0].target,
path_targets[0].thinking_level,
path_targets[0].reasoning_effort,
) not in canon_keys_set:
raise SelectorInputError(
code, "promotion_path must begin at the initial policy target"
)
for previous, current in zip(path_targets, path_targets[1:]):
if policy.promotion_target(previous) != current:
raise SelectorInputError(
code, "promotion_path contains a non-adjacent transition"
)
if path_targets[-1] != matching_cand:
raise SelectorInputError(
code, "promotion_path tail does not match selected target"
)
if "used_candidates" in prior_decision:
raise SelectorInputError(
code, "promotion decision must not carry failover used_candidates"
)
return
if "used_candidates" in prior_decision:
used = prior_decision["used_candidates"]
if not isinstance(used, list):
raise SelectorInputError(code, "prior_decision.used_candidates must be a list")
used_keys = []
for idx, entry in enumerate(used):
if not isinstance(entry, dict):
raise SelectorInputError(
code, f"prior_decision.used_candidates[{idx}] must be an object"
)
u_key = (
entry.get("adapter"),
entry.get("target"),
entry.get("thinking_level"),
entry.get("reasoning_effort"),
)
if u_key not in canon_keys_set:
raise SelectorInputError(
code,
f"prior_decision.used_candidates[{idx}] {u_key} is not in canonical policy targets {canon_keys_set}",
)
used_keys.append(u_key)
if len(used_keys) != len(set(used_keys)):
raise SelectorInputError(
code, "prior_decision.used_candidates contains duplicate targets"
)
indices = [canon_keys_list.index(k) for k in used_keys]
if indices != sorted(indices):
raise SelectorInputError(
code, "prior_decision.used_candidates order does not match candidate rank order"
)
if used_keys and sel_key != used_keys[-1]:
raise SelectorInputError(
code,
f"prior_decision.selected {sel_key} does not match tail of used_candidates {used_keys[-1]}",
)
else:
prior_cands = prior_decision.get("candidates", [])
eligible_cands = [
(
c.get("adapter"),
c.get("target"),
c.get("thinking_level"),
c.get("reasoning_effort"),
)
for c in prior_cands
if isinstance(c, dict) and c.get("eligibility") == "eligible"
]
if eligible_cands and sel_key != eligible_cands[0]:
raise SelectorInputError(
code,
f"prior_decision.selected {sel_key} does not match first eligible candidate {eligible_cands[0]} when used_candidates is absent",
)
def _runtime_key(entry: dict) -> tuple[object, ...]:
return state_validation.runtime_key(entry)
def _validate_prior_candidate_identity(
@ -1077,86 +992,15 @@ def _validate_prior_candidate_identity(
lane: str,
grade: int,
) -> None:
code = "malformed_prior_decision"
decision_info = prior_decision.get("decision")
if not isinstance(decision_info, dict):
raise SelectorInputError(code, "prior_decision.decision must be an object")
eval_str = decision_info.get("evaluated_at")
if not isinstance(eval_str, str):
raise SelectorInputError(code, "prior_decision.decision.evaluated_at must be a string")
try:
prior_eval_at = datetime.fromisoformat(eval_str)
except (ValueError, TypeError) as exc:
raise SelectorInputError(
code, f"prior_decision.decision.evaluated_at is not a valid ISO datetime: {eval_str!r}"
) from exc
if prior_eval_at.tzinfo is None or prior_eval_at.utcoffset() is None:
raise SelectorInputError(
code, f"prior_decision.decision.evaluated_at must be timezone-aware: {eval_str!r}"
)
try:
canonical_decision = policy.select_policy(
stage=stage, lane=lane, grade=grade, evaluated_at=prior_eval_at
)
except ValueError as exc:
raise SelectorInputError(code, str(exc)) from exc
if decision_info.get("rule_id") != canonical_decision.rule_id:
raise SelectorInputError(
code,
f"prior_decision.decision.rule_id ({decision_info.get('rule_id')!r}) "
f"does not match canonical policy ({canonical_decision.rule_id!r})",
)
if decision_info.get("policy_priority") != canonical_decision.policy_priority:
raise SelectorInputError(
code,
f"prior_decision.decision.policy_priority ({decision_info.get('policy_priority')!r}) "
f"does not match canonical policy ({canonical_decision.policy_priority!r})",
)
if list(decision_info.get("reason_codes", [])) != list(canonical_decision.reason_codes):
raise SelectorInputError(
code,
f"prior_decision.decision.reason_codes ({decision_info.get('reason_codes')!r}) "
f"does not match canonical policy ({list(canonical_decision.reason_codes)!r})",
)
if decision_info.get("time_window") != canonical_decision.time_window:
raise SelectorInputError(
code,
f"prior_decision.decision.time_window ({decision_info.get('time_window')!r}) "
f"does not match canonical policy ({canonical_decision.time_window!r})",
)
canonical_targets = canonical_decision.candidates
prior_candidates = prior_decision.get("candidates")
if not isinstance(prior_candidates, list) or len(prior_candidates) != len(canonical_targets):
raise SelectorInputError(
code,
f"prior_decision.candidates length ({len(prior_candidates) if isinstance(prior_candidates, list) else 0}) "
f"does not match canonical policy candidates length ({len(canonical_targets)})",
)
for idx, (p_cand, c_target) in enumerate(zip(prior_candidates, canonical_targets)):
if not isinstance(p_cand, dict):
raise SelectorInputError(code, f"prior_decision.candidates[{idx}] must be an object")
if (
p_cand.get("adapter") != c_target.adapter
or p_cand.get("target") != c_target.target
or p_cand.get("execution_class") != c_target.execution_class
or p_cand.get("selfcheck_required") != c_target.selfcheck_required
or p_cand.get("thinking_level") != c_target.thinking_level
or p_cand.get("reasoning_effort") != c_target.reasoning_effort
):
raise SelectorInputError(
code,
f"prior_decision.candidates[{idx}] identity ({p_cand.get('adapter')}, {p_cand.get('target')}) "
f"does not match canonical policy candidate ({c_target.adapter}, {c_target.target})",
)
_validate_selected_and_used_history(prior_decision, canonical_targets)
state_validation.validate_prior_candidate_identity(
prior_decision,
stage=stage,
lane=lane,
grade=grade,
policy=policy,
error_type=SelectorInputError,
validate_used_candidates=_validate_used_candidates,
)
def _resume(
@ -1195,6 +1039,11 @@ def _resume(
"stage": stage,
"lane": lane,
"grade": grade,
**(
{"catalog": prior_decision["catalog"]}
if "catalog" in prior_decision
else {}
),
"selected": selected,
"candidates": prior_decision["candidates"],
"decision": decision,
@ -1284,15 +1133,7 @@ def _failover(
used = _validate_used_candidates(prior.get("used_candidates"))
if previous not in used:
used.append(previous)
used_set = {
(
entry["adapter"],
entry["target"],
entry.get("thinking_level"),
entry.get("reasoning_effort"),
)
for entry in used
}
used_set = {_runtime_key(entry) for entry in used}
selected_candidate = None
selected_probed_snapshot = None
candidates = []
@ -1325,30 +1166,13 @@ def _failover(
current["eligibility"] = "ineligible" if status == "exhausted" else "eligible"
current["rejection_reason"] = "quota_exhausted" if status == "exhausted" else None
candidates.append(current)
key = (
current["adapter"],
current["target"],
current.get("thinking_level"),
current.get("reasoning_effort"),
)
key = _runtime_key(current)
if index > previous_index and key not in used_set and current["eligibility"] == "eligible" and selected_candidate is None:
selected_candidate = current
selected_probed_snapshot = current_snapshot
if selected_candidate is None:
raise SelectorInputError("no_failover_candidate", "no unused eligible candidate remains for this work unit")
selected = {
field: selected_candidate[field]
for field in (
"adapter",
"target",
"execution_class",
"selfcheck_required",
)
}
if selected_candidate.get("thinking_level") is not None:
selected["thinking_level"] = selected_candidate["thinking_level"]
if selected_candidate.get("reasoning_effort") is not None:
selected["reasoning_effort"] = selected_candidate["reasoning_effort"]
selected = _selected_fields(selected_candidate)
next_target = _target_ref(selected)
used.append(next_target)
decision = dict(prior["decision"])
@ -1357,6 +1181,7 @@ def _failover(
return {
"schema_version": SCHEMA_VERSION, "work_unit_id": work_unit_id, "stage": stage,
"lane": lane, "grade": grade, "selected": selected, "candidates": candidates,
**({"catalog": prior["catalog"]} if "catalog" in prior else {}),
"decision": decision,
"quota": _selected_quota(
selected_target,
@ -1459,22 +1284,8 @@ def _promotion(
"stage": stage,
"lane": lane,
"grade": grade,
"selected": {
"adapter": promoted.adapter,
"target": promoted.target,
"execution_class": promoted.execution_class,
"selfcheck_required": promoted.selfcheck_required,
**(
{"thinking_level": promoted.thinking_level}
if promoted.thinking_level is not None
else {}
),
**(
{"reasoning_effort": promoted.reasoning_effort}
if promoted.reasoning_effort is not None
else {}
),
},
**({"catalog": prior["catalog"]} if "catalog" in prior else {}),
"selected": _selected_fields(promoted),
"candidates": prior["candidates"],
"decision": decision,
"promotion_path": promotion_path,
@ -1492,6 +1303,74 @@ def _promotion(
}
def select_execution_target_for_route(
*,
work_unit_id: str,
stage: str,
lane: str,
grade: int,
evaluated_at: datetime,
transition: str = "initial",
prior_decision: dict | None = None,
quota_snapshot: dict | None = None,
quota_probe_command: str = DEFAULT_QUOTA_PROBE_COMMAND,
failure_class: str | None = None,
) -> dict:
"""Select a target from an already validated task generation identity."""
if not isinstance(work_unit_id, str) or not work_unit_id:
raise SelectorInputError(
"invalid_work_unit_id", "work_unit_id must be a non-empty string"
)
_validate_evaluated_at(evaluated_at)
try:
policy._validate(stage, lane, grade, evaluated_at)
except ValueError as exc:
raise SelectorInputError("invalid_route", str(exc)) from exc
quota_snapshot = _validate_quota_snapshot(quota_snapshot)
if not isinstance(quota_probe_command, str) or not quota_probe_command:
raise SelectorInputError(
"invalid_quota_probe_command",
"quota_probe_command must be a non-empty string",
)
if transition == "initial":
return _initial(
work_unit_id=work_unit_id,
stage=stage,
lane=lane,
grade=grade,
evaluated_at=evaluated_at,
quota_snapshot=quota_snapshot,
quota_probe_command=quota_probe_command,
)
if transition == "resume":
return _resume(
prior_decision,
work_unit_id=work_unit_id,
stage=stage,
lane=lane,
grade=grade,
)
if transition == "failover":
return _failover(
prior_decision, work_unit_id=work_unit_id, stage=stage,
lane=lane, grade=grade, evaluated_at=evaluated_at,
quota_snapshot=quota_snapshot, quota_probe_command=quota_probe_command,
failure_class=failure_class,
)
if transition == "promotion":
return _promotion(
prior_decision, work_unit_id=work_unit_id, stage=stage,
lane=lane, grade=grade, evaluated_at=evaluated_at,
quota_snapshot=quota_snapshot,
quota_probe_command=quota_probe_command,
failure_class=failure_class,
)
raise SelectorInputError(
"invalid_transition", f"unknown transition: {transition!r}"
)
def select_execution_target(
task_file: Path,
*,
@ -1503,8 +1382,7 @@ def select_execution_target(
quota_probe_command: str = DEFAULT_QUOTA_PROBE_COMMAND,
failure_class: str | None = None,
) -> dict:
"""Return the stable JSON-serializable selector decision for one call."""
"""Parse one task file and return a stable selector decision."""
kind, lane, grade = _parse_filename(task_file)
prefix_stage = _STAGE_BY_KIND[kind]
if stage is not None and stage != prefix_stage:
@ -1512,53 +1390,17 @@ def select_execution_target(
"stage_mismatch",
f"explicit stage {stage!r} conflicts with filename stage {prefix_stage!r}",
)
resolved_stage = stage or prefix_stage
header = _parse_header(task_file)
work_unit_id = _work_unit_id(header)
_validate_evaluated_at(evaluated_at)
quota_snapshot = _validate_quota_snapshot(quota_snapshot)
if not isinstance(quota_probe_command, str) or not quota_probe_command:
raise SelectorInputError(
"invalid_quota_probe_command",
"quota_probe_command must be a non-empty string",
)
if transition == "initial":
return _initial(
work_unit_id=work_unit_id,
stage=resolved_stage,
lane=lane,
grade=grade,
evaluated_at=evaluated_at,
quota_snapshot=quota_snapshot,
quota_probe_command=quota_probe_command,
)
if transition == "resume":
return _resume(
prior_decision,
work_unit_id=work_unit_id,
stage=resolved_stage,
lane=lane,
grade=grade,
)
if transition == "failover":
return _failover(
prior_decision, work_unit_id=work_unit_id, stage=resolved_stage,
lane=lane, grade=grade, evaluated_at=evaluated_at,
quota_snapshot=quota_snapshot, quota_probe_command=quota_probe_command,
failure_class=failure_class,
)
if transition == "promotion":
return _promotion(
prior_decision, work_unit_id=work_unit_id, stage=resolved_stage,
lane=lane, grade=grade, evaluated_at=evaluated_at,
quota_snapshot=quota_snapshot,
quota_probe_command=quota_probe_command,
failure_class=failure_class,
)
raise SelectorInputError(
"invalid_transition", f"unknown transition: {transition!r}"
return select_execution_target_for_route(
work_unit_id=_work_unit_id(_parse_header(task_file)),
stage=stage or prefix_stage,
lane=lane,
grade=grade,
evaluated_at=evaluated_at,
transition=transition,
prior_decision=prior_decision,
quota_snapshot=quota_snapshot,
quota_probe_command=quota_probe_command,
failure_class=failure_class,
)

290
agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py
View file

@ -8068,6 +8068,52 @@ class RouteDecisionPersistenceTest(unittest.TestCase):
)
return next(task for task in dispatch.scan_tasks(workspace, None) if task.name == name)
def test_catalog_target_edit_keeps_running_task_spec_snapshot(self):
evaluated = datetime(2026, 7, 26, 14, 0, tzinfo=dispatch.KST)
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
(workspace / ".git").mkdir()
task = self.make_task(workspace, lane="cloud", grade=3)
selector = dispatch._selector_module()
quota = {
"schema_version": "1.0",
"snapshot_id": "catalog-pin-test",
"source": "test",
"checked_at": evaluated.isoformat(),
"targets": [],
"required_caps": [],
"reason_codes": [],
}
initial = dispatch.select_execution_decision(
task,
stage="worker",
evaluated_at=evaluated,
quota_snapshot=quota,
)
data = json.loads(
selector.policy.CATALOG_PATH.read_text(encoding="utf-8")
)
data["targets"]["agy-gemini-medium"]["target"] = (
"Gemini replacement model"
)
catalog_path = workspace / "changed-catalog.json"
catalog_path.write_text(json.dumps(data), encoding="utf-8")
changed = selector.policy.load_catalog(catalog_path)
with mock.patch.object(selector.policy, "CATALOG", changed):
resumed = dispatch.select_execution_decision(
task,
stage="worker",
evaluated_at=evaluated,
transition="resume",
prior_decision=initial,
)
spec = dispatch.agent_spec_from_decision(resumed)
self.assertTrue(resumed["decision"]["pinned"])
self.assertEqual(spec.cli, "agy")
self.assertEqual(spec.model, "Gemini 3.6 Flash (Medium)")
def test_reopen_body_edit_and_generation_reset_preserve_or_reset_pin(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
@ -8099,7 +8145,7 @@ class RouteDecisionPersistenceTest(unittest.TestCase):
reopened, reopened_task, stage="worker"
)
self.assertEqual(resumed["transition"]["trigger"], "resume")
self.assertEqual(resumed_spec.display, "pi/iop/ornith:35b")
self.assertEqual(resumed_spec.display, "pi/iop/ornith:35b high")
assert reopened_task.plan is not None
reopened_task.plan.write_text(
@ -8195,6 +8241,10 @@ class RouteDecisionPersistenceTest(unittest.TestCase):
"adapter": "claude",
"target": "claude-opus-4-8",
"status": "exhausted",
}, {
"adapter": "codex",
"target": "gpt-5.6-terra",
"status": "exhausted",
}],
},
)
@ -8851,7 +8901,11 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
self.assertEqual(
decision["used_candidates"],
[
{"adapter": "codex", "target": "gpt-5.3-codex-spark"},
{
"adapter": "codex",
"target": "gpt-5.3-codex-spark",
"reasoning_effort": "xhigh",
},
{"adapter": "agy", "target": "Gemini 3.6 Flash (Low)"},
{
"adapter": "opencode",
@ -8863,6 +8917,106 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
finally:
store.close()
async def test_archived_review_recovery_uses_review_lane_fallback(self):
evaluated = datetime(2026, 7, 26, 14, 0, tzinfo=dispatch.KST)
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
(workspace / ".git").mkdir()
active = self.make_task(workspace, lane="local", grade=8)
assert active.plan is not None and active.review is not None
active.review.write_text(
active.review.read_text(encoding="utf-8")
+ "\n## 코드리뷰 결과\n- 종합 판정: FAIL\n",
encoding="utf-8",
)
active.plan.rename(active.directory / "plan_local_G08_0.log")
active.review.rename(
active.directory / "code_review_local_G08_0.log"
)
task = next(
item
for item in dispatch.scan_tasks(workspace, None)
if item.name == active.name
)
self.assertTrue(task.recovery)
self.assertIsNone(task.plan)
self.assertIsNone(task.review)
selector = dispatch._selector_module()
data = json.loads(
selector.policy.CATALOG_PATH.read_text(encoding="utf-8")
)
data["lanes"]["review"]["local-G08"]["candidates"] = [
"codex-sol-xhigh",
"claude-haiku-xhigh",
]
catalog_path = workspace / "review-catalog.json"
catalog_path.write_text(json.dumps(data), encoding="utf-8")
changed = selector.policy.load_catalog(catalog_path)
store = dispatch.StateStore(workspace)
try:
with mock.patch.object(selector.policy, "CATALOG", changed):
decision, initial_spec = dispatch.persisted_execution_decision(
store,
task,
stage="review",
evaluated_at=evaluated,
)
self.assertEqual(len(decision["candidates"]), 2)
locators = {
"codex": self.make_attempt_locator(
workspace, task, initial_spec
),
"claude": self.make_attempt_locator(
workspace,
task,
dispatch.AgentSpec(
"claude",
"claude-haiku-4-5",
"claude/claude-haiku-4-5 xhigh",
reasoning_effort="xhigh",
),
),
}
invoked = []
async def fake_invoke(*args, **kwargs):
spec = args[4]
invoked.append(spec)
if spec.cli == "codex":
return 1, "provider-quota", locators["codex"]
return 0, None, locators["claude"]
with (
mock.patch.object(dispatch, "invoke", new=fake_invoke),
mock.patch.object(
dispatch,
"build_context_package",
side_effect=AssertionError(
"review fallback must restart from review artifacts"
),
),
mock.patch.object(
dispatch.asyncio, "sleep", new=mock.AsyncMock()
),
):
success, locator = await dispatch.run_escalating(
workspace,
store,
task,
"review",
initial_spec,
)
self.assertTrue(success)
self.assertEqual(locator, locators["claude"])
self.assertEqual([spec.cli for spec in invoked], ["codex", "claude"])
selected = store.task_state(task)["execution_decisions"]["review"]
self.assertEqual(selected["selected"]["target_id"], "claude-haiku-xhigh")
self.assertEqual(selected["transition"]["trigger"], "provider-quota")
finally:
store.close()
async def test_invalid_logical_context_does_not_commit_or_promote(self):
daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9)))
@ -8985,7 +9139,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
finally:
store.close()
async def test_cloud_g07_provider_quota_promotes_claude_to_codex_without_no_failover_block(self):
async def test_cloud_g07_provider_quota_follows_lane_array_to_codex(self):
daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9)))
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
@ -9054,7 +9208,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
decision = state["execution_decisions"]["worker"]
self.assertEqual(decision["selected"]["adapter"], "codex")
self.assertEqual(decision["selected"]["target"], "gpt-5.6-terra")
self.assertEqual(decision["transition"]["kind"], "promotion")
self.assertNotIn("kind", decision["transition"])
self.assertEqual(
decision["transition"]["trigger"], "provider-quota"
)
@ -9999,8 +10153,8 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
self.assertTrue(task.recovery)
store = dispatch.StateStore(workspace)
try:
# 1. Active review uses the PLAN generation/route but the fixed
# official-review policy and a complete canonical schema.
# 1. Active review uses the PLAN generation/route and that
# review lane's catalog candidates in a canonical schema.
dec_rev, spec_rev = dispatch.persisted_execution_decision(
store, task, stage="review", evaluated_at=daytime
)
@ -10009,12 +10163,13 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
self.assertEqual(dec_rev["lane"], "local")
self.assertEqual(dec_rev["grade"], 8)
self.assertEqual(
dec_rev["decision"]["rule_id"], "official-review-codex"
dec_rev["decision"]["rule_id"],
"review-local-g08-catalog",
)
self.assertEqual(dec_rev["decision"]["policy_priority"], 10)
self.assertEqual(
dec_rev["decision"]["reason_codes"],
["official_review_fixed"],
["review_catalog_lane"],
)
self.assertEqual(dec_rev["decision"]["timezone"], "Asia/Seoul")
self.assertFalse(dec_rev["decision"]["pinned"])
@ -10024,7 +10179,7 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
"snapshot_id": None,
"mode": "bounded",
"status": "unknown",
"source": "official_review_fixed_policy",
"source": "official_review_catalog_policy",
"checked_at": None,
"targets": [],
},
@ -10056,8 +10211,8 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
self.assertEqual(reused, dec_rev)
self.assertEqual(reused_spec, spec_rev)
# A qualified cloud failure restarts the same fixed Codex target
# without selector failover, promotion, quota probe, or local CLI.
# With one configured candidate, a qualified cloud failure
# retries that candidate without selecting an unavailable next one.
retry_locator = self.make_attempt_locator(
workspace, task, spec_rev
)
@ -10077,7 +10232,7 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
dispatch,
"select_execution_decision",
side_effect=AssertionError(
"fixed review recovery must not reselect"
"single-candidate review recovery must not reselect"
),
) as selector_mock,
mock.patch.object(
@ -10119,14 +10274,14 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
# 4. Audit consumers read canonical nested decision/quota and
# only expose legacy flat fields through read-only fallback.
evidence = dispatch.selector_evidence_lines(dec_rev)
self.assertIn("rule_id=official-review-codex", evidence)
self.assertIn("rule_id=review-local-g08-catalog", evidence)
self.assertIn("priority=10", evidence)
self.assertIn("transition=initial", evidence)
self.assertIn("quota_status=unknown", evidence)
status = dispatch.status_lines(
task, "review", "ready", decision=dec_rev
)
self.assertIn("rule_id=official-review-codex", status)
self.assertIn("rule_id=review-local-g08-catalog", status)
runtime_evidence = dispatch.selector_runtime_evidence(dec_rev)
self.assertIn("decision", runtime_evidence)
self.assertIn("quota", runtime_evidence)
@ -10489,9 +10644,12 @@ class ThroughputQuotaBatchTest(unittest.TestCase):
batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now)
self.assertIsNotNone(batch_snap)
# Cloud G7 candidate target: claude/claude-opus-4-8
# Total unique probe keys = 1. Probed EXACTLY 1 time across all 3 tasks!
self.assertEqual(len(probe_calls), 1)
# Every unique target in the shared lane is probed once across all tasks.
self.assertEqual(len(probe_calls), 2)
self.assertEqual(
len({(call["adapter"], call["target"]) for call in probe_calls}),
2,
)
# Evaluate decisions for all tasks using batch_snap
d1, _ = dispatch.persisted_execution_decision(store, t1, stage="worker", quota_snapshot=batch_snap)
@ -10546,8 +10704,8 @@ class ThroughputQuotaBatchTest(unittest.TestCase):
batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now)
self.assertIsNotNone(batch_snap)
# Ensure no duplicate probes were called and exactly 2 unique targets were probed
self.assertEqual(len(probed_keys), 2)
# cloud-G07 contributes Claude/Terra and cloud-G09 adds Sol.
self.assertEqual(len(probed_keys), 3)
self.assertEqual(len(probed_keys), len(set(probed_keys)))
d1, _ = dispatch.persisted_execution_decision(store, t1, stage="worker", quota_snapshot=batch_snap)
@ -10630,14 +10788,77 @@ class ThroughputQuotaBatchTest(unittest.TestCase):
ready = [(t_night, "worker"), (t_review, "review")]
batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now)
# Night local-G08 probes Gemini High plus its Codex Terra backup.
# OpenCode GLM has no dedicated quota probe; review remains excluded.
# Night local-G08 probes Gemini High plus Terra; the independent
# review lane contributes its configured target once.
self.assertIsNotNone(batch_snap)
self.assertEqual(len(probe_calls), 2)
self.assertEqual(len(probe_calls), 3)
self.assertEqual(probe_calls[0]["adapter"], "agy")
self.assertEqual(probe_calls[0]["target"], "Gemini 3.6 Flash (High)")
self.assertEqual(probe_calls[1]["adapter"], "codex")
self.assertEqual(probe_calls[1]["target"], "gpt-5.6-terra")
self.assertEqual(probe_calls[2]["adapter"], "codex")
self.assertEqual(probe_calls[2]["target"], "gpt-5.6-sol")
finally:
store.close()
def test_review_batch_quota_selects_next_catalog_candidate(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
(workspace / ".git").mkdir()
task = self.make_task(
workspace, "route/01_review", lane="local", grade=8
)
selector = dispatch._selector_module()
data = json.loads(
selector.policy.CATALOG_PATH.read_text(encoding="utf-8")
)
data["lanes"]["review"]["local-G08"]["candidates"] = [
"codex-sol-xhigh",
"claude-haiku-xhigh",
]
catalog_path = workspace / "review-catalog.json"
catalog_path.write_text(json.dumps(data), encoding="utf-8")
changed = selector.policy.load_catalog(catalog_path)
store = dispatch.StateStore(workspace)
def probe(*args, **kwargs):
adapter, target = kwargs["adapter"], kwargs["target"]
status = "exhausted" if adapter == "codex" else "available"
return {
"schema_version": "1.0",
"snapshot_id": f"review-{adapter}",
"source": "test",
"checked_at": kwargs["checked_at"].isoformat(),
"targets": [
{"adapter": adapter, "target": target, "status": status}
],
"required_caps": [],
"reason_codes": [],
}
try:
with (
mock.patch.object(selector.policy, "CATALOG", changed),
mock.patch.object(
selector, "probe_candidate_quota", side_effect=probe
) as probe_mock,
):
evaluated = datetime(2026, 7, 26, 14, 0, tzinfo=dispatch.KST)
snapshot = dispatch.build_admission_batch_snapshot(
store, [(task, "review")], evaluated
)
decision, spec = dispatch.persisted_execution_decision(
store,
task,
stage="review",
evaluated_at=evaluated,
quota_snapshot=snapshot,
)
self.assertEqual(probe_mock.call_count, 2)
self.assertEqual(decision["selected"]["target_id"], "claude-haiku-xhigh")
self.assertEqual(decision["quota"]["status"], "available")
self.assertEqual(spec.cli, "claude")
finally:
store.close()
@ -11021,16 +11242,23 @@ class ThroughputQuotaBatchTest(unittest.TestCase):
ready = [(t1, "worker")]
retry_batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now)
self.assertIsNone(retry_batch_snap)
self.assertEqual(len(probe_calls), 0)
self.assertIsNotNone(retry_batch_snap)
self.assertEqual(len(probe_calls), 1)
self.assertEqual(
(probe_calls[0]["adapter"], probe_calls[0]["target"]),
("codex", "gpt-5.6-terra"),
)
# With no persisted unused alternate, retry consumes no quota snapshot and resumes.
# The retry refresh observes the persisted unused lane alternate.
with mock.patch("subprocess.run", side_effect=AssertionError) as run:
d, spec = dispatch.persisted_execution_decision(
store, t1, stage="worker", quota_snapshot=retry_batch_snap
)
run.assert_not_called()
self.assertEqual(store.task_state(t1).get("quota_snapshot")["snapshot_id"], "snap-initial")
self.assertEqual(
store.task_state(t1).get("quota_snapshot")["snapshot_id"],
retry_batch_snap["snapshot_id"],
)
# retry context is preserved through decision commit so that
# invoke() can read handoff_id and atomically consume it.
# In production run_worker() always calls invoke() after this.
@ -11163,9 +11391,13 @@ class ThroughputQuotaBatchTest(unittest.TestCase):
res = await dispatch.dispatch_with_store(args, workspace, store)
run_sub.assert_not_called()
self.assertEqual(len(probe_calls), 1)
self.assertEqual(probe_calls[0]["adapter"], "codex")
self.assertEqual(probe_calls[0]["target"], "gpt-5.6-terra")
self.assertEqual(
{(call["adapter"], call["target"]) for call in probe_calls},
{
("codex", "gpt-5.6-terra"),
("codex", "gpt-5.6-sol"),
},
)
st_blocked_after = store.task_state(t_blocked)
self.assertIsNone(st_blocked_after.get("blocked"))

111
agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_execution_target_policy.py
View file

@ -1,5 +1,7 @@
import importlib.util
import json
import sys
from tempfile import TemporaryDirectory
import unittest
from unittest import mock
from datetime import datetime, timezone
@ -23,6 +25,100 @@ def at_utc(hour: int, minute: int = 0, second: int = 0) -> datetime:
class ExecutionTargetPolicyTests(unittest.TestCase):
def test_catalog_defines_every_stage_lane_grade_independently(self):
expected = {
f"{lane}-G{grade:02d}"
for lane in policy.VALID_LANES
for grade in range(1, 11)
}
self.assertEqual(set(policy.CATALOG.lanes), policy.VALID_STAGES)
for stage in policy.VALID_STAGES:
with self.subTest(stage=stage):
self.assertEqual(set(policy.CATALOG.lanes[stage]), expected)
def test_changing_one_lane_candidate_array_requires_no_python_change(self):
data = json.loads(policy.CATALOG_PATH.read_text(encoding="utf-8"))
data["lanes"]["worker"]["cloud-G03"]["candidates"] = [
"codex-sol-xhigh",
"agy-gemini-medium",
]
with TemporaryDirectory() as tmp:
path = Path(tmp) / "catalog.json"
path.write_text(json.dumps(data), encoding="utf-8")
catalog = policy.load_catalog(path)
with mock.patch.object(policy, "CATALOG", catalog):
decision = policy.select_policy(
stage="worker", lane="cloud", grade=3, evaluated_at=at_utc(3)
)
self.assertEqual(
[target.catalog_id for target in decision.candidates],
["codex-sol-xhigh", "agy-gemini-medium"],
)
def test_catalog_rejects_a_missing_grade_lane(self):
data = json.loads(policy.CATALOG_PATH.read_text(encoding="utf-8"))
del data["lanes"]["worker"]["cloud-G03"]
with TemporaryDirectory() as tmp:
path = Path(tmp) / "catalog.json"
path.write_text(json.dumps(data), encoding="utf-8")
with self.assertRaisesRegex(
policy.CatalogError, "define every grade independently"
):
policy.load_catalog(path)
def test_catalog_owns_explicit_driver_options(self):
self.assertEqual(policy.PI_ORNITH.thinking_level, "high")
self.assertEqual(policy.PI_LAGUNA.thinking_level, "high")
for target in (
policy.CLAUDE_GLM,
policy.CLAUDE_OPUS,
policy.CLAUDE_HAIKU_XHIGH,
policy.CODEX_SPARK_XHIGH,
policy.CODEX_SOL_XHIGH,
):
with self.subTest(target=target.catalog_id):
self.assertEqual(target.reasoning_effort, "xhigh")
def test_catalog_rejects_unsupported_runtime_combinations(self):
cases = (
(
"unknown adapter",
lambda data: data["targets"]["agy-gemini-low"].update(
adapter="unknown-cli"
),
"new adapter requires dispatcher driver support",
),
(
"implicit pi thinking",
lambda data: data["targets"]["pi-ornith-high"].pop(
"thinking_level"
),
"requires thinking_level",
),
(
"mixed execution classes",
lambda data: data["lanes"]["worker"]["local-G01"].update(
candidates=["pi-ornith-high", "codex-sol-xhigh"]
),
"cannot mix local_model and cloud_model",
),
(
"promotion cycle",
lambda data: data["promotions"].update(
{"codex-terra-high": "claude-opus-xhigh"}
),
"contain a cycle",
),
)
for name, mutate, message in cases:
with self.subTest(name=name), TemporaryDirectory() as tmp:
data = json.loads(policy.CATALOG_PATH.read_text(encoding="utf-8"))
mutate(data)
path = Path(tmp) / "catalog.json"
path.write_text(json.dumps(data), encoding="utf-8")
with self.assertRaisesRegex(policy.CatalogError, message):
policy.load_catalog(path)
def test_local_g07_route_uses_kst_boundaries(self):
cases = [
(at_utc(21, 59, 59), "agy", "Gemini 3.6 Flash (High)", "kst-night-[23:00,07:00)"),
@ -48,7 +144,7 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
decision = policy.select_policy(
stage="worker", lane="local", grade=8, evaluated_at=night_time
)
self.assertEqual(decision.rule_id, "worker-local-g07-g08-kst-night")
self.assertEqual(decision.rule_id, "worker-local-g08-kst-night-catalog")
self.assertEqual(
decision.candidates,
(
@ -130,10 +226,10 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
)
self.assertEqual(
decision.reason_codes,
("cloud_spark_priority_grade",),
("worker_catalog_lane",),
)
def test_review_matrix_is_fixed_to_codex(self):
def test_review_catalog_defines_every_lane(self):
for lane in ("local", "cloud"):
for grade in range(1, 11):
with self.subTest(lane=lane, grade=grade):
@ -143,7 +239,14 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
grade=grade,
evaluated_at=at_utc(3),
)
self.assertEqual(decision.rule_id, "official-review-codex")
self.assertEqual(
decision.rule_id,
f"review-{lane}-g{grade:02d}-catalog",
)
self.assertEqual(
decision.reason_codes,
("review_catalog_lane",),
)
self.assertEqual(decision.candidates, (policy.CODEX_SOL_XHIGH,))
def test_local_g07_g08_candidate_order_uses_gemini_high_then_glm_max(self):

124
agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_select_execution_target.py
View file

@ -122,6 +122,39 @@ MALFORMED_NESTED_VARIANTS = [
class SelectorContractTests(unittest.TestCase):
def test_catalog_edit_applies_to_new_work_and_keeps_prior_route_pinned(self):
with TemporaryDirectory() as tmp:
root = Path(tmp)
task_file = write_task_file(root, "PLAN", "cloud", 3)
prior = selector.select_execution_target(
task_file, evaluated_at=kst(12)
)
data = json.loads(
selector.policy.CATALOG_PATH.read_text(encoding="utf-8")
)
data["lanes"]["worker"]["cloud-G03"]["candidates"] = [
"codex-sol-xhigh",
"agy-gemini-medium",
]
catalog_path = root / "catalog.json"
catalog_path.write_text(json.dumps(data), encoding="utf-8")
changed = selector.policy.load_catalog(catalog_path)
with mock.patch.object(selector.policy, "CATALOG", changed), mock.patch.object(
selector.policy, "CATALOG_REVISION", changed.revision
):
resumed = selector.select_execution_target(
task_file,
evaluated_at=kst(12),
transition="resume",
prior_decision=prior,
)
new_work = selector.select_execution_target(
task_file, evaluated_at=kst(12)
)
self.assertEqual(resumed["selected"], prior["selected"])
self.assertEqual(resumed["candidates"], prior["candidates"])
self.assertEqual(new_work["selected"]["target_id"], "codex-sol-xhigh")
def test_worker_contract_shape_and_types(self):
with TemporaryDirectory() as tmp:
task_file = write_task_file(Path(tmp), "PLAN", "cloud", 7)
@ -143,8 +176,15 @@ class SelectorContractTests(unittest.TestCase):
"target": "claude-opus-4-8",
"execution_class": "cloud_model",
"selfcheck_required": False,
"target_id": "claude-opus-xhigh",
"reasoning_effort": "xhigh",
},
)
self.assertEqual(result["catalog"]["route_id"], "worker:cloud-G07")
self.assertEqual(
result["catalog"]["schema_version"],
selector.policy.CATALOG_SCHEMA_VERSION,
)
for key in ("rule_id", "policy_priority", "reason_codes", "pinned"):
self.assertIn(key, result["decision"])
self.assertIs(result["decision"]["pinned"], False)
@ -579,24 +619,26 @@ class SelectorQuotaRepresentationTests(unittest.TestCase):
for name, status in snapshots.items():
with self.subTest(status=name):
if status == "exhausted":
with self.assertRaises(selector.SelectorInputError) as ctx:
selector.select_execution_target(
cloud,
evaluated_at=kst(12),
quota_snapshot={
"snapshot_id": f"probe-{name}",
"source": "iop-node quota-probe",
"checked_at": "2026-07-25T03:00:00+09:00",
"targets": [
{
"adapter": "claude",
"target": "claude-opus-4-8",
"status": status,
}
],
},
)
self.assertEqual(ctx.exception.code, "no_eligible_target")
result = selector.select_execution_target(
cloud,
evaluated_at=kst(12),
quota_snapshot={
"snapshot_id": f"probe-{name}",
"source": "iop-node quota-probe",
"checked_at": "2026-07-25T03:00:00+09:00",
"targets": [
{
"adapter": "claude",
"target": "claude-opus-4-8",
"status": status,
}
],
},
)
self.assertEqual(result["selected"]["adapter"], "codex")
self.assertEqual(
result["selected"]["target"], "gpt-5.6-terra"
)
continue
result = selector.select_execution_target(
cloud,
@ -639,14 +681,11 @@ class SelectorQuotaRepresentationTests(unittest.TestCase):
"subprocess.run", return_value=completed
):
if status == "exhausted":
with self.assertRaises(
selector.SelectorInputError
) as ctx:
selector.select_execution_target(
cloud, evaluated_at=kst(12)
)
result = selector.select_execution_target(
cloud, evaluated_at=kst(12)
)
self.assertEqual(
ctx.exception.code, "no_eligible_target"
result["selected"]["target"], "gpt-5.6-terra"
)
continue
result = selector.select_execution_target(
@ -702,7 +741,12 @@ class SelectorQuotaRepresentationTests(unittest.TestCase):
"adapter": "claude",
"target": "claude-opus-4-8",
"status": "exhausted",
}
},
{
"adapter": "codex",
"target": "gpt-5.6-terra",
"status": "exhausted",
},
],
}
with TemporaryDirectory() as tmp:
@ -1233,7 +1277,11 @@ class SelectorFailoverContractTests(unittest.TestCase):
self.assertEqual(
glm["used_candidates"],
[
{"adapter": "codex", "target": "gpt-5.3-codex-spark"},
{
"adapter": "codex",
"target": "gpt-5.3-codex-spark",
"reasoning_effort": "xhigh",
},
{"adapter": "agy", "target": "Gemini 3.6 Flash (Low)"},
{
"adapter": "opencode",
@ -1548,7 +1596,7 @@ class SelectorFailoverContractTests(unittest.TestCase):
self.assertEqual(result["quota"]["status"], "unknown")
self.assertEqual(run_mock.call_count, 2)
def test_policy_owned_cloud_promotion_chain_and_no_bounce(self):
def test_policy_owned_cloud_lane_fallback_chain_and_no_bounce(self):
with TemporaryDirectory() as tmp:
task_file = write_task_file(Path(tmp), "PLAN", "cloud", 7)
initial = selector.select_execution_target(
@ -1559,7 +1607,7 @@ class SelectorFailoverContractTests(unittest.TestCase):
terra = selector.select_execution_target(
task_file,
evaluated_at=kst(12),
transition="promotion",
transition="failover",
prior_decision=initial,
failure_class="provider-quota",
)
@ -1573,39 +1621,29 @@ class SelectorFailoverContractTests(unittest.TestCase):
(initial["selected"]["adapter"], initial["selected"]["target"]),
("claude", "claude-opus-4-8"),
)
self.assertEqual(terra["transition"]["kind"], "promotion")
self.assertEqual(terra["transition"]["trigger"], "provider-quota")
self.assertEqual(
(terra["selected"]["adapter"], terra["selected"]["target"]),
("codex", "gpt-5.6-terra"),
)
self.assertEqual(
terra["promotion_path"],
[
{"adapter": "claude", "target": "claude-opus-4-8"},
{"adapter": "codex", "target": "gpt-5.6-terra"},
],
)
with self.assertRaises(selector.SelectorInputError) as exhausted:
selector.select_execution_target(
task_file,
evaluated_at=kst(23),
transition="promotion",
transition="failover",
prior_decision=resumed,
failure_class="provider-quota",
)
self.assertEqual(exhausted.exception.code, "no_promotion_target")
self.assertEqual(exhausted.exception.code, "no_failover_candidate")
with self.assertRaises(selector.SelectorInputError) as generic:
selector.select_execution_target(
task_file,
evaluated_at=kst(12),
transition="promotion",
transition="failover",
prior_decision=initial,
failure_class="generic-error",
)
self.assertEqual(
generic.exception.code, "unqualified_promotion_trigger"
)
self.assertEqual(generic.exception.code, "unqualified_failover_trigger")
def test_probe_candidate_quota_argv_and_normalization(self):
eval_time = kst(14, 0, 0)