CASAN/AINative_OKR_CASAN5/packages/casan-harness/scripts/bash/loop_common.py

"""CASAN loop-engineering shared helpers (Plan-17, harness-owned).

Deny-by-default, fail-closed primitives shared by the loop primitives
(`loop-governor.py`, `loop-convergence.py`, `loop-gate.py`). No third-party
dependency is required to *function safely*: if PyYAML is unavailable the loop
still runs under the strictest built-in ceiling (fail-closed) — never an
"unlimited" fallback.

Distinction (Plan-17 T1 17.1/17.2):
  * policy file ABSENT      -> use STRICT_CEILING (evaluate normally).
  * policy file PRESENT but unreadable/corrupt -> raise PolicyError (caller HALTs).

State is written under a redirectable, tenant-aware root so it never pollutes the
repo (tests set CASAN_LOOP_STATE_ROOT / CASAN_TENANT_STATE_ROOT to a tmp dir).
"""
import hashlib
import json
import os
import re
import subprocess
from datetime import datetime, timezone

# Strictest possible ceiling. Used when no policy file exists or no rule matches
# a run (deny-by-default: absence of an explicit grant means the tightest budget,
# not "infinite").
STRICT_CEILING = {
    "max_steps": 3,
    "max_tokens": 8000,
    "max_wall_clock_sec": 120,
    "max_cost_usd": 0.10,
    "max_corrections_per_step": 1,
    "on_exceed": "halt",  # halt | escalate
}

_BUDGET_KEYS = (
    "max_steps",
    "max_tokens",
    "max_wall_clock_sec",
    "max_cost_usd",
    "max_corrections_per_step",
)

# Strictest convergence thresholds (Plan-17 T2). Small windows => detect a stuck /
# oscillating loop *sooner* when no policy grants a looser window (deny-by-default).
STRICT_CONVERGENCE = {
    "oscillation_repeat": 3,   # N identical consecutive actions => OSCILLATING
    "thrash_window": 4,        # A,B,A,B... over this many steps => OSCILLATING
    "no_progress_window": 3,   # W steps with zero forward progress => STALLED
    "on_stall": "escalate",    # escalate | halt
}

_CONVERGENCE_INT_KEYS = ("oscillation_repeat", "thrash_window", "no_progress_window")


class PolicyError(Exception):
    """A policy file exists but cannot be trusted (unreadable / malformed).
    Callers must treat this as fail-closed (HALT), never fall back to open."""


def project_root() -> str:
    d = os.path.abspath(os.path.dirname(__file__))
    p = d
    while p != os.path.dirname(p):
        if os.path.isdir(os.path.join(p, ".specify")):
            return p
        p = os.path.dirname(p)
    return os.path.abspath(os.path.join(d, "..", "..", ".."))


def now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def profile() -> str:
    return os.environ.get("CASAN_PROFILE", "dev").strip() or "dev"


def git_commit() -> str:
    try:
        r = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            cwd=project_root(),
            capture_output=True,
            text=True,
            timeout=5,
        )
        if r.returncode == 0:
            return r.stdout.strip()
    except Exception:
        pass
    return "unknown"


def _tenant_id():
    t = os.environ.get("CASAN_TENANT_ID", "").strip()
    if not t:
        return None
    if not re.fullmatch(r"[A-Za-z0-9_-]+", t):
        # Same fail-closed contract as control-plane-settings.py (SEC-23 MT-01).
        raise PolicyError("tenant_id_invalid")
    return t


def state_root() -> str:
    explicit = os.environ.get("CASAN_LOOP_STATE_ROOT")
    if explicit:
        return explicit
    tenant = _tenant_id()
    if tenant:
        base = os.environ.get(
            "CASAN_TENANT_STATE_ROOT",
            os.path.join(project_root(), ".specify/state/tenants"),
        )
        return os.path.join(base, tenant, "loops")
    return os.path.join(project_root(), ".specify/state/loops")


def run_dir(run_id: str) -> str:
    if not run_id or not re.fullmatch(r"[A-Za-z0-9._-]+", run_id):
        raise PolicyError("run_id_invalid")
    return os.path.join(state_root(), "runs", run_id)


def provenance(source: str, artifact_path=None, verified: bool = False) -> dict:
    """Every primitive output carries this envelope (Plan-13 §8.6 data-contract)."""
    return {
        "source": source,
        "artifact_path": artifact_path,
        "commit": git_commit(),
        "run_at": now_iso(),
        "verified": bool(verified),
    }


def policy_path() -> str:
    explicit = os.environ.get("CASAN_LOOP_POLICY_FILE")
    if explicit:
        return explicit
    return os.path.join(project_root(), ".specify/config/loop-policy.yaml")


