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

fix(agent-ops): Pi quota failover와 GLM cloud 경로를 보강한다

Browse source
This commit is contained in:
toki 2026-08-04 10:28:28 +09:00
parent 95a81ca65f
commit 9459c896fc
6 changed files with 337 additions and 86 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1,6 +1,6 @@
---
name: orchestrate-agent-task-loop
description: Run agent-task work and autonomously execute active PLAN/CODE_REVIEW loops on request. Use when dispatching dependency-ready work in parallel by predecessor completion and workspace write claims, running lane/G-specific Codex, Claude, agy, and Pi workers, adding Pi self-checks, converging official Codex reviews, and escalating cloud context until the task loop finishes.
description: Run agent-task work and autonomously execute active PLAN/CODE_REVIEW loops on request. Use when dispatching dependency-ready work in parallel by predecessor completion and workspace write claims, running lane/G-specific Codex, Claude, agy, and Pi workers, adding local-Pi self-checks, converging official Codex reviews, and escalating cloud context until the task loop finishes.
---
# Orchestrate Agent Task Loop
@ -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 |
| `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-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 |
| `cloud-G03`–`cloud-G04` | agy `Gemini 3.6 Flash (Medium)` → Pi `iop/glm-5.2`, thinking medium |
| `cloud-G05`–`cloud-G06` | agy `Gemini 3.6 Flash (High)` → Pi `iop/glm-5.2`, thinking high |
| `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-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`.
Do not treat 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`; 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.
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, 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 that next GLM 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, 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 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.
- 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 Pi to a cloud model.
- Never escalate local Pi models to a cloud model. Pi-hosted cloud 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.

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

