iop/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_state.py

#!/usr/bin/env python3
"""Persisted execution-target snapshot validation helpers."""

from __future__ import annotations

from datetime import datetime


ERROR_CODE = "malformed_prior_decision"


def runtime_key(entry: dict) -> tuple[object, ...]:
    return (
        entry.get("adapter"),
        entry.get("target"),
        entry.get("thinking_level"),
        entry.get("reasoning_effort"),
    )


def route_key(target) -> tuple[object, ...]:
    return (
        target.adapter,
        target.target,
        target.thinking_level,
        target.reasoning_effort,
    )


def _validate_selected_target(prior_decision, canonical_targets, policy, error_type):
    selected = prior_decision.get("selected")
    if not isinstance(selected, dict):
        raise error_type(ERROR_CODE, "prior_decision.selected must be an object")
    selected_key = runtime_key(selected)
    matching = policy.canonical_target(*selected_key)
    if matching is None:
        raise error_type(
            ERROR_CODE,
            f"prior_decision.selected {selected_key} is not a policy-owned target",
        )
    if (
        selected.get("execution_class") != matching.execution_class
        or selected.get("selfcheck_required") != matching.selfcheck_required
    ):
        raise error_type(
            ERROR_CODE,
            f"prior_decision.selected attributes do not match canonical target for {selected_key}",
        )
    if isinstance(prior_decision.get("catalog"), dict) and (
        selected.get("target_id") != matching.catalog_id
        or selected.get("command_model") != matching.command_model
    ):
        raise error_type(
            ERROR_CODE,
            f"prior_decision.selected catalog attributes do not match canonical target for {selected_key}",
        )
    return selected_key, matching, [route_key(target) for target in canonical_targets]


def _validate_promotion_history(
    prior_decision, matching, canonical_keys, policy, error_type
):
    promotion_path = prior_decision.get("promotion_path")
    if not isinstance(promotion_path, list) or len(promotion_path) < 2:
        raise error_type(
            ERROR_CODE, "promoted prior_decision requires promotion_path evidence"
        )
    path_targets = []
    for index, entry in enumerate(promotion_path):
        if not isinstance(entry, dict):
            raise error_type(ERROR_CODE, f"promotion_path[{index}] must be an object")
        target = policy.canonical_target(*runtime_key(entry))
        if target is None:
            raise error_type(ERROR_CODE, f"promotion_path[{index}] is not policy-owned")
        path_targets.append(target)
    if route_key(path_targets[0]) not in set(canonical_keys):
        raise error_type(
            ERROR_CODE, "promotion_path must begin at the initial policy target"
        )
    for previous, current in zip(path_targets, path_targets[1:]):
        if policy.promotion_target(previous) != current:
            raise error_type(
                ERROR_CODE, "promotion_path contains a non-adjacent transition"
            )
    if path_targets[-1] != matching:
        raise error_type(ERROR_CODE, "promotion_path tail does not match selected target")
    if "used_candidates" in prior_decision:
        raise error_type(
            ERROR_CODE, "promotion decision must not carry failover used_candidates"
        )


def _validate_used_history(prior_decision, selected_key, canonical_keys, error_type):
    used = prior_decision["used_candidates"]
    if not isinstance(used, list):
        raise error_type(ERROR_CODE, "prior_decision.used_candidates must be a list")
    used_keys = []
    for index, entry in enumerate(used):
        if not isinstance(entry, dict):
            raise error_type(
                ERROR_CODE,
                f"prior_decision.used_candidates[{index}] must be an object",
            )
        key = runtime_key(entry)
        if key not in set(canonical_keys):
            raise error_type(
                ERROR_CODE,
                f"prior_decision.used_candidates[{index}] {key} is not in canonical policy targets {set(canonical_keys)}",
            )
        used_keys.append(key)
    if len(used_keys) != len(set(used_keys)):
        raise error_type(
            ERROR_CODE, "prior_decision.used_candidates contains duplicate targets"
        )
    if [canonical_keys.index(key) for key in used_keys] != sorted(
        canonical_keys.index(key) for key in used_keys
    ):
        raise error_type(
            ERROR_CODE,
            "prior_decision.used_candidates order does not match candidate rank order",
        )
    if used_keys and selected_key != used_keys[-1]:
        raise error_type(
            ERROR_CODE,
            f"prior_decision.selected {selected_key} does not match tail of used_candidates {used_keys[-1]}",
        )


