sync: agent-ops from agentic-framework v1.1.197
This commit is contained in:
parent
5519013e85
commit
01c44b1431
7 changed files with 680 additions and 152 deletions
|
|
@ -1 +1 @@
|
|||
1.1.196
|
||||
1.1.197
|
||||
|
|
|
|||
|
|
@ -26,6 +26,7 @@ Monitor the file-backed workflow under `agent-task/` and converge ready PLAN imp
|
|||
- `dry_run`: inspect routes, dependencies, claims, and catalog validity without launching an agent.
|
||||
- `max_parallel`: workspace-wide active task-stage limit; defaults to `3`; `0` means unlimited.
|
||||
- `retry_blocked`: retry eligible blocked tasks without changing their catalog route history.
|
||||
- `accept_catalog_revision`: explicit operator acceptance that replaces only a stale, inactive, incomplete worker selection in the required `task_group`; dry-run previews the reset and live execution records it in audit history.
|
||||
|
||||
`--validate-plan` validates one PLAN without launching orchestration and therefore does not require an execution catalog.
|
||||
|
||||
|
|
@ -66,7 +67,7 @@ Before work starts, the dispatcher:
|
|||
4. runs an optional target `preflight_command` for live execution;
|
||||
5. records the catalog source and SHA-256 revision in the decision.
|
||||
|
||||
A persisted decision is valid only while the resolved catalog revision and selected target snapshot still match. Catalog changes fail closed instead of silently changing an active work unit.
|
||||
A persisted decision is valid only while the resolved catalog revision and selected target snapshot still match. Catalog changes fail closed instead of silently changing an active work unit. After an intentional catalog update, `--accept-catalog-revision` may reset only an inactive worker decision that has not completed; it requires `--task-group`, refuses live attempts and review decisions, clears failures charged to the obsolete worker selection, and preserves an audit record of the prior decision. Use it with `--dry-run` first.
|
||||
|
||||
## Selection and failover
|
||||
|
||||
|
|
@ -75,7 +76,7 @@ A persisted decision is valid only while the resolved catalog revision and selec
|
|||
- The dispatcher never queries quota before admission and never accepts a quota snapshot as selector input.
|
||||
- Classify actual terminal output after an attempt. `provider-quota`, `context-limit`, `model-unavailable`, `provider-stream-disconnect`, and `provider-connection` may advance to the next unused route candidate.
|
||||
- In particular, a confirmed quota/rate-limit error advances directly to the next candidate. A plain mention of quota in source text, model prose, or non-terminal output is not sufficient evidence.
|
||||
- `generic-error`, process termination, work-log failure, and review-control failure do not imply quota and do not change the selected target.
|
||||
- `generic-error` does not imply quota. Persist its count per work-unit, stage, and selected target. Retry the same target for the first two generic failures; on the third, move to the next unused candidate immediately. If no candidate remains, block at `3/3` instead of spending the 10-failure stage budget. Process termination, work-log failure, and review-control failure do not change the selected target.
|
||||
- Never use a hidden promotion table or provider-specific fallback. If no next catalog candidate exists, keep recovery within the stage budget or block the task with evidence.
|
||||
- Transfer logical context using the prior locator, normalized output, raw stream, workspace, and PLAN. Use native resume only when both targets opt into the same catalog-declared native-session mechanism and the session belongs to the current workspace.
|
||||
|
||||
|
|
@ -100,6 +101,8 @@ 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.
|
||||
|
||||
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.
|
||||
|
||||
## Runtime evidence and recovery
|
||||
|
|
@ -107,7 +110,7 @@ Accept self-check completion only when `## Implementation Checklist` or its supp
|
|||
- Store each attempt under the dispatcher state directory with `locator.json`, `stream.log`, `normalized-output.log`, and `heartbeat.log`.
|
||||
- Record the target id, opaque agent/model identity, execution class, runtime contract, catalog evidence, process identity, workspace identity, timestamps, result, and exact failure evidence.
|
||||
- Treat stderr as terminal diagnostic evidence. For JSONL, recognize generic terminal event fields such as error/fatal type or severity, rejected/failed status with an error code, explicit error flags, and a non-retrying `agent_end` whose last assistant message ends with `error` or `aborted`.
|
||||
- Determine liveness from PID/start-token/process-marker evidence and actual stream or native-session progress. Heartbeat mtime is never agent progress.
|
||||
- Determine liveness from PID/start-token/process-marker evidence and actual stream or native-session progress. Heartbeat mtime is never agent progress. For Codex JSONL, an unmatched `item.started` `command_execution` is an active tool interval: suspend the model-response silence timer until its matching `item.completed`, then restore normal stall detection.
|
||||
- Never start a duplicate attempt while owned live evidence remains.
|
||||
- Keep a 10-consecutive-failure budget per task stage. Reset only that stage's budget after success.
|
||||
- Preserve failed attempt logs. Delete successful attempt logs only after verified archive completion and no live evidence.
|
||||
|
|
@ -129,6 +132,18 @@ python3 agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py
|
|||
|
||||
Remove `--dry-run` to start execution. Add `--execution-catalog <path>` only to override the bundled default. Add `--task-group <name>`, `--max-parallel <n>`, or `--retry-blocked` only when requested by the workflow.
|
||||
|
||||
After an intentional catalog replacement invalidates a persisted incomplete worker decision, preview and accept it explicitly:
|
||||
|
||||
```bash
|
||||
python3 agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py \
|
||||
--workspace /absolute/repository \
|
||||
--task-group <group> \
|
||||
--accept-catalog-revision \
|
||||
--dry-run
|
||||
```
|
||||
|
||||
Remove `--dry-run` only after the preview shows the intended task scope.
|
||||
|
||||
Launch the live dispatcher as one persistent foreground process. Do not wrap it in an arbitrary timeout and do not start a second dispatcher after a normal tool yield. Wait on the same execution handle until an attention event or terminal exit.
|
||||
|
||||
## Completion checklist
|
||||
|
|
|
|||
|
|
@ -50,72 +50,6 @@
|
|||
"terminal_success": "agent_end"
|
||||
}
|
||||
},
|
||||
"agy-gemini-low": {
|
||||
"agent": "agy",
|
||||
"model": "Gemini 3.6 Flash (Low)",
|
||||
"execution_class": "cloud_model",
|
||||
"selfcheck_required": false,
|
||||
"runtime": {
|
||||
"command": [
|
||||
"agy",
|
||||
"--print",
|
||||
"{prompt}",
|
||||
"--print-timeout",
|
||||
"8h",
|
||||
"--model",
|
||||
"{model}",
|
||||
"--dangerously-skip-permissions",
|
||||
"--log-file",
|
||||
"{attempt_dir}/agy-cli.log"
|
||||
],
|
||||
"output_format": "text",
|
||||
"auxiliary_logs": ["{attempt_dir}/agy-cli.log"]
|
||||
}
|
||||
},
|
||||
"agy-gemini-medium": {
|
||||
"agent": "agy",
|
||||
"model": "Gemini 3.6 Flash (Medium)",
|
||||
"execution_class": "cloud_model",
|
||||
"selfcheck_required": false,
|
||||
"runtime": {
|
||||
"command": [
|
||||
"agy",
|
||||
"--print",
|
||||
"{prompt}",
|
||||
"--print-timeout",
|
||||
"8h",
|
||||
"--model",
|
||||
"{model}",
|
||||
"--dangerously-skip-permissions",
|
||||
"--log-file",
|
||||
"{attempt_dir}/agy-cli.log"
|
||||
],
|
||||
"output_format": "text",
|
||||
"auxiliary_logs": ["{attempt_dir}/agy-cli.log"]
|
||||
}
|
||||
},
|
||||
"agy-gemini-high": {
|
||||
"agent": "agy",
|
||||
"model": "Gemini 3.6 Flash (High)",
|
||||
"execution_class": "cloud_model",
|
||||
"selfcheck_required": false,
|
||||
"runtime": {
|
||||
"command": [
|
||||
"agy",
|
||||
"--print",
|
||||
"{prompt}",
|
||||
"--print-timeout",
|
||||
"8h",
|
||||
"--model",
|
||||
"{model}",
|
||||
"--dangerously-skip-permissions",
|
||||
"--log-file",
|
||||
"{attempt_dir}/agy-cli.log"
|
||||
],
|
||||
"output_format": "text",
|
||||
"auxiliary_logs": ["{attempt_dir}/agy-cli.log"]
|
||||
}
|
||||
},
|
||||
"opencode-glm-medium": {
|
||||
"agent": "opencode",
|
||||
"model": "glm-5.2",
|
||||
|
|
@ -194,31 +128,6 @@
|
|||
"output_format": "jsonl"
|
||||
}
|
||||
},
|
||||
"claude-opus-xhigh": {
|
||||
"agent": "claude",
|
||||
"model": "claude-opus-5",
|
||||
"reasoning_effort": "xhigh",
|
||||
"execution_class": "cloud_model",
|
||||
"selfcheck_required": false,
|
||||
"runtime": {
|
||||
"command": [
|
||||
"claude",
|
||||
"-p",
|
||||
"--output-format",
|
||||
"stream-json",
|
||||
"--verbose",
|
||||
"--session-id",
|
||||
"{session_id}",
|
||||
"--model",
|
||||
"{model}",
|
||||
"--effort",
|
||||
"{reasoning_effort}",
|
||||
"--dangerously-skip-permissions",
|
||||
"{prompt}"
|
||||
],
|
||||
"output_format": "jsonl"
|
||||
}
|
||||
},
|
||||
"codex-spark-xhigh": {
|
||||
"agent": "codex",
|
||||
"model": "gpt-5.3-codex-spark",
|
||||
|
|
@ -357,7 +266,7 @@
|
|||
"timezone": "Asia/Seoul",
|
||||
"start": "07:00",
|
||||
"end": "23:00",
|
||||
"candidates": ["agy-gemini-high", "opencode-glm-max", "codex-terra-high"],
|
||||
"candidates": ["opencode-glm-max", "codex-terra-high"],
|
||||
"rule_id": "worker-local-g07-kst-day-catalog",
|
||||
"reason_codes": ["worker_catalog_lane_kst_day"]
|
||||
},
|
||||
|
|
@ -365,7 +274,7 @@
|
|||
"timezone": "Asia/Seoul",
|
||||
"start": "23:00",
|
||||
"end": "07:00",
|
||||
"candidates": ["agy-gemini-high", "opencode-glm-max", "codex-terra-high"],
|
||||
"candidates": ["opencode-glm-max", "codex-terra-high"],
|
||||
"rule_id": "worker-local-g07-kst-night-catalog",
|
||||
"reason_codes": ["worker_catalog_lane_kst_night"]
|
||||
}
|
||||
|
|
@ -378,7 +287,7 @@
|
|||
"timezone": "Asia/Seoul",
|
||||
"start": "07:00",
|
||||
"end": "23:00",
|
||||
"candidates": ["agy-gemini-high", "opencode-glm-max", "codex-terra-high"],
|
||||
"candidates": ["opencode-glm-max", "codex-terra-high"],
|
||||
"rule_id": "worker-local-g08-kst-day-catalog",
|
||||
"reason_codes": ["worker_catalog_lane_kst_day"]
|
||||
},
|
||||
|
|
@ -386,68 +295,68 @@
|
|||
"timezone": "Asia/Seoul",
|
||||
"start": "23:00",
|
||||
"end": "07:00",
|
||||
"candidates": ["agy-gemini-high", "opencode-glm-max", "codex-terra-high"],
|
||||
"candidates": ["opencode-glm-max", "codex-terra-high"],
|
||||
"rule_id": "worker-local-g08-kst-night-catalog",
|
||||
"reason_codes": ["worker_catalog_lane_kst_night"]
|
||||
}
|
||||
]
|
||||
},
|
||||
"local-G09": {
|
||||
"candidates": ["claude-opus-xhigh", "codex-terra-high"],
|
||||
"candidates": ["codex-sol-xhigh", "codex-terra-high"],
|
||||
"rule_id": "worker-local-g09-catalog",
|
||||
"policy_priority": 30,
|
||||
"reason_codes": ["worker_catalog_lane"]
|
||||
},
|
||||
"local-G10": {
|
||||
"candidates": ["claude-opus-xhigh", "codex-terra-high"],
|
||||
"candidates": ["codex-sol-xhigh", "codex-terra-high"],
|
||||
"rule_id": "worker-local-g10-catalog",
|
||||
"policy_priority": 30,
|
||||
"reason_codes": ["worker_catalog_lane"]
|
||||
},
|
||||
"cloud-G01": {
|
||||
"candidates": ["codex-spark-xhigh", "agy-gemini-low", "opencode-glm-medium", "codex-terra-high"],
|
||||
"candidates": ["codex-spark-xhigh", "opencode-glm-medium", "codex-terra-high"],
|
||||
"rule_id": "worker-cloud-g01-catalog",
|
||||
"policy_priority": 30,
|
||||
"reason_codes": ["worker_catalog_lane"]
|
||||
},
|
||||
"cloud-G02": {
|
||||
"candidates": ["codex-spark-xhigh", "agy-gemini-low", "opencode-glm-medium", "codex-terra-high"],
|
||||
"candidates": ["codex-spark-xhigh", "opencode-glm-medium", "codex-terra-high"],
|
||||
"rule_id": "worker-cloud-g02-catalog",
|
||||
"policy_priority": 30,
|
||||
"reason_codes": ["worker_catalog_lane"]
|
||||
},
|
||||
"cloud-G03": {
|
||||
"candidates": ["agy-gemini-medium", "opencode-glm-high", "codex-terra-high"],
|
||||
"candidates": ["opencode-glm-high", "codex-terra-high"],
|
||||
"rule_id": "worker-cloud-g03-catalog",
|
||||
"policy_priority": 30,
|
||||
"reason_codes": ["worker_catalog_lane"]
|
||||
},
|
||||
"cloud-G04": {
|
||||
"candidates": ["agy-gemini-medium", "opencode-glm-high", "codex-terra-high"],
|
||||
"candidates": ["opencode-glm-high", "codex-terra-high"],
|
||||
"rule_id": "worker-cloud-g04-catalog",
|
||||
"policy_priority": 30,
|
||||
"reason_codes": ["worker_catalog_lane"]
|
||||
},
|
||||
"cloud-G05": {
|
||||
"candidates": ["agy-gemini-high", "opencode-glm-max", "codex-terra-high"],
|
||||
"candidates": ["opencode-glm-max", "codex-terra-high"],
|
||||
"rule_id": "worker-cloud-g05-catalog",
|
||||
"policy_priority": 30,
|
||||
"reason_codes": ["worker_catalog_lane"]
|
||||
},
|
||||
"cloud-G06": {
|
||||
"candidates": ["agy-gemini-high", "opencode-glm-max", "codex-terra-high"],
|
||||
"candidates": ["opencode-glm-max", "codex-terra-high"],
|
||||
"rule_id": "worker-cloud-g06-catalog",
|
||||
"policy_priority": 30,
|
||||
"reason_codes": ["worker_catalog_lane"]
|
||||
},
|
||||
"cloud-G07": {
|
||||
"candidates": ["claude-opus-xhigh", "codex-terra-high"],
|
||||
"candidates": ["codex-sol-high", "codex-terra-high"],
|
||||
"rule_id": "worker-cloud-g07-catalog",
|
||||
"policy_priority": 30,
|
||||
"reason_codes": ["worker_catalog_lane"]
|
||||
},
|
||||
"cloud-G08": {
|
||||
"candidates": ["claude-opus-xhigh", "codex-terra-high"],
|
||||
"candidates": ["codex-sol-high", "codex-terra-high"],
|
||||
"rule_id": "worker-cloud-g08-catalog",
|
||||
"policy_priority": 30,
|
||||
"reason_codes": ["worker_catalog_lane"]
|
||||
|
|
|
|||
|
|
@ -181,6 +181,7 @@ def validated_max_parallel(value: int) -> int:
|
|||
STREAM_HEARTBEAT_SECONDS = 30
|
||||
MODEL_RESPONSE_STALL_SECONDS = 3 * 60
|
||||
RECOVERY_FAILURE_LIMIT = 10
|
||||
GENERIC_FAILURE_LIMIT_PER_TARGET = 3
|
||||
SELF_CHECK_UNCHECKED_RETRY_LIMIT = 10
|
||||
REVIEW_NO_PROGRESS_LIMIT = 10
|
||||
PROVIDER_TRANSPORT_FAILURES = frozenset(
|
||||
|
|
@ -230,6 +231,8 @@ TARGET_FAILOVER_FAILURES = frozenset(
|
|||
)
|
||||
RECOVERABLE_RUNTIME_FAILURES = TARGET_FAILOVER_FAILURES | PROVIDER_TRANSPORT_FAILURES
|
||||
QUALIFIED_FAILOVER_FAILURES = RECOVERABLE_RUNTIME_FAILURES
|
||||
BOUNDED_TARGET_FAILURES = frozenset({"generic-error", "session-stall"})
|
||||
SELECTOR_FAILOVER_FAILURES = QUALIFIED_FAILOVER_FAILURES | BOUNDED_TARGET_FAILURES
|
||||
|
||||
|
||||
class DispatcherAlreadyRunning(RuntimeError):
|
||||
|
|
@ -845,6 +848,7 @@ class StateStore:
|
|||
"execution_decisions": {},
|
||||
"route_transition_history": [],
|
||||
"stage_failure_budgets": {},
|
||||
"generic_failure_budgets": {},
|
||||
"retry_failover_pending": False,
|
||||
"retry_failover_context": None,
|
||||
"blocker_evidence": None,
|
||||
|
|
@ -872,6 +876,7 @@ class StateStore:
|
|||
"recovery_failures": {},
|
||||
"execution_decisions": {},
|
||||
"route_transition_history": [],
|
||||
"generic_failure_budgets": {},
|
||||
"retry_failover_pending": False,
|
||||
"retry_failover_context": None,
|
||||
"blocker_evidence": None,
|
||||
|
|
@ -1027,9 +1032,114 @@ class StateStore:
|
|||
value["selfcheck_context_locator"] = None
|
||||
value["recovery_failures"] = {}
|
||||
value["stage_failure_budgets"] = {}
|
||||
value["generic_failure_budgets"] = {}
|
||||
value["retry_failover_pending"] = False
|
||||
self.save()
|
||||
|
||||
def accept_catalog_revision(
|
||||
self,
|
||||
tasks: list[Task],
|
||||
*,
|
||||
catalog_revision: str,
|
||||
catalog_source: str,
|
||||
persist: bool,
|
||||
) -> list[str]:
|
||||
"""Reset stale worker selection only after an explicit operator action."""
|
||||
accepted: list[str] = []
|
||||
staged: list[tuple[Task, dict[str, Any], dict[str, Any], str]] = []
|
||||
for task in tasks:
|
||||
state = self.data.get("tasks", {}).get(task.name)
|
||||
if not isinstance(state, dict) or state.get("plan_hash") != task.plan_hash:
|
||||
continue
|
||||
decisions = state.get("execution_decisions")
|
||||
if not isinstance(decisions, dict):
|
||||
continue
|
||||
mismatched: list[tuple[str, dict[str, Any]]] = []
|
||||
for stage, decision in decisions.items():
|
||||
if not isinstance(decision, dict):
|
||||
continue
|
||||
evidence = decision.get("catalog")
|
||||
prior_revision = (
|
||||
evidence.get("revision") if isinstance(evidence, dict) else None
|
||||
)
|
||||
if prior_revision and prior_revision != catalog_revision:
|
||||
mismatched.append((stage, decision))
|
||||
if not mismatched:
|
||||
continue
|
||||
if [stage for stage, _ in mismatched] != ["worker"]:
|
||||
raise DispatcherTerminalStateError(
|
||||
"catalog revision 수락은 미완료 worker 선택에만 허용된다: "
|
||||
f"task={task.name} stages={[stage for stage, _ in mismatched]}"
|
||||
)
|
||||
if state.get("worker_done"):
|
||||
raise DispatcherTerminalStateError(
|
||||
"완료된 worker의 catalog revision은 재선택할 수 없다: "
|
||||
f"task={task.name}"
|
||||
)
|
||||
active_live, active_detail = external_active_is_live(
|
||||
state,
|
||||
expected_workspace=self.workspace,
|
||||
expected_workspace_id=self.workspace_id,
|
||||
expected_runs_root=self.runs,
|
||||
)
|
||||
if active_live:
|
||||
raise DispatcherTerminalStateError(
|
||||
"실행 중 worker가 있어 catalog revision을 재선택할 수 없다: "
|
||||
f"task={task.name}; {active_detail}"
|
||||
)
|
||||
staged.append((task, state, mismatched[0][1], active_detail))
|
||||
|
||||
for task, state, prior_decision, active_detail in staged:
|
||||
decisions = dict(state.get("execution_decisions", {}))
|
||||
decisions.pop("worker", None)
|
||||
recovery_failures = dict(state.get("recovery_failures", {}))
|
||||
recovery_failures.pop("worker", None)
|
||||
stage_budgets = dict(state.get("stage_failure_budgets", {}))
|
||||
work_unit_id = prior_decision.get("work_unit_id")
|
||||
if isinstance(work_unit_id, str):
|
||||
stage_budgets.pop(f"{work_unit_id}|worker", None)
|
||||
generic_budgets = dict(state.get("generic_failure_budgets", {}))
|
||||
if isinstance(work_unit_id, str):
|
||||
prefix = f"{work_unit_id}|worker|"
|
||||
generic_budgets = {
|
||||
key: value
|
||||
for key, value in generic_budgets.items()
|
||||
if not key.startswith(prefix)
|
||||
}
|
||||
history = list(state.get("catalog_revision_acceptance_history", []))
|
||||
history.append(
|
||||
{
|
||||
"accepted_at": now_iso(),
|
||||
"stage": "worker",
|
||||
"work_unit_id": work_unit_id,
|
||||
"previous_catalog": prior_decision.get("catalog"),
|
||||
"previous_selected": prior_decision.get("selected"),
|
||||
"accepted_catalog": {
|
||||
"revision": catalog_revision,
|
||||
"source": catalog_source,
|
||||
},
|
||||
"stale_active_detail": active_detail,
|
||||
}
|
||||
)
|
||||
state.update(
|
||||
execution_decisions=decisions,
|
||||
catalog_revision_acceptance_history=history,
|
||||
recovery_failures=recovery_failures,
|
||||
stage_failure_budgets=stage_budgets,
|
||||
generic_failure_budgets=generic_budgets,
|
||||
retry_failover_pending=False,
|
||||
retry_failover_context=None,
|
||||
blocker_evidence=None,
|
||||
blocked=None,
|
||||
active_stage=None,
|
||||
active_locator=None,
|
||||
active_started_at=None,
|
||||
)
|
||||
accepted.append(task.name)
|
||||
if persist and accepted:
|
||||
self.save()
|
||||
return accepted
|
||||
|
||||
def mark_retry_failover(self, task_group: str | None = None, workspace: Path | None = None) -> None:
|
||||
prefix = f"{task_group}/" if task_group else None
|
||||
for task_name, value in self.data.get("tasks", {}).items():
|
||||
|
|
@ -1589,6 +1699,80 @@ class StageFailureBudget:
|
|||
self.store.update_task(self.task, stage_failure_budgets=budgets)
|
||||
|
||||
|
||||
@dataclass
|
||||
class GenericFailureBudget:
|
||||
"""Persistent generic-error counter scoped to one selected target."""
|
||||
|
||||
store: StateStore
|
||||
task: Task
|
||||
work_unit_id: str
|
||||
stage: str
|
||||
target_id: str
|
||||
|
||||
@classmethod
|
||||
def from_decision(
|
||||
cls,
|
||||
store: StateStore,
|
||||
task: Task,
|
||||
decision: dict[str, Any],
|
||||
) -> "GenericFailureBudget":
|
||||
work_unit_id = decision.get("work_unit_id")
|
||||
stage = decision.get("stage")
|
||||
selected = decision.get("selected")
|
||||
target_id = selected.get("target_id") if isinstance(selected, dict) else None
|
||||
if not all(
|
||||
isinstance(value, str) and value
|
||||
for value in (work_unit_id, stage, target_id)
|
||||
):
|
||||
raise ExecutionDecisionError(
|
||||
"generic failure budget identity가 유효하지 않다"
|
||||
)
|
||||
return cls(store, task, work_unit_id, stage, target_id)
|
||||
|
||||
@property
|
||||
def key(self) -> str:
|
||||
return f"{self.work_unit_id}|{self.stage}|{self.target_id}"
|
||||
|
||||
def _budgets(self) -> dict[str, Any]:
|
||||
state = self.store.task_state(self.task)
|
||||
budgets = state.get("generic_failure_budgets", {})
|
||||
if not isinstance(budgets, dict):
|
||||
raise ExecutionDecisionError(
|
||||
"persisted generic failure budgets schema가 유효하지 않다"
|
||||
)
|
||||
return dict(budgets)
|
||||
|
||||
def count(self) -> int:
|
||||
entry = self._budgets().get(self.key, {})
|
||||
if not isinstance(entry, dict):
|
||||
raise ExecutionDecisionError(
|
||||
"persisted generic failure budget entry가 유효하지 않다"
|
||||
)
|
||||
return int(entry.get("count", 0))
|
||||
|
||||
def record_failure(self) -> int:
|
||||
budgets = self._budgets()
|
||||
entry = dict(budgets.get(self.key, {}))
|
||||
count = int(entry.get("count", 0)) + 1
|
||||
entry.update(
|
||||
work_unit_id=self.work_unit_id,
|
||||
stage=self.stage,
|
||||
target_id=self.target_id,
|
||||
count=count,
|
||||
)
|
||||
budgets[self.key] = entry
|
||||
self.store.update_task(self.task, generic_failure_budgets=budgets)
|
||||
return count
|
||||
|
||||
def reset_stage_on_success(self) -> None:
|
||||
budgets = self._budgets()
|
||||
prefix = f"{self.work_unit_id}|{self.stage}|"
|
||||
budgets = {
|
||||
key: value for key, value in budgets.items() if not key.startswith(prefix)
|
||||
}
|
||||
self.store.update_task(self.task, generic_failure_budgets=budgets)
|
||||
|
||||
|
||||
def scan_tasks(
|
||||
workspace: Path,
|
||||
task_group: str | None,
|
||||
|
|
@ -2294,6 +2478,11 @@ def task_stage(task: Task, state: dict[str, Any]) -> str:
|
|||
if verdict_from_text(text):
|
||||
return "review"
|
||||
if state.get("worker_done"):
|
||||
# A provider exit code is not implementation completion. Re-enter the
|
||||
# worker stage until the implementation-owned review contract is
|
||||
# actually materialized (or contains complete blocker evidence).
|
||||
if implementation_review_errors(task):
|
||||
return "worker"
|
||||
if not _completing_decision_is_valid(task, state):
|
||||
return "blocked"
|
||||
if completing_decision_requires_selfcheck(state) and not state.get("selfcheck_done"):
|
||||
|
|
@ -2550,6 +2739,27 @@ def collaboration_tool(line: str) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
def json_agent_command_transition(line: str) -> tuple[str, str] | None:
|
||||
"""Return the lifecycle transition for one Codex command execution item."""
|
||||
try:
|
||||
value = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
if not isinstance(value, dict):
|
||||
return None
|
||||
event_type = value.get("type")
|
||||
item = value.get("item")
|
||||
if (
|
||||
event_type not in {"item.started", "item.completed"}
|
||||
or not isinstance(item, dict)
|
||||
or item.get("type") != "command_execution"
|
||||
or not isinstance(item.get("id"), str)
|
||||
or not item["id"]
|
||||
):
|
||||
return None
|
||||
return ("started" if event_type == "item.started" else "completed", item["id"])
|
||||
|
||||
|
||||
async def terminate_process_group(
|
||||
process: asyncio.subprocess.Process,
|
||||
grace_seconds: float = 5,
|
||||
|
|
@ -3207,6 +3417,9 @@ async def invoke(
|
|||
prompt: str,
|
||||
resume_locator: Path | None = None,
|
||||
) -> tuple[int, str | None, Path]:
|
||||
worker_signature_before = (
|
||||
task_signature(workspace, task) if role == "worker" else None
|
||||
)
|
||||
attempt, identity = next_execution_identity(store, task, role)
|
||||
attempt_dir = store.runs / f"{datetime.now(KST).strftime('%Y%m%dT%H%M%S%z')}__{identity}"
|
||||
attempt_dir.mkdir(parents=True, exist_ok=False)
|
||||
|
|
@ -3515,6 +3728,7 @@ async def invoke(
|
|||
last_stream_mtime: int | None = None
|
||||
last_native_progress_at = loop.time()
|
||||
last_stream_progress_at = loop.time()
|
||||
active_command_execution_ids: set[str] = set()
|
||||
with (
|
||||
stream_path.open("w", encoding="utf-8") as stream_log,
|
||||
normalized_output_path.open("w", encoding="utf-8") as normalized_output_log,
|
||||
|
|
@ -3566,6 +3780,7 @@ async def invoke(
|
|||
)
|
||||
native_phase = native_state.phase
|
||||
is_native_tool_execution = native_phase == "tool-running"
|
||||
is_command_execution = bool(active_command_execution_ids)
|
||||
# Outside a toolCall->toolResult interval, model stdout/stderr
|
||||
# is the liveness signal. A completed tool result changes phase
|
||||
# but must not reset the model-response silence clock.
|
||||
|
|
@ -3593,6 +3808,7 @@ async def invoke(
|
|||
if (
|
||||
spec.native_resume
|
||||
and not is_native_tool_execution
|
||||
and not is_command_execution
|
||||
and native_inactive_seconds >= MODEL_RESPONSE_STALL_SECONDS
|
||||
and session_stall_seconds is None
|
||||
):
|
||||
|
|
@ -3618,6 +3834,7 @@ async def invoke(
|
|||
)
|
||||
if (
|
||||
not spec.native_resume
|
||||
and not is_command_execution
|
||||
and non_native_inactive_seconds
|
||||
>= MODEL_RESPONSE_STALL_SECONDS
|
||||
and session_stall_seconds is None
|
||||
|
|
@ -3649,6 +3866,11 @@ async def invoke(
|
|||
f" native_activity={record.get('native_activity_state')}"
|
||||
f" native_phase={native_phase}"
|
||||
)
|
||||
if active_command_execution_ids:
|
||||
heartbeat += (
|
||||
" command_execution_ids="
|
||||
+ ",".join(sorted(active_command_execution_ids))
|
||||
)
|
||||
heartbeat_log.write(f"[heartbeat] {heartbeat}\n")
|
||||
heartbeat_log.flush()
|
||||
persist_locator_record()
|
||||
|
|
@ -3665,6 +3887,17 @@ async def invoke(
|
|||
line = raw.decode("utf-8", errors="replace").rstrip("\n")
|
||||
stream_log.write(f"[{channel}] {line}\n")
|
||||
stream_log.flush()
|
||||
if channel == "stdout":
|
||||
command_transition = json_agent_command_transition(line)
|
||||
if command_transition is not None:
|
||||
transition, command_id = command_transition
|
||||
if transition == "started":
|
||||
active_command_execution_ids.add(command_id)
|
||||
else:
|
||||
active_command_execution_ids.discard(command_id)
|
||||
record["active_command_execution_ids"] = sorted(
|
||||
active_command_execution_ids
|
||||
)
|
||||
if (
|
||||
channel == "stdout"
|
||||
and json_agent_terminal_outcome_from_line(line) == "succeeded"
|
||||
|
|
@ -3827,6 +4060,23 @@ async def invoke(
|
|||
failure_source = "cli-terminal-diagnostic"
|
||||
elif return_code != 0:
|
||||
failure_source = "cli-exit"
|
||||
if (
|
||||
role == "worker"
|
||||
and task.review is not None
|
||||
and return_code == 0
|
||||
and failure_class is None
|
||||
):
|
||||
worker_errors = implementation_review_errors(task)
|
||||
worker_signature_after = task_signature(workspace, task)
|
||||
if worker_signature_before == worker_signature_after or worker_errors:
|
||||
failure_class = "generic-error"
|
||||
failure_source = "dispatcher-worker-completion-contract"
|
||||
details = []
|
||||
if worker_signature_before == worker_signature_after:
|
||||
details.append("no claimed file or implementation evidence changed")
|
||||
details.extend(worker_errors)
|
||||
failure_evidence = "; ".join(details)
|
||||
failure_evidence_source = "dispatcher:worker-completion-contract"
|
||||
try:
|
||||
append_milestone_event(
|
||||
task,
|
||||
|
|
@ -4163,6 +4413,14 @@ async def run_escalating(
|
|||
store.update_task(task, recovery_failures=persisted)
|
||||
if stage_budget is not None:
|
||||
stage_budget.reset_on_success()
|
||||
decisions = state.get("execution_decisions", {})
|
||||
completing = (
|
||||
decisions.get(role) if isinstance(decisions, dict) else None
|
||||
)
|
||||
if isinstance(completing, dict):
|
||||
GenericFailureBudget.from_decision(
|
||||
store, task, completing
|
||||
).reset_stage_on_success()
|
||||
return True, locator
|
||||
failure = failure or "generic-error"
|
||||
if failure in {
|
||||
|
|
@ -4231,7 +4489,19 @@ async def run_escalating(
|
|||
if isinstance(decisions, dict):
|
||||
current_decision = decisions.get(role)
|
||||
|
||||
if canonical_selector_failover_route(current_decision) and failure in QUALIFIED_FAILOVER_FAILURES:
|
||||
generic_failure_count = 0
|
||||
if failure in BOUNDED_TARGET_FAILURES and isinstance(current_decision, dict):
|
||||
generic_failure_count = GenericFailureBudget.from_decision(
|
||||
store, task, current_decision
|
||||
).record_failure()
|
||||
|
||||
generic_failover_ready = (
|
||||
failure in BOUNDED_TARGET_FAILURES
|
||||
and generic_failure_count >= GENERIC_FAILURE_LIMIT_PER_TARGET
|
||||
)
|
||||
if canonical_selector_failover_route(current_decision) and (
|
||||
failure in QUALIFIED_FAILOVER_FAILURES or generic_failover_ready
|
||||
):
|
||||
try:
|
||||
next_decision = select_execution_decision(
|
||||
task,
|
||||
|
|
@ -4269,6 +4539,41 @@ async def run_escalating(
|
|||
code = "no_failover_candidate"
|
||||
else:
|
||||
code = exc.__class__.__name__
|
||||
if generic_failover_ready and code == "no_failover_candidate":
|
||||
reason = (
|
||||
f"{role} generic failure limit exhausted for target: "
|
||||
f"{generic_failure_count}/"
|
||||
f"{GENERIC_FAILURE_LIMIT_PER_TARGET}"
|
||||
)
|
||||
selected = (
|
||||
current_decision.get("selected")
|
||||
if isinstance(current_decision, dict)
|
||||
else None
|
||||
)
|
||||
store.update_task(
|
||||
task,
|
||||
blocked=f"{reason} locator={locator}",
|
||||
blocker_evidence={
|
||||
"role": role,
|
||||
"failure_class": failure,
|
||||
"locator": str(locator) if locator else None,
|
||||
"selected": selected,
|
||||
"work_unit_id": current_decision.get("work_unit_id")
|
||||
if isinstance(current_decision, dict)
|
||||
else None,
|
||||
},
|
||||
)
|
||||
banner(
|
||||
"작업차단",
|
||||
task.name,
|
||||
[
|
||||
"reason=generic-failure-limit",
|
||||
*failure_report_lines(failure, locator),
|
||||
f"retry={generic_failure_count}/"
|
||||
f"{GENERIC_FAILURE_LIMIT_PER_TARGET}",
|
||||
],
|
||||
)
|
||||
return False, locator
|
||||
store.update_task(
|
||||
task, blocked=f"{role} selector decision 실패 [{code}]: {exc}"
|
||||
)
|
||||
|
|
@ -4278,6 +4583,40 @@ async def run_escalating(
|
|||
[f"reason={code}", *failure_report_lines(failure, locator)],
|
||||
)
|
||||
return False, locator
|
||||
if generic_failover_ready:
|
||||
reason = (
|
||||
f"{role} generic failure limit exhausted for target: "
|
||||
f"{generic_failure_count}/{GENERIC_FAILURE_LIMIT_PER_TARGET}"
|
||||
)
|
||||
selected = (
|
||||
current_decision.get("selected")
|
||||
if isinstance(current_decision, dict)
|
||||
else None
|
||||
)
|
||||
store.update_task(
|
||||
task,
|
||||
blocked=f"{reason} locator={locator}",
|
||||
blocker_evidence={
|
||||
"role": role,
|
||||
"failure_class": failure,
|
||||
"locator": str(locator) if locator else None,
|
||||
"selected": selected,
|
||||
"work_unit_id": current_decision.get("work_unit_id")
|
||||
if isinstance(current_decision, dict)
|
||||
else None,
|
||||
},
|
||||
)
|
||||
banner(
|
||||
"작업차단",
|
||||
task.name,
|
||||
[
|
||||
"reason=generic-failure-limit",
|
||||
*failure_report_lines(failure, locator),
|
||||
f"retry={generic_failure_count}/"
|
||||
f"{GENERIC_FAILURE_LIMIT_PER_TARGET}",
|
||||
],
|
||||
)
|
||||
return False, locator
|
||||
if spec.native_resume:
|
||||
if failure in {"context-limit", "session-stall"}:
|
||||
native_recovery_retries += 1
|
||||
|
|
@ -4316,7 +4655,7 @@ async def run_escalating(
|
|||
previous_locator = locator
|
||||
await asyncio.sleep(min(30, 2 ** min(native_recovery_retries, 5)))
|
||||
continue
|
||||
if failure == "generic-error":
|
||||
if failure in BOUNDED_TARGET_FAILURES:
|
||||
generic_retries += 1
|
||||
banner(
|
||||
"작업복구재시도",
|
||||
|
|
@ -4324,7 +4663,8 @@ async def run_escalating(
|
|||
[
|
||||
f"model={spec.display}",
|
||||
*failure_report_lines(failure, locator),
|
||||
f"retry={recovery_failures}/{RECOVERY_FAILURE_LIMIT}",
|
||||
f"retry={generic_failure_count}/"
|
||||
f"{GENERIC_FAILURE_LIMIT_PER_TARGET}",
|
||||
],
|
||||
)
|
||||
previous_locator = locator
|
||||
|
|
@ -5090,6 +5430,7 @@ def _mark_worker_done(
|
|||
completing_decision=validated_decision,
|
||||
execution_class=execution_class,
|
||||
selfcheck_done=not selected["selfcheck_required"],
|
||||
review_no_progress=0,
|
||||
blocked=None,
|
||||
)
|
||||
|
||||
|
|
@ -5608,6 +5949,7 @@ async def dispatch_with_store(
|
|||
max_parallel = validated_max_parallel(
|
||||
getattr(args, "max_parallel", DEFAULT_MAX_PARALLEL)
|
||||
)
|
||||
catalog_revision_accepted = False
|
||||
|
||||
while True:
|
||||
if task_cache is None:
|
||||
|
|
@ -5615,6 +5957,28 @@ async def dispatch_with_store(
|
|||
task_cache = {task.name: task for task in tasks}
|
||||
else:
|
||||
tasks = sorted(task_cache.values(), key=lambda task: (task.index, task.name))
|
||||
if (
|
||||
getattr(args, "accept_catalog_revision", False)
|
||||
and not catalog_revision_accepted
|
||||
):
|
||||
selector = _selector_module()
|
||||
catalog = selector.load_runtime_catalog(EXECUTION_CATALOG_PATH)
|
||||
accepted = store.accept_catalog_revision(
|
||||
tasks,
|
||||
catalog_revision=catalog.revision,
|
||||
catalog_source=str(catalog.source),
|
||||
persist=not args.dry_run,
|
||||
)
|
||||
banner(
|
||||
"카탈로그재선택",
|
||||
args.task_group,
|
||||
[
|
||||
f"mode={'preview' if args.dry_run else 'accepted'}",
|
||||
f"catalog_revision={catalog.revision}",
|
||||
f"tasks={','.join(accepted) if accepted else 'none'}",
|
||||
],
|
||||
)
|
||||
catalog_revision_accepted = True
|
||||
if args.dry_run:
|
||||
persistent_errors: dict[str, str] = {}
|
||||
observed_tasks: set[str] = set()
|
||||
|
|
@ -6349,6 +6713,14 @@ def parse_args() -> argparse.Namespace:
|
|||
)
|
||||
parser.add_argument("--dry-run", action="store_true", help="classify and print without launching CLIs")
|
||||
parser.add_argument("--retry-blocked", action="store_true", help="clear dispatcher-local blocked state")
|
||||
parser.add_argument(
|
||||
"--accept-catalog-revision",
|
||||
action="store_true",
|
||||
help=(
|
||||
"explicitly replace stale, inactive, incomplete worker selection "
|
||||
"for the required --task-group after a catalog update"
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-parallel",
|
||||
type=int,
|
||||
|
|
@ -6370,6 +6742,12 @@ def parse_args() -> argparse.Namespace:
|
|||
def main() -> int:
|
||||
global EXECUTION_CATALOG_PATH
|
||||
args = parse_args()
|
||||
if args.accept_catalog_revision and not args.task_group:
|
||||
print(
|
||||
"dispatcher error: --accept-catalog-revision requires --task-group",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 2
|
||||
try:
|
||||
validated_max_parallel(
|
||||
getattr(args, "max_parallel", DEFAULT_MAX_PARALLEL)
|
||||
|
|
|
|||
|
|
@ -38,6 +38,8 @@ _HEADER_RE = re.compile(
|
|||
_STAGE_BY_KIND = {"PLAN": "worker", "CODE_REVIEW": "review"}
|
||||
_VALID_TRANSITIONS = {"initial", "resume", "failover"}
|
||||
_QUALIFIED_FAILOVER_FAILURES = {
|
||||
"generic-error",
|
||||
"session-stall",
|
||||
"provider-quota",
|
||||
"context-limit",
|
||||
"model-unavailable",
|
||||
|
|
|
|||
|
|
@ -275,6 +275,86 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
|
|||
self.assertEqual(second_spec.model, "model-alternate")
|
||||
self.assertNotIn("quota", failed)
|
||||
|
||||
def test_generic_failure_budget_is_persistent_per_target(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
catalog = write_catalog(root)
|
||||
plan = write_plan(root)
|
||||
task = task_from_plan(root, plan)
|
||||
dispatch.EXECUTION_CATALOG_PATH = catalog
|
||||
with mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}):
|
||||
store = dispatch.StateStore(root)
|
||||
try:
|
||||
first, _ = dispatch.persisted_execution_decision(
|
||||
store, task, stage="worker"
|
||||
)
|
||||
first_budget = dispatch.GenericFailureBudget.from_decision(
|
||||
store, task, first
|
||||
)
|
||||
self.assertEqual(first_budget.record_failure(), 1)
|
||||
self.assertEqual(first_budget.record_failure(), 2)
|
||||
self.assertEqual(first_budget.record_failure(), 3)
|
||||
second, _ = dispatch.persisted_execution_decision(
|
||||
store,
|
||||
task,
|
||||
stage="worker",
|
||||
transition="failover",
|
||||
failure_class="generic-error",
|
||||
)
|
||||
second_budget = dispatch.GenericFailureBudget.from_decision(
|
||||
store, task, second
|
||||
)
|
||||
self.assertEqual(second_budget.count(), 0)
|
||||
self.assertEqual(first_budget.count(), 3)
|
||||
finally:
|
||||
store.close()
|
||||
|
||||
def test_catalog_revision_acceptance_resets_only_stale_worker_state(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
catalog = write_catalog(root)
|
||||
plan = write_plan(root)
|
||||
task = task_from_plan(root, plan)
|
||||
dispatch.EXECUTION_CATALOG_PATH = catalog
|
||||
with mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}):
|
||||
store = dispatch.StateStore(root)
|
||||
try:
|
||||
decision, _ = dispatch.persisted_execution_decision(
|
||||
store, task, stage="worker"
|
||||
)
|
||||
state = store.task_state(task)
|
||||
state.update(
|
||||
active_stage="worker",
|
||||
active_locator="/missing/locator.json",
|
||||
recovery_failures={"worker": 2},
|
||||
stage_failure_budgets={
|
||||
f'{decision["work_unit_id"]}|worker': {"count": 2}
|
||||
},
|
||||
generic_failure_budgets={
|
||||
f'{decision["work_unit_id"]}|worker|primary': {"count": 2}
|
||||
},
|
||||
)
|
||||
store.save()
|
||||
accepted = store.accept_catalog_revision(
|
||||
[task],
|
||||
catalog_revision="replacement-revision",
|
||||
catalog_source="/runtime/replacement.json",
|
||||
persist=True,
|
||||
)
|
||||
state = store.task_state(task)
|
||||
finally:
|
||||
store.close()
|
||||
self.assertEqual(accepted, [task.name])
|
||||
self.assertNotIn("worker", state["execution_decisions"])
|
||||
self.assertNotIn("worker", state["recovery_failures"])
|
||||
self.assertEqual(state["stage_failure_budgets"], {})
|
||||
self.assertEqual(state["generic_failure_budgets"], {})
|
||||
self.assertIsNone(state["active_stage"])
|
||||
self.assertEqual(
|
||||
state["catalog_revision_acceptance_history"][-1]["previous_selected"],
|
||||
decision["selected"],
|
||||
)
|
||||
|
||||
def test_retry_blocked_marks_failover_without_quota_state(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
|
|
@ -491,6 +571,78 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
|
|||
self.assertEqual(record["status"], "succeeded")
|
||||
self.assertIn("succeeded:0", work_log)
|
||||
|
||||
def test_zero_exit_worker_without_implementation_progress_is_generic_error(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
|
||||
runner = root / "fake_json_runner.py"
|
||||
runner.write_text(
|
||||
"import json\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",
|
||||
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"))
|
||||
work_log = (
|
||||
root / "agent-task" / "group" / "WORK_LOG.md"
|
||||
).read_text(encoding="utf-8")
|
||||
finally:
|
||||
store.close()
|
||||
|
||||
self.assertEqual(return_code, 0)
|
||||
self.assertEqual(failure, "generic-error")
|
||||
self.assertEqual(
|
||||
record["failure_source"],
|
||||
"dispatcher-worker-completion-contract",
|
||||
)
|
||||
self.assertEqual(record["status"], "failed")
|
||||
self.assertNotIn("succeeded:0", work_log)
|
||||
|
||||
def test_worker_done_with_incomplete_evidence_returns_to_worker_stage(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
|
||||
|
||||
stage = dispatch.task_stage(task, {"worker_done": True})
|
||||
|
||||
self.assertEqual(stage, "worker")
|
||||
|
||||
def test_silent_native_session_is_terminated_and_classified_as_stall(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
|
|
@ -546,6 +698,93 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
|
|||
self.assertGreaterEqual(record["session_stall_seconds"], 0.05)
|
||||
self.assertIn("native_silence_inspection", record)
|
||||
|
||||
def test_non_native_active_command_execution_is_not_classified_as_stall(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
plan = write_plan(root)
|
||||
task = task_from_plan(root, plan)
|
||||
runner = root / "command_runner.py"
|
||||
runner.write_text(
|
||||
"import json, time\n"
|
||||
"item = {'id': 'command-1', 'type': 'command_execution'}\n"
|
||||
"print(json.dumps({'type': 'item.started', 'item': item}), flush=True)\n"
|
||||
"time.sleep(0.15)\n"
|
||||
"print(json.dumps({'type': 'item.completed', 'item': item}), flush=True)\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
agent = dispatch.AgentSpec(
|
||||
"command-runner",
|
||||
"command-model",
|
||||
"command-runner/command-model",
|
||||
target_id="command-target",
|
||||
runtime={
|
||||
"command": [sys.executable, str(runner)],
|
||||
"output_format": "jsonl",
|
||||
},
|
||||
)
|
||||
with (
|
||||
mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}),
|
||||
mock.patch.object(dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01),
|
||||
mock.patch.object(dispatch, "MODEL_RESPONSE_STALL_SECONDS", 0.05),
|
||||
):
|
||||
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.assertEqual(record["active_command_execution_ids"], [])
|
||||
self.assertNotIn("session_stall_seconds", record)
|
||||
|
||||
def test_completed_non_native_command_returns_to_stall_detection(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
plan = write_plan(root)
|
||||
task = task_from_plan(root, plan)
|
||||
runner = root / "completed_command_runner.py"
|
||||
runner.write_text(
|
||||
"import json, time\n"
|
||||
"item = {'id': 'command-1', 'type': 'command_execution'}\n"
|
||||
"print(json.dumps({'type': 'item.started', 'item': item}), flush=True)\n"
|
||||
"print(json.dumps({'type': 'item.completed', 'item': item}), flush=True)\n"
|
||||
"time.sleep(30)\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
agent = dispatch.AgentSpec(
|
||||
"completed-command-runner",
|
||||
"command-model",
|
||||
"completed-command-runner/command-model",
|
||||
target_id="command-target",
|
||||
runtime={
|
||||
"command": [sys.executable, str(runner)],
|
||||
"output_format": "jsonl",
|
||||
},
|
||||
)
|
||||
with (
|
||||
mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}),
|
||||
mock.patch.object(dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01),
|
||||
mock.patch.object(dispatch, "MODEL_RESPONSE_STALL_SECONDS", 0.05),
|
||||
):
|
||||
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.assertLess(return_code, 0)
|
||||
self.assertEqual(failure, "session-stall")
|
||||
self.assertEqual(record["active_command_execution_ids"], [])
|
||||
self.assertIn("stream_silence_inspection", record)
|
||||
|
||||
def test_catalog_source_is_in_runtime_audit_evidence(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
|
|
|
|||
|
|
@ -75,9 +75,10 @@ class SelectorTests(unittest.TestCase):
|
|||
with TemporaryDirectory() as tmp, mock.patch.dict(os.environ, {}, clear=True):
|
||||
task = write_task(Path(tmp))
|
||||
result = selector.select_execution_target(task)
|
||||
self.assertEqual(result["selected"]["target_id"], "agy-gemini-high")
|
||||
self.assertEqual(result["selected"]["agent"], "agy")
|
||||
self.assertEqual(result["selected"]["model"], "Gemini 3.6 Flash (High)")
|
||||
self.assertEqual(result["selected"]["target_id"], "opencode-glm-max")
|
||||
self.assertEqual(result["selected"]["agent"], "opencode")
|
||||
self.assertEqual(result["selected"]["model"], "glm-5.2")
|
||||
self.assertEqual(result["selected"]["reasoning_effort"], "high")
|
||||
self.assertEqual(
|
||||
result["catalog"]["source"],
|
||||
str(selector.DEFAULT_CATALOG_PATH.resolve()),
|
||||
|
|
@ -89,30 +90,26 @@ class SelectorTests(unittest.TestCase):
|
|||
f"local-G{grade:02d}": ["pi-ornith-high"]
|
||||
for grade in range(1, 7)
|
||||
},
|
||||
"local-G07": ["agy-gemini-high", "opencode-glm-max", "codex-terra-high"],
|
||||
"local-G08": ["agy-gemini-high", "opencode-glm-max", "codex-terra-high"],
|
||||
"local-G09": ["claude-opus-xhigh", "codex-terra-high"],
|
||||
"local-G10": ["claude-opus-xhigh", "codex-terra-high"],
|
||||
"cloud-G01": ["codex-spark-xhigh", "agy-gemini-low", "opencode-glm-medium", "codex-terra-high"],
|
||||
"cloud-G02": ["codex-spark-xhigh", "agy-gemini-low", "opencode-glm-medium", "codex-terra-high"],
|
||||
"cloud-G03": ["agy-gemini-medium", "opencode-glm-high", "codex-terra-high"],
|
||||
"cloud-G04": ["agy-gemini-medium", "opencode-glm-high", "codex-terra-high"],
|
||||
"cloud-G05": ["agy-gemini-high", "opencode-glm-max", "codex-terra-high"],
|
||||
"cloud-G06": ["agy-gemini-high", "opencode-glm-max", "codex-terra-high"],
|
||||
"cloud-G07": ["claude-opus-xhigh", "codex-terra-high"],
|
||||
"cloud-G08": ["claude-opus-xhigh", "codex-terra-high"],
|
||||
"local-G07": ["opencode-glm-max", "codex-terra-high"],
|
||||
"local-G08": ["opencode-glm-max", "codex-terra-high"],
|
||||
"local-G09": ["codex-sol-xhigh", "codex-terra-high"],
|
||||
"local-G10": ["codex-sol-xhigh", "codex-terra-high"],
|
||||
"cloud-G01": ["codex-spark-xhigh", "opencode-glm-medium", "codex-terra-high"],
|
||||
"cloud-G02": ["codex-spark-xhigh", "opencode-glm-medium", "codex-terra-high"],
|
||||
"cloud-G03": ["opencode-glm-high", "codex-terra-high"],
|
||||
"cloud-G04": ["opencode-glm-high", "codex-terra-high"],
|
||||
"cloud-G05": ["opencode-glm-max", "codex-terra-high"],
|
||||
"cloud-G06": ["opencode-glm-max", "codex-terra-high"],
|
||||
"cloud-G07": ["codex-sol-high", "codex-terra-high"],
|
||||
"cloud-G08": ["codex-sol-high", "codex-terra-high"],
|
||||
"cloud-G09": ["codex-sol-xhigh"],
|
||||
"cloud-G10": ["codex-sol-xhigh"],
|
||||
}
|
||||
expected_targets = {
|
||||
"pi-ornith-high",
|
||||
"agy-gemini-low",
|
||||
"agy-gemini-medium",
|
||||
"agy-gemini-high",
|
||||
"opencode-glm-medium",
|
||||
"opencode-glm-high",
|
||||
"opencode-glm-max",
|
||||
"claude-opus-xhigh",
|
||||
"codex-spark-xhigh",
|
||||
"codex-sol-high",
|
||||
"codex-sol-xhigh",
|
||||
|
|
@ -179,16 +176,10 @@ class SelectorTests(unittest.TestCase):
|
|||
self.assertIn("--session", pi.runtime["resume_command"])
|
||||
self.assertIn("{resume_session_dir}", pi.runtime["resume_command"])
|
||||
self.assertIn("{session_id}", pi.runtime["session_path"])
|
||||
agy = catalog.targets["agy-gemini-high"]
|
||||
self.assertEqual(agy.runtime["auxiliary_logs"], ["{attempt_dir}/agy-cli.log"])
|
||||
opencode = catalog.targets["opencode-glm-max"]
|
||||
self.assertIn("iop-glm/glm-5.2", opencode.runtime["command"])
|
||||
self.assertEqual(opencode.reasoning_effort, "high")
|
||||
self.assertIn("{reasoning_effort}", opencode.runtime["command"])
|
||||
claude = catalog.targets["claude-opus-xhigh"]
|
||||
self.assertEqual((claude.agent, claude.model), ("claude", "claude-opus-5"))
|
||||
self.assertEqual(claude.reasoning_effort, "xhigh")
|
||||
self.assertIn("{reasoning_effort}", claude.runtime["command"])
|
||||
terra = catalog.targets["codex-terra-high"]
|
||||
self.assertEqual(terra.reasoning_effort, "high")
|
||||
self.assertIn(
|
||||
|
|
@ -248,34 +239,28 @@ class SelectorTests(unittest.TestCase):
|
|||
self.assertEqual(second["transition"]["trigger"], "failover")
|
||||
self.assertEqual(second["transition"]["previous_target"]["target_id"], "first")
|
||||
|
||||
def test_failover_requires_runtime_failure_and_unused_candidate(self):
|
||||
def test_failover_accepts_bounded_failures_and_requires_unused_candidate(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
catalog = write_catalog(root)
|
||||
task = write_task(root)
|
||||
first = selector.select_execution_target(task, catalog_path=catalog)
|
||||
with self.assertRaises(selector.SelectorInputError) as ctx:
|
||||
selector.select_execution_target(
|
||||
for failure in ("generic-error", "session-stall"):
|
||||
first = selector.select_execution_target(task, catalog_path=catalog)
|
||||
second = selector.select_execution_target(
|
||||
task,
|
||||
catalog_path=catalog,
|
||||
transition="failover",
|
||||
prior_decision=first,
|
||||
failure_class="generic-error",
|
||||
failure_class=failure,
|
||||
)
|
||||
self.assertEqual(ctx.exception.code, "unqualified_failover")
|
||||
second = selector.select_execution_target(
|
||||
task,
|
||||
catalog_path=catalog,
|
||||
transition="failover",
|
||||
prior_decision=first,
|
||||
failure_class="model-unavailable",
|
||||
)
|
||||
self.assertEqual(second["selected"]["target_id"], "second")
|
||||
generic_second = second
|
||||
with self.assertRaises(selector.SelectorInputError) as ctx:
|
||||
selector.select_execution_target(
|
||||
task,
|
||||
catalog_path=catalog,
|
||||
transition="failover",
|
||||
prior_decision=second,
|
||||
prior_decision=generic_second,
|
||||
failure_class="provider-quota",
|
||||
)
|
||||
self.assertEqual(ctx.exception.code, "no_failover_candidate")
|
||||
|
|
@ -336,7 +321,7 @@ class SelectorTests(unittest.TestCase):
|
|||
self.assertEqual(completed.stderr, "")
|
||||
self.assertEqual(
|
||||
json.loads(completed.stdout)["selected"]["target_id"],
|
||||
"agy-gemini-high",
|
||||
"opencode-glm-max",
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue