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

Merge dev into feature/openai-compatible-output-validation-filters

Browse source
This commit is contained in:
toki 2026-08-05 05:07:28 +09:00
commit 02beab1c2c
9 changed files with 893 additions and 120 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
README.md
View file

@ -33,3 +33,9 @@ make test-e2e
```
Start with `agent-contract/index.md` for protocol and runtime contracts, and `agent-spec/index.md` for living implementation summaries.
Operator and client setup guides:
- [Edge Local Quickstart](docs/edge-local-dev-guide.md)
- [dev OpenCode Settings Guide](docs/dev-opencode-settings-guide.md)
- [dev-corp Pi Settings Guide](docs/dev-corp-pi-settings-guide.md)

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

@ -59,7 +59,7 @@ Treat Korean text inside code spans or fenced examples as exact runtime or file-
## Preconditions
- [ ] Read the current state contracts in `agent-ops/skills/common/plan/SKILL.md` and `agent-ops/skills/common/code-review/SKILL.md`.
- [ ] Verify that `codex`, `claude`, `agy`, and `pi` are on PATH and their login/provider configuration is valid.
- [ ] Verify that `codex`, `claude`, `agy`, `opencode`, and `pi` are on PATH and their login/provider configuration is valid.
- [ ] Limit automatic approval to PLAN execution inside the current workspace; do not expand scope to external-system changes or destructive work.
- [ ] Verify that no other dispatcher is running in the same workspace. Never bypass a workspace-lock failure.
- [ ] Run `--dry-run` before the first live run to inspect active-task classification and dependency state.
@ -69,11 +69,11 @@ Treat Korean text inside code spans or fenced examples as exact runtime or file-
| 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)` → `claude-glm` `glm-5.2` → Codex `gpt-5.6-terra`, reasoning 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)` → `claude-glm` `glm-5.2` → Codex `gpt-5.6-terra`, reasoning high |
| `cloud-G03`–`cloud-G04` | agy `Gemini 3.6 Flash (Medium)` → `claude-glm` `glm-5.2` → Codex `gpt-5.6-terra`, reasoning high |
| `cloud-G05`–`cloud-G06` | agy `Gemini 3.6 Flash (High)` → `claude-glm` `glm-5.2` → Codex `gpt-5.6-terra`, reasoning high |
| `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 |
@ -111,7 +111,7 @@ 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`; `claude-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 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.
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.
@ -229,13 +229,13 @@ 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 `claude-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 Claude and `claude-glm`, classify a `rate_limit_event` with `rate_limit_info.status=rejected`, an error `result` with `api_error_status=429` or `error=rate_limit`, or a `You've hit your session limit · resets ...` terminal diagnostic as `provider-quota`. Cloud-only escalation remains `Claude/claude-glm -> Codex`; never escalate from an assistant message, source text, tool/test output, or a plain quota-configuration string in an AGY log.
- 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.
- 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. Cloud `claude-glm` may fail over to the policy-owned Codex Terra backup.
- 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.
- 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.

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