def _validate_selected_and_history(
    prior_decision, canonical_targets, policy, error_type
):
    selected_key, matching, canonical_keys = _validate_selected_target(
        prior_decision, canonical_targets, policy, error_type
    )
    if selected_key not in set(canonical_keys):
        _validate_promotion_history(
            prior_decision, matching, canonical_keys, policy, error_type
        )
        return
    if "used_candidates" in prior_decision:
        _validate_used_history(
            prior_decision, selected_key, canonical_keys, error_type
        )
        return
    eligible = [
        runtime_key(candidate)
        for candidate in prior_decision.get("candidates", [])
        if isinstance(candidate, dict) and candidate.get("eligibility") == "eligible"
    ]
    if eligible and selected_key != eligible[0]:
        raise error_type(
            ERROR_CODE,
            f"prior_decision.selected {selected_key} does not match first eligible candidate {eligible[0]} when used_candidates is absent",
        )


def _validate_pinned_catalog_snapshot(
    prior_decision, error_type, validate_used_candidates
):
    candidates = prior_decision.get("candidates")
    selected = prior_decision.get("selected")
    if not isinstance(candidates, list) or not isinstance(selected, dict):
        raise error_type(ERROR_CODE, "pinned catalog snapshot is incomplete")
    keys = [runtime_key(candidate) for candidate in candidates]
    if len(keys) != len(set(keys)):
        raise error_type(ERROR_CODE, "pinned catalog snapshot has duplicate targets")
    selected_key = runtime_key(selected)
    if selected_key not in keys:
        raise error_type(
            ERROR_CODE, "pinned selected target is not present in the candidate snapshot"
        )
    selected_candidate = candidates[keys.index(selected_key)]
    identity_fields = (
        "target_id", "adapter", "target", "execution_class",
        "selfcheck_required", "thinking_level", "reasoning_effort", "command_model",
    )
    for field in identity_fields:
        if selected.get(field) != selected_candidate.get(field):
            raise error_type(
                ERROR_CODE,
                f"pinned selected.{field} does not match its candidate snapshot",
            )
    if "promotion_path" in prior_decision:
        raise error_type(
            ERROR_CODE,
            "catalog-backed decisions must express fallback in the lane candidate array",
        )
    if "used_candidates" not in prior_decision:
        eligible = [
            runtime_key(candidate)
            for candidate in candidates
            if candidate.get("eligibility") == "eligible"
        ]
        if not eligible or selected_key != eligible[0]:
            raise error_type(
                ERROR_CODE, "selected target is not the first eligible pinned candidate"
            )
        return
    used = validate_used_candidates(prior_decision.get("used_candidates"))
    used_keys = [runtime_key(entry) for entry in used]
    if len(used_keys) != len(set(used_keys)):
        raise error_type(ERROR_CODE, "used_candidates contains duplicate targets")
    if any(key not in keys for key in used_keys):
        raise error_type(
            ERROR_CODE, "used_candidates contains a target outside the pinned snapshot"
        )
    if [keys.index(key) for key in used_keys] != sorted(
        keys.index(key) for key in used_keys
    ):
        raise error_type(
            ERROR_CODE, "used_candidates order does not match the pinned candidate order"
        )
    if not used_keys or used_keys[-1] != selected_key:
        raise error_type(
            ERROR_CODE, "selected target does not match used_candidates tail"
        )


def _validate_catalog(catalog, expected_route_id, policy, error_type):
    if not isinstance(catalog, dict):
        return False
    if (
        catalog.get("schema_version") != policy.CATALOG_SCHEMA_VERSION
        or not isinstance(catalog.get("revision"), str)
        or not catalog.get("revision")
    ):
        raise error_type(
            ERROR_CODE,
            "prior_decision.catalog must contain the current schema_version and a non-empty revision",
        )
    if catalog.get("route_id") != expected_route_id:
        raise error_type(
            ERROR_CODE,
            f"prior_decision.catalog.route_id ({catalog.get('route_id')!r}) does not match {expected_route_id!r}",
        )
    return catalog.get("revision") != policy.CATALOG.revision