@ -1814,7 +1814,7 @@ def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
if resolved_path[-1] != canonical:
raise ExecutionDecisionError("selector promotion path tail이 selected와 다르다")
if adapter == "pi":
if not target.startswith("iop/") or not local_pi:
if not target.startswith("iop/"):
raise ExecutionDecisionError("Pi selector target/schema가 유효하지 않다")
model = target.removeprefix("iop/")
return AgentSpec(
@ -1876,15 +1876,24 @@ def _spec_from_completing_decision(decision: dict[str, Any]) -> AgentSpec:
raise ExecutionDecisionError(
f"Pi completing decision target이 iop/ prefix가 아니다: {target}"
)
if execution_class != "local_model":
raise ExecutionDecisionError(
f"Pi completing decision execution_class이 local_model이 아니다: {execution_class}"
)
if not selfcheck_required:
raise ExecutionDecisionError(
"Pi completing decision selfcheck_required가 False이다"
)
model = target.removeprefix("iop/")
glm_cloud = model == "glm-5.2"
expected_class = "cloud_model" if glm_cloud else "local_model"
expected_selfcheck = not glm_cloud
legacy_glm_contract = (
glm_cloud
and execution_class == "local_model"
and selfcheck_required
)
if not legacy_glm_contract and (
execution_class != expected_class
or selfcheck_required != expected_selfcheck
):
raise ExecutionDecisionError(
"Pi completing decision execution/selfcheck 계약이 유효하지 않다: "
f"target={target} execution_class={execution_class} "
f"selfcheck_required={selfcheck_required}"
)
if thinking_level is not None and thinking_level not in {"low", "medium", "high"}:
raise ExecutionDecisionError(
f"Pi completing decision thinking_level이 유효하지 않다: {thinking_level!r}"
@ -2434,6 +2443,14 @@ def completing_decision_requires_selfcheck(state: dict[str, Any]) -> bool:
selected = completing.get("selected")
if not isinstance(selected, dict):
return False
if (
selected.get("adapter") == "pi"
and selected.get("target") == "iop/glm-5.2"
):
return False
selfcheck_required = selected.get("selfcheck_required")
if isinstance(selfcheck_required, bool):
return selfcheck_required
return selected.get("execution_class") == "local_model"
@ -2735,6 +2752,22 @@ def terminal_diagnostic(cli: str, channel: str, line: str) -> str | None:
return line
return None
event_type = str(value.get("type", ""))
if cli == "pi":
if event_type == "auto_retry_end" and value.get("success") is False:
final_error = value.get("finalError")
return final_error if isinstance(final_error, str) and final_error else None
if event_type == "agent_end" and value.get("willRetry") is False:
messages = value.get("messages")
if isinstance(messages, list) and messages:
message = messages[-1]
if (
isinstance(message, dict)
and message.get("stopReason") == "error"
):
error_message = message.get("errorMessage")
if isinstance(error_message, str) and error_message:
return error_message
return None
if cli == "codex" and event_type in {"turn.failed", "error"}:
return json.dumps(value.get("error", value), ensure_ascii=False)
if cli == "agy":

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

@ -46,13 +46,13 @@ AGY_GEMINI_HIGH = RouteTarget(
)
PI_LAGUNA = RouteTarget("pi", "iop/laguna-s:2.1", "local_model", True)
PI_GLM_LOW = RouteTarget(
"pi", "iop/glm-5.2", "local_model", True, thinking_level="low"
"pi", "iop/glm-5.2", "cloud_model", False, thinking_level="low"
)
PI_GLM_MEDIUM = RouteTarget(
"pi", "iop/glm-5.2", "local_model", True, thinking_level="medium"
"pi", "iop/glm-5.2", "cloud_model", False, thinking_level="medium"
)
PI_GLM_HIGH = RouteTarget(
"pi", "iop/glm-5.2", "local_model", True, thinking_level="high"
"pi", "iop/glm-5.2", "cloud_model", False, thinking_level="high"
)
CLAUDE_OPUS = RouteTarget("claude", "claude-opus-4-8", "cloud_model", False)
CLAUDE_HAIKU_XHIGH = RouteTarget(
@ -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)
candidates = (AGY_GEMINI_HIGH, PI_GLM_HIGH, 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)
candidates = (AGY_GEMINI_HIGH, PI_GLM_HIGH, CODEX_TERRA_HIGH)
return PolicyDecision(
rule_id=rule_id,
policy_priority=20,
@ -206,15 +206,16 @@ def select_policy(
CODEX_SPARK_XHIGH,
AGY_GEMINI_LOW,
PI_GLM_LOW,
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)
candidates = (AGY_GEMINI_MEDIUM, PI_GLM_MEDIUM, 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)
candidates = (AGY_GEMINI_HIGH, PI_GLM_HIGH, CODEX_TERRA_HIGH)
rule_id = "worker-cloud-g05-g06"
reason_code = "cloud_gemini_high_grade"
elif grade <= 8:

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

@ -772,6 +772,84 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase):
finally:
store.close()
async def test_glm_cloud_completion_skips_selfcheck(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
(workspace / ".git").mkdir()
task = self.make_task(workspace)
store = dispatch.StateStore(workspace)
try:
decision = {
"work_unit_id": self._WORK_UNIT_ID,
"stage": "worker",
"selected": {
"adapter": "pi",
"target": "iop/glm-5.2",
"execution_class": "cloud_model",
"selfcheck_required": False,
"thinking_level": "high",
},
}
store.update_task(
task,
worker_done=True,
worker_cli="pi",
worker_model="glm-5.2",
completing_decision=decision,
execution_class="cloud_model",
selfcheck_done=True,
blocked=None,
)
self.assertFalse(
dispatch.completing_decision_requires_selfcheck(
store.task_state(task)
)
)
self.assertEqual(
dispatch.task_stage(task, store.task_state(task)),
"review",
)
finally:
store.close()
async def test_legacy_glm_local_completion_skips_obsolete_selfcheck(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
(workspace / ".git").mkdir()
task = self.make_task(workspace)
store = dispatch.StateStore(workspace)
try:
legacy_decision = {
"work_unit_id": self._WORK_UNIT_ID,
"stage": "worker",
"selected": {
"adapter": "pi",
"target": "iop/glm-5.2",
"execution_class": "local_model",
"selfcheck_required": True,
"thinking_level": "high",
},
}
store.update_task(
task,
worker_done=True,
worker_cli="pi",
worker_model="glm-5.2",
completing_decision=legacy_decision,
execution_class="local_model",
selfcheck_done=False,
blocked="selfcheck-incomplete-limit",
)
state = store.task_state(task)
self.assertTrue(dispatch._completing_decision_is_valid(task, state))
self.assertFalse(dispatch.completing_decision_requires_selfcheck(state))
state["blocked"] = None
self.assertEqual(dispatch.task_stage(task, state), "review")
finally:
store.close()
async def test_cloud_completing_decision_skips_selfcheck(self):
"""execution_class=cloud_model skips selfcheck entirely."""
with tempfile.TemporaryDirectory() as temporary:
@ -3093,6 +3171,62 @@ 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):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
(workspace / ".git").mkdir()
task = TaskStageTest().make_task(workspace)
store = dispatch.StateStore(workspace)
error_message = (
'429: {"code":"1308","message":"Usage limit reached for '
'5 hour. Your limit will reset at 2026-08-04 08:43:16"}'
)
terminal_event = json.dumps(
{
"type": "auto_retry_end",
"success": False,
"attempt": 3,
"finalError": error_message,
}
)
command = [
sys.executable,
"-c",
"import sys; print(sys.argv[1])",
terminal_event,
]
spec = dispatch.AgentSpec(
"pi",
"glm-5.2",
"pi/iop/glm-5.2 high",
local_pi=True,
thinking_level="high",
)
try:
with mock.patch.object(
dispatch,
"build_command",
return_value=command,
):
rc, failure, locator = await dispatch.invoke(
workspace,
store,
task,
"worker",
spec,
"Read the plan.",
)
finally:
store.close()
self.assertEqual(rc, 0)
self.assertEqual(failure, "provider-quota")
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)
async def test_exit_143_is_process_termination_not_provider_failure(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
@ -3331,6 +3465,19 @@ class ReviewControlTest(unittest.TestCase):
line = "provider_tunnel_error: connection refused"
self.assertIsNone(dispatch.terminal_diagnostic("pi", "stdout", line))
def test_pi_intermediate_retry_error_is_not_terminal_diagnostic(self):
event = json.dumps(
{
"type": "message_end",
"message": {
"role": "assistant",
"stopReason": "error",
"errorMessage": "429: Usage limit reached",
},
}
)
self.assertIsNone(dispatch.terminal_diagnostic("pi", "stdout", event))
def test_dispatcher_source_provenance_detects_hot_edit(self):
changed_sha256 = "f" * 64
self.assertNotEqual(changed_sha256, dispatch.DISPATCHER_SOURCE_SHA256)
@ -9071,7 +9218,9 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
async def mock_invoke(*args, **kwargs):
spec = args[4]
invoked_specs.append(spec)
return (1, "provider-quota", locator)
if spec.cli == "pi":
return (1, "provider-quota", locator)
return (0, None, locator)
with (
mock.patch.object(dispatch, "invoke", new=mock_invoke),
@ -9079,12 +9228,18 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas
):
success, final_loc = await dispatch.run_escalating(workspace, store, task, "worker", glm_spec)
self.assertFalse(success)
self.assertEqual(len(invoked_specs), 1)
self.assertTrue(success)
self.assertEqual(len(invoked_specs), 2)
self.assertEqual(invoked_specs[0].cli, "pi")
self.assertEqual(invoked_specs[1].cli, "codex")
self.assertEqual(invoked_specs[1].model, "gpt-5.6-terra")
state = store.task_state(task)
self.assertIn("no_failover_candidate", state.get("blocked", ""))
self.assertIsNone(state.get("blocked"))
self.assertEqual(
state["execution_decisions"]["worker"]["selected"]["target"],
"gpt-5.6-terra",
)
finally:
store.close()
@ -9578,25 +9733,53 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
task = self.make_task(workspace, lane="local", grade=8)
store = dispatch.StateStore(workspace)
try:
# 1. Initial decision local G08 -> agy Gemini Medium (primary) & pi Laguna (fallback)
# 1. Initial decision local G08 -> Gemini High, GLM High, 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 Laguna
# 2. Qualified failover (provider-quota) -> transitions to Pi 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")
# 3. Subsequent failover when no candidate remains -> raises no_failover_candidate
terra_available = {
"schema_version": "1.0",
"snapshot_id": "terra-available",
"source": "test",
"checked_at": daytime.isoformat(),
"targets": [
{
"adapter": "codex",
"target": "gpt-5.6-terra",
"status": "available",
}
],
"required_caps": [],
"reason_codes": [],
}
# 3. GLM quota failover continues to the final Codex Terra backup.
dec3 = dispatch.select_execution_decision(
task, stage="worker", prior_decision=dec2,
evaluated_at=daytime, transition="failover", failure_class="provider-quota",
quota_snapshot=terra_available,
)
self.assertEqual(dec3["transition"]["trigger"], "provider-quota")
self.assertEqual(dec3["selected"]["adapter"], "codex")
self.assertEqual(dec3["selected"]["target"], "gpt-5.6-terra")
# 4. No candidate remains after Terra.
with self.assertRaises(dispatch.ExecutionDecisionError) as ctx:
dispatch.select_execution_decision(
task, stage="worker", prior_decision=dec2,
evaluated_at=daytime, transition="failover", failure_class="provider-quota"
task, stage="worker", prior_decision=dec3,
evaluated_at=daytime, transition="failover", failure_class="provider-quota",
quota_snapshot=terra_available,
)
self.assertIn("no_failover_candidate", str(ctx.exception))
finally:
@ -10078,35 +10261,15 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase):
self.assertEqual([s.cli for s in invoked_specs], ["agy", "pi"])
state = store.task_state(task)
self.assertEqual(state["execution_class"], "local_model")
self.assertFalse(state["selfcheck_done"])
self.assertEqual(dispatch.task_stage(task, state), "selfcheck")
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["completing_decision"]["selected"]["execution_class"], "local_model"
state["completing_decision"]["selected"]["execution_class"], "cloud_model"
)
hist1 = list(state["route_transition_history"])
self.assertEqual([h["transition"] for h in hist1], ["initial", "resume", "provider-quota"])
selfcheck_specs = []
async def mock_invoke_selfcheck(*args, **kwargs):
spec = args[4]
selfcheck_specs.append(spec)
return (0, None, loc_glm)
with (
mock.patch.object(dispatch, "invoke", new=mock_invoke_selfcheck),
mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()),
mock.patch.object(dispatch, "implementation_review_errors", return_value=[]),
):
await dispatch.run_selfcheck(workspace, store, task)
self.assertEqual([s.cli for s in selfcheck_specs], ["pi"])
state2 = store.task_state(task)
self.assertTrue(state2["selfcheck_done"])
self.assertEqual(dispatch.task_stage(task, state2), "review")
hist2 = state2["route_transition_history"]
self.assertEqual([h["transition"] for h in hist2], ["initial", "resume", "provider-quota"])
finally:
store.close()
@ -10374,11 +10537,14 @@ class ThroughputQuotaBatchTest(unittest.TestCase):
ready = [(t_night, "worker"), (t_review, "review")]
batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now)
# Night local-G08 now starts on Gemini High; review remains excluded.
# Night local-G08 probes Gemini High plus its Codex Terra backup.
# Pi-hosted GLM has no dedicated quota probe; review remains excluded.
self.assertIsNotNone(batch_snap)
self.assertEqual(len(probe_calls), 1)
self.assertEqual(len(probe_calls), 2)
self.assertEqual(probe_calls[0]["adapter"], "agy")
self.assertEqual(probe_calls[0]["target"], "Gemini 3.6 Flash (High)")
self.assertEqual(probe_calls[1]["adapter"], "codex")
self.assertEqual(probe_calls[1]["target"], "gpt-5.6-terra")
finally:
store.close()
@ -10904,7 +11070,9 @@ class ThroughputQuotaBatchTest(unittest.TestCase):
res = await dispatch.dispatch_with_store(args, workspace, store)
run_sub.assert_not_called()
self.assertEqual(len(probe_calls), 0)
self.assertEqual(len(probe_calls), 1)
self.assertEqual(probe_calls[0]["adapter"], "codex")
self.assertEqual(probe_calls[0]["target"], "gpt-5.6-terra")
st_blocked_after = store.task_state(t_blocked)
self.assertIsNone(st_blocked_after.get("blocked"))
@ -10922,10 +11090,9 @@ class ThroughputQuotaBatchTest(unittest.TestCase):
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]
self.assertEqual(len(blocked_invocations), 2)
self.assertEqual(len(blocked_invocations), 1)
self.assertEqual(blocked_invocations[0][1], "worker")
self.assertEqual(blocked_invocations[0][4], loc_path)
self.assertEqual(blocked_invocations[1][1], "selfcheck")
st_normal_after = store.task_state(t_normal)
self.assertFalse(st_normal_after.get("retry_quota_refresh_pending"))

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

