sync: agent-ops from agentic-framework v1.1.205

.clinerules

@@ -3,6 +3,7 @@
 **현재 문서를 반드시 끝까지 정독하고 작업한다. 다 읽지 않고 즉각 작업은 금지한다.**
 
 - 기존 구조를 우선한다. 새 파일 생성보다 기존 파일 수정을 우선한다.
+- **과설계를 금지한다.** 기능을 먼저 최소 구현하며, 구현 단계의 검증은 보안·데이터 손상 방지처럼 반드시 필요한 경우 외에는 두지 않는다. 추가 검증은 실제 문제가 확인된 뒤에만 보강한다.
 - `agent-ops/rules/common/**`와 `agent-ops/skills/common/**`은 중앙 관리되는 공통 영역이므로 어떤 프로젝트 작업에서도 사용자가 직접 지시하지 않는 이상 절대 직접 수정하지 않는다. 프로젝트별 규칙과 스킬은 반드시 대응하는 `project/**` 영역에만 반영한다.
 - 최종 답변은 한국어로 한다.
 - 코드 변경 전 관련 domain rule을 먼저 확인한다.

.cursorrules

@@ -3,6 +3,7 @@
 **현재 문서를 반드시 끝까지 정독하고 작업한다. 다 읽지 않고 즉각 작업은 금지한다.**
 
 - 기존 구조를 우선한다. 새 파일 생성보다 기존 파일 수정을 우선한다.
+- **과설계를 금지한다.** 기능을 먼저 최소 구현하며, 구현 단계의 검증은 보안·데이터 손상 방지처럼 반드시 필요한 경우 외에는 두지 않는다. 추가 검증은 실제 문제가 확인된 뒤에만 보강한다.
 - `agent-ops/rules/common/**`와 `agent-ops/skills/common/**`은 중앙 관리되는 공통 영역이므로 어떤 프로젝트 작업에서도 사용자가 직접 지시하지 않는 이상 절대 직접 수정하지 않는다. 프로젝트별 규칙과 스킬은 반드시 대응하는 `project/**` 영역에만 반영한다.
 - 최종 답변은 한국어로 한다.
 - 코드 변경 전 관련 domain rule을 먼저 확인한다.

AGENTS.md

@@ -3,6 +3,7 @@
 **현재 문서를 반드시 끝까지 정독하고 작업한다. 다 읽지 않고 즉각 작업은 금지한다.**
 
 - 기존 구조를 우선한다. 새 파일 생성보다 기존 파일 수정을 우선한다.
+- **과설계를 금지한다.** 기능을 먼저 최소 구현하며, 구현 단계의 검증은 보안·데이터 손상 방지처럼 반드시 필요한 경우 외에는 두지 않는다. 추가 검증은 실제 문제가 확인된 뒤에만 보강한다.
 - `agent-ops/rules/common/**`와 `agent-ops/skills/common/**`은 중앙 관리되는 공통 영역이므로 어떤 프로젝트 작업에서도 사용자가 직접 지시하지 않는 이상 절대 직접 수정하지 않는다. 프로젝트별 규칙과 스킬은 반드시 대응하는 `project/**` 영역에만 반영한다.
 - 최종 답변은 한국어로 한다.
 - 코드 변경 전 관련 domain rule을 먼저 확인한다.

CLAUDE.md

@@ -3,6 +3,7 @@
 **현재 문서를 반드시 끝까지 정독하고 작업한다. 다 읽지 않고 즉각 작업은 금지한다.**
 
 - 기존 구조를 우선한다. 새 파일 생성보다 기존 파일 수정을 우선한다.
+- **과설계를 금지한다.** 기능을 먼저 최소 구현하며, 구현 단계의 검증은 보안·데이터 손상 방지처럼 반드시 필요한 경우 외에는 두지 않는다. 추가 검증은 실제 문제가 확인된 뒤에만 보강한다.
 - `agent-ops/rules/common/**`와 `agent-ops/skills/common/**`은 중앙 관리되는 공통 영역이므로 어떤 프로젝트 작업에서도 사용자가 직접 지시하지 않는 이상 절대 직접 수정하지 않는다. 프로젝트별 규칙과 스킬은 반드시 대응하는 `project/**` 영역에만 반영한다.
 - 최종 답변은 한국어로 한다.
 - 코드 변경 전 관련 domain rule을 먼저 확인한다.