def _canonical_decision(decision_info, stage, lane, grade, policy, error_type):
    evaluated_at = decision_info.get("evaluated_at")
    if not isinstance(evaluated_at, str):
        raise error_type(
            ERROR_CODE, "prior_decision.decision.evaluated_at must be a string"
        )
    try:
        parsed = datetime.fromisoformat(evaluated_at)
    except (ValueError, TypeError) as exc:
        raise error_type(
            ERROR_CODE,
            f"prior_decision.decision.evaluated_at is not a valid ISO datetime: {evaluated_at!r}",
        ) from exc
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        raise error_type(
            ERROR_CODE,
            f"prior_decision.decision.evaluated_at must be timezone-aware: {evaluated_at!r}",
        )
    try:
        return policy.select_policy(
            stage=stage, lane=lane, grade=grade, evaluated_at=parsed
        )
    except ValueError as exc:
        raise error_type(ERROR_CODE, str(exc)) from exc


def _validate_decision_metadata(decision_info, canonical, error_type):
    expected = {
        "rule_id": canonical.rule_id,
        "policy_priority": canonical.policy_priority,
        "reason_codes": list(canonical.reason_codes),
        "time_window": canonical.time_window,
    }
    actual = {
        "rule_id": decision_info.get("rule_id"),
        "policy_priority": decision_info.get("policy_priority"),
        "reason_codes": list(decision_info.get("reason_codes", [])),
        "time_window": decision_info.get("time_window"),
    }
    for field, expected_value in expected.items():
        if actual[field] != expected_value:
            raise error_type(
                ERROR_CODE,
                f"prior_decision.decision.{field} ({actual[field]!r}) does not match canonical policy ({expected_value!r})",
            )


def _validate_candidates(prior_decision, canonical_targets, catalog, error_type):
    candidates = prior_decision.get("candidates")
    if not isinstance(candidates, list) or len(candidates) != len(canonical_targets):
        actual_length = len(candidates) if isinstance(candidates, list) else 0
        raise error_type(
            ERROR_CODE,
            f"prior_decision.candidates length ({actual_length}) does not match canonical policy candidates length ({len(canonical_targets)})",
        )
    for index, (candidate, target) in enumerate(zip(candidates, canonical_targets)):
        if not isinstance(candidate, dict):
            raise error_type(
                ERROR_CODE, f"prior_decision.candidates[{index}] must be an object"
            )
        expected = {
            "adapter": target.adapter,
            "target": target.target,
            "execution_class": target.execution_class,
            "selfcheck_required": target.selfcheck_required,
            "thinking_level": target.thinking_level,
            "reasoning_effort": target.reasoning_effort,
        }
        if isinstance(catalog, dict):
            expected.update(
                target_id=target.catalog_id, command_model=target.command_model
            )
        if any(candidate.get(field) != value for field, value in expected.items()):
            raise error_type(
                ERROR_CODE,
                f"prior_decision.candidates[{index}] identity ({candidate.get('adapter')}, {candidate.get('target')}) does not match canonical policy candidate ({target.adapter}, {target.target})",
            )


def validate_prior_candidate_identity(
    prior_decision,
    *,
    stage,
    lane,
    grade,
    policy,
    error_type,
    validate_used_candidates,
):
    decision_info = prior_decision.get("decision")
    if not isinstance(decision_info, dict):
        raise error_type(ERROR_CODE, "prior_decision.decision must be an object")
    catalog = prior_decision.get("catalog")
    changed = _validate_catalog(
        catalog, f"{stage}:{lane}-G{grade:02d}", policy, error_type
    )
    if changed:
        _validate_pinned_catalog_snapshot(
            prior_decision, error_type, validate_used_candidates
        )
        return
    canonical = _canonical_decision(
        decision_info, stage, lane, grade, policy, error_type
    )
    _validate_decision_metadata(decision_info, canonical, error_type)
    _validate_candidates(prior_decision, canonical.candidates, catalog, error_type)
    _validate_selected_and_history(
        prior_decision, canonical.candidates, policy, error_type
    )