@ -508,6 +508,8 @@ class AgentSpec:
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
@ -538,9 +540,11 @@ def agent_spec_from_record(record: dict[str, Any]) -> AgentSpec | 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"}:
effort = reasoning_effort or "xhigh"
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)
@ -1609,7 +1613,16 @@ class StageFailureBudget:
count = int(entry.get("count", 0)) + 1
entry.update(
work_unit_id=self.work_unit_id, stage=self.stage, count=count,
last_target={"adapter": target.get("adapter"), "target": target.get("target")},
last_target={
key: target.get(key)
for key in (
"adapter",
"target",
"thinking_level",
"reasoning_effort",
)
if target.get(key) is not None
},
last_transition=transition,
)
budgets[self.key] = entry
@ -1754,6 +1767,7 @@ def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
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
@ -1775,7 +1789,9 @@ def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
lane=decision["lane"],
grade=decision["grade"],
)
canonical = selector.policy.canonical_target(adapter, target, thinking_level)
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 (
@ -1784,10 +1800,15 @@ def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
):
raise ExecutionDecisionError("selector selected가 canonical policy target이 아니다")
initial_keys = {
(item.adapter, item.target, item.thinking_level)
(
item.adapter,
item.target,
item.thinking_level,
item.reasoning_effort,
)
for item in policy_targets
}
if (adapter, target, thinking_level) not in initial_keys:
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가 없다")
@ -1801,6 +1822,7 @@ def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
entry.get("adapter"),
entry.get("target"),
entry.get("thinking_level"),
entry.get("reasoning_effort"),
)
if resolved is None:
raise ExecutionDecisionError(
@ -1811,6 +1833,7 @@ def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
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(
@ -1831,11 +1854,13 @@ def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
local_pi=True,
thinking_level=canonical.thinking_level,
)
if adapter not in {"agy", "claude", "claude-glm", "codex"} or local_pi:
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 = "high" if canonical == selector.policy.CODEX_TERRA_HIGH else None
suffix = " high" if reasoning_effort == "high" else (
reasoning_effort = canonical.reasoning_effort or (
"high" if canonical == selector.policy.CODEX_TERRA_HIGH else None
)
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}"
@ -1868,6 +1893,7 @@ def _spec_from_completing_decision(decision: dict[str, Any]) -> AgentSpec:
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이어야 한다"
@ -1915,7 +1941,7 @@ def _spec_from_completing_decision(decision: dict[str, Any]) -> AgentSpec:
local_pi=True,
thinking_level=thinking_level,
)
if adapter not in {"agy", "claude", "claude-glm", "codex"}:
if adapter not in {"agy", "claude", "claude-glm", "codex", "opencode"}:
raise ExecutionDecisionError(
f"completing decision adapter가 유효하지 않다: {adapter!r}"
)
@ -1939,6 +1965,24 @@ def _spec_from_completing_decision(decision: dict[str, Any]) -> AgentSpec:
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)
@ -2178,8 +2222,12 @@ def selector_evidence_lines(decision: dict[str, Any] | None) -> list[str]:
rank = c.get("candidate_rank", "?")
adapter = c.get("adapter", "?")
target = c.get("target", "?")
effort = c.get("reasoning_effort")
elig = c.get("eligibility", "?")
cand_strs.append(f"#{rank}:{adapter}/{target}({elig})")
effort_suffix = f" {effort}" if effort else ""
cand_strs.append(
f"#{rank}:{adapter}/{target}{effort_suffix}({elig})"
)
reasons = decision_info.get(
"reason_codes", selected.get("reason_codes", [])
@ -2843,6 +2891,13 @@ def terminal_diagnostic(cli: str, channel: str, line: str) -> str | None:
return json.dumps(value, ensure_ascii=False)
if event_type == "system" and subtype.startswith("error"):
return json.dumps(value, ensure_ascii=False)
if cli == "opencode" and event_type.lower() in {
"error",
"session.error",
"request.failed",
"turn.failed",
}:
return json.dumps(value.get("error", value), ensure_ascii=False)
return None
@ -3186,7 +3241,7 @@ def attempt_terminal_diagnostics(
def promoted_spec(spec: AgentSpec, recovery_count: int) -> AgentSpec | None:
if spec.cli == "agy":
return AgentSpec("claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh")
if spec.cli in {"claude", "claude-glm"}:
if spec.cli in {"claude", "claude-glm", "opencode"}:
return AgentSpec(
"codex",
"gpt-5.6-terra",
@ -3228,6 +3283,18 @@ def render_json_line(cli: str, line: str) -> tuple[list[str], str | None]:
if value.get("type") == "result" and value.get("result"):
rendered.extend(str(value["result"]).splitlines())
session_id = session_id or value.get("session_id")
elif cli == "opencode":
part = value.get("part")
if isinstance(part, dict):
if part.get("type") == "text" and part.get("text"):
rendered.extend(str(part["text"]).splitlines())
elif part.get("type") == "tool" and part.get("tool"):
rendered.append(f"tool={part['tool']}")
session_id = (
value.get("sessionID")
or value.get("sessionId")
or value.get("session_id")
)
return rendered, str(session_id) if session_id else None
@ -3835,6 +3902,23 @@ def build_command(
"--effort", str(effective_reasoning_effort(spec)),
"--dangerously-skip-permissions", prompt,
]
if spec.cli == "opencode":
return [
"opencode",
"run",
"--format",
"json",
"--dir",
str(workspace),
"--agent",
"build",
"--model",
spec.command_model or spec.model,
"--variant",
str(effective_reasoning_effort(spec)),
"--auto",
prompt,
]
if spec.cli == "agy":
return [
# `--print` consumes its immediately following argument as the prompt.
@ -3956,7 +4040,11 @@ async def invoke(
"agent_process_marker": process_marker,
"plan_path": str(task.plan) if task.plan else None,
"review_path": str(task.review) if task.review else None,
"session_id": session_id if spec.cli in {"claude", "claude-glm", "pi"} else None,
"session_id": (
session_id
if spec.cli in {"claude", "claude-glm", "opencode", "pi"}
else None
),
"native_session_path": (
str(pi_resume_session)
if pi_resume_session is not None

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

@ -14,6 +14,7 @@ KST = ZoneInfo("Asia/Seoul")
VALID_STAGES = {"worker", "review"}
VALID_LANES = {"local", "cloud"}
VALID_PI_THINKING_LEVELS = frozenset({"low", "medium", "high"})
VALID_REASONING_EFFORTS = frozenset({"medium", "high", "max", "xhigh"})
@dataclass(frozen=True)
@ -23,6 +24,7 @@ class RouteTarget:
execution_class: str
selfcheck_required: bool
thinking_level: str | None = None
reasoning_effort: str | None = None
command_model: str | None = None
@ -46,6 +48,34 @@ 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",
@ -73,6 +103,9 @@ CANONICAL_TARGETS = (
AGY_GEMINI_MEDIUM,
AGY_GEMINI_HIGH,
PI_LAGUNA,
OPENCODE_GLM_MEDIUM,
OPENCODE_GLM_HIGH,
OPENCODE_GLM_MAX,
CLAUDE_GLM,
CLAUDE_OPUS,
CLAUDE_HAIKU_XHIGH,
@ -83,7 +116,10 @@ CANONICAL_TARGETS = (
def canonical_target(
adapter: str, target: str, thinking_level: str | None = None,
adapter: str,
target: str,
thinking_level: str | None = None,
reasoning_effort: str | None = None,
) -> RouteTarget | None:
"""Resolve one policy-owned adapter + target + thinking identity."""
return next(
@ -94,6 +130,7 @@ def canonical_target(
candidate.adapter == adapter
and candidate.target == target
and candidate.thinking_level == thinking_level
and candidate.reasoning_effort == reasoning_effort
)
),
None,
@ -181,11 +218,19 @@ def select_policy(
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, CLAUDE_GLM, CODEX_TERRA_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, CLAUDE_GLM, CODEX_TERRA_HIGH)
candidates = (
AGY_GEMINI_HIGH,
OPENCODE_GLM_MAX,
CODEX_TERRA_HIGH,
)
return PolicyDecision(
rule_id=rule_id,
policy_priority=20,
@ -205,17 +250,25 @@ def select_policy(
candidates = (
CODEX_SPARK_XHIGH,
AGY_GEMINI_LOW,
CLAUDE_GLM,
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, CLAUDE_GLM, CODEX_TERRA_HIGH)
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, CLAUDE_GLM, CODEX_TERRA_HIGH)
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:

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

@ -191,6 +191,16 @@ def _validate_prior_selected(selected: object) -> None:
"prior_decision.selected.thinking_level must be null or one of "
f"{sorted(policy.VALID_PI_THINKING_LEVELS)}",
)
reasoning_effort = selected.get("reasoning_effort")
if reasoning_effort is not None and (
not isinstance(reasoning_effort, str)
or reasoning_effort not in policy.VALID_REASONING_EFFORTS
):
raise SelectorInputError(
code,
"prior_decision.selected.reasoning_effort must be null or one of "
f"{sorted(policy.VALID_REASONING_EFFORTS)}",
)
def _require_non_empty_string(
@ -272,6 +282,16 @@ def _validate_prior_candidates(candidates: object) -> None:
f"{prefix}.thinking_level must be null or one of "
f"{sorted(policy.VALID_PI_THINKING_LEVELS)}",
)
reasoning_effort = entry.get("reasoning_effort")
if reasoning_effort is not None and (
not isinstance(reasoning_effort, str)
or reasoning_effort not in policy.VALID_REASONING_EFFORTS
):
raise SelectorInputError(
code,
f"{prefix}.reasoning_effort must be null or one of "
f"{sorted(policy.VALID_REASONING_EFFORTS)}",
)
_require_string_enum(entry, "quota_mode", _VALID_QUOTA_MODES, prefix, code)
_require_string_enum(
entry, "quota_status", _VALID_QUOTA_STATUSES, prefix, code
@ -856,6 +876,8 @@ def _initial(
}
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
@ -873,6 +895,8 @@ def _initial(
}
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
return {
"schema_version": SCHEMA_VERSION,
"work_unit_id": work_unit_id,
@ -916,9 +940,11 @@ def _validate_selected_and_used_history(
selected.get("adapter"),
selected.get("target"),
selected.get("thinking_level"),
selected.get("reasoning_effort"),
)
canon_keys_list = [
(c.adapter, c.target, c.thinking_level) for c in canonical_targets
(c.adapter, c.target, c.thinking_level, c.reasoning_effort)
for c in canonical_targets
]
canon_keys_set = set(canon_keys_list)
@ -954,6 +980,7 @@ def _validate_selected_and_used_history(
entry.get("adapter"),
entry.get("target"),
entry.get("thinking_level"),
entry.get("reasoning_effort"),
)
if target is None:
raise SelectorInputError(
@ -964,6 +991,7 @@ def _validate_selected_and_used_history(
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"
@ -998,6 +1026,7 @@ def _validate_selected_and_used_history(
entry.get("adapter"),
entry.get("target"),
entry.get("thinking_level"),
entry.get("reasoning_effort"),
)
if u_key not in canon_keys_set:
raise SelectorInputError(
@ -1029,6 +1058,7 @@ def _validate_selected_and_used_history(
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"
@ -1118,6 +1148,7 @@ def _validate_prior_candidate_identity(
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,
@ -1157,7 +1188,7 @@ def _resume(
selected = prior_decision["selected"]
decision = dict(prior_decision["decision"])
decision["pinned"] = True
target_ref = {"adapter": selected["adapter"], "target": selected["target"]}
target_ref = _target_ref(selected)
return {
"schema_version": SCHEMA_VERSION,
"work_unit_id": work_unit_id,
@ -1183,6 +1214,8 @@ def _target_ref(candidate: dict) -> dict:
ref = {"adapter": candidate["adapter"], "target": candidate["target"]}
if candidate.get("thinking_level") is not None:
ref["thinking_level"] = candidate["thinking_level"]
if candidate.get("reasoning_effort") is not None:
ref["reasoning_effort"] = candidate["reasoning_effort"]
return ref
@ -1209,9 +1242,22 @@ def _validate_used_candidates(value: object) -> list[dict]:
f"{index}].thinking_level must be null or one of "
f"{sorted(policy.VALID_PI_THINKING_LEVELS)}",
)
reasoning_effort = entry.get("reasoning_effort")
if reasoning_effort is not None and (
not isinstance(reasoning_effort, str)
or reasoning_effort not in policy.VALID_REASONING_EFFORTS
):
raise SelectorInputError(
"malformed_prior_decision",
"used_candidates["
f"{index}].reasoning_effort must be null or one of "
f"{sorted(policy.VALID_REASONING_EFFORTS)}",
)
ref = {"adapter": adapter, "target": target}
if thinking_level is not None:
ref["thinking_level"] = thinking_level
if reasoning_effort is not None:
ref["reasoning_effort"] = reasoning_effort
refs.append(ref)
return refs
@ -1238,7 +1284,15 @@ def _failover(
used = _validate_used_candidates(prior.get("used_candidates"))
if previous not in used:
used.append(previous)
used_set = {(entry["adapter"], entry["target"]) for entry in used}
used_set = {
(
entry["adapter"],
entry["target"],
entry.get("thinking_level"),
entry.get("reasoning_effort"),
)
for entry in used
}
selected_candidate = None
selected_probed_snapshot = None
candidates = []
@ -1271,7 +1325,12 @@ 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"])
key = (
current["adapter"],
current["target"],
current.get("thinking_level"),
current.get("reasoning_effort"),
)
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
@ -1288,6 +1347,8 @@ def _failover(
}
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"]
next_target = _target_ref(selected)
used.append(next_target)
decision = dict(prior["decision"])
@ -1359,6 +1420,7 @@ def _promotion(
prior["selected"]["adapter"],
prior["selected"]["target"],
prior["selected"].get("thinking_level"),
prior["selected"].get("reasoning_effort"),
)
promoted = policy.promotion_target(current) if current is not None else None
if promoted is None:
@ -1371,6 +1433,7 @@ def _promotion(
"adapter": current.adapter,
"target": current.target,
"thinking_level": current.thinking_level,
"reasoning_effort": current.reasoning_effort,
}
)
next_target = _target_ref(
@ -1378,6 +1441,7 @@ def _promotion(
"adapter": promoted.adapter,
"target": promoted.target,
"thinking_level": promoted.thinking_level,
"reasoning_effort": promoted.reasoning_effort,
}
)
promotion_path = list(prior.get("promotion_path", [previous_target]))
@ -1405,6 +1469,11 @@ def _promotion(
if promoted.thinking_level is not None
else {}
),
**(
{"reasoning_effort": promoted.reasoning_effort}
if promoted.reasoning_effort is not None
else {}
),
},
"candidates": prior["candidates"],
"decision": decision,

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

@ -142,6 +142,36 @@ class CommandConstructionTest(unittest.TestCase):
effort_index = command.index("--effort")
self.assertEqual(command[effort_index + 1], "xhigh")
def test_opencode_glm_uses_provider_model_and_requested_variant(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
for effort in ("medium", "high", "max"):
with self.subTest(effort=effort):
command = dispatch.build_command(
dispatch.AgentSpec(
"opencode",
"glm-5.2",
f"opencode/glm-5.2 {effort}",
reasoning_effort=effort,
command_model="iop-glm/glm-5.2",
),
"Implement the active plan.",
workspace,
"test-session",
workspace / "attempt",
)
self.assertEqual(command[:2], ["opencode", "run"])
self.assertEqual(command[command.index("--format") + 1], "json")
self.assertEqual(command[command.index("--dir") + 1], str(workspace))
self.assertEqual(command[command.index("--agent") + 1], "build")
self.assertEqual(
command[command.index("--model") + 1],
"iop-glm/glm-5.2",
)
self.assertEqual(command[command.index("--variant") + 1], effort)
self.assertIn("--auto", command)
class TaskStageTest(unittest.TestCase):
def make_task(self, root: Path, review_text: str = ""):
@ -3346,6 +3376,30 @@ class ReviewControlTest(unittest.TestCase):
dispatch.terminal_diagnostic("agy", "stdout", event)
)
def test_opencode_error_event_is_terminal_diagnostic(self):
event = json.dumps(
{
"type": "error",
"sessionID": "session-opencode",
"error": {"statusCode": 429, "message": "quota exhausted"},
}
)
diagnostic = dispatch.terminal_diagnostic("opencode", "stdout", event)
self.assertIsNotNone(diagnostic)
self.assertIn("quota exhausted", diagnostic)
def test_opencode_json_event_preserves_text_and_session(self):
event = json.dumps(
{
"type": "text",
"sessionID": "session-opencode",
"part": {"type": "text", "text": "first\nsecond"},
}
)
rendered, session_id = dispatch.render_json_line("opencode", event)
self.assertEqual(rendered, ["first", "second"])
self.assertEqual(session_id, "session-opencode")
def test_agy_log_diagnostic_requires_strong_quota_evidence(self):
with tempfile.TemporaryDirectory() as temporary:
log = Path(temporary) / "agy-cli.log"
@ -8503,10 +8557,16 @@ class DynamicFailoverBudgetTest(unittest.TestCase):
workspace = Path(temporary)
(workspace / ".git").mkdir()
task = self.make_task(workspace)
gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)")
gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)")
locator_gemini = self.make_attempt_locator(workspace, task, gemini_spec)
laguna_spec = dispatch.AgentSpec("pi", "laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True)
locator_laguna = self.make_attempt_locator(workspace, task, laguna_spec)
glm_spec = dispatch.AgentSpec(
"opencode",
"glm-5.2",
"opencode/glm-5.2 max",
reasoning_effort="max",
command_model="iop-glm/glm-5.2",
)
locator_glm = self.make_attempt_locator(workspace, task, glm_spec)
initial_store = dispatch.StateStore(workspace)
try:
@ -8523,7 +8583,7 @@ class DynamicFailoverBudgetTest(unittest.TestCase):
if spec.cli == "agy":
return (1, "provider-quota", locator_gemini)
if len(invoked_specs) == 2:
return (1, "generic-error", locator_laguna)
return (1, "generic-error", locator_glm)
raise asyncio.CancelledError()
with (
@ -8539,7 +8599,7 @@ class DynamicFailoverBudgetTest(unittest.TestCase):
decisions1 = state1["execution_decisions"]
history1 = state1["route_transition_history"]
worker_budget1 = dispatch.StageFailureBudget.from_decision(store, task, decisions1["worker"])
self.assertEqual([s.cli for s in invoked_specs[:2]], ["agy", "claude-glm"])
self.assertEqual([s.cli for s in invoked_specs[:2]], ["agy", "opencode"])
self.assertEqual([h["transition"] for h in history1], ["initial", "provider-quota"])
self.assertEqual(worker_budget1.count(), 2)
finally:
@ -8563,11 +8623,11 @@ class DynamicFailoverBudgetTest(unittest.TestCase):
mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()),
):
success, final_loc = asyncio.run(
dispatch.run_escalating(workspace, reopened, task, "worker", laguna_spec)
dispatch.run_escalating(workspace, reopened, task, "worker", glm_spec)
)
self.assertFalse(success)
self.assertEqual(invoke.await_count, 8)
self.assertTrue(all(call.args[4] == laguna_spec for call in invoke.await_args_list))
self.assertTrue(all(call.args[4] == glm_spec for call in invoke.await_args_list))
self.assertEqual(final_loc, locators[-1])
self.assertIn("recovery failure limit exhausted", reopened.task_state(task)["blocked"])
@ -8578,7 +8638,14 @@ class DynamicFailoverBudgetTest(unittest.TestCase):
self.assertEqual(worker_budget2.count(), 10)
self.assertEqual(review_budget2.count(), 0)
raw_entry = state2.get("stage_failure_budgets", {}).get(worker_budget2.key, {})
self.assertEqual(raw_entry.get("last_target"), {"adapter": "claude-glm", "target": "glm-5.2"})
self.assertEqual(
raw_entry.get("last_target"),
{
"adapter": "opencode",
"target": "glm-5.2",
"reasoning_effort": "max",
},
)
self.assertEqual(raw_entry.get("last_transition"), "provider-quota")
self.assertEqual(state2["execution_decisions"], decisions1)
self.assertEqual([h["transition"] for h in state2["route_transition_history"]], ["initial", "provider-quota"])
@ -8684,7 +8751,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
locator.write_text(json.dumps(record), encoding="utf-8")
return locator
async def test_cloud_g01_g02_quota_failover_runs_spark_gemini_glm_low(self):
async def test_cloud_g01_g02_quota_failover_runs_spark_gemini_glm_medium(self):
daytime = datetime(
2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))
)
@ -8733,11 +8800,12 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
"Gemini 3.6 Flash (Low)",
"agy/Gemini 3.6 Flash (Low)",
),
"claude-glm": dispatch.AgentSpec(
"claude-glm",
"opencode": dispatch.AgentSpec(
"opencode",
"glm-5.2",
"claude-glm/glm-5.2 xhigh",
command_model="sonnet",
"opencode/glm-5.2 medium",
reasoning_effort="medium",
command_model="iop-glm/glm-5.2",
),
}
locators = {
@ -8749,7 +8817,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
async def mock_invoke(*args, **kwargs):
spec = args[4]
invoked_specs.append(spec)
if spec.cli == "claude-glm":
if spec.cli == "opencode":
return 0, None, locators[spec.cli]
return 1, "provider-quota", locators[spec.cli]
@ -8770,13 +8838,13 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
)
self.assertTrue(success)
self.assertEqual(final_locator, locators["claude-glm"])
self.assertEqual(final_locator, locators["opencode"])
self.assertEqual(
[(spec.cli, spec.model) for spec in invoked_specs],
[
("codex", "gpt-5.3-codex-spark"),
("agy", "Gemini 3.6 Flash (Low)"),
("claude-glm", "glm-5.2"),
("opencode", "glm-5.2"),
],
)
decision = store.task_state(task)["execution_decisions"]["worker"]
@ -8785,7 +8853,11 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
[
{"adapter": "codex", "target": "gpt-5.3-codex-spark"},
{"adapter": "agy", "target": "Gemini 3.6 Flash (Low)"},
{"adapter": "claude-glm", "target": "glm-5.2"},
{
"adapter": "opencode",
"target": "glm-5.2",
"reasoning_effort": "medium",
},
],
)
finally:
@ -8864,7 +8936,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
finally:
store.close()
async def test_day_gemini_zero_exit_quota_continues_on_claude_glm_with_logical_context(self):
async def test_day_gemini_zero_exit_quota_continues_on_opencode_glm_with_logical_context(self):
daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9)))
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
@ -8872,7 +8944,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
task = self.make_task(workspace)
store = dispatch.StateStore(workspace)
try:
gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)")
gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)")
locator = self.make_attempt_locator(workspace, task, gemini_spec)
invoked_specs = []
invoked_prompts = []
@ -8896,7 +8968,8 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
self.assertTrue(success)
self.assertEqual(len(invoked_specs), 2)
self.assertEqual(invoked_specs[0].cli, "agy")
self.assertEqual(invoked_specs[1].cli, "claude-glm")
self.assertEqual(invoked_specs[1].cli, "opencode")
self.assertEqual(invoked_specs[1].reasoning_effort, "max")
self.assertFalse(invoked_specs[1].local_pi)
continuation_prompt = invoked_prompts[1]
self.assertIn(str(task.plan.resolve()), continuation_prompt)
@ -8906,7 +8979,8 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
self.assertIn(str((locator.parent / "normalized-output.log").resolve()), continuation_prompt)
state = store.task_state(task)
decisions = state["execution_decisions"]["worker"]
self.assertEqual(decisions["selected"]["adapter"], "claude-glm")
self.assertEqual(decisions["selected"]["adapter"], "opencode")
self.assertEqual(decisions["selected"]["reasoning_effort"], "max")
self.assertEqual(decisions["transition"]["trigger"], "provider-quota")
finally:
store.close()
@ -8996,7 +9070,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
finally:
store.close()
async def test_cloud_agy_quota_failover_commits_glm_high(self):
async def test_cloud_agy_quota_failover_commits_glm_max(self):
daytime = datetime(
2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))
)
@ -9012,10 +9086,11 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
"agy/Gemini 3.6 Flash (High)",
)
glm_spec = dispatch.AgentSpec(
"claude-glm",
"opencode",
"glm-5.2",
"claude-glm/glm-5.2 xhigh",
command_model="sonnet",
"opencode/glm-5.2 max",
reasoning_effort="max",
command_model="iop-glm/glm-5.2",
)
locators = {
spec.cli: self.make_attempt_locator(workspace, task, spec)
@ -9037,7 +9112,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
store, task, decision
)
transition_budget_counts.append(budget.count())
return (0, None, locators["claude-glm"])
return (0, None, locators["opencode"])
with (
mock.patch.object(dispatch, "invoke", new=mock_invoke),
@ -9053,7 +9128,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
)
self.assertTrue(success)
self.assertEqual(final_locator, locators["claude-glm"])
self.assertEqual(final_locator, locators["opencode"])
self.assertEqual(invoked_specs, [agy_spec, glm_spec])
self.assertEqual(transition_budget_counts, [1])
self.assertIn(
@ -9069,8 +9144,9 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
"target": "Gemini 3.6 Flash (High)",
},
{
"adapter": "claude-glm",
"adapter": "opencode",
"target": "glm-5.2",
"reasoning_effort": "max",
},
],
)
@ -9087,7 +9163,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
finally:
store.close()
async def test_night_gemini_quota_fails_over_to_glm_high(self):
async def test_night_gemini_quota_fails_over_to_glm_max(self):
nighttime = datetime(2026, 7, 26, 1, 0, 0, tzinfo=timezone(timedelta(hours=9)))
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
@ -9097,8 +9173,9 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
try:
gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)")
glm_spec = dispatch.AgentSpec(
"claude-glm", "glm-5.2", "claude-glm/glm-5.2 xhigh",
command_model="sonnet",
"opencode", "glm-5.2", "opencode/glm-5.2 max",
reasoning_effort="max",
command_model="iop-glm/glm-5.2",
)
locator = self.make_attempt_locator(workspace, task, gemini_spec)
invoked_specs = []
@ -9122,14 +9199,15 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
self.assertEqual(invoked_specs, [gemini_spec, glm_spec])
state = store.task_state(task)
decisions = state["execution_decisions"]["worker"]
self.assertEqual(decisions["selected"]["adapter"], "claude-glm")
self.assertEqual(decisions["selected"]["adapter"], "opencode")
self.assertEqual(decisions["selected"]["target"], "glm-5.2")
self.assertEqual(decisions["selected"]["reasoning_effort"], "max")
self.assertNotIn("thinking_level", decisions["selected"])
self.assertEqual(decisions["transition"]["trigger"], "provider-quota")
finally:
store.close()
async def test_night_gemini_quota_initially_exhausted_selects_glm_high(self):
async def test_night_gemini_quota_initially_exhausted_selects_glm_max(self):
nighttime = datetime(2026, 7, 26, 1, 0, 0, tzinfo=timezone(timedelta(hours=9)))
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
@ -9138,8 +9216,9 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
store = dispatch.StateStore(workspace)
try:
glm_spec = dispatch.AgentSpec(
"claude-glm", "glm-5.2", "claude-glm/glm-5.2 xhigh",
command_model="sonnet",
"opencode", "glm-5.2", "opencode/glm-5.2 max",
reasoning_effort="max",
command_model="iop-glm/glm-5.2",
)
locator = self.make_attempt_locator(workspace, task, glm_spec)
invoked_specs = []
@ -9195,8 +9274,9 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
)
glm_spec = dispatch.AgentSpec(
"claude-glm", "glm-5.2", "claude-glm/glm-5.2 xhigh",
command_model="sonnet",
"opencode", "glm-5.2", "opencode/glm-5.2 max",
reasoning_effort="max",
command_model="iop-glm/glm-5.2",
)
decision, spec = dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime)
self.assertEqual(spec, glm_spec)
@ -9219,7 +9299,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
async def mock_invoke(*args, **kwargs):
spec = args[4]
invoked_specs.append(spec)
if spec.cli == "claude-glm":
if spec.cli == "opencode":
return (1, "provider-quota", locator)
return (0, None, locator)
@ -9231,7 +9311,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
self.assertTrue(success)
self.assertEqual(len(invoked_specs), 2)
self.assertEqual(invoked_specs[0].cli, "claude-glm")
self.assertEqual(invoked_specs[0].cli, "opencode")
self.assertEqual(invoked_specs[1].cli, "codex")
self.assertEqual(invoked_specs[1].model, "gpt-5.6-terra")
@ -9360,8 +9440,8 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
finally:
store.close()
async def test_cloud_g05_g06_gemini_quota_fails_over_to_glm_high(self):
"""Cloud G05–G06 sends qualified Gemini failures to Claude GLM 5.2."""
async def test_cloud_g05_g06_gemini_quota_fails_over_to_glm_max(self):
"""Cloud G05–G06 sends qualified Gemini failures to OpenCode GLM max."""
daytime = datetime(
2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))
)
@ -9375,8 +9455,9 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
"agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)"
)
glm_spec = dispatch.AgentSpec(
"claude-glm", "glm-5.2", "claude-glm/glm-5.2 xhigh",
command_model="sonnet",
"opencode", "glm-5.2", "opencode/glm-5.2 max",
reasoning_effort="max",
command_model="iop-glm/glm-5.2",
)
loc_agy = self.make_attempt_locator(workspace, task, agy_spec)
loc_glm = self.make_attempt_locator(workspace, task, glm_spec)
@ -9403,7 +9484,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
self.assertTrue(success)
self.assertEqual(
[s.cli for s in invoked_specs],
["agy", "claude-glm"],
["agy", "opencode"],
)
self.assertEqual(invoked_specs[1], glm_spec)
@ -9413,12 +9494,16 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
decision["used_candidates"],
[
{"adapter": "agy", "target": "Gemini 3.6 Flash (High)"},
{"adapter": "claude-glm", "target": "glm-5.2"},
{
"adapter": "opencode",
"target": "glm-5.2",
"reasoning_effort": "max",
},
],
)
transitions = [h["transition"] for h in state["route_transition_history"]]
self.assertIn("provider-quota", transitions)
self.assertEqual(decision["selected"]["adapter"], "claude-glm")
self.assertEqual(decision["selected"]["adapter"], "opencode")
self.assertEqual(decision["selected"]["target"], "glm-5.2")
self.assertNotIn("thinking_level", decision["selected"])
finally:
@ -9734,20 +9819,21 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
task = self.make_task(workspace, lane="local", grade=8)
store = dispatch.StateStore(workspace)
try:
# 1. Initial decision local G08 -> Gemini High, Claude GLM 5.2, Terra High.
# 1. Initial decision local G08 -> Gemini High, OpenCode GLM max, Terra High.
dec1, spec1 = dispatch.persisted_execution_decision(
store, task, stage="worker", evaluated_at=daytime
)
self.assertEqual(spec1.cli, "agy")
# 2. Qualified failover (provider-quota) -> transitions to Claude GLM.
# 2. Qualified failover (provider-quota) -> transitions to OpenCode GLM.
dec2 = dispatch.select_execution_decision(
task, stage="worker", prior_decision=dec1,
evaluated_at=daytime, transition="failover", failure_class="provider-quota"
)
self.assertEqual(dec2["transition"]["trigger"], "provider-quota")
self.assertEqual(dec2["selected"]["adapter"], "claude-glm")
self.assertEqual(dec2["selected"]["adapter"], "opencode")
self.assertEqual(dec2["selected"]["target"], "glm-5.2")
self.assertEqual(dec2["selected"]["reasoning_effort"], "max")
terra_available = {
"schema_version": "1.0",
@ -9816,7 +9902,7 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
)
self.assertEqual(spec1.cli, "agy")
# 2. Record primary failure (count=1) -> failover to Claude GLM.
# 2. Record primary failure (count=1) -> failover to OpenCode GLM.
budget = dispatch.StageFailureBudget.from_decision(store, task, dec1)
count1 = budget.record_failure(target=dec1["selected"], transition="provider-quota")
self.assertEqual(count1, 1)
@ -9826,7 +9912,8 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
evaluated_at=daytime, transition="failover", failure_class="provider-quota"
)
dispatch.commit_execution_decision(store, task, "worker", dec2)
self.assertEqual(dec2["selected"]["adapter"], "claude-glm")
self.assertEqual(dec2["selected"]["adapter"], "opencode")
self.assertEqual(dec2["selected"]["reasoning_effort"], "max")
# 3. Alternate fails 9 times -> budget count reaches 10, task is blocked
budget2 = dispatch.StageFailureBudget.from_decision(store, task, dec2)
@ -10230,7 +10317,7 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9)))
nighttime = datetime(2026, 7, 26, 1, 0, 0, tzinfo=timezone(timedelta(hours=9)))
# Case 1: Day local G08 Gemini quota failover completes on pinned Claude GLM.
# Case 1: Day local G08 Gemini quota failover completes on pinned OpenCode GLM.
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
(workspace / ".git").mkdir()
@ -10239,8 +10326,9 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
try:
gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)")
glm_spec = dispatch.AgentSpec(
"claude-glm", "glm-5.2", "claude-glm/glm-5.2 xhigh",
command_model="sonnet",
"opencode", "glm-5.2", "opencode/glm-5.2 max",
reasoning_effort="max",
command_model="iop-glm/glm-5.2",
)
loc_gemini = self.make_attempt_locator(workspace, task, gemini_spec)
loc_glm = self.make_attempt_locator(workspace, task, glm_spec)
@ -10260,12 +10348,16 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime)
await dispatch.run_worker(workspace, store, task)
self.assertEqual([s.cli for s in invoked_specs], ["agy", "claude-glm"])
self.assertEqual([s.cli for s in invoked_specs], ["agy", "opencode"])
state = store.task_state(task)
self.assertEqual(state["execution_class"], "cloud_model")
self.assertTrue(state["selfcheck_done"])
self.assertEqual(dispatch.task_stage(task, state), "review")
self.assertEqual(state["execution_decisions"]["worker"]["selected"]["adapter"], "claude-glm")
self.assertEqual(state["execution_decisions"]["worker"]["selected"]["adapter"], "opencode")
self.assertEqual(
state["execution_decisions"]["worker"]["selected"]["reasoning_effort"],
"max",
)
self.assertEqual(
state["completing_decision"]["selected"]["execution_class"], "cloud_model"
)
@ -10539,7 +10631,7 @@ class ThroughputQuotaBatchTest(unittest.TestCase):
batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now)
# Night local-G08 probes Gemini High plus its Codex Terra backup.
# claude-glm has no dedicated quota probe; review remains excluded.
# OpenCode GLM has no dedicated quota probe; review remains excluded.
self.assertIsNotNone(batch_snap)
self.assertEqual(len(probe_calls), 2)
self.assertEqual(probe_calls[0]["adapter"], "agy")
@ -11079,15 +11171,16 @@ class ThroughputQuotaBatchTest(unittest.TestCase):
self.assertIsNone(st_blocked_after.get("blocked"))
self.assertFalse(st_blocked_after.get("retry_quota_refresh_pending"))
dec_after = st_blocked_after["execution_decisions"]["worker"]
self.assertEqual(dec_after["selected"]["adapter"], "claude-glm")
self.assertEqual(dec_after["selected"]["adapter"], "opencode")
self.assertEqual(dec_after["selected"]["target"], "glm-5.2")
self.assertEqual(dec_after["selected"]["reasoning_effort"], "max")
self.assertNotIn("thinking_level", dec_after["selected"])
self.assertEqual(dec_after["transition"]["trigger"], "provider-quota")
self.assertEqual(dec_after["work_unit_id"], d_blocked["work_unit_id"])
used = dec_after.get("used_candidates", [])
used_adapters = [u.get("adapter") for u in used]
self.assertIn("claude-glm", used_adapters)
self.assertIn("opencode", used_adapters)
self.assertIn("agy", used_adapters)
self.assertTrue(len(st_blocked_after.get("route_transition_history", [])) >= 2)
blocked_invocations = [call for call in invoke_calls if call[0] == t_blocked.name]

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