GEMINI.md

@@ -3,6 +3,7 @@
 **현재 문서를 반드시 끝까지 정독하고 작업한다. 다 읽지 않고 즉각 작업은 금지한다.**
 
 - 기존 구조를 우선한다. 새 파일 생성보다 기존 파일 수정을 우선한다.
+- **과설계를 금지한다.** 기능을 먼저 최소 구현하며, 구현 단계의 검증은 보안·데이터 손상 방지처럼 반드시 필요한 경우 외에는 두지 않는다. 추가 검증은 실제 문제가 확인된 뒤에만 보강한다.
 - `agent-ops/rules/common/**`와 `agent-ops/skills/common/**`은 중앙 관리되는 공통 영역이므로 어떤 프로젝트 작업에서도 사용자가 직접 지시하지 않는 이상 절대 직접 수정하지 않는다. 프로젝트별 규칙과 스킬은 반드시 대응하는 `project/**` 영역에만 반영한다.
 - 최종 답변은 한국어로 한다.
 - 코드 변경 전 관련 domain rule을 먼저 확인한다.

agent-ops/.version

@@ -1 +1 @@
-1.1.204
+1.1.205

agent-ops/rules/common/rules.md

@@ -3,6 +3,7 @@
 **현재 문서를 반드시 끝까지 정독하고 작업한다. 다 읽지 않고 즉각 작업은 금지한다.**
 
 - 기존 구조를 우선한다. 새 파일 생성보다 기존 파일 수정을 우선한다.
+- **과설계를 금지한다.** 기능을 먼저 최소 구현하며, 구현 단계의 검증은 보안·데이터 손상 방지처럼 반드시 필요한 경우 외에는 두지 않는다. 추가 검증은 실제 문제가 확인된 뒤에만 보강한다.
 - `agent-ops/rules/common/**`와 `agent-ops/skills/common/**`은 중앙 관리되는 공통 영역이므로 어떤 프로젝트 작업에서도 사용자가 직접 지시하지 않는 이상 절대 직접 수정하지 않는다. 프로젝트별 규칙과 스킬은 반드시 대응하는 `project/**` 영역에만 반영한다.
 - 최종 답변은 한국어로 한다.
 - 코드 변경 전 관련 domain rule을 먼저 확인한다.

agent-ops/skills/common/orchestrate-agent-task-loop/SKILL.md

@@ -48,7 +48,8 @@ Each target has:
 - an opaque `model` identity;
 - optional `reasoning_effort`, stored as a separate opaque catalog value rather than embedded in dispatcher code or a literal command argument;
 - `execution_class`: `local_model` or `cloud_model`;
-- optional `selfcheck_required` boolean;
+- optional `selfcheck_required` boolean for an unconditional post-worker self-check;
+- optional `completion_selfcheck_on_incomplete` boolean that routes a changed, zero-exit worker with incomplete implementation checklist evidence to the self-check stage before charging a generic worker failure;
 - `runtime.command`: a non-empty argv template executed without a shell;
 - optional `runtime.resume_command`, `preflight_command`, `environment`, `session_path`, `native_session_monitor`, `session_stall_resume`, `terminal_success`, and `auxiliary_logs`;
 - optional `runtime.output_format`: `text` or `jsonl`.
@@ -101,7 +102,9 @@ Never ask a child to create, edit, or summarize `WORK_LOG.md`; that file is disp
 
 Run self-check only when the selected catalog target declares `selfcheck_required=true`. The completing decision, not a fixed agent identity or execution class, determines the requirement.
 
