From 3e67613360e3f86f33d7af699008716f64ce8aa0 Mon Sep 17 00:00:00 2001 From: toki Date: Wed, 5 Aug 2026 09:54:10 +0900 Subject: [PATCH] =?UTF-8?q?feat(dispatcher):=20selfcheck=20=EB=8B=A8?= =?UTF-8?q?=EA=B3=84=EB=A5=BC=20=EC=84=A4=EC=A0=95=EC=9C=BC=EB=A1=9C=20?= =?UTF-8?q?=EB=B6=84=EB=A6=AC=ED=95=9C=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 작업 모델별로 전체 검토와 코드리뷰 체크 검토를 독립적으로 켜고 끌 수 있어야 하며, 다음 단계 진입 시 변경된 카탈로그 설정을 재시작 없이 반영해야 한다. --- .../orchestrate-agent-task-loop/SKILL.md | 44 +- .../scripts/dispatch.py | 330 +++++++++++++-- .../scripts/execution_target_catalog.json | 70 +++- .../scripts/execution_target_contract.py | 6 +- .../scripts/execution_target_policy.py | 107 ++++- .../tests/test_dispatch.py | 387 +++++++++++++++++- .../tests/test_execution_target_policy.py | 41 ++ 7 files changed, 883 insertions(+), 102 deletions(-) diff --git a/agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md b/agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md index 3868f4b1..1f13fb6c 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md @@ -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 catalog-selected lane/G workers and reviewers, adding local-Pi self-checks, converging official reviews, and escalating cloud context until the task loop finishes. +description: Run agent-task work and autonomously execute active PLAN/CODE_REVIEW loops on request. Use when dispatching dependency-ready work in parallel by predecessor completion and workspace write claims, running catalog-selected lane/G workers and reviewers, applying target-configured self-check stages, converging official reviews, and escalating cloud context until the task loop finishes. --- # Orchestrate Agent Task Loop @@ -75,12 +75,16 @@ reinterpret a target from task prose or environment variables. eligible target is the default and qualified terminal failures advance to the next unused eligible target without returning to an earlier rank. - `targets` owns adapter, model, command model, execution class, self-check, - thinking, and reasoning options. Reorder or replace existing target ids by - editing only the lane array. Add a model for an existing adapter by adding one - target entry and referencing its id. Only a new CLI/driver requires Python - dispatcher support. -- The dispatcher validates the catalog at process start. New work uses that - revision; already persisted work keeps its ordered candidate snapshot. + thinking, and reasoning options. `selfcheck.full_review` and + `selfcheck.checklist_review` independently enable the full-work review and + implementation-checklist-only review for that exact worker target. Reorder or + replace existing target ids by editing only the lane array. Add a model for an + existing adapter by adding one target entry and referencing its id. Only a new + CLI/driver requires Python dispatcher support. +- The dispatcher reloads and validates the catalog before each scheduler + admission and again immediately before a self-check starts. A running model + invocation keeps its pinned decision, while the next task or self-check stage + uses the latest switches without a dispatcher source change or restart. - Missing grade lanes, unknown target ids, duplicate candidates or runtime identities, invalid options, and incomplete time-window metadata fail closed. @@ -105,25 +109,29 @@ Keep control prompts in English, insert absolute paths only, and do not expand t - A dispatcher child runs only while `IOP_AGENT_TASK_EXECUTION_ID` is present. - Prefix every worker and review prompt with: `You are a child agent already launched by the dispatcher, not the orchestration caller. Execute only the assigned role directly. Do not start, monitor, or wait for orchestration through dispatch.py or orchestrate-agent-task-loop. You may run dispatch.py --validate-plan only when required by plan or code-review finalization because that mode validates one candidate PLAN without starting or monitoring orchestration.` -- Keep local self-check prompts short. Start fresh self-check and recovery prompts with: `Think in English. Final in Korean.` The same-session unchecked-item retry uses the exact terse prompt below. +- Keep self-check prompts short. Start full-review, checklist-review, and recovery prompts with: `Think in English. Final in Korean.` - Cloud worker: `Read {PLAN_PATH} and complete the task. Keep artifact content in English. Final in Korean.` - Pi worker: `Think in English. Keep artifact content in English. Final in Korean. Read {PLAN_PATH} and complete the task.` -- Pi self-check full pass: `Think in English. Final in Korean. Read {PLAN_PATH}; review all work once, fix omissions, and update {CODE_REVIEW_PATH}. Keep files in English.` -- Pi self-check unchecked-item retry: `The code review file is incomplete! Complete it now!` +- Self-check full review: `Think in English. Final in Korean. Read {PLAN_PATH}; review all work once, fix omissions, and update {CODE_REVIEW_PATH}. Keep files in English.` +- Self-check checklist review: `Think in English. Final in Korean. Read {CODE_REVIEW_PATH}. Review only its Implementation Checklist section. Mark every completed item, finish any missing implementation or evidence required by those items, and leave all official-review-only sections untouched. Keep files in English.` - Official review: `Read {CODE_REVIEW_PATH} and start the review. Keep artifact content in English. Final in Korean.` - Review-exit recovery: `Continue the review for {TASK_PATH}. Keep artifact content in English. Final in Korean.` - Context escalation: `Continue from {LOCATOR_PATH}. Check the saved context and current workspace. Keep artifact content in English. Final in Korean.` Never ask a worker, self-check, or review model to create, edit, or summarize `WORK_LOG.md`. -Run Pi self-check only for catalog targets marked `local_model` and `selfcheck_required=true`; cloud targets skip self-check. Do not treat a local Pi self-check exit code `0` as success by itself. Set `selfcheck_done=true` only when `## Implementation Checklist` (or legacy `## 구현 체크리스트`) in `CODE_REVIEW_PATH` contains at least one Markdown list checkbox and every `[...]` checkbox value has at least one non-whitespace character. If both canonical and legacy checklist headings are present in the same file, fail closed. Accept any non-empty value, including `x`, `v`, and `✅`. Do not inspect `## Implementation Item Completion`, `Deviations from Plan`, `Key Design Decisions`, `Verification Results`, or final CODE_REVIEW synchronization text. Run the full self-check prompt exactly once. If its checklist condition fails, resume that successful pass's Pi native session and run the unchecked-item retry prompt up to 10 times. Each retry must resume the locator returned by the preceding successful pass so the same conversation context is preserved; never repeat the full review prompt or start a fresh retry session. Persist the latest successful context locator for dispatcher restart, and block instead of starting fresh when that context cannot be resumed. Block that task after the 10th unchecked-item retry remains incomplete, and continue draining independent work. +Resolve self-check stages from the completing worker target's live catalog entry; do not rerun target selection or substitute another model. Treat `selfcheck.full_review` and `selfcheck.checklist_review` as separate scheduler stages and persist `selfcheck_full_review_done` and `selfcheck_checklist_review_done` independently. A disabled stage is skipped. A newly enabled unfinished stage runs before official review on the next scheduler entry. + +Treat catalog `selfcheck_required` as a legacy persisted-decision compatibility field only; operators configure the two runtime stages through the nested `selfcheck` object. When migrating old Pi state, `selfcheck_done=true` means both stages completed, while `selfcheck_done=false` with `selfcheck_incomplete > 0` means full review completed and checklist review remains unfinished. Legacy cloud `selfcheck_done=true` means the old dispatcher skipped self-check and does not satisfy a newly enabled stage. + +The full-review stage runs its prompt exactly once. The checklist-review stage first evaluates `## Implementation Checklist` (or legacy `## 구현 체크리스트`) in `CODE_REVIEW_PATH`; if it already contains at least one Markdown list checkbox and every `[...]` checkbox value has at least one non-whitespace character, complete the stage without invoking a model. 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. When incomplete, run one checklist-only pass plus up to 10 checklist-only retries. For Pi, each retry resumes the locator returned by the preceding successful pass; persist the latest successful context locator across dispatcher restart and block instead of starting fresh when it cannot be resumed. For cloud targets, each checklist-only retry starts fresh on the same completing target. Never promote or substitute a different self-check target after a provider failure. Keep the two stages' process-recovery budgets independent. Block that task after the 10th checklist 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 ordered lane failover chain; `available` or `unknown` remains a completion-evidence recovery on Gemini. -For Pi worker recovery attempts, pass only `Read {PLAN_PATH}. Continue.` without a locator explanation. Pi self-check recovery must preserve the current full-pass or unchecked-item role and use its concise prompt. For other CLI escalation attempts, pass `Continue from {LOCATOR_PATH}. Check the saved context and current workspace. Keep artifact content in English. Final in Korean.` Preserve the collaboration prohibition and next-state-materialization sentence in official-review escalation and recovery prompts. Do not ask the model to write a separate handoff summary. +For Pi worker recovery attempts, pass only `Read {PLAN_PATH}. Continue.` without a locator explanation. Pi self-check recovery must preserve the current full-review or checklist-review role and use its concise prompt. For other CLI escalation attempts, pass `Continue from {LOCATOR_PATH}. Check the saved context and current workspace. Keep artifact content in English. Final in Korean.` Preserve the collaboration prohibition and next-state-materialization sentence in official-review escalation and recovery prompts. Do not ask the model to write a separate handoff summary. -When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a terminal `session-stall` locator left by an earlier dispatcher, first require the locator and native session to belong to the current physical workspace. Do not create a fresh session ID for an owned locator. Resume its native session file with `pi --session` and the existing `--session-dir`. For worker recovery pass `Think in English. Keep artifact content in English. Final in Korean. Continue this session and complete the current task.` For interrupted full self-check recovery pass `Think in English. Final in Korean. Continue. Keep files in English.` For an unchecked-item retry, pass its normal concise prompt while resuming the existing native session. After a dispatcher restart, find the owned locator and resume the same session. Count this same-session restart toward the same stage's 10-consecutive-failure limit. +When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a terminal `session-stall` locator left by an earlier dispatcher, first require the locator and native session to belong to the current physical workspace. Do not create a fresh session ID for an owned locator. Resume its native session file with `pi --session` and the existing `--session-dir`. For worker recovery pass `Think in English. Keep artifact content in English. Final in Korean. Continue this session and complete the current task.` For interrupted full-review recovery pass `Think in English. Final in Korean. Continue. Keep files in English.` For a checklist-review retry, pass its normal concise prompt while resuming the existing native session. After a dispatcher restart, find the owned locator and resume the same session. Count this same-session restart toward the same stage's 10-consecutive-failure limit. ## Work-Log Contract @@ -161,7 +169,7 @@ When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a termin - Determine every CLI's health/progress primarily from actual stdout/stderr in `stream.log`, plus native session events when available. Before accepting PID, marker, native-session, or stream evidence, require the locator path and recorded workspace identity to belong to the current physical workspace; accept an identity-less legacy locator only under the current store's `runs` root. Never use heartbeat mtime as progress evidence. Record workspace id, dispatcher PID, agent PID, each process start token, and the per-attempt process environment marker in the locator; namespace that marker by workspace. Another dispatcher must not start a duplicate attempt merely because the stream is quiet when the PID/start token or marker shows the same process is alive. For a locator without an agent PID, never infer stale state or duplicate recovery from elapsed time while any stream/native progress evidence exists; use only an actual terminal error or confirmed process exit as recovery evidence for every model. Run Pi with `--mode json` so `thinking_delta`, `text_delta`, and tool streams reach stdout. End an **exact** Pi toolCall-to-all-toolResult interval only when every `toolCall.id` in the preceding assistant event matches a later `toolResult.toolCallId`; never terminate the process on a time limit. If the locator lacks an agent PID during this interval, never classify it as stale or duplicate recovery based on log age; require recorded process evidence to show termination. Do not infer tool execution from `starting`, `unknown`, model reasoning, or post-toolResult state. Outside this interval, use only `stream.log` updates for Pi liveness; toolResult alone does not reset the model-response silence clock. If the stream stops for three minutes outside tool execution, store the final stream excerpt as `pi_silence_inspection` for Pi or `stream_silence_inspection` for another CLI, emit `모델응답점검`, and do not terminate the model process. Recover only from an actual terminal error or process exit. - Detect a local-model `repetition-loop` only when the same normalized chunk repeats three consecutive times with no new tool event or file/state change. Do not infer it from similarity or semantic duplication in `thinking_delta`/`text_delta`. This signal alone must not terminate the process, block the task, trigger recovery/retry, or escalate the model; keep observing for substantive progress or an actual terminal error. - Keep `provider-connection`, `provider-stream-disconnect`, `session-stall`, `generic-error`, `process-terminated`, context/quota/model errors, and review-control violations distinct, but make them share a budget of 10 consecutive automatic recovery failures for the same task stage. On the 10th failure, block that task and do not auto-resume after cooldown. Reset the stage counter after success. -- Record an explicit terminal blocker when the initial Pi full self-check plus 10 same-context unchecked-item retries leave the implementation checklist incomplete, or official review makes no change 10 consecutive times. +- Record an explicit terminal blocker when a checklist-review initial pass plus 10 retries leaves the implementation checklist incomplete, or official review makes no change 10 consecutive times. - While one task recovers or becomes blocked, continue every ready/running task that neither requires it as a predecessor nor collides with its retained workspace claim. Internal recovery or blocking must not trigger an arbitrary complete-candidate rescan. - If review shared-state preflight fails, block only ready review tasks and still start every worker/self-check with a disjoint claim in the same pass. The complete scan after `complete.log` must preserve the existing snapshot rather than reread already running task directories, avoiding races with parallel archive moves that could stop another process. - For KST-night `local-G07`–`local-G08` Laguna locator `context-limit`/`session-stall`, prefer the Prompt Contract's same-session resume and display `Pi세션연속재시작`. Use a fresh session and `세션응답복구재시도` only for other legacy Pi `session-stall` recovery. @@ -227,9 +235,9 @@ When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a termin - If a worker/self-check/review future ends without `complete.log`, reread only that task and run its next stage. Do not rescan the complete candidate set. - Persist `active_stage` for a running task. After dispatcher restart, exclude that task from candidates, restore or conservatively adopt its workspace write claim, and immediately dispatch every other dependency-ready task whose claim does not collide. - **ABSOLUTE RULE:** Scan the complete candidate set only at initial entry and immediately after creating a verified `complete.log`. In that scan, exclude tasks shown as running by current-workspace state and native session/locator evidence, then atomically admit every dependency-ready task with a non-colliding write claim. An unmet dependency or write collision excludes only that task. Exit instead of polling when no candidate remains. - - Persist Pi worker success, Pi self-check success, and official review as separate stages. If restart state is `worker_done=true` and `selfcheck_done=false`, resume on the same Pi model, not with worker or review. Run the full pass when `selfcheck_incomplete=0`; otherwise resume the persisted successful self-check context locator with an unchecked-item retry. Never replace a missing or invalid persisted context with a fresh session. + - Persist worker success, full-review self-check, checklist-review self-check, and official review as separate stages. If restart state has an enabled unfinished self-check stage, use the completing worker target rather than rerunning worker selection or advancing to official review. For Pi checklist-review retries, resume the persisted successful context locator; never replace a missing or invalid persisted context with a fresh session. - Key persistent state to the first-line `task/plan/tag` generation and, for `m-*`, its `milestone-task` scope. Checklist/body edits to the same PLAN do not reset the stage; a new plan number or changed Milestone Task scope does. - - Send an already completed review stub with no dispatcher execution record to review. Never send dispatcher-recorded Pi worker success to review before self-check completes. + - Send an already completed review stub with no dispatcher execution record to review. Never send dispatcher-recorded worker success to review while a configured self-check stage remains unfinished. - Start official review and worker/self-check together when they belong to different dependency-ready tasks with disjoint workspace claims. Wait for a claim owner to reach verified completion before admitting a colliding task. - Let the dispatcher record every worker/self-check/review attempt start and finish in the task-group `WORK_LOG.md`. - 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. @@ -257,7 +265,7 @@ When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a termin - [ ] Scan the complete candidate set only on initial entry and immediately after verified `complete.log`; atomically claim and start every non-running, dependency-ready, non-colliding candidate in the same pass. - [ ] Confirm the actual CLI/model for each route matches its catalog lane array. -- [ ] Run exactly one full fresh-session self-check only for Pi work, followed by at most 10 unchecked-item retries in that same Pi native session context when its checklist remains incomplete. +- [ ] Reload the catalog before admission, then run only the completing target's enabled full-review/checklist-review stages; allow one checklist-only pass plus at most 10 retries, preserving Pi native context between retries. - [ ] Resolve every official review from its explicit `lanes.review` grade entry and dispatch dependency-ready reviews with disjoint workspace claims in parallel, subject to the global `--max-parallel` cap (no separate review-only limit). - [ ] Locate the native session and output log for every attempt locator. - [ ] Record every worker/self-check/review attempt `START`/`FINISH` in one task-group `WORK_LOG.md`. @@ -294,7 +302,7 @@ Use the same separator format for `작업대기`, `작업수행중`, `자가검 - Never infer dependency from numeric order when no predecessor index is present. - Never scan the complete archive or read archive files outside dependency candidates. - Never ask a worker to perform official review, archive work, or create `complete.log`. -- Never treat Pi self-check as official review. +- Never treat either self-check stage as official review. - Never depend on a model-authored handoff summary for context recovery. - Never treat a generic failure as token/quota failure and escalate it to a higher model. - Never resolve `USER_REVIEW.md` automatically or guess a user decision. diff --git a/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py b/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py index e1d7f5ec..35b12132 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py @@ -207,6 +207,8 @@ PI_MODEL_RESPONSE_STALL_SECONDS = 3 * 60 PI_SESSION_SCHEMA_VERSION = 3 RECOVERY_FAILURE_LIMIT = 10 SELF_CHECK_UNCHECKED_RETRY_LIMIT = 10 +SELF_CHECK_FULL_REVIEW_FAILURE_KEY = "selfcheck-full-review" +SELF_CHECK_CHECKLIST_REVIEW_FAILURE_KEY = "selfcheck-checklist-review" REVIEW_NO_PROGRESS_LIMIT = 10 PROVIDER_TRANSPORT_FAILURES = frozenset( {"provider-connection", "provider-stream-disconnect"} @@ -276,6 +278,18 @@ class ExecutionDecisionError(RuntimeError): """A selector decision is invalid for this task and must fail closed.""" +@dataclass(frozen=True) +class SelfcheckStages: + full_review: bool + checklist_review: bool + catalog_revision: str | None = None + target_id: str | None = None + + @property + def required(self) -> bool: + return self.full_review or self.checklist_review + + def now_iso() -> str: return datetime.now(timezone.utc).isoformat() @@ -827,6 +841,9 @@ class StateStore: "worker_cli": None, "worker_model": None, "selfcheck_done": False, + "selfcheck_full_review_done": False, + "selfcheck_checklist_review_done": False, + "selfcheck_config": None, "blocked": None, "active_stage": None, "active_locator": None, @@ -855,6 +872,9 @@ class StateStore: "worker_cli": None, "worker_model": None, "selfcheck_done": False, + "selfcheck_full_review_done": False, + "selfcheck_checklist_review_done": False, + "selfcheck_config": None, "blocked": None, "active_stage": None, "active_locator": None, @@ -2262,22 +2282,103 @@ def plan_number(task: Task) -> int: return 0 -def completing_decision_requires_selfcheck(state: dict[str, Any]) -> bool: +def reload_execution_target_catalog(): + selector = _selector_module() + return selector.policy.reload_catalog() + + +def completing_decision_selfcheck_stages( + state: dict[str, Any], +) -> SelfcheckStages: completing = state.get("completing_decision") if not isinstance(completing, dict): - return False + return SelfcheckStages(False, False) selected = completing.get("selected") if not isinstance(selected, dict): - return False + return SelfcheckStages(False, False) if ( selected.get("adapter") == "pi" and selected.get("target") == "iop/glm-5.2" + ): + return SelfcheckStages(False, False) + try: + policy = _selector_module().policy + target_id = selected.get("target_id") + target = ( + policy.CATALOG.targets.get(target_id) + if isinstance(target_id, str) and target_id + else None + ) + if target is not None and ( + target.adapter != selected.get("adapter") + or target.target != selected.get("target") + or target.thinking_level != selected.get("thinking_level") + or target.reasoning_effort != selected.get("reasoning_effort") + ): + target = None + if target is None: + target = policy.canonical_target( + selected.get("adapter"), + selected.get("target"), + selected.get("thinking_level"), + selected.get("reasoning_effort"), + ) + if target is not None: + return SelfcheckStages( + full_review=target.selfcheck_full_review, + checklist_review=target.selfcheck_checklist_review, + catalog_revision=policy.CATALOG.revision, + target_id=target.catalog_id, + ) + except (AttributeError, TypeError, ValueError): + pass + # Persisted decisions from before the two-stage catalog keep their old + # behavior when their target can no longer be resolved in the live catalog. + legacy_required = selected.get("selfcheck_required") + if not isinstance(legacy_required, bool): + legacy_required = selected.get("execution_class") == "local_model" + return SelfcheckStages(legacy_required, legacy_required) + + +def completing_decision_requires_selfcheck(state: dict[str, Any]) -> bool: + return completing_decision_selfcheck_stages(state).required + + +def selfcheck_step_done(state: dict[str, Any], field: str) -> bool: + value = state.get(field) + if isinstance(value, bool): + return value + completing = state.get("completing_decision") + selected = completing.get("selected") if isinstance(completing, dict) else None + # Legacy cloud tasks used selfcheck_done=true to mean "skipped", while + # legacy local tasks used it to mean the self-check actually ran. In the + # old single-loop state, selfcheck_incomplete > 0 means the full pass had + # succeeded and only its checklist completion loop remained. + if ( + not isinstance(selected, dict) + or selected.get("execution_class") != "local_model" ): return False - selfcheck_required = selected.get("selfcheck_required") - if isinstance(selfcheck_required, bool): - return selfcheck_required - return selected.get("execution_class") == "local_model" + if state.get("selfcheck_done"): + return True + if field != "selfcheck_full_review_done": + return False + try: + return int(state.get("selfcheck_incomplete", 0)) > 0 + except (TypeError, ValueError): + return False + + +def selfcheck_pipeline_done( + state: dict[str, Any], stages: SelfcheckStages +) -> bool: + return ( + not stages.full_review + or selfcheck_step_done(state, "selfcheck_full_review_done") + ) and ( + not stages.checklist_review + or selfcheck_step_done(state, "selfcheck_checklist_review_done") + ) def _validated_completing_decision( @@ -2437,7 +2538,8 @@ def task_stage(task: Task, state: dict[str, Any]) -> str: if state.get("worker_done"): if not _completing_decision_is_valid(task, state): return "blocked" - if completing_decision_requires_selfcheck(state) and not state.get("selfcheck_done"): + stages = completing_decision_selfcheck_stages(state) + if stages.required and not selfcheck_pipeline_done(state, stages): return "selfcheck" return "review" return "worker" @@ -4318,7 +4420,13 @@ def selfcheck_prompt(task: Task, *, unchecked_items: bool = False) -> str: if task.review is None: raise RuntimeError("selfcheck CODE_REVIEW 파일이 없다") if unchecked_items: - return "The code review file is incomplete! Complete it now!" + body = ( + f"Read {task.review.resolve()}. Review only its Implementation " + "Checklist section. Mark every completed item, finish any missing " + "implementation or evidence required by those items, and leave all " + "official-review-only sections untouched. Keep files in English." + ) + return f"{SELF_CHECK_PROMPT_PREFIX} {body}" body = ( f"Read {task.plan.resolve()}; review all work once, fix omissions, " f"and update {task.review.resolve()}. Keep files in English." @@ -4491,7 +4599,9 @@ def continuation_prompt( context: dict[str, Any] | None = None, unchecked_items: bool = False, ) -> str: - if local_pi and role == "selfcheck": + if role == "selfcheck": + if not local_pi: + return selfcheck_prompt(task, unchecked_items=unchecked_items) if resume_same_pi_session: if unchecked_items: return selfcheck_prompt(task, unchecked_items=True) @@ -4535,8 +4645,10 @@ async def run_escalating( initial_resume_locator: Path | None = None, *, unchecked_items: bool = False, + recovery_state_key: str | None = None, ) -> tuple[bool, Path | None]: spec = initial + recovery_key = recovery_state_key or role previous_locator = initial_resume_locator codex_recovery_count = 0 codex_session_stall_retries = 0 @@ -4551,14 +4663,23 @@ async def run_escalating( state = store.task_state(task) persisted = state.get("recovery_failures", {}) if isinstance(persisted, dict): - recovery_failures = int(persisted.get(role, 0)) + persisted_count = persisted.get(recovery_key) + if persisted_count is None and recovery_key != role: + # Migrate the old single selfcheck recovery counter into the + # first independently scheduled selfcheck stage that resumes. + persisted_count = persisted.get(role, 0) + recovery_failures = int(persisted_count or 0) decisions = state.get("execution_decisions", {}) decision = decisions.get(role) if isinstance(decisions, dict) and role in {"worker", "review"} else None if isinstance(decision, dict): stage_budget = StageFailureBudget.from_decision(store, task, decision) recovery_failures = stage_budget.count() legacy_recovery: LegacyPromotionRecovery | None = None - if initial_resume_locator is not None and isinstance(store, StateStore): + if ( + initial_resume_locator is not None + and isinstance(store, StateStore) + and role != "selfcheck" + ): state = store.task_state(task) legacy_recovery = legacy_promotion_recovery( store.runs, @@ -4674,7 +4795,9 @@ async def run_escalating( if isinstance(store, StateStore): state = store.task_state(task) persisted = dict(state.get("recovery_failures", {})) - persisted.pop(role, None) + persisted.pop(recovery_key, None) + if recovery_key != role: + persisted.pop(role, None) store.update_task(task, recovery_failures=persisted) if stage_budget is not None: stage_budget.reset_on_success() @@ -4696,7 +4819,9 @@ async def run_escalating( if isinstance(store, StateStore): state = store.task_state(task) persisted = dict(state.get("recovery_failures", {})) - persisted[role] = recovery_failures + if recovery_key != role: + persisted.pop(role, None) + persisted[recovery_key] = recovery_failures store.update_task(task, recovery_failures=persisted) if recovery_failures >= RECOVERY_FAILURE_LIMIT: reason = ( @@ -4888,14 +5013,18 @@ async def run_escalating( min(30, 2 ** min(terminal_recovery_retries, 5)) ) continue - if role == "review": + if role in {"review", "selfcheck"}: terminal_recovery_retries += 1 banner( - "리뷰재시도", + "리뷰재시도" if role == "review" else "자가검증재시도", task.name, [ f"model={spec.display}", - "reason=review-catalog-target-retry", + ( + "reason=review-catalog-target-retry" + if role == "review" + else "reason=selfcheck-completing-target-retry" + ), *failure_report_lines(failure, locator), f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}", ], @@ -5720,6 +5849,9 @@ def _mark_worker_done( ) _require_same_runtime_identity(expected_spec, worker_cli, worker_model) execution_class = validated_decision["selected"]["execution_class"] + stages = completing_decision_selfcheck_stages( + {"completing_decision": validated_decision} + ) store.update_task( task, worker_done=True, @@ -5727,7 +5859,16 @@ def _mark_worker_done( worker_model=worker_model, completing_decision=validated_decision, execution_class=execution_class, - selfcheck_done=(execution_class == "cloud_model"), + selfcheck_done=not stages.required, + selfcheck_full_review_done=False, + selfcheck_checklist_review_done=False, + selfcheck_config={ + "full_review": stages.full_review, + "checklist_review": stages.checklist_review, + "catalog_revision": stages.catalog_revision, + "target_id": stages.target_id, + "evaluated_at": now_iso(), + }, blocked=None, ) @@ -5738,6 +5879,13 @@ async def run_selfcheck( task: Task, resume_locator: Path | None = None, ) -> None: + try: + reload_execution_target_catalog() + except (OSError, ValueError) as exc: + reason = f"selfcheck runtime catalog reload failed: {exc}" + store.update_task(task, blocked=reason) + banner("작업차단", task.name, [f"reason={reason}"]) + return completing = store.task_state(task).get("completing_decision") if not isinstance(completing, dict): store.update_task( @@ -5754,32 +5902,107 @@ async def run_selfcheck( store.update_task(task, blocked=str(exc)) banner("작업차단", task.name, [f"reason={exc}"]) return - if not spec.local_pi: - raise RuntimeError("Pi가 아닌 route에 selfcheck stage가 배정됐다") + stages = completing_decision_selfcheck_stages( + {"completing_decision": completing} + ) + state = store.task_state(task) + store.update_task( + task, + selfcheck_config={ + "full_review": stages.full_review, + "checklist_review": stages.checklist_review, + "catalog_revision": stages.catalog_revision, + "target_id": stages.target_id, + "evaluated_at": now_iso(), + }, + ) + if not stages.required: + store.update_task(task, selfcheck_done=True, blocked=None) + return work_log = milestone_work_log_path(task) + if stages.full_review and not selfcheck_step_done( + state, "selfcheck_full_review_done" + ): + banner( + "자가검증시작", + task.name, + [ + "mode=full-review", + f"model={spec.display}", + f"plan={task.plan.resolve()}", + f"work_log={work_log.resolve()}", + *task_observation_lines(task), + ], + ) + success, locator = await run_escalating( + workspace, + store, + task, + "selfcheck", + spec, + initial_resume_locator=resume_locator, + unchecked_items=False, + recovery_state_key=SELF_CHECK_FULL_REVIEW_FAILURE_KEY, + ) + if not success: + current = store.task_state(task).get("blocked") + store.update_task( + task, blocked=current or f"selfcheck failure locator={locator}" + ) + return + updated = dict(store.task_state(task)) + updated["selfcheck_full_review_done"] = True + done = selfcheck_pipeline_done(updated, stages) + store.update_task( + task, + selfcheck_full_review_done=True, + selfcheck_done=done, + blocked=None, + ) + return + + if not stages.checklist_review or selfcheck_step_done( + state, "selfcheck_checklist_review_done" + ): + store.update_task(task, selfcheck_done=True, blocked=None) + return + + errors = implementation_review_errors(task) + if not errors: + store.update_task( + task, + selfcheck_checklist_review_done=True, + selfcheck_done=True, + selfcheck_incomplete=0, + selfcheck_context_locator=None, + blocked=None, + ) + return + banner( "자가검증시작", task.name, [ + "mode=checklist-review", f"model={spec.display}", - f"plan={task.plan.resolve()}", + f"review={task.review.resolve()}", f"work_log={work_log.resolve()}", *task_observation_lines(task), ], ) - # 0 means the full pass is pending. After it fails the checklist gate, - # each additional count represents one completed unchecked-item retry. + # Count failed checklist-only passes. One initial pass plus ten retries is + # allowed. Full review has its own persisted completion flag and does not + # consume this budget. incomplete_results = 0 if isinstance(store, StateStore): incomplete_results = int( store.task_state(task).get("selfcheck_incomplete", 0) ) - incomplete_retries = max(0, incomplete_results - 1) - if incomplete_retries >= SELF_CHECK_UNCHECKED_RETRY_LIMIT: + if incomplete_results >= SELF_CHECK_UNCHECKED_RETRY_LIMIT + 1: locator = resume_locator reason = ( - "selfcheck unchecked-item retry limit already exhausted: " - f"{incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}" + "selfcheck checklist-review retry limit already exhausted: " + f"{SELF_CHECK_UNCHECKED_RETRY_LIMIT}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}" ) store.update_task(task, blocked=f"{reason} locator={locator}") banner( @@ -5787,13 +6010,13 @@ async def run_selfcheck( task.name, [ "reason=selfcheck-incomplete-limit", - "mode=unchecked-items", - f"retry={incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}", + "mode=checklist-review", + f"retry={SELF_CHECK_UNCHECKED_RETRY_LIMIT}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}", f"locator={locator}", ], ) return - if incomplete_results > 0 and resume_locator is None: + if incomplete_results > 0 and resume_locator is None and spec.local_pi: resume_locator, context_error = selfcheck_context_resume_locator( store.task_state(task), task, @@ -5811,7 +6034,6 @@ async def run_selfcheck( ) return while True: - unchecked_items = incomplete_results > 0 success, locator = await run_escalating( workspace, store, @@ -5819,7 +6041,8 @@ async def run_selfcheck( "selfcheck", spec, initial_resume_locator=resume_locator, - unchecked_items=unchecked_items, + unchecked_items=True, + recovery_state_key=SELF_CHECK_CHECKLIST_REVIEW_FAILURE_KEY, ) if not success: current = store.task_state(task).get("blocked") @@ -5830,7 +6053,7 @@ async def run_selfcheck( errors = implementation_review_errors(task) if not errors: break - if locator is None: + if locator is None and spec.local_pi: reason = "selfcheck 성공 locator가 없어 context를 이어갈 수 없다" store.update_task(task, blocked=reason) banner( @@ -5840,25 +6063,30 @@ async def run_selfcheck( ) return incomplete_results += 1 - incomplete_retries = max(0, incomplete_results - 1) - if incomplete_retries >= SELF_CHECK_UNCHECKED_RETRY_LIMIT: + retries = max(0, incomplete_results - 1) + if incomplete_results >= SELF_CHECK_UNCHECKED_RETRY_LIMIT + 1: reason = ( - "selfcheck checklist remains incomplete after unchecked-item retry: " - f"{incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}" + "selfcheck checklist remains incomplete after checklist-review retry: " + f"{SELF_CHECK_UNCHECKED_RETRY_LIMIT}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}" ) store.update_task( task, blocked=f"{reason} locator={locator}", selfcheck_incomplete=incomplete_results, - selfcheck_context_locator=str(locator), + selfcheck_context_locator=( + str(locator) + if locator is not None and spec.local_pi + else None + ), ) banner( "작업차단", task.name, [ "reason=selfcheck-incomplete-limit", + "mode=checklist-review", f"detail={'; '.join(errors)}", - f"retry={incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}", + f"retry={SELF_CHECK_UNCHECKED_RETRY_LIMIT}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}", f"locator={locator}", ], ) @@ -5866,21 +6094,24 @@ async def run_selfcheck( store.update_task( task, selfcheck_incomplete=incomplete_results, - selfcheck_context_locator=str(locator), + selfcheck_context_locator=( + str(locator) if locator is not None and spec.local_pi else None + ), ) - resume_locator = locator + resume_locator = locator if spec.local_pi else None banner( "자가검증재시도", task.name, [ f"reason={'; '.join(errors)}", - "mode=unchecked-items", - f"retry={incomplete_retries + 1}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}", + "mode=checklist-review", + f"retry={retries + 1}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}", f"locator={locator}", ], ) store.update_task( task, + selfcheck_checklist_review_done=True, selfcheck_done=True, selfcheck_incomplete=0, selfcheck_context_locator=None, @@ -6244,6 +6475,19 @@ async def dispatch_with_store( ) while True: + try: + reload_execution_target_catalog() + except (OSError, ValueError) as exc: + banner( + "디스패치차단", + args.task_group or "agent-task", + [f"execution target catalog reload failed: {exc}"], + ) + if running: + raise DispatcherTerminalStateError( + f"execution target catalog reload failed: {exc}" + ) from exc + return 2 if task_cache is None: tasks = scan_tasks(workspace, args.task_group) task_cache = {task.name: task for task in tasks} diff --git a/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_catalog.json b/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_catalog.json index 023adb31..e83b8557 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_catalog.json +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_catalog.json @@ -5,40 +5,58 @@ "adapter": "pi", "target": "iop/ornith:35b", "execution_class": "local_model", - "selfcheck_required": true, + "selfcheck": { + "full_review": true, + "checklist_review": true + }, "thinking_level": "high" }, "pi-laguna-high": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", - "selfcheck_required": true, + "selfcheck": { + "full_review": true, + "checklist_review": true + }, "thinking_level": "high" }, "agy-gemini-low": { "adapter": "agy", "target": "Gemini 3.6 Flash (Low)", "execution_class": "cloud_model", - "selfcheck_required": false + "selfcheck": { + "full_review": false, + "checklist_review": false + } }, "agy-gemini-medium": { "adapter": "agy", "target": "Gemini 3.6 Flash (Medium)", "execution_class": "cloud_model", - "selfcheck_required": false + "selfcheck": { + "full_review": false, + "checklist_review": false + } }, "agy-gemini-high": { "adapter": "agy", "target": "Gemini 3.6 Flash (High)", "execution_class": "cloud_model", - "selfcheck_required": false + "selfcheck": { + "full_review": false, + "checklist_review": false + } }, "opencode-glm-medium": { "adapter": "opencode", "target": "glm-5.2", "command_model": "iop-glm/glm-5.2", "execution_class": "cloud_model", - "selfcheck_required": false, + "selfcheck": { + "full_review": false, + "checklist_review": true + }, "reasoning_effort": "medium" }, "opencode-glm-high": { @@ -46,7 +64,10 @@ "target": "glm-5.2", "command_model": "iop-glm/glm-5.2", "execution_class": "cloud_model", - "selfcheck_required": false, + "selfcheck": { + "full_review": false, + "checklist_review": true + }, "reasoning_effort": "high" }, "opencode-glm-max": { @@ -54,7 +75,10 @@ "target": "glm-5.2", "command_model": "iop-glm/glm-5.2", "execution_class": "cloud_model", - "selfcheck_required": false, + "selfcheck": { + "full_review": false, + "checklist_review": true + }, "reasoning_effort": "max" }, "legacy-claude-glm": { @@ -62,42 +86,60 @@ "target": "glm-5.2", "command_model": "sonnet", "execution_class": "cloud_model", - "selfcheck_required": false, + "selfcheck": { + "full_review": false, + "checklist_review": true + }, "reasoning_effort": "xhigh" }, "claude-opus-xhigh": { "adapter": "claude", "target": "claude-opus-4-8", "execution_class": "cloud_model", - "selfcheck_required": false, + "selfcheck": { + "full_review": false, + "checklist_review": false + }, "reasoning_effort": "xhigh" }, "claude-haiku-xhigh": { "adapter": "claude", "target": "claude-haiku-4-5", "execution_class": "cloud_model", - "selfcheck_required": false, + "selfcheck": { + "full_review": false, + "checklist_review": false + }, "reasoning_effort": "xhigh" }, "codex-spark-xhigh": { "adapter": "codex", "target": "gpt-5.3-codex-spark", "execution_class": "cloud_model", - "selfcheck_required": false, + "selfcheck": { + "full_review": false, + "checklist_review": false + }, "reasoning_effort": "xhigh" }, "codex-sol-xhigh": { "adapter": "codex", "target": "gpt-5.6-sol", "execution_class": "cloud_model", - "selfcheck_required": false, + "selfcheck": { + "full_review": false, + "checklist_review": false + }, "reasoning_effort": "xhigh" }, "codex-terra-high": { "adapter": "codex", "target": "gpt-5.6-terra", "execution_class": "cloud_model", - "selfcheck_required": false, + "selfcheck": { + "full_review": false, + "checklist_review": false + }, "reasoning_effort": "high" } }, diff --git a/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_contract.py b/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_contract.py index 301195b1..52360807 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_contract.py +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_contract.py @@ -24,8 +24,12 @@ def validate_target_contract(target, path: str, error_type) -> None: if target.selfcheck_required != local: raise error_type( f"{path}: selfcheck_required must be {str(local).lower()} " - f"for {target.adapter}" + f"for {target.adapter} as the persisted-decision compatibility field" ) + if not isinstance(target.selfcheck_full_review, bool): + raise error_type(f"{path}: selfcheck.full_review must be a boolean") + if not isinstance(target.selfcheck_checklist_review, bool): + raise error_type(f"{path}: selfcheck.checklist_review must be a boolean") if target.adapter == "pi": if not target.target.startswith("iop/"): raise error_type(f"{path}: pi target must start with iop/") diff --git a/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_policy.py b/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_policy.py index 5c81a2f1..62817dca 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_policy.py +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_policy.py @@ -35,6 +35,8 @@ class RouteTarget: target: str execution_class: str selfcheck_required: bool + selfcheck_full_review: bool = False + selfcheck_checklist_review: bool = False thinking_level: str | None = None reasoning_effort: str | None = None command_model: str | None = None @@ -123,6 +125,7 @@ def _target_from_config(target_id: str, value: object) -> RouteTarget: "target", "execution_class", "selfcheck_required", + "selfcheck", "thinking_level", "reasoning_effort", "command_model", @@ -137,9 +140,40 @@ def _target_from_config(target_id: str, value: object) -> RouteTarget: raise CatalogError( f"{path}.execution_class must be one of {sorted(VALID_EXECUTION_CLASSES)}" ) - selfcheck = item.get("selfcheck_required") - if not isinstance(selfcheck, bool): - raise CatalogError(f"{path}.selfcheck_required must be a boolean") + legacy_selfcheck = item.get("selfcheck_required") + raw_selfcheck = item.get("selfcheck") + if raw_selfcheck is not None and legacy_selfcheck is not None: + raise CatalogError( + f"{path} cannot combine selfcheck and selfcheck_required" + ) + if raw_selfcheck is not None: + selfcheck_config = _object(raw_selfcheck, f"{path}.selfcheck") + expected_fields = {"full_review", "checklist_review"} + if set(selfcheck_config) != expected_fields: + raise CatalogError( + f"{path}.selfcheck must contain exactly {sorted(expected_fields)}" + ) + full_review = selfcheck_config.get("full_review") + checklist_review = selfcheck_config.get("checklist_review") + if not isinstance(full_review, bool): + raise CatalogError( + f"{path}.selfcheck.full_review must be a boolean" + ) + if not isinstance(checklist_review, bool): + raise CatalogError( + f"{path}.selfcheck.checklist_review must be a boolean" + ) + # Keep the old decision field stable for persisted-state compatibility. + # Runtime self-check admission uses the two explicit stage flags below. + selfcheck_required = item.get("adapter") == "pi" + else: + if not isinstance(legacy_selfcheck, bool): + raise CatalogError( + f"{path}.selfcheck must be an object with boolean stages" + ) + full_review = legacy_selfcheck + checklist_review = legacy_selfcheck + selfcheck_required = legacy_selfcheck command_model = item.get("command_model") if command_model is not None: command_model = _non_empty_string(command_model, f"{path}.command_model") @@ -147,7 +181,9 @@ def _target_from_config(target_id: str, value: object) -> RouteTarget: adapter=_non_empty_string(item.get("adapter"), f"{path}.adapter"), target=_non_empty_string(item.get("target"), f"{path}.target"), execution_class=execution_class, - selfcheck_required=selfcheck, + selfcheck_required=selfcheck_required, + selfcheck_full_review=full_review, + selfcheck_checklist_review=checklist_review, thinking_level=_optional_enum( item.get("thinking_level"), VALID_PI_THINKING_LEVELS, @@ -368,6 +404,69 @@ CATALOG_TARGETS_BY_ID = CATALOG.targets CANONICAL_TARGETS = tuple(CATALOG.targets.values()) +def reload_catalog(path: Path = CATALOG_PATH) -> ExecutionTargetCatalog: + """Atomically publish the latest operator-owned catalog. + + The dispatcher calls this before each scheduler admission and immediately + before a self-check stage starts. Existing model invocations keep their + pinned decision; the next stage observes the newest self-check switches. + """ + catalog = load_catalog(path) + compatibility_ids = ( + "pi-ornith-high", + "pi-laguna-high", + "agy-gemini-low", + "agy-gemini-medium", + "agy-gemini-high", + "opencode-glm-medium", + "opencode-glm-high", + "opencode-glm-max", + "legacy-claude-glm", + "claude-opus-xhigh", + "claude-haiku-xhigh", + "codex-spark-xhigh", + "codex-sol-xhigh", + "codex-terra-high", + ) + missing = [ + target_id + for target_id in compatibility_ids + if target_id not in catalog.targets + ] + if missing: + raise CatalogError( + "catalog is missing compatibility targets required by the " + f"dispatcher: {missing}" + ) + compatibility_targets = { + target_id: catalog.targets[target_id] for target_id in compatibility_ids + } + global CATALOG, CATALOG_REVISION, CATALOG_TARGETS_BY_ID, CANONICAL_TARGETS + global PI_ORNITH, PI_LAGUNA, AGY_GEMINI_LOW, AGY_GEMINI_MEDIUM + global AGY_GEMINI_HIGH, OPENCODE_GLM_MEDIUM, OPENCODE_GLM_HIGH + global OPENCODE_GLM_MAX, CLAUDE_GLM, CLAUDE_OPUS, CLAUDE_HAIKU_XHIGH + global CODEX_SPARK_XHIGH, CODEX_SOL_XHIGH, CODEX_TERRA_HIGH + CATALOG = catalog + CATALOG_REVISION = catalog.revision + CATALOG_TARGETS_BY_ID = catalog.targets + CANONICAL_TARGETS = tuple(catalog.targets.values()) + PI_ORNITH = compatibility_targets["pi-ornith-high"] + PI_LAGUNA = compatibility_targets["pi-laguna-high"] + AGY_GEMINI_LOW = compatibility_targets["agy-gemini-low"] + AGY_GEMINI_MEDIUM = compatibility_targets["agy-gemini-medium"] + AGY_GEMINI_HIGH = compatibility_targets["agy-gemini-high"] + OPENCODE_GLM_MEDIUM = compatibility_targets["opencode-glm-medium"] + OPENCODE_GLM_HIGH = compatibility_targets["opencode-glm-high"] + OPENCODE_GLM_MAX = compatibility_targets["opencode-glm-max"] + CLAUDE_GLM = compatibility_targets["legacy-claude-glm"] + CLAUDE_OPUS = compatibility_targets["claude-opus-xhigh"] + CLAUDE_HAIKU_XHIGH = compatibility_targets["claude-haiku-xhigh"] + CODEX_SPARK_XHIGH = compatibility_targets["codex-spark-xhigh"] + CODEX_SOL_XHIGH = compatibility_targets["codex-sol-xhigh"] + CODEX_TERRA_HIGH = compatibility_targets["codex-terra-high"] + return catalog + + def catalog_target(target_id: str) -> RouteTarget: try: return CATALOG.targets[target_id] diff --git a/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py b/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py index e4b3b7b1..da6eef29 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py @@ -558,7 +558,7 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): """Verify selfcheck is determined by the completing decision's execution_class. - Worker success persists the actual completing decision with execution_class. - - selfcheck schedules exactly once when execution_class=local_model. + - target-configured selfcheck stages schedule independently. - local selfcheck reuses the completing target without re-evaluating selector. - Gemini→Laguna, Laguna→Gemini, cloud completions follow the policy. - Restart does not duplicate selfcheck execution. @@ -745,8 +745,8 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): finally: store.close() - async def test_local_completing_decision_triggers_selfcheck(self): - """execution_class=local_model schedules exactly one selfcheck.""" + async def test_local_completing_decision_runs_two_selfcheck_stages(self): + """Pi runs full-review and checklist-review as separate stages.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() @@ -783,7 +783,7 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): dispatch.completing_decision_requires_selfcheck(state) ) - # Run selfcheck once + # Full review and checklist review are separate scheduler entries. with ( mock.patch.object( dispatch, "run_escalating", @@ -797,14 +797,20 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): await dispatch.run_selfcheck( workspace, store, task ) + self.assertTrue( + store.task_state(task)["selfcheck_full_review_done"] + ) + self.assertFalse(store.task_state(task)["selfcheck_done"]) + await dispatch.run_selfcheck(workspace, store, task) state2 = store.task_state(task) self.assertTrue(state2["selfcheck_done"]) + self.assertTrue(state2["selfcheck_checklist_review_done"]) self.assertEqual(dispatch.task_stage(task, state2), "review") finally: store.close() - async def test_claude_glm_completion_skips_selfcheck(self): + async def test_claude_glm_completion_runs_checklist_only(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() @@ -832,14 +838,194 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): blocked=None, ) - self.assertFalse( + self.assertTrue( dispatch.completing_decision_requires_selfcheck( store.task_state(task) ) ) self.assertEqual( dispatch.task_stage(task, store.task_state(task)), - "review", + "selfcheck", + ) + stages = dispatch.completing_decision_selfcheck_stages( + store.task_state(task) + ) + self.assertFalse(stages.full_review) + self.assertTrue(stages.checklist_review) + finally: + store.close() + + async def test_opencode_glm_runs_checklist_with_completing_worker_target(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + task = self.make_task(workspace, lane="cloud", grade=3) + store = dispatch.StateStore(workspace) + decision = { + "work_unit_id": self._WORK_UNIT_ID, + "stage": "worker", + "selected": { + "target_id": "opencode-glm-high", + "adapter": "opencode", + "target": "glm-5.2", + "command_model": "iop-glm/glm-5.2", + "reasoning_effort": "high", + "execution_class": "cloud_model", + "selfcheck_required": False, + }, + } + store.update_task( + task, + worker_done=True, + worker_cli="opencode", + worker_model="glm-5.2", + completing_decision=decision, + execution_class="cloud_model", + selfcheck_done=False, + blocked=None, + ) + locator = self.make_locator(workspace, "opencode", "glm-5.2") + try: + with ( + mock.patch.object( + dispatch, + "run_escalating", + new=mock.AsyncMock(return_value=(True, locator)), + ) as run_escalating, + mock.patch.object( + dispatch, + "implementation_review_errors", + side_effect=[["구현 체크리스트 미완료"], []], + ), + ): + await dispatch.run_selfcheck(workspace, store, task) + + self.assertEqual(run_escalating.await_count, 1) + self.assertEqual(run_escalating.await_args.args[4].cli, "opencode") + self.assertTrue(run_escalating.await_args.kwargs["unchecked_items"]) + state = store.task_state(task) + self.assertFalse(state["selfcheck_full_review_done"]) + self.assertTrue(state["selfcheck_checklist_review_done"]) + self.assertTrue(state["selfcheck_done"]) + finally: + store.close() + + async def test_cloud_selfcheck_retries_without_promoting_completing_target(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + task = self.make_task(workspace, lane="cloud", grade=3) + store = dispatch.StateStore(workspace) + spec = dispatch.AgentSpec( + "opencode", + "glm-5.2", + "opencode/glm-5.2 high", + reasoning_effort="high", + command_model="iop-glm/glm-5.2", + ) + first_locator = self.make_locator( + workspace, + "opencode-first", + "glm-5.2", + ) + second_locator = self.make_locator( + workspace, + "opencode-second", + "glm-5.2", + ) + try: + with ( + mock.patch.object( + dispatch, + "invoke", + new=mock.AsyncMock( + side_effect=[ + (1, "provider-quota", first_locator), + (0, None, second_locator), + ] + ), + ) as invoke, + mock.patch.object( + dispatch.asyncio, + "sleep", + new=mock.AsyncMock(), + ), + mock.patch.object( + dispatch, + "promoted_spec", + side_effect=AssertionError( + "selfcheck must not promote its completing target" + ), + ), + ): + success, locator = await dispatch.run_escalating( + workspace, + store, + task, + "selfcheck", + spec, + unchecked_items=True, + recovery_state_key=( + dispatch.SELF_CHECK_CHECKLIST_REVIEW_FAILURE_KEY + ), + ) + + self.assertTrue(success) + self.assertEqual(locator, second_locator) + self.assertEqual(invoke.await_count, 2) + self.assertEqual( + [call.args[4].cli for call in invoke.await_args_list], + ["opencode", "opencode"], + ) + finally: + store.close() + + async def test_selfcheck_steps_have_independent_recovery_budgets(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + task = self.make_task(workspace) + store = dispatch.StateStore(workspace) + store.update_task( + task, + recovery_failures={ + dispatch.SELF_CHECK_FULL_REVIEW_FAILURE_KEY: + dispatch.RECOVERY_FAILURE_LIMIT, + }, + ) + locator = self.make_locator(workspace, "pi", "laguna-s:2.1") + spec = dispatch.AgentSpec( + "pi", + "laguna-s:2.1", + "pi/iop/laguna-s:2.1", + local_pi=True, + ) + try: + with mock.patch.object( + dispatch, + "invoke", + new=mock.AsyncMock(return_value=(0, None, locator)), + ) as invoke: + success, _ = await dispatch.run_escalating( + workspace, + store, + task, + "selfcheck", + spec, + unchecked_items=True, + recovery_state_key=( + dispatch.SELF_CHECK_CHECKLIST_REVIEW_FAILURE_KEY + ), + ) + + self.assertTrue(success) + self.assertEqual(invoke.await_count, 1) + self.assertEqual( + store.task_state(task)["recovery_failures"], + { + dispatch.SELF_CHECK_FULL_REVIEW_FAILURE_KEY: + dispatch.RECOVERY_FAILURE_LIMIT, + }, ) finally: store.close() @@ -969,6 +1155,7 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): await dispatch.run_selfcheck( workspace, store, task ) + await dispatch.run_selfcheck(workspace, store, task) # persisted_execution_decision must NOT be called during selfcheck self.assertEqual(len(selector_calls), 0) @@ -1003,6 +1190,8 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): completing_decision=local_decision, execution_class="local_model", selfcheck_done=True, + selfcheck_full_review_done=True, + selfcheck_checklist_review_done=True, blocked=None, ) @@ -1016,6 +1205,127 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): finally: store.close() + async def test_legacy_partial_selfcheck_resumes_at_checklist_stage(self): + """Old incomplete state must not repeat its already-successful full pass.""" + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + task = self.make_task(workspace) + store = dispatch.StateStore(workspace) + decision = { + "work_unit_id": self._WORK_UNIT_ID, + "stage": "worker", + "selected": { + "adapter": "pi", + "target": "iop/laguna-s:2.1", + "execution_class": "local_model", + "selfcheck_required": True, + }, + } + locator = self.make_locator(workspace, "pi", "laguna-s:2.1") + store.update_task( + task, + worker_done=True, + worker_cli="pi", + worker_model="laguna-s:2.1", + completing_decision=decision, + execution_class="local_model", + selfcheck_done=False, + selfcheck_incomplete=1, + selfcheck_context_locator=str(locator), + blocked=None, + ) + legacy_state = store.task_state(task) + legacy_state.pop("selfcheck_full_review_done") + legacy_state.pop("selfcheck_checklist_review_done") + store.save() + try: + with ( + mock.patch.object( + dispatch, + "run_escalating", + new=mock.AsyncMock(return_value=(True, locator)), + ) as run_escalating, + mock.patch.object( + dispatch, + "implementation_review_errors", + side_effect=[["구현 체크리스트 미완료"], []], + ), + ): + await dispatch.run_selfcheck( + workspace, + store, + task, + resume_locator=locator, + ) + + self.assertEqual(run_escalating.await_count, 1) + self.assertTrue( + run_escalating.await_args.kwargs["unchecked_items"] + ) + state = store.task_state(task) + self.assertTrue(state["selfcheck_checklist_review_done"]) + self.assertTrue(state["selfcheck_done"]) + finally: + store.close() + + async def test_live_catalog_toggle_changes_the_next_stage(self): + """A catalog-only switch applies to persisted work without source reload.""" + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + task = self.make_task(workspace, lane="cloud", grade=7) + store = dispatch.StateStore(workspace) + decision = { + "work_unit_id": self._WORK_UNIT_ID, + "stage": "worker", + "selected": { + "target_id": "codex-sol-xhigh", + "adapter": "codex", + "target": "gpt-5.6-sol", + "reasoning_effort": "xhigh", + "execution_class": "cloud_model", + "selfcheck_required": False, + }, + } + store.update_task( + task, + worker_done=True, + worker_cli="codex", + worker_model="gpt-5.6-sol", + completing_decision=decision, + execution_class="cloud_model", + selfcheck_done=True, + blocked=None, + ) + policy = dispatch._selector_module().policy + data = json.loads(policy.CATALOG_PATH.read_text(encoding="utf-8")) + data["targets"]["codex-sol-xhigh"]["selfcheck"] = { + "full_review": True, + "checklist_review": False, + } + catalog_path = workspace / "execution-target-catalog.json" + catalog_path.write_text(json.dumps(data), encoding="utf-8") + try: + self.assertEqual( + dispatch.task_stage(task, store.task_state(task)), + "review", + ) + reloaded = policy.reload_catalog(catalog_path) + stages = dispatch.completing_decision_selfcheck_stages( + store.task_state(task) + ) + self.assertTrue(stages.full_review) + self.assertFalse(stages.checklist_review) + self.assertEqual(stages.catalog_revision, reloaded.revision) + self.assertEqual( + dispatch.task_stage(task, store.task_state(task)), + "selfcheck", + ) + finally: + policy.reload_catalog() + store.close() + async def test_identity_mismatch_fails_closed(self): """Missing or malformed completing decision blocks selfcheck.""" with tempfile.TemporaryDirectory() as temporary: @@ -1040,6 +1350,7 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): await dispatch.run_selfcheck( workspace, store, task ) + await dispatch.run_selfcheck(workspace, store, task) self.assertEqual(run_escalating.await_count, 0) state = store.task_state(task) @@ -1215,6 +1526,7 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): await dispatch.run_selfcheck( workspace, store, task ) + await dispatch.run_selfcheck(workspace, store, task) self.assertEqual(len(selector_select_calls), 0) self.assertEqual(len(quota_probe_calls), 0) @@ -3850,10 +4162,15 @@ class ReviewControlTest(unittest.TestCase): ) self.assertEqual( unchecked_retry, - "The code review file is incomplete! Complete it now!", + f"{dispatch.SELF_CHECK_PROMPT_PREFIX} Read " + f"{task.review.resolve()}. Review only its Implementation " + "Checklist section. Mark every completed item, finish any " + "missing implementation or evidence required by those items, " + "and leave all official-review-only sections untouched. Keep " + "files in English.", ) self.assertNotIn(str(task.plan.resolve()), unchecked_retry) - self.assertNotIn(str(task.review.resolve()), unchecked_retry) + self.assertIn(str(task.review.resolve()), unchecked_retry) self.assertNotIn("dispatcher child", selfcheck.lower()) self.assertEqual( dispatch.continuation_prompt( @@ -4712,6 +5029,7 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase): } store.update_task( task, + selfcheck_full_review_done=True, selfcheck_incomplete=( dispatch.SELF_CHECK_UNCHECKED_RETRY_LIMIT + 1 ), @@ -4769,6 +5087,7 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase): } store.update_task( task, + selfcheck_full_review_done=True, selfcheck_incomplete=dispatch.SELF_CHECK_UNCHECKED_RETRY_LIMIT, completing_decision=completing_decision, ) @@ -4837,6 +5156,7 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase): ), ): await dispatch.run_selfcheck(root, store, task) + await dispatch.run_selfcheck(root, store, task) self.assertEqual(run_escalating.await_count, 2) self.assertFalse( @@ -4854,7 +5174,7 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase): run_escalating.await_args_list[1].kwargs[ "initial_resume_locator" ], - locators[0], + None, ) state = store.task_state(task) self.assertTrue(state["selfcheck_done"]) @@ -4901,6 +5221,7 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase): store.update_task( task, completing_decision=completing_decision, + selfcheck_full_review_done=True, selfcheck_incomplete=1, selfcheck_context_locator=str(locator), ) @@ -4913,7 +5234,9 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase): new=mock.AsyncMock(return_value=(True, retry_locator)), ) as run_escalating, mock.patch.object( - dispatch, "implementation_review_errors", return_value=[] + dispatch, + "implementation_review_errors", + side_effect=[["구현 체크리스트 미완료"], []], ), ): await dispatch.run_selfcheck(root, store, task) @@ -4948,6 +5271,7 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase): store.update_task( task, completing_decision=completing_decision, + selfcheck_full_review_done=True, selfcheck_incomplete=1, ) try: @@ -4980,7 +5304,11 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase): "selfcheck_required": True, }, } - store.update_task(task, completing_decision=completing_decision) + store.update_task( + task, + completing_decision=completing_decision, + selfcheck_full_review_done=True, + ) locator = root / "locator.json" try: with ( @@ -5001,13 +5329,10 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase): run_escalating.await_count, 1 + dispatch.SELF_CHECK_UNCHECKED_RETRY_LIMIT, ) - self.assertFalse( - run_escalating.await_args_list[0].kwargs["unchecked_items"] - ) self.assertTrue( all( call.kwargs["unchecked_items"] - for call in run_escalating.await_args_list[1:] + for call in run_escalating.await_args_list ) ) self.assertTrue( @@ -8345,7 +8670,13 @@ class DispatcherConvergenceSimulationTest(unittest.IsolatedAsyncioTestCase): enter("selfcheck", task.name) try: await asyncio.sleep(0.005) - store.update_task(task, selfcheck_done=True, blocked=None) + store.update_task( + task, + selfcheck_done=True, + selfcheck_full_review_done=True, + selfcheck_checklist_review_done=True, + blocked=None, + ) finally: leave("selfcheck", task.name) @@ -10506,8 +10837,14 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase): 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.assertFalse(state["selfcheck_done"]) + self.assertEqual(dispatch.task_stage(task, state), "selfcheck") + self.assertEqual( + state["selfcheck_config"]["checklist_review"], True + ) + self.assertEqual( + state["selfcheck_config"]["full_review"], False + ) self.assertEqual(state["execution_decisions"]["worker"]["selected"]["adapter"], "opencode") self.assertEqual( state["execution_decisions"]["worker"]["selected"]["reasoning_effort"], @@ -12758,16 +13095,22 @@ class ArtifactLanguageContractTest(unittest.TestCase): "pi_selfcheck_continuation", "pi_selfcheck_native_continuation", } - terse_unchecked_prompts = { + checklist_only_prompts = { "selfcheck_unchecked", "pi_selfcheck_unchecked_continuation", } for name, prompt in prompts.items(): with self.subTest(prompt=name): - if name in terse_unchecked_prompts: + if name in checklist_only_prompts: self.assertEqual( prompt, - "The code review file is incomplete! Complete it now!", + f"{dispatch.SELF_CHECK_PROMPT_PREFIX} Read " + f"{task.review.resolve()}. Review only its " + "Implementation Checklist section. Mark every " + "completed item, finish any missing implementation " + "or evidence required by those items, and leave all " + "official-review-only sections untouched. Keep " + "files in English.", ) continue self.assertTrue( diff --git a/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_execution_target_policy.py b/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_execution_target_policy.py index 101b9ba7..07a80f24 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_execution_target_policy.py +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_execution_target_policy.py @@ -55,6 +55,38 @@ class ExecutionTargetPolicyTests(unittest.TestCase): ["codex-sol-xhigh", "agy-gemini-medium"], ) + def test_selfcheck_stages_reload_from_catalog_without_python_change(self): + data = json.loads(policy.CATALOG_PATH.read_text(encoding="utf-8")) + data["targets"]["codex-sol-xhigh"]["selfcheck"] = { + "full_review": True, + "checklist_review": False, + } + with TemporaryDirectory() as tmp: + path = Path(tmp) / "catalog.json" + path.write_text(json.dumps(data), encoding="utf-8") + try: + reloaded = policy.reload_catalog(path) + target = reloaded.targets["codex-sol-xhigh"] + self.assertTrue(target.selfcheck_full_review) + self.assertFalse(target.selfcheck_checklist_review) + self.assertEqual(policy.CATALOG.revision, reloaded.revision) + finally: + policy.reload_catalog() + + def test_failed_reload_preserves_the_published_catalog(self): + data = json.loads(policy.CATALOG_PATH.read_text(encoding="utf-8")) + del data["targets"]["legacy-claude-glm"] + published = policy.CATALOG + with TemporaryDirectory() as tmp: + path = Path(tmp) / "catalog.json" + path.write_text(json.dumps(data), encoding="utf-8") + with self.assertRaisesRegex( + policy.CatalogError, + "missing compatibility targets", + ): + policy.reload_catalog(path) + self.assertIs(policy.CATALOG, published) + def test_catalog_rejects_a_missing_grade_lane(self): data = json.loads(policy.CATALOG_PATH.read_text(encoding="utf-8")) del data["lanes"]["worker"]["cloud-G03"] @@ -109,6 +141,13 @@ class ExecutionTargetPolicyTests(unittest.TestCase): ), "contain a cycle", ), + ( + "incomplete selfcheck stages", + lambda data: data["targets"]["opencode-glm-high"][ + "selfcheck" + ].pop("checklist_review"), + "must contain exactly", + ), ) for name, mutate, message in cases: with self.subTest(name=name), TemporaryDirectory() as tmp: @@ -284,6 +323,8 @@ class ExecutionTargetPolicyTests(unittest.TestCase): self.assertEqual(target.reasoning_effort, effort) self.assertEqual(target.execution_class, "cloud_model") self.assertFalse(target.selfcheck_required) + self.assertFalse(target.selfcheck_full_review) + self.assertTrue(target.selfcheck_checklist_review) def test_invalid_inputs_are_rejected(self): cases = [