@ -51,7 +51,11 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
self.assertEqual(decision.rule_id, "worker-local-g07-g08-kst-night")
self.assertEqual(
decision.candidates,
(policy.AGY_GEMINI_HIGH, policy.CLAUDE_GLM, policy.CODEX_TERRA_HIGH),
(
policy.AGY_GEMINI_HIGH,
policy.OPENCODE_GLM_MAX,
policy.CODEX_TERRA_HIGH,
),
)
self.assertEqual(decision.time_window, "kst-night-[23:00,07:00)")
self.assertEqual(decision.candidates[0].target, "Gemini 3.6 Flash (High)")
@ -120,7 +124,7 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
(
policy.CODEX_SPARK_XHIGH,
policy.AGY_GEMINI_LOW,
policy.CLAUDE_GLM,
policy.OPENCODE_GLM_MEDIUM,
policy.CODEX_TERRA_HIGH,
),
)
@ -142,7 +146,7 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
self.assertEqual(decision.rule_id, "official-review-codex")
self.assertEqual(decision.candidates, (policy.CODEX_SOL_XHIGH,))
def test_local_g07_g08_candidate_order_uses_gemini_high_then_glm_high(self):
def test_local_g07_g08_candidate_order_uses_gemini_high_then_glm_max(self):
daytime = policy.select_policy(
stage="worker",
lane="local",
@ -157,19 +161,26 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
)
expected = (
policy.AGY_GEMINI_HIGH,
policy.CLAUDE_GLM,
policy.OPENCODE_GLM_MAX,
policy.CODEX_TERRA_HIGH,
)
self.assertEqual(daytime.candidates, expected)
self.assertEqual(nighttime.candidates, expected)
def test_claude_glm_is_cloud_without_selfcheck(self):
target = policy.CLAUDE_GLM
self.assertEqual(target.adapter, "claude-glm")
self.assertEqual(target.target, "glm-5.2")
self.assertEqual(target.command_model, "sonnet")
self.assertEqual(target.execution_class, "cloud_model")
self.assertFalse(target.selfcheck_required)
def test_opencode_glm_effort_is_one_step_above_gemini(self):
cases = (
(policy.AGY_GEMINI_LOW, policy.OPENCODE_GLM_MEDIUM, "medium"),
(policy.AGY_GEMINI_MEDIUM, policy.OPENCODE_GLM_HIGH, "high"),
(policy.AGY_GEMINI_HIGH, policy.OPENCODE_GLM_MAX, "max"),
)
for gemini, target, effort in cases:
with self.subTest(gemini=gemini.target):
self.assertEqual(target.adapter, "opencode")
self.assertEqual(target.target, "glm-5.2")
self.assertEqual(target.command_model, "iop-glm/glm-5.2")
self.assertEqual(target.reasoning_effort, effort)
self.assertEqual(target.execution_class, "cloud_model")
self.assertFalse(target.selfcheck_required)
def test_invalid_inputs_are_rejected(self):
cases = [
@ -202,6 +213,9 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
(policy.CODEX_TERRA_HIGH, None),
(policy.PI_ORNITH, None),
(policy.PI_LAGUNA, None),
(policy.OPENCODE_GLM_MEDIUM, None),
(policy.OPENCODE_GLM_HIGH, None),
(policy.OPENCODE_GLM_MAX, None),
(policy.CLAUDE_GLM, None),
]
for current, expected in cases:
@ -212,7 +226,10 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
with self.subTest(identity=target.target):
self.assertEqual(
policy.canonical_target(
target.adapter, target.target, target.thinking_level
target.adapter,
target.target,
target.thinking_level,
target.reasoning_effort,
),
target,
)
@ -222,6 +239,9 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
cases = [
(policy.PI_ORNITH, None),
(policy.PI_LAGUNA, None),
(policy.OPENCODE_GLM_MEDIUM, None),
(policy.OPENCODE_GLM_HIGH, None),
(policy.OPENCODE_GLM_MAX, None),
(policy.CLAUDE_GLM, None),
(
policy.AGY_GEMINI_LOW,

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

@ -527,10 +527,10 @@ class SelectorRouteMatrixTests(unittest.TestCase):
[c["candidate_rank"] for c in daytime], [1, 2, 3]
)
self.assertEqual(
[c["adapter"] for c in daytime], ["agy", "claude-glm", "codex"]
[c["adapter"] for c in daytime], ["agy", "opencode", "codex"]
)
self.assertEqual(
[c["adapter"] for c in nighttime], ["agy", "claude-glm", "codex"]
[c["adapter"] for c in nighttime], ["agy", "opencode", "codex"]
)
single = write_task_file(Path(tmp), "PLAN", "cloud", 5)
candidates = selector.select_execution_target(
@ -538,6 +538,34 @@ class SelectorRouteMatrixTests(unittest.TestCase):
)["candidates"]
self.assertEqual([c["candidate_rank"] for c in candidates], [1, 2, 3])
def test_each_gemini_candidate_is_immediately_followed_by_higher_effort_glm(self):
cases = (
("cloud", 1, "Low", "medium"),
("cloud", 3, "Medium", "high"),
("cloud", 5, "High", "max"),
("local", 7, "High", "max"),
)
with TemporaryDirectory() as tmp:
for lane, grade, gemini_effort, glm_effort in cases:
with self.subTest(lane=lane, grade=grade):
task_file = write_task_file(Path(tmp), "PLAN", lane, grade)
candidates = selector.select_execution_target(
task_file, evaluated_at=kst(12)
)["candidates"]
gemini_index = next(
index
for index, candidate in enumerate(candidates)
if candidate["adapter"] == "agy"
)
self.assertEqual(
candidates[gemini_index]["target"],
f"Gemini 3.6 Flash ({gemini_effort})",
)
glm = candidates[gemini_index + 1]
self.assertEqual(glm["adapter"], "opencode")
self.assertEqual(glm["target"], "glm-5.2")
self.assertEqual(glm["reasoning_effort"], glm_effort)
class SelectorQuotaRepresentationTests(unittest.TestCase):
def test_quota_probe_tri_state(self):
@ -641,7 +669,7 @@ class SelectorQuotaRepresentationTests(unittest.TestCase):
result["quota"]["targets"], snapshot["targets"]
)
def test_exhausted_gemini_falls_back_to_glm_high(self):
def test_exhausted_gemini_high_falls_back_to_glm_max(self):
snapshot = {
"snapshot_id": "gemini-exhausted",
"source": "iop-node quota-probe",
@ -659,8 +687,9 @@ class SelectorQuotaRepresentationTests(unittest.TestCase):
result = selector.select_execution_target(
task_file, evaluated_at=kst(12), quota_snapshot=snapshot
)
self.assertEqual(result["selected"]["adapter"], "claude-glm")
self.assertEqual(result["selected"]["adapter"], "opencode")
self.assertEqual(result["selected"]["target"], "glm-5.2")
self.assertEqual(result["selected"]["reasoning_effort"], "max")
self.assertNotIn("thinking_level", result["selected"])
def test_all_candidates_exhausted_returns_no_eligible_target(self):
@ -788,7 +817,7 @@ class SelectorQuotaRepresentationTests(unittest.TestCase):
self.assertEqual(local_result["quota"]["mode"], "unbounded")
self.assertEqual(local_result["quota"]["status"], "not_applicable")
# Local G07 has Gemini, Claude GLM, and Terra candidates.
# Local G07 has Gemini, OpenCode GLM, and Terra candidates.
dynamic = write_task_file(Path(tmp), "PLAN", "local", 7)
candidates = selector.select_execution_target(
dynamic, evaluated_at=kst(12)
@ -796,8 +825,9 @@ class SelectorQuotaRepresentationTests(unittest.TestCase):
self.assertEqual(len(candidates), 3)
self.assertEqual(candidates[0]["adapter"], "agy")
self.assertEqual(candidates[0]["quota_status"], "unknown")
self.assertEqual(candidates[1]["adapter"], "claude-glm")
self.assertEqual(candidates[1]["adapter"], "opencode")
self.assertEqual(candidates[1]["target"], "glm-5.2")
self.assertEqual(candidates[1]["reasoning_effort"], "max")
self.assertEqual(candidates[1]["quota_status"], "unknown")
self.assertEqual(candidates[1]["execution_class"], "cloud_model")
self.assertFalse(candidates[1]["selfcheck_required"])
@ -1154,7 +1184,7 @@ class SelectorIdentityAndQuotaRoundtripTests(unittest.TestCase):
class SelectorFailoverContractTests(unittest.TestCase):
def test_cloud_g01_g02_quota_failover_follows_spark_gemini_glm_low_order(self):
def test_cloud_g01_g02_quota_failover_follows_spark_gemini_glm_medium_order(self):
with TemporaryDirectory() as tmp:
task_file = write_task_file(Path(tmp), "PLAN", "cloud", 1)
initial = selector.select_execution_target(
@ -1187,7 +1217,7 @@ class SelectorFailoverContractTests(unittest.TestCase):
[
("codex", "gpt-5.3-codex-spark"),
("agy", "Gemini 3.6 Flash (Low)"),
("claude-glm", "glm-5.2"),
("opencode", "glm-5.2"),
("codex", "gpt-5.6-terra"),
],
)
@ -1197,7 +1227,7 @@ class SelectorFailoverContractTests(unittest.TestCase):
)
self.assertEqual(
(glm["selected"]["adapter"], glm["selected"]["target"]),
("claude-glm", "glm-5.2"),
("opencode", "glm-5.2"),
)
self.assertNotIn("thinking_level", glm["selected"])
self.assertEqual(
@ -1205,7 +1235,11 @@ class SelectorFailoverContractTests(unittest.TestCase):
[
{"adapter": "codex", "target": "gpt-5.3-codex-spark"},
{"adapter": "agy", "target": "Gemini 3.6 Flash (Low)"},
{"adapter": "claude-glm", "target": "glm-5.2"},
{
"adapter": "opencode",
"target": "glm-5.2",
"reasoning_effort": "medium",
},
],
)
terra = selector.select_execution_target(
@ -1240,8 +1274,9 @@ class SelectorFailoverContractTests(unittest.TestCase):
task_file, evaluated_at=kst(12), transition="failover",
prior_decision=prior, failure_class="provider-quota",
)
self.assertEqual(result["selected"]["adapter"], "claude-glm")
self.assertEqual(result["selected"]["adapter"], "opencode")
self.assertEqual(result["selected"]["target"], "glm-5.2")
self.assertEqual(result["selected"]["reasoning_effort"], "max")
self.assertEqual(result["transition"]["context_transfer"], "logical")
self.assertEqual(result["transition"]["trigger"], "provider-quota")
self.assertEqual(len(result["used_candidates"]), 2)
@ -1319,8 +1354,9 @@ class SelectorFailoverContractTests(unittest.TestCase):
prior = selector.select_execution_target(
task_file, evaluated_at=kst(12), quota_snapshot=gemini_exhausted_snapshot
)
self.assertEqual(prior["selected"]["adapter"], "claude-glm")
self.assertEqual(prior["selected"]["adapter"], "opencode")
self.assertEqual(prior["selected"]["target"], "glm-5.2")
self.assertEqual(prior["selected"]["reasoning_effort"], "max")
self.assertNotIn("thinking_level", prior["selected"])
result = selector.select_execution_target(
@ -1449,7 +1485,7 @@ class SelectorFailoverContractTests(unittest.TestCase):
)
self.assertEqual(day_initial["selected"]["adapter"], "agy")
# 23:00 KST is nighttime -> failover to claude-glm
# 23:00 KST is nighttime -> failover to OpenCode GLM max.
night_failover = selector.select_execution_target(
task_file,
evaluated_at=kst(23, 0, 0),
@ -1457,23 +1493,27 @@ class SelectorFailoverContractTests(unittest.TestCase):
prior_decision=day_initial,
failure_class="provider-quota",
)
self.assertEqual(night_failover["selected"]["adapter"], "claude-glm")
self.assertEqual(night_failover["selected"]["adapter"], "opencode")
self.assertEqual(
night_failover["used_candidates"],
[
{"adapter": "agy", "target": "Gemini 3.6 Flash (High)"},
{"adapter": "claude-glm", "target": "glm-5.2"},
{
"adapter": "opencode",
"target": "glm-5.2",
"reasoning_effort": "max",
},
],
)
# 23:01 KST nighttime resume -> preserved pinned claude-glm decision
# 23:01 KST nighttime resume -> preserved pinned OpenCode decision.
night_resume = selector.select_execution_target(
task_file,
evaluated_at=kst(23, 1, 0),
transition="resume",
prior_decision=night_failover,
)
self.assertEqual(night_resume["selected"]["adapter"], "claude-glm")
self.assertEqual(night_resume["selected"]["adapter"], "opencode")
self.assertIs(night_resume["decision"]["pinned"], True)
self.assertEqual(night_resume["used_candidates"], night_failover["used_candidates"])
@ -1501,8 +1541,9 @@ class SelectorFailoverContractTests(unittest.TestCase):
failure_class="provider-stream-disconnect",
)
self.assertEqual(result["selected"]["adapter"], "claude-glm")
self.assertEqual(result["selected"]["adapter"], "opencode")
self.assertEqual(result["selected"]["target"], "glm-5.2")
self.assertEqual(result["selected"]["reasoning_effort"], "max")
self.assertNotIn("thinking_level", result["selected"])
self.assertEqual(result["quota"]["status"], "unknown")
self.assertEqual(run_mock.call_count, 2)

