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

Merge dev into feature/node-provider-execution-liveness-recovery

Browse source
This commit is contained in:
toki 2026-08-04 16:15:28 +09:00
commit f727b8f4a2
6 changed files with 187 additions and 168 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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)` → Pi `iop/glm-5.2`, thinking high → Codex `gpt-5.6-terra`, reasoning 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-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)` → Pi `iop/glm-5.2`, thinking low → Codex `gpt-5.6-terra`, reasoning high |
| `cloud-G03`–`cloud-G04` | agy `Gemini 3.6 Flash (Medium)` → Pi `iop/glm-5.2`, thinking medium → Codex `gpt-5.6-terra`, reasoning high |
| `cloud-G05`–`cloud-G06` | agy `Gemini 3.6 Flash (High)` → Pi `iop/glm-5.2`, thinking high → Codex `gpt-5.6-terra`, reasoning high |
| `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-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`; Pi-hosted `iop/glm-5.2` is a cloud model 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`; `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.
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 Pi 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 Pi, accept a terminal `auto_retry_end` with `success=false` and `finalError`, or the final non-retrying `agent_end` whose last message has `stopReason=error` and `errorMessage`; do not treat intermediate retry events or message content as terminal evidence. 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, 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 -> 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 `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.
- 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. Pi-hosted cloud GLM may fail over to the policy-owned Codex Terra backup.
- Never escalate local Pi models to a cloud model. Cloud `claude-glm` may fail over to the policy-owned Codex Terra backup.
- 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.

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

@ -502,10 +502,11 @@ class AgentSpec:
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"}:
if spec.cli in {"codex", "claude", "claude-glm"}:
return spec.reasoning_effort or "xhigh"
return None
@ -532,8 +533,13 @@ def agent_spec_from_record(record: dict[str, Any]) -> AgentSpec | None:
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"
local_pi = cli == "pi"
if cli in {"codex", "claude"}:
if cli in {"codex", "claude", "claude-glm"}:
effort = reasoning_effort or "xhigh"
display = f"{cli}/{model} {effort}"
elif cli == "pi":
@ -547,6 +553,7 @@ def agent_spec_from_record(record: dict[str, Any]) -> AgentSpec | None:
local_pi=local_pi,
reasoning_effort=reasoning_effort,
thinking_level=thinking_level,
command_model=command_model,
)
@ -1824,21 +1831,23 @@ def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
local_pi=True,
thinking_level=canonical.thinking_level,
)
if adapter not in {"agy", "claude", "codex"} or local_pi:
if adapter not in {"agy", "claude", "claude-glm", "codex"} 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 (
" xhigh" if adapter in {"claude", "codex"} else ""
" xhigh" if adapter in {"claude", "claude-glm", "codex"} else ""
)
display = f"{adapter}/{target}{suffix}"
if reasoning_effort is not None:
return AgentSpec(
adapter,
target,
model,
display,
reasoning_effort=reasoning_effort,
command_model=canonical.command_model,
)
return AgentSpec(adapter, target, display)
return AgentSpec(adapter, model, display, command_model=canonical.command_model)
def _spec_from_completing_decision(decision: dict[str, Any]) -> AgentSpec:
@ -1906,7 +1915,7 @@ def _spec_from_completing_decision(decision: dict[str, Any]) -> AgentSpec:
local_pi=True,
thinking_level=thinking_level,
)
if adapter not in {"agy", "claude", "codex"}:
if adapter not in {"agy", "claude", "claude-glm", "codex"}:
raise ExecutionDecisionError(
f"completing decision adapter가 유효하지 않다: {adapter!r}"
)
@ -1918,6 +1927,18 @@ def _spec_from_completing_decision(decision: dict[str, Any]) -> AgentSpec:
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",
)
display = f"{adapter}/{target}"
return AgentSpec(adapter, target, display, local_pi=False)
@ -2805,7 +2826,7 @@ def terminal_diagnostic(cli: str, channel: str, line: str) -> str | None:
)
):
return json.dumps(value, ensure_ascii=False)
if cli == "claude":
if cli in {"claude", "claude-glm"}:
subtype = str(value.get("subtype", ""))
if event_type == "rate_limit_event":
rate_limit_info = value.get("rate_limit_info")
@ -3165,7 +3186,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 == "claude":
if spec.cli in {"claude", "claude-glm"}:
return AgentSpec(
"codex",
"gpt-5.6-terra",
@ -3197,7 +3218,7 @@ def render_json_line(cli: str, line: str) -> tuple[list[str], str | None]:
rendered.append(f"{item_type}: {item.get('server', '')} {item.get('tool', item.get('query', ''))}")
elif value.get("type") == "turn.failed":
rendered.append(str(value.get("error", value)))
elif cli == "claude":
elif cli in {"claude", "claude-glm"}:
message = value.get("message") or {}
for block in message.get("content") or []:
if block.get("type") == "text":
@ -3211,7 +3232,7 @@ def render_json_line(cli: str, line: str) -> tuple[list[str], str | None]:
def native_session_path(cli: str, workspace: Path, session_id: str | None, attempt_dir: Path) -> str | None:
if cli == "claude" and session_id:
if cli in {"claude", "claude-glm"} and session_id:
encoded = str(workspace).replace("/", "-")
return str(Path.home() / ".claude" / "projects" / encoded / f"{session_id}.jsonl")
if cli == "pi" and session_id:
@ -3807,10 +3828,10 @@ def build_command(
"-c", f'model_reasoning_effort="{effective_reasoning_effort(spec)}"',
"--dangerously-bypass-approvals-and-sandbox", prompt,
]
if spec.cli == "claude":
if spec.cli in {"claude", "claude-glm"}:
return [
"claude", "-p", "--output-format", "stream-json", "--verbose",
"--session-id", session_id, "--model", spec.model,
spec.cli, "-p", "--output-format", "stream-json", "--verbose",
"--session-id", session_id, "--model", spec.command_model or spec.model,
"--effort", str(effective_reasoning_effort(spec)),
"--dangerously-skip-permissions", prompt,
]
@ -3929,12 +3950,13 @@ async def invoke(
**dispatcher_source_provenance(),
"cli": spec.cli,
"model": spec.model,
"command_model": spec.command_model,
"reasoning_effort": effective_reasoning_effort(spec),
"thinking_level": effective_pi_thinking_level(spec),
"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", "pi"} else None,
"session_id": session_id if spec.cli in {"claude", "claude-glm", "pi"} else None,
"native_session_path": (
str(pi_resume_session)
if pi_resume_session is not None

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

@ -23,6 +23,7 @@ class RouteTarget:
execution_class: str
selfcheck_required: bool
thinking_level: str | None = None
command_model: str | None = None
@dataclass(frozen=True)
@ -45,14 +46,15 @@ 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)
PI_GLM_LOW = RouteTarget(
"pi", "iop/glm-5.2", "cloud_model", False, thinking_level="low"
)
PI_GLM_MEDIUM = RouteTarget(
"pi", "iop/glm-5.2", "cloud_model", False, thinking_level="medium"
)
PI_GLM_HIGH = RouteTarget(
"pi", "iop/glm-5.2", "cloud_model", False, thinking_level="high"
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(
@ -71,9 +73,7 @@ CANONICAL_TARGETS = (
AGY_GEMINI_MEDIUM,
AGY_GEMINI_HIGH,
PI_LAGUNA,
PI_GLM_LOW,
PI_GLM_MEDIUM,
PI_GLM_HIGH,
CLAUDE_GLM,
CLAUDE_OPUS,
CLAUDE_HAIKU_XHIGH,
CODEX_SPARK_XHIGH,
@ -181,11 +181,11 @@ 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, PI_GLM_HIGH, CODEX_TERRA_HIGH)
candidates = (AGY_GEMINI_HIGH, CLAUDE_GLM, CODEX_TERRA_HIGH)
else:
rule_id = "worker-local-g07-g08-kst-night"
reason_code = "kst_night_gemini_high"
candidates = (AGY_GEMINI_HIGH, PI_GLM_HIGH, CODEX_TERRA_HIGH)
candidates = (AGY_GEMINI_HIGH, CLAUDE_GLM, CODEX_TERRA_HIGH)
return PolicyDecision(
rule_id=rule_id,
policy_priority=20,
@ -205,17 +205,17 @@ def select_policy(
candidates = (
CODEX_SPARK_XHIGH,
AGY_GEMINI_LOW,
PI_GLM_LOW,
CLAUDE_GLM,
CODEX_TERRA_HIGH,
)
rule_id = "worker-cloud-g01-g02"
reason_code = "cloud_spark_priority_grade"
elif grade <= 4:
candidates = (AGY_GEMINI_MEDIUM, PI_GLM_MEDIUM, CODEX_TERRA_HIGH)
candidates = (AGY_GEMINI_MEDIUM, CLAUDE_GLM, CODEX_TERRA_HIGH)
rule_id = "worker-cloud-g03-g04"
reason_code = "cloud_gemini_medium_grade"
elif grade <= 6:
candidates = (AGY_GEMINI_HIGH, PI_GLM_HIGH, CODEX_TERRA_HIGH)
candidates = (AGY_GEMINI_HIGH, CLAUDE_GLM, CODEX_TERRA_HIGH)
rule_id = "worker-cloud-g05-g06"
reason_code = "cloud_gemini_high_grade"
elif grade <= 8:

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

@ -120,16 +120,15 @@ class CommandConstructionTest(unittest.TestCase):
command[-2:], ["--log-file", str(workspace / "attempt" / "agy-cli.log")]
)
def test_pi_glm_preserves_policy_thinking_level(self):
def test_claude_glm_uses_glm_policy_identity_and_sonnet_command_alias(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
command = dispatch.build_command(
dispatch.AgentSpec(
"pi",
"claude-glm",
"glm-5.2",
"pi/iop/glm-5.2 medium",
local_pi=True,
thinking_level="medium",
"claude-glm/glm-5.2 xhigh",
command_model="sonnet",
),
"Implement the active plan.",
workspace,
@ -137,8 +136,11 @@ class CommandConstructionTest(unittest.TestCase):
workspace / "attempt",
)
thinking_index = command.index("--thinking")
self.assertEqual(command[thinking_index + 1], "medium")
self.assertEqual(command[0], "claude-glm")
model_index = command.index("--model")
self.assertEqual(command[model_index + 1], "sonnet")
effort_index = command.index("--effort")
self.assertEqual(command[effort_index + 1], "xhigh")
class TaskStageTest(unittest.TestCase):
@ -772,7 +774,7 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase):
finally:
store.close()
async def test_glm_cloud_completion_skips_selfcheck(self):
async def test_claude_glm_completion_skips_selfcheck(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
(workspace / ".git").mkdir()
@ -783,17 +785,16 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase):
"work_unit_id": self._WORK_UNIT_ID,
"stage": "worker",
"selected": {
"adapter": "pi",
"target": "iop/glm-5.2",
"adapter": "claude-glm",
"target": "glm-5.2",
"execution_class": "cloud_model",
"selfcheck_required": False,
"thinking_level": "high",
},
}
store.update_task(
task,
worker_done=True,
worker_cli="pi",
worker_cli="claude-glm",
worker_model="glm-5.2",
completing_decision=decision,
execution_class="cloud_model",
@ -3171,7 +3172,7 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase):
self.assertEqual(record["failure_evidence_excerpt"], diagnostic)
self.assertFalse(record["provider_transport_failure_confirmed"])
async def test_pi_structured_quota_with_zero_exit_records_provider_quota(self):
async def test_claude_glm_structured_quota_with_zero_exit_records_provider_quota(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
(workspace / ".git").mkdir()
@ -3183,10 +3184,11 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase):
)
terminal_event = json.dumps(
{
"type": "auto_retry_end",
"success": False,
"attempt": 3,
"finalError": error_message,
"type": "result",
"subtype": "error_rate_limit",
"is_error": True,
"api_error_status": 429,
"result": error_message,
}
)
command = [
@ -3196,11 +3198,10 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase):
terminal_event,
]
spec = dispatch.AgentSpec(
"pi",
"claude-glm",
"glm-5.2",
"pi/iop/glm-5.2 high",
local_pi=True,
thinking_level="high",
"claude-glm/glm-5.2 xhigh",
command_model="sonnet",
)
try:
with mock.patch.object(
@ -3224,8 +3225,8 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase):
record = json.loads(locator.read_text(encoding="utf-8"))
self.assertEqual(record["status"], "failed")
self.assertEqual(record["failure_source"], "cli-terminal-diagnostic")
self.assertEqual(record["failure_evidence_source"], "pi:stdout")
self.assertEqual(record["failure_evidence_excerpt"], error_message)
self.assertEqual(record["failure_evidence_source"], "claude-glm:stdout")
self.assertEqual(record["failure_evidence_excerpt"], terminal_event)
async def test_exit_143_is_process_termination_not_provider_failure(self):
with tempfile.TemporaryDirectory() as temporary:
@ -8538,7 +8539,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", "pi"])
self.assertEqual([s.cli for s in invoked_specs[:2]], ["agy", "claude-glm"])
self.assertEqual([h["transition"] for h in history1], ["initial", "provider-quota"])
self.assertEqual(worker_budget1.count(), 2)
finally:
@ -8577,7 +8578,7 @@ 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": "pi", "target": "iop/glm-5.2"})
self.assertEqual(raw_entry.get("last_target"), {"adapter": "claude-glm", "target": "glm-5.2"})
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"])
@ -8732,12 +8733,11 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
"Gemini 3.6 Flash (Low)",
"agy/Gemini 3.6 Flash (Low)",
),
"pi": dispatch.AgentSpec(
"pi",
"claude-glm": dispatch.AgentSpec(
"claude-glm",
"glm-5.2",
"pi/iop/glm-5.2 low",
local_pi=True,
thinking_level="low",
"claude-glm/glm-5.2 xhigh",
command_model="sonnet",
),
}
locators = {
@ -8749,7 +8749,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
async def mock_invoke(*args, **kwargs):
spec = args[4]
invoked_specs.append(spec)
if spec.cli == "pi":
if spec.cli == "claude-glm":
return 0, None, locators[spec.cli]
return 1, "provider-quota", locators[spec.cli]
@ -8770,13 +8770,13 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
)
self.assertTrue(success)
self.assertEqual(final_locator, locators["pi"])
self.assertEqual(final_locator, locators["claude-glm"])
self.assertEqual(
[(spec.cli, spec.model) for spec in invoked_specs],
[
("codex", "gpt-5.3-codex-spark"),
("agy", "Gemini 3.6 Flash (Low)"),
("pi", "glm-5.2"),
("claude-glm", "glm-5.2"),
],
)
decision = store.task_state(task)["execution_decisions"]["worker"]
@ -8785,7 +8785,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
[
{"adapter": "codex", "target": "gpt-5.3-codex-spark"},
{"adapter": "agy", "target": "Gemini 3.6 Flash (Low)"},
{"adapter": "pi", "target": "iop/glm-5.2", "thinking_level": "low"},
{"adapter": "claude-glm", "target": "glm-5.2"},
],
)
finally:
@ -8864,7 +8864,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
finally:
store.close()
async def test_day_gemini_zero_exit_quota_continues_on_laguna_with_logical_context(self):
async def test_day_gemini_zero_exit_quota_continues_on_claude_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)
@ -8896,17 +8896,17 @@ 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, "pi")
self.assertTrue(invoked_specs[1].local_pi)
laguna_prompt = invoked_prompts[1]
self.assertIn(str(task.plan.resolve()), laguna_prompt)
self.assertIn(str(locator.resolve()), laguna_prompt)
self.assertIn(str(workspace.resolve()), laguna_prompt)
self.assertIn(str((locator.parent / "stream.log").resolve()), laguna_prompt)
self.assertIn(str((locator.parent / "normalized-output.log").resolve()), laguna_prompt)
self.assertEqual(invoked_specs[1].cli, "claude-glm")
self.assertFalse(invoked_specs[1].local_pi)
continuation_prompt = invoked_prompts[1]
self.assertIn(str(task.plan.resolve()), continuation_prompt)
self.assertIn(str(locator.resolve()), continuation_prompt)
self.assertIn(str(workspace.resolve()), continuation_prompt)
self.assertIn(str((locator.parent / "stream.log").resolve()), continuation_prompt)
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"], "pi")
self.assertEqual(decisions["selected"]["adapter"], "claude-glm")
self.assertEqual(decisions["transition"]["trigger"], "provider-quota")
finally:
store.close()
@ -9012,11 +9012,10 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
"agy/Gemini 3.6 Flash (High)",
)
glm_spec = dispatch.AgentSpec(
"pi",
"claude-glm",
"glm-5.2",
"pi/iop/glm-5.2 high",
local_pi=True,
thinking_level="high",
"claude-glm/glm-5.2 xhigh",
command_model="sonnet",
)
locators = {
spec.cli: self.make_attempt_locator(workspace, task, spec)
@ -9038,7 +9037,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
store, task, decision
)
transition_budget_counts.append(budget.count())
return (0, None, locators["pi"])
return (0, None, locators["claude-glm"])
with (
mock.patch.object(dispatch, "invoke", new=mock_invoke),
@ -9054,7 +9053,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
)
self.assertTrue(success)
self.assertEqual(final_locator, locators["pi"])
self.assertEqual(final_locator, locators["claude-glm"])
self.assertEqual(invoked_specs, [agy_spec, glm_spec])
self.assertEqual(transition_budget_counts, [1])
self.assertIn(
@ -9070,9 +9069,8 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
"target": "Gemini 3.6 Flash (High)",
},
{
"adapter": "pi",
"target": "iop/glm-5.2",
"thinking_level": "high",
"adapter": "claude-glm",
"target": "glm-5.2",
},
],
)
@ -9099,8 +9097,8 @@ 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(
"pi", "glm-5.2", "pi/iop/glm-5.2 high",
local_pi=True, thinking_level="high",
"claude-glm", "glm-5.2", "claude-glm/glm-5.2 xhigh",
command_model="sonnet",
)
locator = self.make_attempt_locator(workspace, task, gemini_spec)
invoked_specs = []
@ -9124,9 +9122,9 @@ 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"], "pi")
self.assertEqual(decisions["selected"]["target"], "iop/glm-5.2")
self.assertEqual(decisions["selected"]["thinking_level"], "high")
self.assertEqual(decisions["selected"]["adapter"], "claude-glm")
self.assertEqual(decisions["selected"]["target"], "glm-5.2")
self.assertNotIn("thinking_level", decisions["selected"])
self.assertEqual(decisions["transition"]["trigger"], "provider-quota")
finally:
store.close()
@ -9140,8 +9138,8 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
store = dispatch.StateStore(workspace)
try:
glm_spec = dispatch.AgentSpec(
"pi", "glm-5.2", "pi/iop/glm-5.2 high",
local_pi=True, thinking_level="high",
"claude-glm", "glm-5.2", "claude-glm/glm-5.2 xhigh",
command_model="sonnet",
)
locator = self.make_attempt_locator(workspace, task, glm_spec)
invoked_specs = []
@ -9168,7 +9166,10 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
self.assertTrue(success)
self.assertEqual(invoked_specs, [glm_spec])
state = store.task_state(task)
self.assertEqual(state["execution_decisions"]["worker"]["selected"]["thinking_level"], "high")
self.assertEqual(
state["execution_decisions"]["worker"]["selected"]["target"],
"glm-5.2",
)
self.assertIsNone(state.get("blocked"))
finally:
store.close()
@ -9194,8 +9195,8 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
)
glm_spec = dispatch.AgentSpec(
"pi", "glm-5.2", "pi/iop/glm-5.2 high",
local_pi=True, thinking_level="high",
"claude-glm", "glm-5.2", "claude-glm/glm-5.2 xhigh",
command_model="sonnet",
)
decision, spec = dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime)
self.assertEqual(spec, glm_spec)
@ -9218,7 +9219,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
async def mock_invoke(*args, **kwargs):
spec = args[4]
invoked_specs.append(spec)
if spec.cli == "pi":
if spec.cli == "claude-glm":
return (1, "provider-quota", locator)
return (0, None, locator)
@ -9230,7 +9231,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
self.assertTrue(success)
self.assertEqual(len(invoked_specs), 2)
self.assertEqual(invoked_specs[0].cli, "pi")
self.assertEqual(invoked_specs[0].cli, "claude-glm")
self.assertEqual(invoked_specs[1].cli, "codex")
self.assertEqual(invoked_specs[1].model, "gpt-5.6-terra")
@ -9360,7 +9361,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
store.close()
async def test_cloud_g05_g06_gemini_quota_fails_over_to_glm_high(self):
"""Cloud G05–G06 sends qualified Gemini failures to Pi GLM High."""
"""Cloud G05–G06 sends qualified Gemini failures to Claude GLM 5.2."""
daytime = datetime(
2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))
)
@ -9374,8 +9375,8 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
"agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)"
)
glm_spec = dispatch.AgentSpec(
"pi", "glm-5.2", "pi/iop/glm-5.2 high",
local_pi=True, thinking_level="high",
"claude-glm", "glm-5.2", "claude-glm/glm-5.2 xhigh",
command_model="sonnet",
)
loc_agy = self.make_attempt_locator(workspace, task, agy_spec)
loc_glm = self.make_attempt_locator(workspace, task, glm_spec)
@ -9402,7 +9403,7 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
self.assertTrue(success)
self.assertEqual(
[s.cli for s in invoked_specs],
["agy", "pi"],
["agy", "claude-glm"],
)
self.assertEqual(invoked_specs[1], glm_spec)
@ -9412,14 +9413,14 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
decision["used_candidates"],
[
{"adapter": "agy", "target": "Gemini 3.6 Flash (High)"},
{"adapter": "pi", "target": "iop/glm-5.2", "thinking_level": "high"},
{"adapter": "claude-glm", "target": "glm-5.2"},
],
)
transitions = [h["transition"] for h in state["route_transition_history"]]
self.assertIn("provider-quota", transitions)
self.assertEqual(decision["selected"]["adapter"], "pi")
self.assertEqual(decision["selected"]["target"], "iop/glm-5.2")
self.assertEqual(decision["selected"]["thinking_level"], "high")
self.assertEqual(decision["selected"]["adapter"], "claude-glm")
self.assertEqual(decision["selected"]["target"], "glm-5.2")
self.assertNotIn("thinking_level", decision["selected"])
finally:
store.close()
@ -9733,20 +9734,20 @@ 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, GLM High, Terra High.
# 1. Initial decision local G08 -> Gemini High, Claude GLM 5.2, 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 Pi GLM.
# 2. Qualified failover (provider-quota) -> transitions to Claude 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"], "pi")
self.assertEqual(dec2["selected"]["target"], "iop/glm-5.2")
self.assertEqual(dec2["selected"]["adapter"], "claude-glm")
self.assertEqual(dec2["selected"]["target"], "glm-5.2")
terra_available = {
"schema_version": "1.0",
@ -9815,7 +9816,7 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
)
self.assertEqual(spec1.cli, "agy")
# 2. Record primary failure (count=1) -> failover to alternate (pi/laguna)
# 2. Record primary failure (count=1) -> failover to Claude GLM.
budget = dispatch.StageFailureBudget.from_decision(store, task, dec1)
count1 = budget.record_failure(target=dec1["selected"], transition="provider-quota")
self.assertEqual(count1, 1)
@ -9825,7 +9826,7 @@ 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"], "pi")
self.assertEqual(dec2["selected"]["adapter"], "claude-glm")
# 3. Alternate fails 9 times -> budget count reaches 10, task is blocked
budget2 = dispatch.StageFailureBudget.from_decision(store, task, dec2)
@ -10229,7 +10230,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 GLM High.
# Case 1: Day local G08 Gemini quota failover completes on pinned Claude GLM.
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
(workspace / ".git").mkdir()
@ -10238,8 +10239,8 @@ 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(
"pi", "glm-5.2", "pi/iop/glm-5.2 high",
local_pi=True, thinking_level="high",
"claude-glm", "glm-5.2", "claude-glm/glm-5.2 xhigh",
command_model="sonnet",
)
loc_gemini = self.make_attempt_locator(workspace, task, gemini_spec)
loc_glm = self.make_attempt_locator(workspace, task, glm_spec)
@ -10259,12 +10260,12 @@ 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", "pi"])
self.assertEqual([s.cli for s in invoked_specs], ["agy", "claude-glm"])
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"], "pi")
self.assertEqual(state["execution_decisions"]["worker"]["selected"]["adapter"], "claude-glm")
self.assertEqual(
state["completing_decision"]["selected"]["execution_class"], "cloud_model"
)
@ -10538,7 +10539,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.
# Pi-hosted GLM has no dedicated quota probe; review remains excluded.
# claude-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")
@ -11078,15 +11079,15 @@ 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"], "pi")
self.assertEqual(dec_after["selected"]["target"], "iop/glm-5.2")
self.assertEqual(dec_after["selected"]["thinking_level"], "high")
self.assertEqual(dec_after["selected"]["adapter"], "claude-glm")
self.assertEqual(dec_after["selected"]["target"], "glm-5.2")
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("pi", used_adapters)
self.assertIn("claude-glm", 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]

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

@ -51,7 +51,7 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
self.assertEqual(decision.rule_id, "worker-local-g07-g08-kst-night")
self.assertEqual(
decision.candidates,
(policy.AGY_GEMINI_HIGH, policy.PI_GLM_HIGH, policy.CODEX_TERRA_HIGH),
(policy.AGY_GEMINI_HIGH, policy.CLAUDE_GLM, 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 +120,7 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
(
policy.CODEX_SPARK_XHIGH,
policy.AGY_GEMINI_LOW,
policy.PI_GLM_LOW,
policy.CLAUDE_GLM,
policy.CODEX_TERRA_HIGH,
),
)
@ -157,21 +157,19 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
)
expected = (
policy.AGY_GEMINI_HIGH,
policy.PI_GLM_HIGH,
policy.CLAUDE_GLM,
policy.CODEX_TERRA_HIGH,
)
self.assertEqual(daytime.candidates, expected)
self.assertEqual(nighttime.candidates, expected)
def test_glm_targets_are_cloud_without_selfcheck(self):
for target in (
policy.PI_GLM_LOW,
policy.PI_GLM_MEDIUM,
policy.PI_GLM_HIGH,
):
with self.subTest(thinking_level=target.thinking_level):
self.assertEqual(target.execution_class, "cloud_model")
self.assertFalse(target.selfcheck_required)
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_invalid_inputs_are_rejected(self):
cases = [
@ -204,9 +202,7 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
(policy.CODEX_TERRA_HIGH, None),
(policy.PI_ORNITH, None),
(policy.PI_LAGUNA, None),
(policy.PI_GLM_LOW, None),
(policy.PI_GLM_MEDIUM, None),
(policy.PI_GLM_HIGH, None),
(policy.CLAUDE_GLM, None),
]
for current, expected in cases:
with self.subTest(current=current):
@ -226,9 +222,7 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
cases = [
(policy.PI_ORNITH, None),
(policy.PI_LAGUNA, None),
(policy.PI_GLM_LOW, None),
(policy.PI_GLM_MEDIUM, None),
(policy.PI_GLM_HIGH, None),
(policy.CLAUDE_GLM, None),
(
policy.AGY_GEMINI_LOW,
policy.QuotaProbeSpec("agy", "Gemini 3.6 Flash (Low)", ("overall", "model:Gemini 3.6 Flash (Low)")),

48
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", "pi", "codex"]
[c["adapter"] for c in daytime], ["agy", "claude-glm", "codex"]
)
self.assertEqual(
[c["adapter"] for c in nighttime], ["agy", "pi", "codex"]
[c["adapter"] for c in nighttime], ["agy", "claude-glm", "codex"]
)
single = write_task_file(Path(tmp), "PLAN", "cloud", 5)
candidates = selector.select_execution_target(
@ -659,9 +659,9 @@ class SelectorQuotaRepresentationTests(unittest.TestCase):
result = selector.select_execution_target(
task_file, evaluated_at=kst(12), quota_snapshot=snapshot
)
self.assertEqual(result["selected"]["adapter"], "pi")
self.assertEqual(result["selected"]["target"], "iop/glm-5.2")
self.assertEqual(result["selected"]["thinking_level"], "high")
self.assertEqual(result["selected"]["adapter"], "claude-glm")
self.assertEqual(result["selected"]["target"], "glm-5.2")
self.assertNotIn("thinking_level", result["selected"])
def test_all_candidates_exhausted_returns_no_eligible_target(self):
snapshot = {
@ -788,7 +788,7 @@ class SelectorQuotaRepresentationTests(unittest.TestCase):
self.assertEqual(local_result["quota"]["mode"], "unbounded")
self.assertEqual(local_result["quota"]["status"], "not_applicable")
# Local G07 has Gemini, cloud GLM, and Terra candidates.
# Local G07 has Gemini, Claude GLM, and Terra candidates.
dynamic = write_task_file(Path(tmp), "PLAN", "local", 7)
candidates = selector.select_execution_target(
dynamic, evaluated_at=kst(12)
@ -796,7 +796,8 @@ 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"], "pi")
self.assertEqual(candidates[1]["adapter"], "claude-glm")
self.assertEqual(candidates[1]["target"], "glm-5.2")
self.assertEqual(candidates[1]["quota_status"], "unknown")
self.assertEqual(candidates[1]["execution_class"], "cloud_model")
self.assertFalse(candidates[1]["selfcheck_required"])
@ -1186,7 +1187,7 @@ class SelectorFailoverContractTests(unittest.TestCase):
[
("codex", "gpt-5.3-codex-spark"),
("agy", "Gemini 3.6 Flash (Low)"),
("pi", "iop/glm-5.2"),
("claude-glm", "glm-5.2"),
("codex", "gpt-5.6-terra"),
],
)
@ -1196,15 +1197,15 @@ class SelectorFailoverContractTests(unittest.TestCase):
)
self.assertEqual(
(glm["selected"]["adapter"], glm["selected"]["target"]),
("pi", "iop/glm-5.2"),
("claude-glm", "glm-5.2"),
)
self.assertEqual(glm["selected"]["thinking_level"], "low")
self.assertNotIn("thinking_level", glm["selected"])
self.assertEqual(
glm["used_candidates"],
[
{"adapter": "codex", "target": "gpt-5.3-codex-spark"},
{"adapter": "agy", "target": "Gemini 3.6 Flash (Low)"},
{"adapter": "pi", "target": "iop/glm-5.2", "thinking_level": "low"},
{"adapter": "claude-glm", "target": "glm-5.2"},
],
)
terra = selector.select_execution_target(
@ -1239,7 +1240,8 @@ class SelectorFailoverContractTests(unittest.TestCase):
task_file, evaluated_at=kst(12), transition="failover",
prior_decision=prior, failure_class="provider-quota",
)
self.assertEqual(result["selected"]["adapter"], "pi")
self.assertEqual(result["selected"]["adapter"], "claude-glm")
self.assertEqual(result["selected"]["target"], "glm-5.2")
self.assertEqual(result["transition"]["context_transfer"], "logical")
self.assertEqual(result["transition"]["trigger"], "provider-quota")
self.assertEqual(len(result["used_candidates"]), 2)
@ -1317,9 +1319,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"], "pi")
self.assertEqual(prior["selected"]["target"], "iop/glm-5.2")
self.assertEqual(prior["selected"]["thinking_level"], "high")
self.assertEqual(prior["selected"]["adapter"], "claude-glm")
self.assertEqual(prior["selected"]["target"], "glm-5.2")
self.assertNotIn("thinking_level", prior["selected"])
result = selector.select_execution_target(
task_file,
@ -1447,7 +1449,7 @@ class SelectorFailoverContractTests(unittest.TestCase):
)
self.assertEqual(day_initial["selected"]["adapter"], "agy")
# 23:00 KST is nighttime -> failover to pi
# 23:00 KST is nighttime -> failover to claude-glm
night_failover = selector.select_execution_target(
task_file,
evaluated_at=kst(23, 0, 0),
@ -1455,23 +1457,23 @@ class SelectorFailoverContractTests(unittest.TestCase):
prior_decision=day_initial,
failure_class="provider-quota",
)
self.assertEqual(night_failover["selected"]["adapter"], "pi")
self.assertEqual(night_failover["selected"]["adapter"], "claude-glm")
self.assertEqual(
night_failover["used_candidates"],
[
{"adapter": "agy", "target": "Gemini 3.6 Flash (High)"},
{"adapter": "pi", "target": "iop/glm-5.2", "thinking_level": "high"},
{"adapter": "claude-glm", "target": "glm-5.2"},
],
)
# 23:01 KST nighttime resume -> preserved pinned pi decision
# 23:01 KST nighttime resume -> preserved pinned claude-glm 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"], "pi")
self.assertEqual(night_resume["selected"]["adapter"], "claude-glm")
self.assertIs(night_resume["decision"]["pinned"], True)
self.assertEqual(night_resume["used_candidates"], night_failover["used_candidates"])
@ -1499,9 +1501,9 @@ class SelectorFailoverContractTests(unittest.TestCase):
failure_class="provider-stream-disconnect",
)
self.assertEqual(result["selected"]["adapter"], "pi")
self.assertEqual(result["selected"]["target"], "iop/glm-5.2")
self.assertEqual(result["selected"]["thinking_level"], "high")
self.assertEqual(result["selected"]["adapter"], "claude-glm")
self.assertEqual(result["selected"]["target"], "glm-5.2")
self.assertNotIn("thinking_level", result["selected"])
self.assertEqual(result["quota"]["status"], "unknown")
self.assertEqual(run_mock.call_count, 2)