def load_policy():
    """Return the parsed policy dict, or None when no policy file exists.

    Fail-closed: a present-but-unreadable/malformed policy raises PolicyError so
    the caller HALTs rather than silently running unbounded.
    """
    path = policy_path()
    if not os.path.isfile(path):
        return None
    try:
        import yaml  # optional dependency
    except ImportError as exc:  # cannot parse a policy we were told to honour
        raise PolicyError("pyyaml_unavailable") from exc
    try:
        with open(path, encoding="utf-8") as fh:
            data = yaml.safe_load(fh)
    except Exception as exc:  # malformed YAML
        raise PolicyError(f"policy_unreadable:{exc}") from exc
    if data is None:
        raise PolicyError("policy_empty")
    if not isinstance(data, dict):
        raise PolicyError("policy_not_mapping")
    return data


def _coerce_budget(raw, base):
    """Overlay only the known, well-typed budget keys from `raw` onto `base`.
    Unknown keys are ignored; a wrong-typed value fails closed."""
    out = dict(base)
    if not isinstance(raw, dict):
        return out
    for k in _BUDGET_KEYS:
        if k in raw and raw[k] is not None:
            v = raw[k]
            if not isinstance(v, (int, float)) or isinstance(v, bool) or v < 0:
                raise PolicyError(f"bad_budget_value:{k}")
            out[k] = v
    if "on_exceed" in raw and raw["on_exceed"] is not None:
        oe = raw["on_exceed"]
        if oe not in ("halt", "escalate"):
            raise PolicyError(f"bad_on_exceed:{oe}")
        out["on_exceed"] = oe
    return out


# --- governed override layer (Plan-17 T5 meta-loop) -------------------------
# A versioned, approved change recorded in the control-plane settings store can
# tighten/loosen the effective ceiling. This is how a governed meta-loop decision
# actually changes the governor's behaviour (not just a proposal on paper).
_GOV_BUDGET_MAP = {
    "loop.max_steps": "max_steps",
    "loop.max_tokens": "max_tokens",
    "loop.max_wall_clock_sec": "max_wall_clock_sec",
    "loop.max_cost_usd": "max_cost_usd",
    "loop.max_corrections_per_step": "max_corrections_per_step",
}
_GOV_CONV_MAP = {
    "loop.oscillation_repeat": "oscillation_repeat",
    "loop.no_progress_window": "no_progress_window",
}


def cp_store_path() -> str:
    """Resolve the control-plane settings store the same way control-plane-settings.py
    does, so governed loop overrides are read from exactly where they were written."""
    explicit = os.environ.get("CASAN_CP_STORE_FILE")
    if explicit:
        return explicit
    tenant = _tenant_id()
    if tenant:
        base = os.environ.get(
            "CASAN_TENANT_STATE_ROOT",
            os.path.join(project_root(), ".specify/state/tenants"),
        )
        return os.path.join(base, tenant, "control-plane", "settings.json")
    return os.path.join(project_root(), ".specify/level5/control-plane-settings.json")


def load_governed_overrides() -> dict:
    """Best-effort read of `loop.*` governed settings. Absent/unreadable store =>
    {} (no override => the YAML/strict ceiling stands, which is already safe, so a
    read failure never *loosens* anything)."""
    path = cp_store_path()
    if not os.path.isfile(path):
        return {}
    try:
        data = json.load(open(path, encoding="utf-8"))
        settings = data.get("settings", {})
        out = {}
        for k, v in settings.items():
            if k.startswith("loop.") and isinstance(v, dict) and "value" in v:
                out[k] = v["value"]
        return out
    except (OSError, ValueError, KeyError, TypeError):
        return {}


def org_ceiling(policy) -> dict:
    """Organization hard cap (17.19): a governed loosen can never exceed these,
    even with approval. Malformed => fail-closed (PolicyError via _coerce_budget)."""
    if not policy:
        return {}
    oc = policy.get("org_ceiling")
    if not isinstance(oc, dict):
        return {}
    return _coerce_budget(oc, {})


def _apply_governed_budget(ceiling, policy):
    overrides = load_governed_overrides()
    if not overrides:
        return ceiling, ""
    oc = org_ceiling(policy)
    applied = []
    for gk, bk in _GOV_BUDGET_MAP.items():
        v = overrides.get(gk)
        if isinstance(v, (int, float)) and not isinstance(v, bool) and v >= 0:
            if bk in oc:
                v = min(v, oc[bk])  # clamp to org hard cap (defense-in-depth)
            ceiling[bk] = v
            applied.append(bk)
    return ceiling, ("+governed(" + ",".join(applied) + ")" if applied else "")


def _apply_governed_convergence(cfg):
    overrides = load_governed_overrides()
    if not overrides:
        return cfg, ""
    applied = []
    for gk, ck in _GOV_CONV_MAP.items():
        v = overrides.get(gk)
        if isinstance(v, int) and not isinstance(v, bool) and v >= 1:
            cfg[ck] = v
            applied.append(ck)
    return cfg, ("+governed(" + ",".join(applied) + ")" if applied else "")


def resolve_budget(policy, prof: str, delegation_level=None, project=None):
    """Compute the effective ceiling for a run.

    Deny-by-default: start from STRICT_CEILING and overlay, in order,
    profile.defaults -> delegation_levels[L] -> projects[name] -> governed
    overrides (clamped to org_ceiling). Any field not granted stays strictest.
    Returns (ceiling_dict, source_tag).
    """
    ceiling = dict(STRICT_CEILING)
    sources = []
    if policy:
        profiles = policy.get("profiles")
        if not isinstance(profiles, dict):
            raise PolicyError("policy_missing_profiles")
        prof_block = profiles.get(prof)
        if isinstance(prof_block, dict):
            defaults = prof_block.get("defaults")
            if isinstance(defaults, dict):
                ceiling = _coerce_budget(defaults, ceiling)
                sources.append(f"{prof}.defaults")
            if delegation_level:
                levels = prof_block.get("delegation_levels")
                if isinstance(levels, dict) and delegation_level in levels:
                    ceiling = _coerce_budget(levels[delegation_level], ceiling)
                    sources.append(f"delegation:{delegation_level}")
            if project:
                projects = prof_block.get("projects")
                if isinstance(projects, dict) and project in projects:
                    ceiling = _coerce_budget(projects[project], ceiling)
                    sources.append(f"project:{project}")
        else:
            # Unknown profile => no matching rule => strictest (deny-by-default).
            sources.append(f"strict-default(no-profile:{prof})")
    else:
        sources.append("strict-default(no-policy)")

    ceiling, gov = _apply_governed_budget(ceiling, policy)
    tag = "+".join(sources) if sources else f"strict-default(empty:{prof})"
    return ceiling, tag + gov


def _coerce_convergence(raw, base):
    out = dict(base)
    if not isinstance(raw, dict):
        return out
    for k in _CONVERGENCE_INT_KEYS:
        if k in raw and raw[k] is not None:
            v = raw[k]
            if not isinstance(v, int) or isinstance(v, bool) or v < 1:
                raise PolicyError(f"bad_convergence_value:{k}")
            out[k] = v
    if "on_stall" in raw and raw["on_stall"] is not None:
        os_ = raw["on_stall"]
        if os_ not in ("halt", "escalate"):
            raise PolicyError(f"bad_on_stall:{os_}")
        out["on_stall"] = os_
    return out


def resolve_convergence(policy, prof: str):
    """Effective convergence thresholds for a profile. Deny-by-default: absent
    policy / profile => strictest (detect stalls soonest). A governed override
    layer (meta-loop) can adjust the windows. Returns (dict, tag)."""
    cfg = dict(STRICT_CONVERGENCE)
    if policy:
        profiles = policy.get("profiles")
        if not isinstance(profiles, dict):
            raise PolicyError("policy_missing_profiles")
        prof_block = profiles.get(prof)
        if isinstance(prof_block, dict):
            raw = prof_block.get("convergence")
            if isinstance(raw, dict):
                cfg = _coerce_convergence(raw, cfg)
                src = f"{prof}.convergence"
            else:
                src = f"strict-default(no-convergence:{prof})"
        else:
            src = f"strict-default(no-profile:{prof})"
    else:
        src = "strict-default(no-policy)"
    cfg, gov = _apply_governed_convergence(cfg)
    return cfg, src + gov


# ---------------------------------------------------------------------------
# Hash-linked loop audit log (self-contained tamper-evidence). Shares the same
# canonical-JSON + SHA-256 scheme as control-plane-settings.py so a future
# unified verifier (Plan-17 T4 / sync-point S2) can adopt it unchanged.
# ---------------------------------------------------------------------------
GENESIS_HASH = "0" * 64


def _canon(entry) -> str:
    return json.dumps(entry, sort_keys=True, ensure_ascii=False)


def _hash_entry(entry) -> str:
    return hashlib.sha256(_canon(entry).encode("utf-8")).hexdigest()


# Public serializer/verifier (sync-point S2: one serializer, one verifier). The
# per-run loop trace (Track 4) reuses these so its hash-chain is byte-compatible
# with the audit log and any future unified verifier.
def canonical(entry) -> str:
    return _canon(entry)


def chain_hash(entry) -> str:
    return _hash_entry(entry)


def audit_log_path() -> str:
    return os.path.join(state_root(), "audit", "loop-audit.jsonl")


def append_audit(event: dict) -> dict:
    """Append a hash-linked audit record. Append-only; each record chains to the
    previous via prev_hash so any later edit breaks the chain."""
    path = audit_log_path()
    os.makedirs(os.path.dirname(path), exist_ok=True)
    prev = GENESIS_HASH
    seq = 0
    if os.path.isfile(path):
        with open(path, encoding="utf-8") as fh:
            last = None
            for line in fh:
                line = line.strip()
                if line:
                    last = line
                    seq += 1
            if last:
                try:
                    prev = json.loads(last).get("hash", GENESIS_HASH)
                except ValueError:
                    prev = GENESIS_HASH
    base = {
        "seq": seq,
        "ts": now_iso(),
        "prev_hash": prev,
        "event": event,
    }
    base["hash"] = _hash_entry(base)
    with open(path, "a", encoding="utf-8") as fh:
        fh.write(_canon(base) + "\n")
    return base