-Treat worker exit `0` as transport completion only. Before marking the worker done, require at least one claimed file or implementation-evidence change and a complete implementation-owned checklist (or concrete blocker evidence). Classify a no-op or incomplete-evidence exit as `generic-error`, apply the same per-target three-error budget, and return persisted `worker_done` state to the worker stage while that contract remains incomplete. Apply that bounded three-attempt target budget to `session-stall` as well, so a repeatedly silent candidate advances instead of consuming the ten-attempt stage recovery budget.
+When `completion_selfcheck_on_incomplete=true`, a zero-exit worker that changed claimed files but left implementation checklist evidence incomplete advances to self-check instead of being counted immediately as `generic-error`. This conditional path does not waive the checklist gate, does not accept a no-op worker, and does not run after a complete handoff. The self-check must fill or concretely block the implementation-owned evidence before official review. If a conditional self-check on a target without native resume still leaves the checklist incomplete, fail over once to the next worker candidate with the current workspace instead of fresh-restarting the same target.
+
+Treat worker exit `0` as transport completion only. Before advancing to official review, require at least one claimed file or implementation-evidence change and a complete implementation-owned checklist (or concrete blocker evidence). Always classify a no-op exit as `generic-error`. Classify incomplete evidence as `generic-error` unless the selected target declares `completion_selfcheck_on_incomplete=true`; that target enters self-check without consuming the worker generic-failure budget. Apply the bounded per-target three-attempt budget to ordinary worker retries and `session-stall`, so a repeatedly silent candidate advances instead of consuming the ten-failure stage recovery budget.
 
 Accept self-check completion only when `## Implementation Checklist` or its supported legacy heading contains at least one checkbox and every checkbox has a non-empty value. Run one full pass, then resume the latest successful native context for at most 10 unchecked-item retries when the target supports native resume. Block instead of silently starting a new context when a required persisted context is unavailable.
 

agent-ops/skills/common/orchestrate-agent-task-loop/assets/default-execution-catalog.json

@@ -50,12 +50,62 @@
         "terminal_success": "agent_end"
       }
     },
+    "pi-ornith-fast-high": {
+      "agent": "pi",
+      "model": "ornith-fast",
+      "reasoning_effort": "high",
+      "execution_class": "local_model",
+      "selfcheck_required": true,
+      "runtime": {
+        "command": [
+          "pi",
+          "-p",
+          "--mode",
+          "json",
+          "--approve",
+          "--provider",
+          "iop",
+          "--model",
+          "{model}",
+          "--thinking",
+          "{reasoning_effort}",
+          "--session-id",
+          "{session_id}",
+          "--session-dir",
+          "{attempt_dir}/pi-sessions",
+          "{prompt}"
+        ],
+        "resume_command": [
+          "pi",
+          "-p",
+          "--mode",
+          "json",
+          "--approve",
+          "--provider",
+          "iop",
+          "--model",
+          "{model}",
+          "--thinking",
+          "{reasoning_effort}",
+          "--session",
+          "{resume_session}",
+          "--session-dir",
+          "{resume_session_dir}",
+          "{prompt}"
+        ],
+        "output_format": "jsonl",
+        "session_path": "{attempt_dir}/pi-sessions/*{session_id}*.jsonl",
+        "native_session_monitor": true,
+        "terminal_success": "agent_end"
+      }
+    },
     "opencode-glm-medium": {
       "agent": "opencode",
       "model": "glm-5.2",
       "reasoning_effort": "medium",
       "execution_class": "cloud_model",
       "selfcheck_required": false,
+      "completion_selfcheck_on_incomplete": true,
       "runtime": {
         "command": [
           "opencode",
@@ -107,6 +157,7 @@
       "reasoning_effort": "high",
       "execution_class": "cloud_model",
       "selfcheck_required": false,
+      "completion_selfcheck_on_incomplete": true,
       "runtime": {
         "command": [
           "opencode",
@@ -158,6 +209,7 @@
       "reasoning_effort": "high",
       "execution_class": "cloud_model",
       "selfcheck_required": false,
+      "completion_selfcheck_on_incomplete": true,
       "runtime": {
         "command": [
           "opencode",
@@ -345,7 +397,7 @@
         "reason_codes": ["worker_catalog_lane"]
       },
       "local-G04": {
-        "candidates": ["pi-ornith-high"],
+        "candidates": ["pi-ornith-fast-high"],
         "rule_id": "worker-local-g04-catalog",
         "policy_priority": 30,
         "reason_codes": ["worker_catalog_lane"]

agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py

@@ -512,6 +512,7 @@ class AgentSpec:
     target_id: str | None = None
     execution_class: str = "cloud_model"
     selfcheck_required: bool = False
+    completion_selfcheck_on_incomplete: bool = False
     reasoning_effort: str | None = None
     runtime: dict[str, Any] = field(default_factory=dict)
 
@@ -532,6 +533,11 @@ def agent_spec_from_record(record: dict[str, Any]) -> AgentSpec | None:
     selfcheck_required = record.get("selfcheck_required", False)
     if not isinstance(selfcheck_required, bool):
         return None
+    completion_selfcheck_on_incomplete = record.get(
+        "completion_selfcheck_on_incomplete", False
+    )
+    if not isinstance(completion_selfcheck_on_incomplete, bool):
+        return None
     reasoning_effort = record.get("reasoning_effort")
     if reasoning_effort is not None and (
         not isinstance(reasoning_effort, str) or not reasoning_effort
@@ -549,6 +555,7 @@ def agent_spec_from_record(record: dict[str, Any]) -> AgentSpec | None:
         target_id=target_id,
         execution_class=execution_class,
         selfcheck_required=selfcheck_required,
+        completion_selfcheck_on_incomplete=completion_selfcheck_on_incomplete,
         reasoning_effort=reasoning_effort,
         runtime=dict(runtime),
     )
@@ -842,6 +849,7 @@ class StateStore:
                 "worker_cli": None,
                 "worker_model": None,
                 "selfcheck_done": False,
+                "completion_selfcheck_pending": False,
                 "blocked": None,
                 "active_stage": None,
                 "active_locator": None,
@@ -871,6 +879,7 @@ class StateStore:
             "worker_cli": None,
             "worker_model": None,
             "selfcheck_done": False,
+            "completion_selfcheck_pending": False,
             "blocked": None,
             "active_stage": None,
             "active_locator": None,
@@ -1134,6 +1143,7 @@ class StateStore:
                 retry_failover_pending=False,
                 retry_failover_context=None,
                 blocker_evidence=None,
+                completion_selfcheck_pending=False,
                 blocked=None,
                 active_stage=None,
                 active_locator=None,
@@ -1188,6 +1198,7 @@ class StateStore:
             value["review_no_progress"] = 0
             value["selfcheck_incomplete"] = 0
             value["selfcheck_context_locator"] = None
+            value["completion_selfcheck_pending"] = False
             value["recovery_failures"] = {}
             value["stage_failure_budgets"] = {}
             value["generic_failure_budgets"] = {}
@@ -1959,6 +1970,9 @@ def agent_spec_from_decision(decision: dict[str, Any]) -> AgentSpec:
         target_id=target.catalog_id,
         execution_class=target.execution_class,
         selfcheck_required=target.selfcheck_required,
+        completion_selfcheck_on_incomplete=(
+            target.completion_selfcheck_on_incomplete
+        ),
         reasoning_effort=target.reasoning_effort,
         runtime=runtime,
     )
@@ -2347,6 +2361,18 @@ def completing_decision_requires_selfcheck(state: dict[str, Any]) -> bool:
     return selected.get("selfcheck_required") is True
 
 
+def completing_decision_allows_completion_selfcheck(
+    state: dict[str, Any],
+) -> bool:
+    completing = state.get("completing_decision")
+    if not isinstance(completing, dict):
+        return False
+    selected = completing.get("selected")
+    if not isinstance(selected, dict):
+        return False
+    return selected.get("completion_selfcheck_on_incomplete") is True
+
+
 def _validated_completing_decision(
     task: Task, decision: dict[str, Any]
 ) -> tuple[dict[str, Any], AgentSpec]:
@@ -2506,6 +2532,13 @@ def task_stage(task: Task, state: dict[str, Any]) -> str:
         # worker stage until the implementation-owned review contract is
         # actually materialized (or contains complete blocker evidence).
         if implementation_review_errors(task):
+            if (
+                state.get("completion_selfcheck_pending")
+                and not state.get("selfcheck_done")
+                and completing_decision_allows_completion_selfcheck(state)
+                and _completing_decision_is_valid(task, state)
+            ):
+                return "selfcheck"
             return "worker"
         if not _completing_decision_is_valid(task, state):
             return "blocked"
@@ -3677,6 +3710,9 @@ async def invoke(
         "target_id": spec.target_id,
         "execution_class": spec.execution_class,
         "selfcheck_required": spec.selfcheck_required,
+        "completion_selfcheck_on_incomplete": (
+            spec.completion_selfcheck_on_incomplete
+        ),
         "reasoning_effort": spec.reasoning_effort,
         "runtime": spec.runtime,
         "agent_process_marker": process_marker,
@@ -4257,13 +4293,18 @@ async def invoke(
     ):
         worker_errors = implementation_review_errors(task)
         worker_signature_after = task_signature(workspace, task)
-        if worker_signature_before == worker_signature_after or worker_errors:
+        unchanged = worker_signature_before == worker_signature_after
+        incomplete_without_selfcheck = (
+            bool(worker_errors) and not spec.completion_selfcheck_on_incomplete
+        )
+        if unchanged or incomplete_without_selfcheck:
             failure_class = "generic-error"
             failure_source = "dispatcher-worker-completion-contract"
             details = []
-            if worker_signature_before == worker_signature_after:
+            if unchanged:
                 details.append("no claimed file or implementation evidence changed")
-            details.extend(worker_errors)
+            if incomplete_without_selfcheck:
+                details.extend(worker_errors)
             failure_evidence = "; ".join(details)
             failure_evidence_source = "dispatcher:worker-completion-contract"
     try:
@@ -5655,6 +5696,13 @@ def _mark_worker_done(
     _require_same_runtime_identity(expected_spec, worker_cli, worker_model)
     selected = validated_decision["selected"]
     execution_class = selected["execution_class"]
+    completion_selfcheck_pending = bool(
+        selected.get("completion_selfcheck_on_incomplete")
+        and implementation_review_errors(task)
+    )
+    selfcheck_pending = bool(
+        selected["selfcheck_required"] or completion_selfcheck_pending
+    )
     store.update_task(
         task,
         worker_done=True,
@@ -5662,7 +5710,8 @@ def _mark_worker_done(
         worker_model=worker_model,
         completing_decision=validated_decision,
         execution_class=execution_class,
-        selfcheck_done=not selected["selfcheck_required"],
+        selfcheck_done=not selfcheck_pending,
+        completion_selfcheck_pending=completion_selfcheck_pending,
         review_no_progress=0,
         blocked=None,
     )
@@ -5690,8 +5739,16 @@ async def run_selfcheck(
         store.update_task(task, blocked=str(exc))
         banner("작업차단", task.name, [f"reason={exc}"])
         return
-    if not spec.selfcheck_required:
+    completion_selfcheck_pending = bool(
-        raise RuntimeError("selfcheck_required가 아닌 route에 selfcheck stage가 배정됐다")
+        store.task_state(task).get("completion_selfcheck_pending")
+    )
+    if not spec.selfcheck_required and not (
+        completion_selfcheck_pending
+        and spec.completion_selfcheck_on_incomplete
+    ):
+        raise RuntimeError(
+            "selfcheck 계약이 없는 route에 selfcheck stage가 배정됐다"
+        )
     work_log = milestone_work_log_path(task)
     banner(
         "자가검증시작",
@@ -5756,7 +5813,7 @@ async def run_selfcheck(
             )
             return
     while True:
-        unchecked_items = incomplete_results > 0
+        unchecked_items = completion_selfcheck_pending or incomplete_results > 0
         success, locator = await run_escalating(
             workspace,
             store,
@@ -5776,6 +5833,45 @@ async def run_selfcheck(
         if not errors:
             break
         if not spec.native_resume:
+            if completion_selfcheck_pending:
+                try:
+                    next_decision = select_execution_decision(
+                        task,
+                        stage="worker",
+                        prior_decision=completing,
+                        transition="failover",
+                        failure_class="generic-error",
+                    )
+                    next_spec = agent_spec_from_decision(next_decision)
+                except (ExecutionDecisionError, OSError, ValueError):
+                    next_decision = None
+                    next_spec = spec
+                if next_decision is not None and next_spec != spec:
+                    commit_execution_decision(store, task, "worker", next_decision)
+                    store.update_task(
+                        task,
+                        worker_done=False,
+                        worker_cli=None,
+                        worker_model=None,
+                        completing_decision=None,
+                        execution_class=next_spec.execution_class,
+                        selfcheck_done=False,
+                        completion_selfcheck_pending=False,
+                        selfcheck_incomplete=0,
+                        selfcheck_context_locator=None,
+                        blocked=None,
+                    )
+                    banner(
+                        "자가검증실행대상전환",
+                        task.name,
+                        [
+                            f"from={spec.display}",
+                            f"to={next_spec.display}",
+                            f"reason={'; '.join(errors)}",
+                            f"locator={locator}",
+                        ],
+                    )
+                    return
             reason = "selfcheck checklist가 미완료지만 target에 native resume 계약이 없다"
             store.update_task(task, blocked=reason)
             banner(
@@ -5836,6 +5932,7 @@ async def run_selfcheck(
     store.update_task(
         task,
         selfcheck_done=True,
+        completion_selfcheck_pending=False,
         selfcheck_incomplete=0,
         selfcheck_context_locator=None,
         blocked=None,

agent-ops/skills/common/orchestrate-agent-task-loop/scripts/execution_target_policy.py

@@ -48,6 +48,7 @@ class RouteTarget:
     reasoning_effort: str | None
     execution_class: str
     selfcheck_required: bool
+    completion_selfcheck_on_incomplete: bool
     runtime: dict[str, Any]
 
 @dataclass(frozen=True)
@@ -211,6 +212,7 @@ def _validate_target(target_id: str, value: object) -> RouteTarget:
         "reasoning_effort",
         "execution_class",
         "selfcheck_required",
+        "completion_selfcheck_on_incomplete",
         "runtime",
     }
     if unknown:
@@ -224,6 +226,13 @@ def _validate_target(target_id: str, value: object) -> RouteTarget:
     selfcheck_required = value.get("selfcheck_required", False)
     if not isinstance(selfcheck_required, bool):
         raise CatalogError(f"{label}.selfcheck_required must be a boolean")
+    completion_selfcheck_on_incomplete = value.get(
+        "completion_selfcheck_on_incomplete", False
+    )
+    if not isinstance(completion_selfcheck_on_incomplete, bool):
+        raise CatalogError(
+            f"{label}.completion_selfcheck_on_incomplete must be a boolean"
+        )
     reasoning_effort_value = value.get("reasoning_effort")
     reasoning_effort = (
         None
@@ -268,6 +277,7 @@ def _validate_target(target_id: str, value: object) -> RouteTarget:
         reasoning_effort=reasoning_effort,
         execution_class=execution_class,
         selfcheck_required=selfcheck_required,
+        completion_selfcheck_on_incomplete=completion_selfcheck_on_incomplete,
         runtime=runtime,
     )
 

agent-ops/skills/common/orchestrate-agent-task-loop/scripts/select_execution_target.py

@@ -18,7 +18,7 @@ from pathlib import Path
 from zoneinfo import ZoneInfo
 
 
-SCHEMA_VERSION = "2.0"
+SCHEMA_VERSION = "3.0"
 CATALOG_ENV = "AGENT_TASK_EXECUTION_CATALOG"
 DEFAULT_CATALOG_PATH = (
     Path(__file__).resolve().parents[1]
@@ -160,6 +160,9 @@ def _target_snapshot(target) -> dict:
         "model": target.model,
         "execution_class": target.execution_class,
         "selfcheck_required": target.selfcheck_required,
+        "completion_selfcheck_on_incomplete": (
+            target.completion_selfcheck_on_incomplete
+        ),
     }
     if target.reasoning_effort is not None:
         snapshot["reasoning_effort"] = target.reasoning_effort
@@ -180,6 +183,7 @@ def _validate_target_snapshot(value: object, prefix: str) -> dict:
         "model",
         "execution_class",
         "selfcheck_required",
+        "completion_selfcheck_on_incomplete",
     }
     missing = required - set(value)
     if missing:
@@ -194,6 +198,11 @@ def _validate_target_snapshot(value: object, prefix: str) -> dict:
         )
     if not isinstance(value["selfcheck_required"], bool):
         raise SelectorInputError(code, f"{prefix}.selfcheck_required must be a boolean")
+    if not isinstance(value["completion_selfcheck_on_incomplete"], bool):
+        raise SelectorInputError(
+            code,
+            f"{prefix}.completion_selfcheck_on_incomplete must be a boolean",
+        )
     reasoning_effort = value.get("reasoning_effort")
     if reasoning_effort is not None and (
         not isinstance(reasoning_effort, str) or not reasoning_effort

agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatch.py

@@ -729,6 +729,7 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
                 "fake-model",
                 "fake-json-runner/fake-model",
                 target_id="fake-json-target",
+                completion_selfcheck_on_incomplete=True,
                 runtime={
                     "command": [sys.executable, str(runner)],
                     "output_format": "jsonl",
@@ -762,6 +763,54 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
         self.assertEqual(record["status"], "failed")
         self.assertNotIn("succeeded:0", work_log)
 
+    def test_changed_worker_with_incomplete_evidence_can_advance_to_completion_selfcheck(self):
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            plan = write_plan(root)
+            task = task_from_plan(root, plan)
+            review = plan.parent / "CODE_REVIEW-cloud-G05.md"
+            review.write_text(
+                "## Implementation Checklist\n\n- [ ] Implement the task.\n",
+                encoding="utf-8",
+            )
+            task.review = review
+            claimed = root / "src" / "item.txt"
+            claimed.parent.mkdir(parents=True)
+            runner = root / "completion_selfcheck_runner.py"
+            runner.write_text(
+                "import json\n"
+                f"open({str(claimed)!r}, 'w', encoding='utf-8').write('changed')\n"
+                "print(json.dumps({'type': 'agent_end', 'willRetry': False, "
+                "'messages': [{'role': 'assistant', 'stopReason': 'stop'}]}))\n",
+                encoding="utf-8",
+            )
+            agent = dispatch.AgentSpec(
+                "fake-json-runner",
+                "fake-model",
+                "fake-json-runner/fake-model",
+                target_id="fake-json-target",
+                completion_selfcheck_on_incomplete=True,
+                runtime={
+                    "command": [sys.executable, str(runner)],
+                    "output_format": "jsonl",
+                    "terminal_success": "agent_end",
+                },
+            )
+            with mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}):
+                store = dispatch.StateStore(root)
+                try:
+                    return_code, failure, locator = asyncio.run(
+                        dispatch.invoke(root, store, task, "worker", agent, "fake prompt")
+                    )
+                    record = json.loads(locator.read_text(encoding="utf-8"))
+                finally:
+                    store.close()
+
+        self.assertEqual(return_code, 0)
+        self.assertIsNone(failure)
+        self.assertEqual(record["status"], "succeeded")
+        self.assertTrue(record["completion_selfcheck_on_incomplete"])
+
     def test_worker_done_with_incomplete_evidence_returns_to_worker_stage(self):
         with TemporaryDirectory() as tmp:
             root = Path(tmp)
@@ -778,6 +827,98 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
 
         self.assertEqual(stage, "worker")
 
+    def test_worker_done_with_conditional_selfcheck_routes_incomplete_evidence_to_selfcheck(self):
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            value = catalog_value()
+            value["targets"]["primary"][
+                "completion_selfcheck_on_incomplete"
+            ] = True
+            catalog = write_catalog(root, value)
+            plan = write_plan(root)
+            task = task_from_plan(root, plan)
+            review = plan.parent / "CODE_REVIEW-cloud-G05.md"
+            review.write_text(
+                "## Implementation Checklist\n\n- [ ] Implement the task.\n",
+                encoding="utf-8",
+            )
+            task.review = review
+            selector = dispatch._selector_module()
+            decision = selector.select_execution_target(plan, catalog_path=catalog)
+
+            stage = dispatch.task_stage(
+                task,
+                {
+                    "worker_done": True,
+                    "selfcheck_done": False,
+                    "completion_selfcheck_pending": True,
+                    "completing_decision": decision,
+                },
+            )
+
+        self.assertEqual(stage, "selfcheck")
+
+    def test_incomplete_conditional_selfcheck_without_resume_fails_over_to_next_worker(self):
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            value = catalog_value()
+            value["targets"]["primary"]["runtime"].pop(
+                "native_session_monitor"
+            )
+            value["targets"]["primary"]["runtime"].pop("resume_command")
+            value["targets"]["primary"][
+                "completion_selfcheck_on_incomplete"
+            ] = True
+            catalog = write_catalog(root, value)
+            plan = write_plan(root)
+            task = task_from_plan(root, plan)
+            review = plan.parent / "CODE_REVIEW-cloud-G05.md"
+            review.write_text(
+                "## Implementation Checklist\n\n- [ ] Implement the task.\n",
+                encoding="utf-8",
+            )
+            task.review = review
+            dispatch.EXECUTION_CATALOG_PATH = catalog
+            selector = dispatch._selector_module()
+            decision = selector.select_execution_target(
+                plan, catalog_path=catalog
+            )
+            with mock.patch.dict(
+                os.environ, {"XDG_STATE_HOME": str(root / "state")}
+            ):
+                store = dispatch.StateStore(root)
+                try:
+                    store.update_task(
+                        task,
+                        worker_done=True,
+                        selfcheck_done=False,
+                        completion_selfcheck_pending=True,
+                        completing_decision=decision,
+                        execution_decisions={"worker": decision},
+                    )
+                    with mock.patch.object(
+                        dispatch,
+                        "run_escalating",
+                        new=mock.AsyncMock(
+                            return_value=(
+                                True,
+                                root / "selfcheck-locator.json",
+                            )
+                        ),
+                    ):
+                        asyncio.run(dispatch.run_selfcheck(root, store, task))
+                    state = store.task_state(task)
+                finally:
+                    store.close()
+
+        self.assertFalse(state["worker_done"])
+        self.assertFalse(state["completion_selfcheck_pending"])
+        self.assertIsNone(state["blocked"])
+        self.assertEqual(
+            state["execution_decisions"]["worker"]["selected"]["target_id"],
+            "alternate",
+        )
+
     def test_silent_native_session_is_terminated_and_classified_as_stall(self):
         with TemporaryDirectory() as tmp:
             root = Path(tmp)

agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_execution_target_policy.py

@@ -105,6 +105,30 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
         self.assertFalse(hasattr(policy, "quota_probe_spec"))
         self.assertFalse(hasattr(policy, "promotion_target"))
 
+    def test_completion_selfcheck_flag_is_optional_and_typed(self):
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            value = catalog_value()
+            value["targets"]["target-a"][
+                "completion_selfcheck_on_incomplete"
+            ] = True
+            catalog = policy.load_catalog(write_catalog(root, value))
+            self.assertTrue(
+                catalog.targets["target-a"].completion_selfcheck_on_incomplete
+            )
+            self.assertFalse(
+                catalog.targets["target-b"].completion_selfcheck_on_incomplete
+            )
+
+            value["targets"]["target-a"][
+                "completion_selfcheck_on_incomplete"
+            ] = "yes"
+            with self.assertRaisesRegex(
+                policy.CatalogError,
+                "completion_selfcheck_on_incomplete must be a boolean",
+            ):
+                policy.load_catalog(write_catalog(root, value))
+
     def test_optional_windows_are_catalog_owned_and_timezone_generic(self):
         with TemporaryDirectory() as tmp:
             catalog = policy.load_catalog(write_catalog(Path(tmp), catalog_value(windows=True)))

agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_select_execution_target.py

@@ -90,6 +90,7 @@ class SelectorTests(unittest.TestCase):
                 f"local-G{grade:02d}": ["pi-ornith-high"]
                 for grade in range(1, 7)
             },
+            "local-G04": ["pi-ornith-fast-high"],
             "local-G07": ["opencode-glm-max", "codex-terra-high"],
             "local-G08": ["opencode-glm-max", "codex-terra-high"],
             "local-G09": ["codex-sol-high", "codex-terra-high"],
@@ -107,6 +108,7 @@ class SelectorTests(unittest.TestCase):
         }
         expected_targets = {
             "pi-ornith-high",
+            "pi-ornith-fast-high",
             "opencode-glm-medium",
             "opencode-glm-high",
             "opencode-glm-max",
@@ -180,6 +182,7 @@ class SelectorTests(unittest.TestCase):
         opencode = catalog.targets["opencode-glm-max"]
         self.assertIn("iop-glm/glm-5.2", opencode.runtime["command"])
         self.assertEqual(opencode.reasoning_effort, "high")
+        self.assertTrue(opencode.completion_selfcheck_on_incomplete)
         self.assertIn("{reasoning_effort}", opencode.runtime["command"])
         for target_id in (
             "opencode-glm-medium",
@@ -220,7 +223,7 @@ class SelectorTests(unittest.TestCase):
                 catalog_path=catalog,
                 evaluated_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
             )
-        self.assertEqual(result["schema_version"], "2.0")
+        self.assertEqual(result["schema_version"], "3.0")
         self.assertEqual(result["selected"]["target_id"], "first")
         self.assertEqual(result["selected"]["agent"], "agent-one")
         self.assertEqual(result["selected"]["model"], "model-one")