@ -49,7 +49,10 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
stage="worker", lane="local", grade=8, evaluated_at=night_time
)
self.assertEqual(decision.rule_id, "worker-local-g07-g08-kst-night")
self.assertEqual(decision.candidates, (policy.AGY_GEMINI_HIGH, policy.PI_GLM_HIGH))
self.assertEqual(
decision.candidates,
(policy.AGY_GEMINI_HIGH, policy.PI_GLM_HIGH, 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)")
@ -118,6 +121,7 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
policy.CODEX_SPARK_XHIGH,
policy.AGY_GEMINI_LOW,
policy.PI_GLM_LOW,
policy.CODEX_TERRA_HIGH,
),
)
self.assertEqual(
@ -151,8 +155,23 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
grade=8,
evaluated_at=at_utc(15),
)
self.assertEqual(daytime.candidates, (policy.AGY_GEMINI_HIGH, policy.PI_GLM_HIGH))
self.assertEqual(nighttime.candidates, (policy.AGY_GEMINI_HIGH, policy.PI_GLM_HIGH))
expected = (
policy.AGY_GEMINI_HIGH,
policy.PI_GLM_HIGH,
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_invalid_inputs_are_rejected(self):
cases = [

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

@ -524,19 +524,19 @@ class SelectorRouteMatrixTests(unittest.TestCase):
dynamic, evaluated_at=kst(2)
)["candidates"]
self.assertEqual(
[c["candidate_rank"] for c in daytime], [1, 2]
[c["candidate_rank"] for c in daytime], [1, 2, 3]
)
self.assertEqual(
[c["adapter"] for c in daytime], ["agy", "pi"]
[c["adapter"] for c in daytime], ["agy", "pi", "codex"]
)
self.assertEqual(
[c["adapter"] for c in nighttime], ["agy", "pi"]
[c["adapter"] for c in nighttime], ["agy", "pi", "codex"]
)
single = write_task_file(Path(tmp), "PLAN", "cloud", 5)
candidates = selector.select_execution_target(
single, evaluated_at=kst(12)
)["candidates"]
self.assertEqual([c["candidate_rank"] for c in candidates], [1, 2])
self.assertEqual([c["candidate_rank"] for c in candidates], [1, 2, 3])
class SelectorQuotaRepresentationTests(unittest.TestCase):
@ -788,15 +788,19 @@ class SelectorQuotaRepresentationTests(unittest.TestCase):
self.assertEqual(local_result["quota"]["mode"], "unbounded")
self.assertEqual(local_result["quota"]["status"], "not_applicable")
# Local G07 has Gemini primary candidate and Laguna fallback.
# Local G07 has Gemini, cloud GLM, and Terra candidates.
dynamic = write_task_file(Path(tmp), "PLAN", "local", 7)
candidates = selector.select_execution_target(
dynamic, evaluated_at=kst(12)
)["candidates"]
self.assertEqual(len(candidates), 2)
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]["quota_status"], "unknown")
self.assertEqual(candidates[1]["execution_class"], "cloud_model")
self.assertFalse(candidates[1]["selfcheck_required"])
self.assertEqual(candidates[2]["adapter"], "codex")
def test_injected_snapshot_is_reflected(self):
snapshot = {
@ -1183,6 +1187,7 @@ class SelectorFailoverContractTests(unittest.TestCase):
("codex", "gpt-5.3-codex-spark"),
("agy", "Gemini 3.6 Flash (Low)"),
("pi", "iop/glm-5.2"),
("codex", "gpt-5.6-terra"),
],
)
self.assertEqual(
@ -1202,12 +1207,24 @@ class SelectorFailoverContractTests(unittest.TestCase):
{"adapter": "pi", "target": "iop/glm-5.2", "thinking_level": "low"},
],
)
terra = selector.select_execution_target(
task_file,
evaluated_at=kst(12),
transition="failover",
prior_decision=glm,
failure_class="provider-quota",
quota_probe_command="missing-probe",
)
self.assertEqual(
(terra["selected"]["adapter"], terra["selected"]["target"]),
("codex", "gpt-5.6-terra"),
)
with self.assertRaises(selector.SelectorInputError) as exhausted:
selector.select_execution_target(
task_file,
evaluated_at=kst(12),
transition="failover",
prior_decision=glm,
prior_decision=terra,
failure_class="provider-quota",
quota_probe_command="missing-probe",
)
@ -1243,9 +1260,14 @@ class SelectorFailoverContractTests(unittest.TestCase):
task_file, evaluated_at=kst(12), transition="failover",
prior_decision=prior, failure_class="provider-quota",
)
second = selector.select_execution_target(
task_file, evaluated_at=kst(12), transition="failover",
prior_decision=first, failure_class="provider-quota",
)
self.assertEqual(second["selected"]["target"], "gpt-5.6-terra")
with self.assertRaises(selector.SelectorInputError) as exhausted:
selector.select_execution_target(
task_file, evaluated_at=kst(12), transition="failover", prior_decision=first, failure_class="provider-quota",
task_file, evaluated_at=kst(12), transition="failover", prior_decision=second, failure_class="provider-quota",
)
self.assertEqual(exhausted.exception.code, "no_failover_candidate")
@ -1256,8 +1278,13 @@ class SelectorFailoverContractTests(unittest.TestCase):
first = selector.select_execution_target(task_file, evaluated_at=kst(12), transition="failover", prior_decision=prior, failure_class="provider-stream-disconnect")
resumed = selector.select_execution_target(task_file, evaluated_at=kst(12), transition="resume", prior_decision=first)
self.assertEqual(resumed["used_candidates"], first["used_candidates"])
second = selector.select_execution_target(
task_file, evaluated_at=kst(12), transition="failover",
prior_decision=resumed, failure_class="provider-quota",
)
self.assertEqual(second["selected"]["target"], "gpt-5.6-terra")
with self.assertRaises(selector.SelectorInputError) as repeated:
selector.select_execution_target(task_file, evaluated_at=kst(12), transition="failover", prior_decision=resumed, failure_class="provider-quota")
selector.select_execution_target(task_file, evaluated_at=kst(12), transition="failover", prior_decision=second, failure_class="provider-quota")
self.assertEqual(repeated.exception.code, "no_failover_candidate")
def test_failover_never_returns_to_an_earlier_candidate_rank(self):
@ -1294,16 +1321,20 @@ class SelectorFailoverContractTests(unittest.TestCase):
self.assertEqual(prior["selected"]["target"], "iop/glm-5.2")
self.assertEqual(prior["selected"]["thinking_level"], "high")
with self.assertRaises(selector.SelectorInputError) as ctx:
selector.select_execution_target(
task_file,
evaluated_at=kst(12),
transition="failover",
prior_decision=prior,
failure_class="provider-stream-disconnect",
quota_snapshot=gemini_available_snapshot,
)
self.assertEqual(ctx.exception.code, "no_failover_candidate")
result = selector.select_execution_target(
task_file,
evaluated_at=kst(12),
transition="failover",
prior_decision=prior,
failure_class="provider-stream-disconnect",
quota_snapshot=gemini_available_snapshot,
)
self.assertEqual(result["selected"]["adapter"], "codex")
self.assertEqual(result["selected"]["target"], "gpt-5.6-terra")
self.assertNotIn(
{"adapter": "agy", "target": "Gemini 3.6 Flash (High)"},
result["used_candidates"][1:],
)
def test_tampered_prior_decision_rejected(self):
with TemporaryDirectory() as tmp:
@ -1471,8 +1502,8 @@ class SelectorFailoverContractTests(unittest.TestCase):
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["quota"]["status"], "not_applicable")
self.assertEqual(run_mock.call_count, 1)
self.assertEqual(result["quota"]["status"], "unknown")
self.assertEqual(run_mock.call_count, 2)
def test_policy_owned_cloud_promotion_chain_and_no_bounce(self):
with TemporaryDirectory() as tmp: