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

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

Browse source
This commit is contained in:
toki 2026-08-15 11:59:47 +09:00
parent 48e918f617
commit a16d55ca72
15 changed files with 358 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.clinerules
View file

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

1
.cursorrules
View file

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

1
AGENTS.md
View file

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

1
CLAUDE.md
View file

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

1
GEMINI.md
View file

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

2
agent-ops/.version
View file

@ -1 +1 @@
1.1.204
1.1.205

1
agent-ops/rules/common/rules.md
View file

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

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

@ -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.

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

@ -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"]

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

@ -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:
raise RuntimeError("selfcheck_required가 아닌 route에 selfcheck stage가 배정됐다")
completion_selfcheck_pending = bool(
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,

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

@ -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,
)

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

@ -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

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

@ -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)

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

@ -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)))

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

@ -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")