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

fix(agent-ops): selfcheck 재시도 문맥을 유지한다

Browse source
This commit is contained in:
toki 2026-08-01 05:25:14 +09:00
parent 174a2a6b13
commit 23b3ad74a2
3 changed files with 496 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -98,24 +98,26 @@ Concurrency limits:
Keep control prompts in English, insert absolute paths only, and do not expand these sentences unnecessarily.
- A dispatcher child runs only while `IOP_AGENT_TASK_EXECUTION_ID` is present.
- Prefix every child 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.`
- 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 them 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: `Think in English. Keep artifact content in English. Final in Korean. Read {CODE_REVIEW_PATH} and fill every missing implementation field. Do not finish until all implementation fields are complete. This is a self-check of completed work, not a review. Read {PLAN_PATH} and finish any missing work. Recheck and fix your work.`
- 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: `Think in English. Final in Korean. Read {PLAN_PATH}; complete every unchecked implementation item and update {CODE_REVIEW_PATH}. 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`.
Do not treat Pi self-check exit code `0` as success by itself. Set `selfcheck_done=true` only when `## Implementation Checklist` (or legacy `## 구현 체크리스트`) in `CODE_REVIEW_PATH` contains at least one Markdown list checkbox and every `[...]` checkbox value has at least one non-whitespace character. If both canonical and legacy checklist headings are present in the same file, fail closed. Accept any non-empty value, including `x`, `v`, and `✅`. Do not inspect `## Implementation Item Completion`, `Deviations from Plan`, `Key Design Decisions`, `Verification Results`, or final CODE_REVIEW synchronization text. If the checklist condition fails, retry with the same prompt. After 10 consecutive incomplete results, block that task and continue draining independent work.
Do not treat Pi self-check exit code `0` as success by itself. Set `selfcheck_done=true` only when `## Implementation Checklist` (or legacy `## 구현 체크리스트`) in `CODE_REVIEW_PATH` contains at least one Markdown list checkbox and every `[...]` checkbox value has at least one non-whitespace character. If both canonical and legacy checklist headings are present in the same file, fail closed. Accept any non-empty value, including `x`, `v`, and `✅`. Do not inspect `## Implementation Item Completion`, `Deviations from Plan`, `Key Design Decisions`, `Verification Results`, or final CODE_REVIEW synchronization text. Run the full self-check prompt exactly once. If its checklist condition fails, resume that successful pass's Pi native session and run the unchecked-item retry prompt up to 10 times. Each retry must resume the locator returned by the preceding successful pass so the same conversation context is preserved; never repeat the full review prompt or start a fresh retry session. Persist the latest successful context locator for dispatcher restart, and block instead of starting fresh when that context cannot be resumed. Block that task after the 10th unchecked-item retry remains incomplete, and continue draining independent work.
After an AGY/Gemini worker exits `0`, apply the same `CODE_REVIEW_PATH` implementation-checklist regex before accepting worker completion. If it is incomplete, run a fresh quota probe: only an `exhausted` target becomes `provider-quota` and enters the existing selector failover/promotion chain; `available` or `unknown` remains a completion-evidence recovery on Gemini.
For Pi recovery attempts, pass only `Read {PLAN_PATH}. Continue.` without a locator explanation. 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-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.
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`, and pass `Think in English. Keep artifact content in English. Final in Korean. Continue this session and complete the current task.` After a dispatcher restart, find the failed 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 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.
## Work-Log Contract
@ -153,7 +155,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 Pi self-check leaves implementation fields incomplete 10 consecutive times or official review makes no change 10 consecutive times.
- 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.
- 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.
@ -219,7 +221,7 @@ 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 with a fresh self-check session on the same Pi model, not with worker or review.
- 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.
- Key persistent state to the `task/plan/tag` generation at the start of PLAN. Checklist/body edits to the same PLAN do not reset the stage; a new plan number in a follow-up PLAN 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.
- 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.
@ -250,7 +252,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 the routing table.
- [ ] Run exactly one fresh-session self-check only for Pi work.
- [ ] 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.
- [ ] Run every official review with Codex `gpt-5.6-sol` xhigh 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`.

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

@ -127,6 +127,7 @@ DISPATCHER_CHILD_BOUNDARY_PROMPT = (
"when required by plan or code-review finalization because that mode "
"validates one candidate PLAN without starting or monitoring orchestration."
)
SELF_CHECK_PROMPT_PREFIX = "Think in English. Final in Korean."
KST = timezone(timedelta(hours=9), name="KST")
DEFAULT_MAX_PARALLEL = 3
@ -153,7 +154,7 @@ STREAM_HEARTBEAT_SECONDS = 30
PI_MODEL_RESPONSE_STALL_SECONDS = 3 * 60
PI_SESSION_SCHEMA_VERSION = 3
RECOVERY_FAILURE_LIMIT = 10
SELF_CHECK_INCOMPLETE_LIMIT = 10
SELF_CHECK_UNCHECKED_RETRY_LIMIT = 10
REVIEW_NO_PROGRESS_LIMIT = 10
PROVIDER_TRANSPORT_FAILURES = frozenset(
{"provider-connection", "provider-stream-disconnect"}
@ -717,6 +718,7 @@ class StateStore:
"active_locator": None,
"review_no_progress": 0,
"selfcheck_incomplete": 0,
"selfcheck_context_locator": None,
"recovery_failures": {},
"execution_decisions": {},
"route_transition_history": [],
@ -744,6 +746,7 @@ class StateStore:
"active_locator": None,
"review_no_progress": 0,
"selfcheck_incomplete": 0,
"selfcheck_context_locator": None,
"recovery_failures": {},
"execution_decisions": {},
"route_transition_history": [],
@ -899,6 +902,7 @@ class StateStore:
value["blocked"] = None
value["review_no_progress"] = 0
value["selfcheck_incomplete"] = 0
value["selfcheck_context_locator"] = None
value["recovery_failures"] = {}
value["stage_failure_budgets"] = {}
value["retry_quota_refresh_pending"] = False
@ -946,6 +950,7 @@ class StateStore:
value["blocked"] = None
value["review_no_progress"] = 0
value["selfcheck_incomplete"] = 0
value["selfcheck_context_locator"] = None
value["recovery_failures"] = {}
value["stage_failure_budgets"] = {}
value["retry_quota_refresh_pending"] = qualified
@ -3506,6 +3511,64 @@ def laguna_resume_locator(
return locator
def selfcheck_context_resume_locator(
state: dict[str, Any],
task: Task,
*,
expected_workspace: Path,
expected_workspace_id: str,
expected_runs_root: Path,
) -> tuple[Path | None, str]:
raw_locator = state.get("selfcheck_context_locator")
if not isinstance(raw_locator, str) or not raw_locator:
return None, "persisted selfcheck context locator가 없다"
target = Path(raw_locator)
locator = target if target.name == "locator.json" else target / "locator.json"
path_owned, detail = locator_workspace_ownership(
locator,
{},
expected_workspace=expected_workspace,
expected_workspace_id=expected_workspace_id,
expected_runs_root=expected_runs_root,
)
if not path_owned:
return None, detail
try:
record = json.loads(locator.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return None, "persisted selfcheck context locator를 읽을 수 없다"
if not isinstance(record, dict):
return None, "persisted selfcheck context locator 형식이 잘못됐다"
owned, detail = locator_workspace_ownership(
locator,
record,
expected_workspace=expected_workspace,
expected_workspace_id=expected_workspace_id,
expected_runs_root=expected_runs_root,
)
if not owned:
return None, detail
if (
record.get("task") != task.name
or record.get("role") != "selfcheck"
or record.get("cli") != "pi"
or record.get("status") != "succeeded"
):
return None, "persisted selfcheck context locator identity가 일치하지 않는다"
native_raw = record.get("native_session_path")
native = Path(str(native_raw)) if native_raw else None
if native is not None and native.is_dir():
sessions = list(native.glob("*.jsonl"))
native = max(sessions, key=lambda path: path.stat().st_mtime_ns) if sessions else None
if native is None or not native.is_file():
return None, "persisted selfcheck native session이 없다"
try:
native.resolve().relative_to(expected_runs_root.resolve())
except (OSError, RuntimeError, ValueError):
return None, "persisted selfcheck native session이 workspace runs 밖에 있다"
return locator, ""
def agy_conversations() -> dict[Path, int]:
root = Path.home() / ".gemini" / "antigravity-cli" / "conversations"
if not root.is_dir():
@ -4159,7 +4222,31 @@ def dispatcher_child_prompt(body: str) -> str:
return f"{DISPATCHER_CHILD_BOUNDARY_PROMPT} {body}"
def base_prompt(task: Task, role: str, spec: AgentSpec) -> str:
def selfcheck_prompt(task: Task, *, unchecked_items: bool = False) -> str:
if task.plan is None:
raise RuntimeError("selfcheck PLAN이 없다")
if task.review is None:
raise RuntimeError("selfcheck CODE_REVIEW 파일이 없다")
if unchecked_items:
body = (
f"Read {task.plan.resolve()}; complete every unchecked implementation "
f"item and update {task.review.resolve()}. Keep files in English."
)
else:
body = (
f"Read {task.plan.resolve()}; review all work once, fix omissions, "
f"and update {task.review.resolve()}. Keep files in English."
)
return f"{SELF_CHECK_PROMPT_PREFIX} {body}"
def base_prompt(
task: Task,
role: str,
spec: AgentSpec,
*,
unchecked_items: bool = False,
) -> str:
if role == "review":
target = task.review or task.directory
if task.review:
@ -4175,14 +4262,7 @@ def base_prompt(task: Task, role: str, spec: AgentSpec) -> str:
raise RuntimeError("worker PLAN이 없다")
target = task.plan.resolve()
if role == "selfcheck":
if task.review is None:
raise RuntimeError("selfcheck CODE_REVIEW 파일이 없다")
return dispatcher_child_prompt(
f"Think in English. Keep artifact content in English. Final in Korean. Read {task.review.resolve()} and fill "
"every missing implementation field. Do not finish until all implementation "
"fields are complete. This is a self-check of completed work, not a review. "
f"Read {target} and finish any missing work. Recheck and fix your work."
)
return selfcheck_prompt(task, unchecked_items=unchecked_items)
if spec.local_pi:
return dispatcher_child_prompt(
f"Think in English. Keep artifact content in English. Final in "
@ -4304,7 +4384,16 @@ def continuation_prompt(
local_pi: bool = False,
resume_same_pi_session: bool = False,
context: dict[str, Any] | None = None,
unchecked_items: bool = False,
) -> str:
if local_pi and role == "selfcheck":
if resume_same_pi_session:
if unchecked_items:
return selfcheck_prompt(task, unchecked_items=True)
return (
f"{SELF_CHECK_PROMPT_PREFIX} Continue. Keep files in English."
)
return selfcheck_prompt(task, unchecked_items=unchecked_items)
if context is not None:
return continuation_prompt_from_package(
context,
@ -4316,14 +4405,6 @@ def continuation_prompt(
"Think in English. Keep artifact content in English. Final in Korean. Continue this session and complete "
"the current task."
)
if role == "selfcheck" and task.plan and task.review:
return dispatcher_child_prompt(
f"Think in English. Keep artifact content in English. Final in Korean. Read {task.review.resolve()} and fill "
"every missing implementation field. Do not finish until all implementation "
"fields are complete. This is a self-check of completed work, not a review. "
f"Read {task.plan.resolve()} and finish any missing work. Recheck and fix "
"your work."
)
target = task.plan or task.directory
return dispatcher_child_prompt(
f"Think in English. Keep artifact content in English. Final in "
@ -4347,6 +4428,8 @@ async def run_escalating(
role: str,
initial: AgentSpec,
initial_resume_locator: Path | None = None,
*,
unchecked_items: bool = False,
) -> tuple[bool, Path | None]:
spec = initial
previous_locator = initial_resume_locator
@ -4459,7 +4542,7 @@ async def run_escalating(
context: dict[str, Any] | None = None
while True:
prompt = (
base_prompt(task, role, spec)
base_prompt(task, role, spec, unchecked_items=unchecked_items)
if previous_locator is None
else continuation_prompt(
task,
@ -4468,6 +4551,7 @@ async def run_escalating(
local_pi=spec.local_pi,
resume_same_pi_session=pi_resume_locator is not None,
context=context,
unchecked_items=unchecked_items,
)
)
context = None
@ -5559,14 +5643,19 @@ async def run_selfcheck(
f"work_log={work_log.resolve()}",
],
)
retry = 0
# 0 means the full pass is pending. After it fails the checklist gate,
# each additional count represents one completed unchecked-item retry.
incomplete_results = 0
if isinstance(store, StateStore):
retry = int(store.task_state(task).get("selfcheck_incomplete", 0))
if retry >= SELF_CHECK_INCOMPLETE_LIMIT:
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:
locator = resume_locator
reason = (
"selfcheck implementation field limit already exhausted: "
f"{retry}/{SELF_CHECK_INCOMPLETE_LIMIT}"
"selfcheck unchecked-item retry limit already exhausted: "
f"{incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}"
)
store.update_task(task, blocked=f"{reason} locator={locator}")
banner(
@ -5574,12 +5663,31 @@ async def run_selfcheck(
task.name,
[
"reason=selfcheck-incomplete-limit",
f"retry={retry}/{SELF_CHECK_INCOMPLETE_LIMIT}",
"mode=unchecked-items",
f"retry={incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}",
f"locator={locator}",
],
)
return
if incomplete_results > 0 and resume_locator is None:
resume_locator, context_error = selfcheck_context_resume_locator(
store.task_state(task),
task,
expected_workspace=store.workspace,
expected_workspace_id=store.workspace_id,
expected_runs_root=store.runs,
)
if resume_locator is None:
reason = f"selfcheck context resume 실패: {context_error}"
store.update_task(task, blocked=reason)
banner(
"작업차단",
task.name,
["reason=selfcheck-context-unavailable", context_error],
)
return
while True:
unchecked_items = incomplete_results > 0
success, locator = await run_escalating(
workspace,
store,
@ -5587,8 +5695,8 @@ async def run_selfcheck(
"selfcheck",
spec,
initial_resume_locator=resume_locator,
unchecked_items=unchecked_items,
)
resume_locator = None
if not success:
current = store.task_state(task).get("blocked")
store.update_task(
@ -5598,16 +5706,27 @@ async def run_selfcheck(
errors = implementation_review_errors(task)
if not errors:
break
retry += 1
if retry >= SELF_CHECK_INCOMPLETE_LIMIT:
if locator is None:
reason = "selfcheck 성공 locator가 없어 context를 이어갈 수 없다"
store.update_task(task, blocked=reason)
banner(
"작업차단",
task.name,
["reason=selfcheck-context-unavailable", reason],
)
return
incomplete_results += 1
incomplete_retries = max(0, incomplete_results - 1)
if incomplete_retries >= SELF_CHECK_UNCHECKED_RETRY_LIMIT:
reason = (
"selfcheck implementation fields remain incomplete: "
f"{retry}/{SELF_CHECK_INCOMPLETE_LIMIT}"
"selfcheck checklist remains incomplete after unchecked-item retry: "
f"{incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}"
)
store.update_task(
task,
blocked=f"{reason} locator={locator}",
selfcheck_incomplete=retry,
selfcheck_incomplete=incomplete_results,
selfcheck_context_locator=str(locator),
)
banner(
"작업차단",
@ -5615,18 +5734,24 @@ async def run_selfcheck(
[
"reason=selfcheck-incomplete-limit",
f"detail={'; '.join(errors)}",
f"retry={retry}/{SELF_CHECK_INCOMPLETE_LIMIT}",
f"retry={incomplete_retries}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}",
f"locator={locator}",
],
)
return
store.update_task(task, selfcheck_incomplete=retry)
store.update_task(
task,
selfcheck_incomplete=incomplete_results,
selfcheck_context_locator=str(locator),
)
resume_locator = locator
banner(
"자가검증재시도",
task.name,
[
f"reason={'; '.join(errors)}",
f"retry={retry}/{SELF_CHECK_INCOMPLETE_LIMIT}",
"mode=unchecked-items",
f"retry={incomplete_retries + 1}/{SELF_CHECK_UNCHECKED_RETRY_LIMIT}",
f"locator={locator}",
],
)
@ -5634,6 +5759,7 @@ async def run_selfcheck(
task,
selfcheck_done=True,
selfcheck_incomplete=0,
selfcheck_context_locator=None,
blocked=None,
)

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

@ -3468,6 +3468,9 @@ class ReviewControlTest(unittest.TestCase):
task = TaskStageTest().make_task(root)
pi = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True)
selfcheck = dispatch.base_prompt(task, "selfcheck", pi)
unchecked_retry = dispatch.base_prompt(
task, "selfcheck", pi, unchecked_items=True
)
review = dispatch.base_prompt(
task,
"review",
@ -3477,13 +3480,46 @@ class ReviewControlTest(unittest.TestCase):
self.assertNotIn("user review", selfcheck.lower())
self.assertEqual(
selfcheck,
dispatch.dispatcher_child_prompt(
f"Think in English. Keep artifact content in English. Final in Korean. Read {task.review.resolve()} "
"and fill every missing implementation field. Do not finish until "
"all implementation fields are complete. This is a self-check of "
f"completed work, not a review. Read {task.plan.resolve()} and "
"finish any missing work. Recheck and fix your work."
f"{dispatch.SELF_CHECK_PROMPT_PREFIX} Read "
f"{task.plan.resolve()}; review all work once, fix omissions, "
f"and update {task.review.resolve()}. Keep files in English.",
)
self.assertEqual(
unchecked_retry,
f"{dispatch.SELF_CHECK_PROMPT_PREFIX} Read "
f"{task.plan.resolve()}; complete every unchecked implementation "
f"item and update {task.review.resolve()}. Keep files in English.",
)
self.assertIn(str(task.plan.resolve()), unchecked_retry)
self.assertNotIn("dispatcher child", selfcheck.lower())
self.assertEqual(
dispatch.continuation_prompt(
task,
"selfcheck",
local_pi=True,
unchecked_items=True,
),
unchecked_retry,
)
self.assertEqual(
dispatch.continuation_prompt(
task,
"selfcheck",
local_pi=True,
resume_same_pi_session=True,
unchecked_items=True,
),
unchecked_retry,
)
self.assertEqual(
dispatch.continuation_prompt(
task,
"selfcheck",
local_pi=True,
resume_same_pi_session=True,
),
f"{dispatch.SELF_CHECK_PROMPT_PREFIX} Continue. Keep files in "
"English.",
)
self.assertEqual(
review,
@ -4313,18 +4349,16 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase):
}
store.update_task(
task,
selfcheck_incomplete=dispatch.SELF_CHECK_INCOMPLETE_LIMIT,
selfcheck_incomplete=(
dispatch.SELF_CHECK_UNCHECKED_RETRY_LIMIT + 1
),
completing_decision=completing_decision,
)
try:
with mock.patch.object(
dispatch, "run_escalating", new=mock.AsyncMock()
) as run_escalating:
await dispatch.run_selfcheck(
root,
store,
task,
)
await dispatch.run_selfcheck(root, store, task)
self.assertEqual(run_escalating.await_count, 0)
self.assertIn(
"limit already exhausted", store.task_state(task)["blocked"]
@ -4354,7 +4388,7 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase):
finally:
store.close()
async def test_selfcheck_incomplete_tenth_pass_blocks_task(self):
async def test_selfcheck_tenth_checklist_retry_blocks_task(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root / ".git").mkdir()
@ -4372,7 +4406,7 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase):
}
store.update_task(
task,
selfcheck_incomplete=dispatch.SELF_CHECK_INCOMPLETE_LIMIT - 1,
selfcheck_incomplete=dispatch.SELF_CHECK_UNCHECKED_RETRY_LIMIT,
completing_decision=completing_decision,
)
locator = root / "locator.json"
@ -4390,18 +4424,247 @@ class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase):
),
):
await dispatch.run_selfcheck(
root,
store,
task,
root, store, task, resume_locator=locator
)
self.assertEqual(run_escalating.await_count, 1)
self.assertTrue(
run_escalating.await_args.kwargs["unchecked_items"]
)
self.assertIn(
"selfcheck implementation fields remain incomplete",
"selfcheck checklist remains incomplete",
store.task_state(task)["blocked"],
)
finally:
store.close()
async def test_selfcheck_switches_to_unchecked_items_after_full_pass(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root / ".git").mkdir()
task = TaskStageTest().make_task(root)
store = dispatch.StateStore(root)
completing_decision = {
"work_unit_id": "test::plan-0::tag-TEST",
"stage": "worker",
"selected": {
"adapter": "pi",
"target": "iop/ornith:35b",
"execution_class": "local_model",
"selfcheck_required": True,
},
}
store.update_task(task, completing_decision=completing_decision)
locators = [root / "full-locator.json", root / "retry-locator.json"]
try:
with (
mock.patch.object(
dispatch,
"run_escalating",
new=mock.AsyncMock(
side_effect=[
(True, locators[0]),
(True, locators[1]),
]
),
) as run_escalating,
mock.patch.object(
dispatch,
"implementation_review_errors",
side_effect=[["구현 체크리스트 미완료"], []],
),
):
await dispatch.run_selfcheck(root, store, task)
self.assertEqual(run_escalating.await_count, 2)
self.assertFalse(
run_escalating.await_args_list[0].kwargs["unchecked_items"]
)
self.assertTrue(
run_escalating.await_args_list[1].kwargs["unchecked_items"]
)
self.assertIsNone(
run_escalating.await_args_list[0].kwargs[
"initial_resume_locator"
]
)
self.assertEqual(
run_escalating.await_args_list[1].kwargs[
"initial_resume_locator"
],
locators[0],
)
state = store.task_state(task)
self.assertTrue(state["selfcheck_done"])
self.assertEqual(state["selfcheck_incomplete"], 0)
self.assertIsNone(state["selfcheck_context_locator"])
finally:
store.close()
async def test_selfcheck_restart_resumes_persisted_context(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root / ".git").mkdir()
task = TaskStageTest().make_task(root)
store = dispatch.StateStore(root)
completing_decision = {
"work_unit_id": "test::plan-0::tag-TEST",
"stage": "worker",
"selected": {
"adapter": "pi",
"target": "iop/ornith:35b",
"execution_class": "local_model",
"selfcheck_required": True,
},
}
attempt = store.runs / "prior-selfcheck"
attempt.mkdir()
native = attempt / "session.jsonl"
native.write_text("{}\n", encoding="utf-8")
locator = attempt / "locator.json"
locator.write_text(
json.dumps(
{
"workspace": str(root.resolve()),
"workspace_id": store.workspace_id,
"task": task.name,
"role": "selfcheck",
"cli": "pi",
"status": "succeeded",
"native_session_path": str(native),
}
),
encoding="utf-8",
)
store.update_task(
task,
completing_decision=completing_decision,
selfcheck_incomplete=1,
selfcheck_context_locator=str(locator),
)
retry_locator = root / "retry-locator.json"
try:
with (
mock.patch.object(
dispatch,
"run_escalating",
new=mock.AsyncMock(return_value=(True, retry_locator)),
) as run_escalating,
mock.patch.object(
dispatch, "implementation_review_errors", return_value=[]
),
):
await dispatch.run_selfcheck(root, store, task)
self.assertEqual(run_escalating.await_count, 1)
self.assertTrue(
run_escalating.await_args.kwargs["unchecked_items"]
)
self.assertEqual(
run_escalating.await_args.kwargs["initial_resume_locator"],
locator,
)
finally:
store.close()
async def test_selfcheck_restart_blocks_without_persisted_context(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root / ".git").mkdir()
task = TaskStageTest().make_task(root)
store = dispatch.StateStore(root)
completing_decision = {
"work_unit_id": "test::plan-0::tag-TEST",
"stage": "worker",
"selected": {
"adapter": "pi",
"target": "iop/ornith:35b",
"execution_class": "local_model",
"selfcheck_required": True,
},
}
store.update_task(
task,
completing_decision=completing_decision,
selfcheck_incomplete=1,
)
try:
with mock.patch.object(
dispatch, "run_escalating", new=mock.AsyncMock()
) as run_escalating:
await dispatch.run_selfcheck(root, store, task)
self.assertEqual(run_escalating.await_count, 0)
self.assertIn(
"selfcheck context resume 실패",
store.task_state(task)["blocked"],
)
finally:
store.close()
async def test_selfcheck_allows_ten_unchecked_item_retries(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root / ".git").mkdir()
task = TaskStageTest().make_task(root)
store = dispatch.StateStore(root)
completing_decision = {
"work_unit_id": "test::plan-0::tag-TEST",
"stage": "worker",
"selected": {
"adapter": "pi",
"target": "iop/ornith:35b",
"execution_class": "local_model",
"selfcheck_required": True,
},
}
store.update_task(task, completing_decision=completing_decision)
locator = root / "locator.json"
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",
return_value=["구현 체크리스트 미완료"],
),
):
await dispatch.run_selfcheck(root, store, task)
self.assertEqual(
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:]
)
)
self.assertTrue(
all(
call.kwargs["initial_resume_locator"] == locator
for call in run_escalating.await_args_list[1:]
)
)
state = store.task_state(task)
self.assertEqual(
state["selfcheck_incomplete"],
1 + dispatch.SELF_CHECK_UNCHECKED_RETRY_LIMIT,
)
self.assertIn(
"selfcheck checklist remains incomplete",
state["blocked"],
)
finally:
store.close()
async def test_review_tenth_no_progress_pass_blocks_task(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
@ -11573,6 +11836,9 @@ class ArtifactLanguageContractTest(unittest.TestCase):
"worker": dispatch.base_prompt(task, "worker", codex),
"pi_worker": dispatch.base_prompt(task, "worker", pi),
"selfcheck": dispatch.base_prompt(task, "selfcheck", pi),
"selfcheck_unchecked": dispatch.base_prompt(
task, "selfcheck", pi, unchecked_items=True
),
"official_review": dispatch.base_prompt(task, "review", codex),
"review_without_stub": dispatch.base_prompt(
review_missing, "review", codex
@ -11588,6 +11854,22 @@ class ArtifactLanguageContractTest(unittest.TestCase):
"pi_selfcheck_continuation": dispatch.continuation_prompt(
task, "selfcheck", local_pi=True
),
"pi_selfcheck_unchecked_continuation": (
dispatch.continuation_prompt(
task,
"selfcheck",
local_pi=True,
unchecked_items=True,
)
),
"pi_selfcheck_native_continuation": (
dispatch.continuation_prompt(
task,
"selfcheck",
local_pi=True,
resume_same_pi_session=True,
)
),
"worker_continuation": dispatch.continuation_prompt(
task, "worker", locator
),
@ -11600,15 +11882,34 @@ class ArtifactLanguageContractTest(unittest.TestCase):
)
),
}
concise_selfcheck_prompts = {
"selfcheck",
"selfcheck_unchecked",
"pi_selfcheck_continuation",
"pi_selfcheck_unchecked_continuation",
"pi_selfcheck_native_continuation",
}
for name, prompt in prompts.items():
with self.subTest(prompt=name):
self.assertTrue(
prompt.startswith(
dispatch.DISPATCHER_CHILD_BOUNDARY_PROMPT
dispatch.SELF_CHECK_PROMPT_PREFIX
if name in concise_selfcheck_prompts
else dispatch.DISPATCHER_CHILD_BOUNDARY_PROMPT
)
)
self.assertIn("Keep artifact content in English.", prompt)
self.assertIn("Final in Korean.", prompt)
if name in concise_selfcheck_prompts:
self.assertIn("Keep files in English.", prompt)
self.assertTrue(
prompt.startswith(
"Think in English. Final in Korean."
)
)
else:
self.assertIn(
"Keep artifact content in English.", prompt
)
self.assertIn("Final in Korean.", prompt)
self.assertIn(
"`IOP_AGENT_TASK_EXECUTION_ID` is present",
@ -11618,6 +11919,10 @@ class ArtifactLanguageContractTest(unittest.TestCase):
dispatch.DISPATCHER_CHILD_BOUNDARY_PROMPT,
orchestrator_skill,
)
self.assertIn(
dispatch.SELF_CHECK_PROMPT_PREFIX,
orchestrator_skill,
)
self.assertIn(
"You may run dispatch.py --validate-plan only when required by "
"plan or code-review finalization",