diff --git a/scripts/await_intent.py b/scripts/await_intent.py new file mode 100644 index 00000000..570d4d54 --- /dev/null +++ b/scripts/await_intent.py @@ -0,0 +1,375 @@ +"""await_intent.py — ASYNC CORRECTIVE PATCH: intent normalization at the +execution boundary. + +PRINCIPLE: MODEL OUTPUT = INTENT. RUNE CONTROLLER = EXECUTION SEMANTICS. + +The model may express continuation intent in any quality — `await_operation +(deadline_secs=1200)`, `await-operation [op_type=master_ci, sha=X]`, +"sleep 300 then check CI", "poll until done". When a matching +PendingOperation exists (or the deterministic resolver can derive the +operation truth from git/Gitea state), the controller executes the +CANONICAL continuation and discards model-supplied scheduling policy. + +Authority provenance for timing: + USER_EXPLICIT / EXTERNAL_CONTRACT / TEST_HARNESS / + GOVERNED_OPERATION_POLICY -> authoritative, retained + MODEL_INVENTED -> non-authoritative, ignored + +This module owns NO storage: it reads the existing W/R wait store +(rune.event_wait) and git/Gitea state. The canonical continuation remains +scripts/ci_event_wait.py run() (the T21.5 single-logical-await). +""" +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +from pathlib import Path +from typing import Any + +_SCRIPT_DIR = Path(__file__).resolve().parent +if str(_SCRIPT_DIR) not in sys.path: + sys.path.insert(0, str(_SCRIPT_DIR)) +_REPO_SRC = str(_SCRIPT_DIR.parent / "src") +if _REPO_SRC not in sys.path: + sys.path.insert(0, _REPO_SRC) + +WR_BRIDGE = Path(os.environ.get("RUNE_WR_ROOT", "/mnt/workdrive/wr-bridge")) +REPO = Path("/mnt/workdrive/issue161-harness") +GIT_TIMEOUT = 10 + +# Default policy deadline per operation type — GOVERNED_OPERATION_POLICY. +# The model's numbers never become policy; these are the controller's. +POLICY_DEADLINES = { + "branch_ci": 1800, + "pull_request_checks": 1800, + "master_ci": 1800, + "deploy": 900, + "release": 900, +} + +VALID_OP_TYPES = ("branch_ci", "pull_request_checks", "master_ci", + "deploy", "release") + + +# ── timing provenance ─────────────────────────────────────────────────────── + +def classify_timing_provenance(model_deadline: Any = None, + user_deadline: Any = None, + contract_deadline: Any = None) -> tuple[int, str]: + """Resolve the effective deadline + its provenance. + + Authority order: USER_EXPLICIT > EXTERNAL_CONTRACT > TEST_HARNESS > + GOVERNED_OPERATION_POLICY. MODEL_INVENTED values are recorded but + NEVER used as policy. + """ + if model_deadline is not None: + try: + model_deadline = int(model_deadline) + except (TypeError, ValueError): + model_deadline = None + if user_deadline is not None: + return int(user_deadline), "USER_EXPLICIT" + if contract_deadline is not None: + return int(contract_deadline), "EXTERNAL_CONTRACT" + return (model_deadline or 0), "MODEL_INVENTED" # caller discards this + + +# ── PendingOperation lookup over the existing W/R wait store ──────────────── + +def _load_wait_records() -> list[dict]: + """All durable wait records from the bridge store (read-only).""" + out = [] + waits = WR_BRIDGE / "waits" + if not waits.is_dir(): + return out + for f in sorted(waits.glob("*.json")): + try: + rec = json.loads(f.read_text()) + rec["_file"] = f.name + out.append(rec) + except (OSError, ValueError): + continue + return out + + +def _rec_operation_id(rec: dict) -> str: + return str((rec.get("ext") or {}).get("operation_id") + or rec.get("operation_id") or "") + + +def _rec_sha(rec: dict) -> str: + return str(rec.get("sha") or "") + + +def pending_operations(sha: str = "", op_type: str = "") -> list[dict]: + """Pending (non-terminal) operations, optionally filtered.""" + terminal = {"COMPLETED", "FAILED", "CLOSED", "CANCELLED", "EXPIRED"} + out = [] + for rec in _load_wait_records(): + status = str(rec.get("status", "")).upper() + if status in terminal: + continue + if sha and _rec_sha(rec) != sha: + continue + out.append(rec) + return out + + +def find_op_type_for_sha(sha: str) -> str | None: + """The op_type a prior canonical wait already recorded for this SHA + (objective text embeds the type: 'event-driven {op_type} qualification').""" + for rec in _load_wait_records(): + if _rec_sha(rec) != sha: + continue + obj = str(rec.get("continuation", {}).get("objective", "")) + m = re.search(r"event-driven (\w+) qualification", obj) + if m and m.group(1) in VALID_OP_TYPES: + return m.group(1) + return None + + +# ── deterministic op_type derivation from repo/Gitea truth ───────────────── + +def _git(*args: str) -> str: + try: + r = subprocess.run(["git", "-C", str(REPO), *args], + capture_output=True, text=True, timeout=GIT_TIMEOUT) + return r.stdout.strip() + except (OSError, subprocess.TimeoutExpired): + return "" + + +def sha_branch(sha: str) -> str: + """Branch(es) containing this commit in the local repo.""" + return _git("branch", "--all", "--contains", sha) + + +def sha_on_master(sha: str) -> bool: + out = _git("branch", "--contains", sha) + return any(l.strip().endswith("master") or l.strip() == "master" + for l in out.splitlines()) + + +def open_pr_for_sha(sha: str) -> int | None: + """Is there an OPEN PR whose head SHA is this one? (Gitea API, read-only.)""" + try: + sys.path.insert(0, str(_SCRIPT_DIR)) + import gitea_auth # noqa: E402 + st, prs = gitea_auth.api("/pulls?state=open&limit=50") + if st != 200: + return None + for p in prs: + hs = str(p.get("head") or {}).get("sha") if isinstance( + p.get("head"), dict) else None + if hs and str(hs).startswith(sha[:12]): + return int(p.get("number")) + except Exception: # noqa: BLE001 — resolver must never crash the tool + return None + return None + + +def derive_op_type(sha: str) -> tuple[str, str]: + """Deterministically derive the canonical op_type for a SHA. + + Returns (op_type, evidence). Order (A-patch §2 amendment): + 1. POST-PATCH normalized record (ext.intent_normalized=true) — true + stored operation truth, wins outright; + 2. open PR for the sha -> pull_request_checks; + 3. sha ancestor of master -> master_ci; + 4. sha on a feature branch -> branch_ci; + 5. legacy (pre-patch) records only CONFIRM, never override — they + were created from model-supplied defaults and cannot be trusted. + Never trusts model input. + """ + for rec in _load_wait_records(): + if _rec_sha(rec) != sha: + continue + if (rec.get("ext") or {}).get("intent_normalized"): + obj = str(rec.get("continuation", {}).get("objective", "")) + m = re.search(r"event-driven (\w+) qualification", obj) + if m and m.group(1) in VALID_OP_TYPES: + return m.group(1), "normalized_record:" + m.group(1) + pr = open_pr_for_sha(sha) + if pr: + return "pull_request_checks", f"open_pr:#{pr}" + if sha_on_master(sha): + return "master_ci", "commit_ancestor_of_master" + return "branch_ci", "commit_on_feature_branch" + + +# ── intent normalization ──────────────────────────────────────────────────── + +def normalize_action(action: dict, *, + model_deadline: Any = None, + user_deadline: Any = None, + contract_deadline: Any = None) -> dict: + """Normalize a model continuation action into the canonical continuation. + + action: {kind: await_operation|sleep_poll|check_status|unknown, + sha?: str, operation_id?: str, op_type?: str, raw?: str} + Returns: + {normalized: bool, action: "ContinuePendingOperation", operation_id, + op_type, authority, deadline_secs, deadline_provenance, + model_deadline_ignored, model_op_type_ignored, evidence} + or {normalized: False, reason: ...} / {normalized: "AMBIGUOUS", candidates} + """ + sha = str(action.get("sha") or "") + op_id = str(action.get("operation_id") or "") + model_op_type = action.get("op_type") + model_deadline_ignored = model_deadline is not None + model_op_type_ignored = model_op_type is not None + + # 1. explicit operation_id wins + if op_id: + from rune import event_wait as ew # noqa: E402 + rec = ew.operation_wait(op_id) + if rec is None: + return {"normalized": False, + "reason": f"operation_id {op_id} has no wait record"} + return _canonical_from_record(rec, model_deadline_ignored, + model_op_type_ignored, "explicit_id", + user_deadline, contract_deadline) + + # 2. resolve by sha / active operation + candidates = pending_operations(sha=sha) + if len(candidates) > 1 and not sha: + return {"normalized": "AMBIGUOUS", + "candidates": [_rec_operation_id(c) for c in candidates], + "reason": "multiple pending operations, no target"} + if len(candidates) == 1: + return _canonical_from_record(candidates[0], model_deadline_ignored, + model_op_type_ignored, "pending_match", + user_deadline, contract_deadline) + + # 3. no stored record: derive truth deterministically + if not sha: + return {"normalized": False, + "reason": "no pending operation and no sha to resolve"} + op_type, evidence = derive_op_type(sha) + return { + "normalized": True, + "action": "ContinuePendingOperation", + "operation_id": f"ci-{sha}", + "op_type": op_type, + "authority": "ci_runs" if op_type in ("branch_ci", + "pull_request_checks") + else "deploy_qualify", + "deadline_secs": POLICY_DEADLINES[op_type], + "deadline_provenance": "GOVERNED_OPERATION_POLICY", + "model_deadline_ignored": model_deadline_ignored, + "model_op_type_ignored": model_op_type_ignored, + "evidence": evidence, + } + + +def _canonical_from_record(rec: dict, model_deadline_ignored: bool, + model_op_type_ignored: bool, how: str, + user_deadline: Any, contract_deadline: Any) -> dict: + op_id = _rec_operation_id(rec) + obj = str(rec.get("continuation", {}).get("objective", "")) + m = re.search(r"event-driven (\w+) qualification", obj) + recorded = m.group(1) if (m and m.group(1) in VALID_OP_TYPES) else None + rec_sha = _rec_sha(rec) + if (rec.get("ext") or {}).get("intent_normalized") or how == "explicit_id": + # normalized record or explicit id: stored truth wins + op_type = recorded or "master_ci" + else: + # legacy record: git/Gitea derivation outranks the model-era default + op_type, _ev = derive_op_type(rec_sha) if rec_sha else \ + (recorded or "master_ci", "no_sha") + deadline, prov = classify_timing_provenance( + user_deadline=user_deadline, contract_deadline=contract_deadline) + if prov == "MODEL_INVENTED" or not deadline: + deadline = POLICY_DEADLINES.get(op_type, 1800) + prov = "GOVERNED_OPERATION_POLICY" + return { + "normalized": True, + "action": "ContinuePendingOperation", + "operation_id": op_id, + "op_type": op_type, + "authority": "ci_runs" if op_type in ("branch_ci", + "pull_request_checks") + else "deploy_qualify", + "deadline_secs": deadline, + "deadline_provenance": prov, + "model_deadline_ignored": model_deadline_ignored, + "model_op_type_ignored": model_op_type_ignored, + "matched_by": how, + } + + +def normalize_bash(command: str, *, sha_hint: str = "") -> dict | None: + """Recognize sleep/check/poll intent aimed at a known pending operation. + + Returns a normalization record if the command's semantic intent is + continuation of a known external operation; None if the command is + legitimate local timing (short sync, user delay, backoff) with no + pending operation to continue. NEVER returns an executed sleep. + """ + if not command: + return None + cmd = command.strip() + # long sleep + (status-check language or an explicit sha hint) + sleep_m = re.search(r"\bsleep\s+(\d{2,5})\b", cmd) + if not sleep_m: + return None + secs = int(sleep_m.group(1)) + check_intent = bool(re.search( + r"\b(ci|deploy|release|status|check|poll|pipeline|run)\b", cmd, + re.IGNORECASE)) + sha = sha_hint or "" + m = re.search(r"\b([0-9a-f]{40}|[0-9a-f]{12})\b", cmd) + if m: + sha = m.group(1) + if not check_intent and not sha: + return None # plain sleep, no continuation intent + cands = pending_operations(sha=sha) if sha else pending_operations() + if not cands and not sha: + return None + if len(cands) > 1: + return {"normalized": "AMBIGUOUS", + "candidates": [_rec_operation_id(c) for c in cands], + "raw_command_kind": "sleep_poll"} + rec = cands[0] if cands else None + if rec is not None: + norm = _canonical_from_record(rec, True, False, "sleep_poll_intent", + None, None) + else: + op_type, evidence = derive_op_type(sha) + norm = { + "normalized": True, "action": "ContinuePendingOperation", + "operation_id": f"ci-{sha}", "op_type": op_type, + "authority": "ci_runs" if op_type in ("branch_ci", + "pull_request_checks") + else "deploy_qualify", + "deadline_secs": POLICY_DEADLINES[op_type], + "deadline_provenance": "GOVERNED_OPERATION_POLICY", + "model_deadline_ignored": True, "model_op_type_ignored": False, + "evidence": evidence, + } + norm["raw_command_kind"] = "sleep_poll" + norm["raw_sleep_seconds_suppressed"] = secs + return norm + + +def emit_normalized_record(record: dict) -> None: + """Compact structured observability — one JSONL line per normalization. + No payloads, no secrets.""" + try: + log = _SCRIPT_DIR.parent / "store" / "await_intent_log.jsonl" + log.parent.mkdir(parents=True, exist_ok=True) + slim = {k: record.get(k) for k in ( + "normalized", "action", "operation_id", "op_type", "authority", + "deadline_secs", "deadline_provenance", "model_deadline_ignored", + "model_op_type_ignored", "matched_by", "evidence", + "raw_command_kind", "raw_sleep_seconds_suppressed", "reason", + "candidates") if k in record} + slim["at"] = __import__("time").strftime( + "%Y-%m-%dT%H:%M:%SZ", __import__("time").gmtime()) + with open(log, "a") as f: + f.write(json.dumps(slim, sort_keys=True) + "\n") + except OSError: + pass diff --git a/scripts/await_intent_cli.py b/scripts/await_intent_cli.py new file mode 100644 index 00000000..5901f3f5 --- /dev/null +++ b/scripts/await_intent_cli.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python3 +"""await_intent_cli.py — CLI surface for the intent normalization layer. + +Used by the await-operation harness tool so the MODEL's scheduling hints +never reach the wait engine. Prints one JSON line. +""" +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import await_intent # noqa: E402 + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + sub = ap.add_subparsers(dest="cmd", required=True) + n = sub.add_parser("normalize", help="normalize a continuation intent") + n.add_argument("--sha", default="") + n.add_argument("--operation-id", default="") + n.add_argument("--model-op-type", default=None) + n.add_argument("--model-deadline", default=None) + n.add_argument("--user-deadline", default=None, + help="USER_EXPLICIT deadline (authoritative)") + n.add_argument("--contract-deadline", default=None, + help="EXTERNAL_CONTRACT deadline (authoritative)") + args = ap.parse_args() + + action = {"kind": "await_operation", "sha": args.sha, + "operation_id": args.operation_id, + "op_type": args.model_op_type} + model_deadline = (int(args.model_deadline) + if args.model_deadline else None) + user_deadline = (int(args.user_deadline) + if args.user_deadline else None) + contract_deadline = (int(args.contract_deadline) + if args.contract_deadline else None) + record = await_intent.normalize_action( + action, model_deadline=model_deadline, + user_deadline=user_deadline, contract_deadline=contract_deadline) + await_intent.emit_normalized_record(record) + print(json.dumps(record, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/store/scaffold/async_intent_gates.json b/store/scaffold/async_intent_gates.json new file mode 100644 index 00000000..8915554e --- /dev/null +++ b/store/scaffold/async_intent_gates.json @@ -0,0 +1,164 @@ +{ + "tranche": "ASYNC_INTENT_NORMALIZATION", + "purpose": "Intent normalization at the execution boundary \u2014 model emits intent, controller owns execution semantics", + "predecessor": "T21.5 (guards retired) + Memory T0 (complete)", + "created_at": "2026-09-17T20:43:47Z", + "gates": [ + { + "gate_id": "A10_LIVE_MASTER_DEPLOY_PROOF", + "mandatory": true, + "requirement": "live master CI -> deploy with normalization evidence", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A11_REGRESSION_TESTS", + "mandatory": true, + "requirement": "deterministic test matrix green; existing wait tests unbroken", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A12_RAW_SELF_AUDIT", + "mandatory": true, + "requirement": "post-fix stream classified; no model scheduling choice affected execution", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A13_FINAL_RUNTIME_CONVERGENCE", + "mandatory": true, + "requirement": "git==release==deployed==runtime==healthy", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A1_INTENT_NORMALIZATION_BOUNDARY_PROVEN", + "mandatory": true, + "requirement": "earliest deterministic boundary identified + normalization implemented there (not prompt, not guard)", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A2_PENDING_OPERATION_IS_WAIT_AUTHORITY", + "mandatory": true, + "requirement": "stored operation truth (op_type/authority/timing) wins over model args", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A3_MODEL_TIMING_NON_AUTHORITATIVE", + "mandatory": true, + "requirement": "MODEL_DURATION_ACTUALLY_USED_COUNT=0 for canonical flows; provenance recorded", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A4_MODEL_OP_TYPE_NON_AUTHORITATIVE", + "mandatory": true, + "requirement": "MODEL_SUPPLIED_OP_TYPE_COUNT_USED=0; stored/derived authority wins", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A5_SLEEP_POLL_INTENT_NORMALIZED", + "mandatory": true, + "requirement": "sleep/check/poll intent for a known pending op compiles to canonical continuation; literal sleep not executed", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A6_EXPLICIT_REAL_DEADLINES_PRESERVED", + "mandatory": true, + "requirement": "USER_EXPLICIT and CONTRACT deadlines retained with provenance", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A7_AMBIGUITY_FAILS_SAFE", + "mandatory": true, + "requirement": "multiple genuinely ambiguous pending ops -> structured ambiguity, no guess", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A8_LIVE_BRANCH_CI_PROOF", + "mandatory": true, + "requirement": "live branch push -> CI with normalization evidence", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + }, + { + "gate_id": "A9_LIVE_PR_MERGE_PROOF", + "mandatory": true, + "requirement": "live PR checks -> merge with normalization evidence", + "status": "UNRUN", + "evidence": "", + "blocker": "", + "waiver_reason": "", + "waived_by": "", + "declared_at": "2026-09-17T20:43:47Z", + "resolved_at": "" + } + ] +} \ No newline at end of file diff --git a/tests/test_await_intent.py b/tests/test_await_intent.py new file mode 100644 index 00000000..df7fdd79 --- /dev/null +++ b/tests/test_await_intent.py @@ -0,0 +1,197 @@ +"""test_await_intent.py — ASYNC CORRECTIVE PATCH test matrix (§12). + +MODEL OUTPUT = INTENT. RUNE CONTROLLER = EXECUTION SEMANTICS. +Every case proves the controller discards model scheduling policy and +executes stored/derived operation truth. +""" +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import pytest + +REPO = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO / "scripts")) + +import await_intent as ai # noqa: E402 + + +@pytest.fixture +def isolated(monkeypatch, tmp_path): + """Isolate from the live bridge: stub wait records + git.""" + monkeypatch.setattr(ai, "WR_BRIDGE", tmp_path / "wr") + waits = tmp_path / "wr" / "waits" + waits.mkdir(parents=True) + monkeypatch.setattr(ai, "REPO", tmp_path / "repo") + return waits + + +def _put(waits, op_id, sha, op_type, *, normalized=False, status="WAITING"): + rec = { + "operation_id": op_id, "status": status, "sha": sha, + "provider": "gitea", "event_type": "ci_completed", + "continuation": {"objective": + f"event-driven {op_type} qualification for {sha}"}, + "ext": {"operation_id": op_id, + **({"intent_normalized": True} if normalized else {})}, + } + (waits / f"{op_id}.json").write_text(json.dumps(rec)) + + +def _record_patch(monkeypatch, waits): + """point _load_wait_records at the isolated dir (it reads WR_BRIDGE).""" + # _load_wait_records reads ai.WR_BRIDGE which the fixture already patched. + + +# ── A. MODEL DEADLINE IGNORED ─────────────────────────────────────────────── + +def test_model_deadline_ignored(isolated, monkeypatch): + _put(isolated, "ci-123", "b" * 40, "branch_ci", normalized=True) + monkeypatch.setattr(ai, "open_pr_for_sha", lambda sha: None) + monkeypatch.setattr(ai, "sha_on_master", lambda sha: False) + r = ai.normalize_action({"kind": "await_operation", + "sha": "b" * 40, + "op_type": "master_ci"}, + model_deadline=1200) + assert r["normalized"] is True + assert r["op_type"] == "branch_ci" # stored truth + assert r["deadline_secs"] == 1800 # policy, not 1200 + assert r["deadline_provenance"] == "GOVERNED_OPERATION_POLICY" + assert r["model_deadline_ignored"] is True + assert r["model_op_type_ignored"] is True + + +# ── B. WRONG OP_TYPE IGNORED ──────────────────────────────────────────────── + +def test_wrong_op_type_ignored(isolated, monkeypatch): + sha = "c" * 40 + _put(isolated, "ci-456", sha, "branch_ci", normalized=True) + monkeypatch.setattr(ai, "open_pr_for_sha", lambda s: None) + monkeypatch.setattr(ai, "sha_on_master", lambda s: False) + r = ai.normalize_action({"kind": "await_operation", "sha": sha, + "op_type": "master_ci"}) + assert r["op_type"] == "branch_ci" + assert r["authority"] == "ci_runs" # deploy_qualify NOT used + + +# ── C. SLEEP+CHECK NORMALIZED ─────────────────────────────────────────────── + +def test_sleep_check_normalized(isolated, monkeypatch): + sha = "d" * 40 + _put(isolated, "ci-789", sha, "branch_ci", normalized=True) + r = ai.normalize_bash(f"sleep 300 && check CI for {sha}") + assert r["normalized"] is True + assert r["action"] == "ContinuePendingOperation" + assert r["raw_sleep_seconds_suppressed"] == 300 + assert r["op_type"] == "branch_ci" + + +def test_sleep_check_no_pending_no_sha(isolated): + # a sleep+check phrase with NO pending op and no sha: not normalized + r = ai.normalize_bash("sleep 300 && check CI") + assert r is None or r.get("normalized") in (True, "AMBIGUOUS") + + +# ── E. USER DEADLINE PRESERVED ────────────────────────────────────────────── + +def test_user_deadline_preserved(isolated, monkeypatch): + sha = "e" * 40 + _put(isolated, "ci-ue", sha, "branch_ci", normalized=True) + r = ai.normalize_action({"kind": "await_operation", "sha": sha}, + user_deadline=600) + assert r["deadline_provenance"] == "USER_EXPLICIT" + assert r["deadline_secs"] == 600 + + +# ── F. CONTRACT DEADLINE PRESERVED ────────────────────────────────────────── + +def test_contract_deadline_preserved(isolated, monkeypatch): + sha = "f" * 40 + _put(isolated, "ci-cd", sha, "deploy", normalized=True) + r = ai.normalize_action({"kind": "await_operation", "sha": sha}, + contract_deadline=450) + assert r["deadline_provenance"] == "EXTERNAL_CONTRACT" + assert r["deadline_secs"] == 450 + + +# ── G. AMBIGUOUS OPERATIONS ───────────────────────────────────────────────── + +def test_ambiguous_fails_safe(isolated): + _put(isolated, "ci-a1", "1" * 40, "branch_ci", normalized=True) + _put(isolated, "ci-a2", "2" * 40, "master_ci", normalized=True) + r = ai.normalize_action({"kind": "await_operation"}) # no sha, no id + assert r["normalized"] == "AMBIGUOUS" + assert set(r["candidates"]) == {"ci-a1", "ci-a2"} + + +# ── H. NO PENDING OPERATION — legitimate local sleep untouched ───────────── + +def test_plain_sleep_not_routed(isolated): + assert ai.normalize_bash("sleep 5") is None + assert ai.normalize_bash("sleep 2 && ls") is None + + +# ── derivation truth table ────────────────────────────────────────────────── + +def test_derivation_pr_over_branch(monkeypatch): + monkeypatch.setattr(ai, "open_pr_for_sha", lambda s: 42) + monkeypatch.setattr(ai, "sha_on_master", lambda s: False) + t, ev = ai.derive_op_type("a" * 40) + assert t == "pull_request_checks" and "open_pr" in ev + + +def test_derivation_master_beats_legacy_record(isolated, monkeypatch): + # legacy (pre-patch) record says master_ci for a sha that IS on master: + # record confirms, does not override + sha = "9" * 40 + _put(isolated, "ci-legacy", sha, "master_ci", normalized=False) + monkeypatch.setattr(ai, "sha_on_master", lambda s: True) + t, ev = ai.derive_op_type(sha) + assert t == "master_ci" + + +def test_derivation_branch_beats_legacy_record(isolated, monkeypatch): + # legacy record says master_ci (model-era default) for a branch sha: + # git truth outranks the stale default + sha = "8" * 40 + _put(isolated, "ci-legacy2", sha, "master_ci", normalized=False) + monkeypatch.setattr(ai, "open_pr_for_sha", lambda s: None) + monkeypatch.setattr(ai, "sha_on_master", lambda s: False) + t, ev = ai.derive_op_type(sha) + assert t == "branch_ci" + + +def test_normalized_record_wins_over_derivation(isolated, monkeypatch): + sha = "7" * 40 + _put(isolated, "ci-norm", sha, "branch_ci", normalized=True) + monkeypatch.setattr(ai, "open_pr_for_sha", + lambda s: 99) # would derive PR, but record wins + t, ev = ai.derive_op_type(sha) + assert t == "branch_ci" and "normalized_record" in ev + + +# ── timing provenance ─────────────────────────────────────────────────────── + +def test_timing_provenance_order(): + assert ai.classify_timing_provenance( + model_deadline=100, user_deadline=600)[1] == "USER_EXPLICIT" + assert ai.classify_timing_provenance( + model_deadline=100, contract_deadline=450)[1] == "EXTERNAL_CONTRACT" + assert ai.classify_timing_provenance(model_deadline=100)[1] == \ + "MODEL_INVENTED" + + +# ── observability ─────────────────────────────────────────────────────────── + +def test_observability_record(tmp_path, monkeypatch): + monkeypatch.setattr(ai, "_SCRIPT_DIR", tmp_path) + rec = {"normalized": True, "action": "ContinuePendingOperation", + "op_type": "branch_ci", "model_deadline_ignored": True} + ai.emit_normalized_record(rec) + log = tmp_path.parent / "store" / "await_intent_log.jsonl" + # _SCRIPT_DIR is tmp_path; parent/store = tmp/store + assert log.exists() + last = json.loads(log.read_text().splitlines()[-1]) + assert last["model_deadline_ignored"] is True