feat(async-intent): intent normalization at the execution boundary

MODEL OUTPUT = INTENT; RUNE CONTROLLER = EXECUTION SEMANTICS.

scripts/await_intent.py: deterministic resolver that derives canonical
op_type from stored normalized records / open PRs / git ancestry — model
supplied op_type and deadline_secs are recorded but NEVER used as policy
(git/Gitea truth outranks even legacy wait records). Sleep/check/poll
intent aimed at a known pending operation compiles to
ContinuePendingOperation; plain legitimate sleeps are untouched.
Timing provenance: USER_EXPLICIT/EXTERNAL_CONTRACT authoritative,
MODEL_INVENTED discarded. Ambiguity fails safe with structured
candidates. await_intent_cli.py = tool surface; compact JSONL
observability per normalization. await-operation.ts rewired to normalize
before dispatch (needs commit of harness tool separately).

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

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())

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": ""
+    }
+  ]
+}

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