403
docs/dev-opencode-settings-guide.md Normal file
View file

@ -0,0 +1,403 @@
# dev OpenCode Settings Guide
IOP dev OpenAI-compatible endpoint를 OpenCode에서 사용하기 위한 설정과 복구 절차다. 이 문서는 저장소 코드나 Edge 런타임을 수정하는 절차가 아니라 사용자 호스트의 OpenCode 설정만 다룬다.
2026-08-05 기준 다음 조합을 실제 호출로 확인했다.
- OpenCode: `1.18.3`
- IOP endpoint: `http://toki-labs.com:18083/v1`
- local model: `iop/ornith:35b`
- cloud model: `iop-glm/glm-5.2`
## 인증 경계
IOP를 OpenCode에 연결할 때 서로 다른 두 토큰을 구분한다.
| 용도 | 요청 위치 | OpenCode 환경 변수 |
| --- | --- | --- |
| IOP 접속 토큰 | `Authorization: Bearer ...` | `IOP_OPENCODE_API_KEY` |
| upstream provider 토큰 | `X-IOP-Provider-Authorization: ...` | `IOP_GLM_CODING_PLAN_TOKEN` |
`/v1/models`부터 실패하면 IOP 접속 토큰 문제다. `/v1/models`는 성공하지만 completion에서 `provider auth token required`가 나오면 upstream provider header 문제다. Edge 설정의 legacy `openai.bearer_token` 값을 사용자용 IOP 토큰 대신 복사하지 않는다.
실제 토큰은 tracked 문서, 저장소 설정, 셸 명령 인자, 로그, 채팅에 남기지 않는다.
## 원격 SOPS 토큰 찾기
dev 사용자 토큰의 현재 source of truth는 원격 호스트 `toki@toki-labs.com`의 다음 파일이다.
```text
~/.config/iop/secrets/dev-openai-toki.sops.yaml
```
이 파일은 원격 호스트 자체의 파일이다. `code-server` 컨테이너의 `/config/.config/sops/iop/*.sops.yaml`과 혼동하지 않는다.
현재 OpenCode에는 IDE agent 용도로 발급된 다음 token entry를 사용한다.
```text
tokens.toki-dev-cline
```
원격 macOS에서는 SOPS와 age key 경로를 명시한다.
```bash
export SOPS_AGE_KEY_FILE="$HOME/.config/sops/age/keys.txt"
secret_file="$HOME/.config/iop/secrets/dev-openai-toki.sops.yaml"
/opt/homebrew/bin/sops -d --output-type json "$secret_file" |
jq -r 'paths(scalars) | map(tostring) | join(".")'
```
위 명령은 key path만 출력한다. 복호화된 전체 JSON이나 token value를 터미널에 출력하지 않는다.
원격 endpoint와 token 유효성은 응답 본문이나 token을 출력하지 않고 상태 코드로 확인한다.
```bash
export SOPS_AGE_KEY_FILE="$HOME/.config/sops/age/keys.txt"
secret_file="$HOME/.config/iop/secrets/dev-openai-toki.sops.yaml"
base_url="$(/opt/homebrew/bin/sops -d --extract '["base_url"]' "$secret_file")"
iop_token="$(/opt/homebrew/bin/sops -d --extract '["tokens"]["toki-dev-cline"]' "$secret_file")"
http_code="$(
curl -sS -o /dev/null -w '%{http_code}' \
-H "Authorization: Bearer $iop_token" \
"$base_url/v1/models"
)"
printf 'status=%s token_present=%s\n' "$http_code" "$([[ -n "$iop_token" ]] && echo yes || echo no)"
unset iop_token
```
정상 기준은 `status=200`이다. SOPS의 `base_url`은 원격 호스트 내부에서 사용하는 `127.0.0.1` 주소일 수 있으므로, 로컬 OpenCode 설정에는 외부에서 접근 가능한 dev endpoint를 사용한다.
## 로컬 SOPS로 안전하게 이전
원격 SOPS 파일의 age recipient와 로컬 age key가 다르면 파일 자체를 복사해도 복호화할 수 없다. token을 평문 임시 파일에 저장하지 않고 SSH pipe에서 바로 로컬 recipient로 재암호화한다.
로컬 기준 경로:
```text
~/.config/sops/age/keys.txt
~/.config/sops/iop/iop-opencode.sops.yaml
```
```bash
local_recipient="$(
"$HOME/.local/bin/age-keygen" -y "$HOME/.config/sops/age/keys.txt"
)"
umask 077
ssh toki@toki-labs.com \
"export SOPS_AGE_KEY_FILE=/Users/toki/.config/sops/age/keys.txt; \
/opt/homebrew/bin/sops -d \
--extract '[\"tokens\"][\"toki-dev-cline\"]' \
/Users/toki/.config/iop/secrets/dev-openai-toki.sops.yaml" |
jq -Rn '{data: input}' |
"$HOME/.local/bin/sops" encrypt \
--age "$local_recipient" \
--input-type json \
--output-type yaml \
--output "$HOME/.config/sops/iop/iop-opencode.sops.yaml" \
/dev/stdin
chmod 600 "$HOME/.config/sops/iop/iop-opencode.sops.yaml"
unset local_recipient
```
복호화 가능 여부는 값 대신 길이와 endpoint 상태만 확인한다.
```bash
iop_token="$(
"$HOME/.local/bin/sops" -d --extract '["data"]' \
"$HOME/.config/sops/iop/iop-opencode.sops.yaml"
)"
http_code="$(
curl -sS -o /dev/null -w '%{http_code}' \
-H "Authorization: Bearer $iop_token" \
http://toki-labs.com:18083/v1/models
)"
printf 'status=%s token_length=%s\n' "$http_code" "${#iop_token}"
unset iop_token
```
## OpenCode 설정 파일
현재 구성은 다음 파일을 사용한다.
```text
~/.config/opencode/opencode.json
~/.config/opencode/iop-glm.json
~/.config/opencode/plugins/iop-quota-guard.js
~/.local/bin/opencode
~/.local/bin/opencode-tmp-wrapper
```
`opencode.json`은 기본 IOP/Ornith provider를, `iop-glm.json`은 GLM provider와 build/plan agent override를 정의한다. 실행 wrapper가 `OPENCODE_CONFIG=~/.config/opencode/iop-glm.json`을 주입하면 OpenCode가 두 설정을 병합한다.
API key는 평문으로 넣지 않고 반드시 환경 변수 참조를 사용한다.
```json
{
"provider": {
"iop": {
"npm": "@ai-sdk/openai-compatible",
"options": {
"baseURL": "http://toki-labs.com:18083/v1",
"apiKey": "{env:IOP_OPENCODE_API_KEY}"
}
}
}
}
```
OpenCode의 custom OpenAI-compatible provider 형식은 공식 [Providers 문서](https://opencode.ai/docs/providers/)를 기준으로 한다.
## Ornith 35B 설정
`~/.config/opencode/opencode.json`의 기준값:
```json
{
"model": "iop/ornith:35b",
"small_model": "iop/ornith:35b",
"agent": {
"ornith": {
"mode": "primary",
"model": "iop/ornith:35b",
"temperature": 0.6,
"top_p": 0.95
}
},
"provider": {
"iop": {
"npm": "@ai-sdk/openai-compatible",
"name": "IOP",
"options": {
"baseURL": "http://toki-labs.com:18083/v1",
"apiKey": "{env:IOP_OPENCODE_API_KEY}"
},
"models": {
"ornith:35b": {
"name": "IOP Ornith 35B",
"family": "ornith",
"reasoning": true,
"temperature": true,
"tool_call": true,
"headers": {
"X-IOP-Provider-Authorization": "iop-local"
},
"interleaved": {
"field": "reasoning_content"
},
"modalities": {
"input": ["text"],
"output": ["text"]
},
"limit": {
"context": 262144,
"output": 32768
}
}
}
}
}
}
```
`temperature=0.6`, `top_p=0.95`는 현재 dev Ornith baseline이다. `X-IOP-Provider-Authorization: iop-local`은 현재 dev Lemonade route에서 확인된 provider marker이며, provider 인증 정책이 바뀌면 실제 upstream credential 요구 여부를 다시 확인한다.
## GLM 5.2 설정
`~/.config/opencode/iop-glm.json`의 기준값:
```json
{
"model": "iop-glm/glm-5.2",
"small_model": "iop/ornith:35b",
"agent": {
"build": {
"model": "iop-glm/glm-5.2",
"variant": "max",
"temperature": 1,
"top_p": 1,
"steps": 24
},
"plan": {
"model": "iop-glm/glm-5.2",
"variant": "max",
"temperature": 1,
"top_p": 1,
"steps": 16
}
},
"provider": {
"iop-glm": {
"npm": "@ai-sdk/openai-compatible",
"name": "IOP GLM",
"options": {
"baseURL": "http://toki-labs.com:18083/v1",
"apiKey": "{env:IOP_OPENCODE_API_KEY}",
"timeout": 300000,
"chunkTimeout": 120000,
"headers": {
"X-IOP-Provider-Authorization": "{env:IOP_GLM_CODING_PLAN_TOKEN}"
}
},
"models": {
"glm-5.2": {
"name": "IOP GLM-5.2",
"family": "glm",
"reasoning": true,
"temperature": true,
"tool_call": true,
"interleaved": {
"field": "reasoning_content"
},
"modalities": {
"input": ["text"],
"output": ["text"]
},
"limit": {
"context": 1000000,
"output": 131072
},
"variants": {
"medium": {
"reasoningEffort": "medium"
},
"high": {
"reasoningEffort": "high"
},
"max": {
"reasoningEffort": "max"
}
}
}
}
}
}
}
```
GLM-5.2의 `1M` context, 최대 `131072` output, reasoning effort, sampling parameter는 [Z.AI parameter 문서](https://docs.z.ai/guides/overview/concept-param)와 [GLM-5.2 안내](https://z.ai/blog/glm-5.2)를 기준으로 한다. OpenCode의 OpenAI-compatible GLM-5.2 변형은 현재 [provider transform](https://github.com/anomalyco/opencode/blob/dev/packages/opencode/src/provider/transform.ts)에서 `high`와 `max`를 제공하며, dispatcher의 Gemini Low fallback은 custom provider 설정에 명시한 `medium` variant를 사용한다.
일반 coding/agent 작업은 `temperature=1`, `top_p=1`, `variant=max`를 기본으로 한다. 지연이나 quota 비용을 줄여야 할 때만 agent별로 `variant=high`를 선택한다.
## 실행 wrapper
`~/.local/bin/opencode`는 최소한 다음 순서를 지킨다.
1. OpenCode XDG data/state/runtime 경로를 준비한다.
2. `IOP_OPENCODE_API_KEY`가 없으면 로컬 SOPS의 `data`를 복호화해 export한다.
3. `IOP_GLM_CODING_PLAN_TOKEN`이 없으면 기존 Pi `glm-5.2` model header command를 실행해 export한다.
4. `OPENCODE_CONFIG=~/.config/opencode/iop-glm.json`을 export한다.
5. 실제 OpenCode binary를 실행한다.
IOP 접속 토큰 부분의 기준 구현:
```bash
if [[ -z "${IOP_OPENCODE_API_KEY:-}" ]]; then
IOP_OPENCODE_API_KEY="$(
"$HOME/.local/bin/sops" -d --extract '["data"]' \
"$HOME/.config/sops/iop/iop-opencode.sops.yaml"
)"
fi
export IOP_OPENCODE_API_KEY
```
`command -v opencode`가 npm symlink나 임시 wrapper를 가리킬 수 있다. 최종 진입점이 위 wrapper를 건너뛰면 SOPS token과 `OPENCODE_CONFIG`가 적용되지 않는다.
```bash
command -v opencode
readlink "$(command -v opencode)"
sed -n '1,40p' "$(command -v opencode)"
```
현재 `~/.local/bin/opencode-tmp-wrapper`는 다른 환경을 다시 구성하지 않고 `~/.local/bin/opencode`에 그대로 위임한다.
## 검증
모델 등록:
```bash
opencode models iop
opencode models iop-glm
```
기대값:
```text
iop/ornith:35b
iop-glm/glm-5.2
```
민감한 provider option을 출력하지 않고 effective model/agent 설정만 확인한다.
```bash
opencode debug config |
jq '{
model,
small_model,
build: (.agent.build | {model, variant, temperature, top_p, steps}),
plan: (.agent.plan | {model, variant, temperature, top_p, steps}),
ornith: (.agent.ornith | {model, temperature, top_p}),
glm_limit: .provider["iop-glm"].models["glm-5.2"].limit,
glm_variants: .provider["iop-glm"].models["glm-5.2"].variants,
ornith_limit: .provider.iop.models["ornith:35b"].limit
}'
```
실제 OpenCode 호출:
```bash
opencode run --agent ornith --model 'iop/ornith:35b' \
'Reply with exactly: ORNITH_OPENCODE_OK'
opencode run --agent build --model 'iop-glm/glm-5.2' \
'Reply with exactly: GLM52_OPENCODE_OK'
```
2026-08-05 검증에서는 두 명령 모두 exit code `0`으로 marker를 정확히 반환했다.
## 장애 대응
### `/v1/models`가 401 또는 403
- OpenCode config의 `apiKey`가 `{env:IOP_OPENCODE_API_KEY}`인지 확인한다.
- wrapper가 로컬 `iop-opencode.sops.yaml`을 실제로 읽는지 확인한다.
- 원격 SOPS token을 상태 코드로 검증하고, 필요하면 평문 파일 없이 다시 재암호화한다.
- Edge runtime config에 들어 있는 bearer 값을 사용자 토큰으로 대체 사용하지 않는다.
### `provider auth token required`
- IOP 접속은 성공한 상태일 수 있다.
- GLM은 `X-IOP-Provider-Authorization={env:IOP_GLM_CODING_PLAN_TOKEN}`을 확인한다.
- Ornith는 현재 dev provider marker가 model header에 들어 있는지 확인한다.
- IOP와 upstream provider 토큰을 서로 바꾸지 않는다.
### OpenCode가 다른 model/config를 사용함
- `command -v opencode`와 symlink/wrapper chain을 확인한다.
- `OPENCODE_CONFIG`가 `iop-glm.json`을 가리키는지 확인한다.
- `opencode debug config`에서 병합된 provider와 agent를 확인한다.
### SOPS가 `identity did not match any recipients`로 실패
- 원격 macOS에서 `SOPS_AGE_KEY_FILE=~/.config/sops/age/keys.txt`를 명시한다.
- 원격 SOPS recipient와 로컬 age recipient가 다르면 파일을 그대로 복사하지 않는다.
- 이 문서의 SSH pipe 재암호화 절차를 사용한다.
### 이전 auth cache가 호출에 개입함
- wrapper가 지정한 `XDG_DATA_HOME`과 다른 경로의 `opencode/auth.json`은 정상 호출에 필요하지 않다.
- 설정에 `options.apiKey={env:IOP_OPENCODE_API_KEY}`가 있으면 그 경로를 source of truth로 유지한다.
- 사용되지 않는 캐시를 정리할 때는 provider key를 먼저 확인하고 다른 provider credential을 삭제하지 않는다.
## 보안 체크리스트
- token 원문을 `jq`, `sed`, `cat`, `set -x`, shell history, 로그에 출력하지 않는다.
- 원격 SOPS와 로컬 SOPS의 age key를 혼동하지 않는다.
- 로컬 encrypted file 권한은 `600`으로 유지한다.
- tracked `opencode.json`이나 문서에 실제 token을 넣지 않는다.
- token이 노출됐으면 파일만 다시 암호화하지 말고 원격 source에서 token을 회전한다.
- 검증 결과에는 HTTP status, token 존재 여부/길이, model id, marker만 남긴다.