conductor
efb99d4682
A just-merged master commit was invisible to branch --contains until refs were fetched: the local clone's stale master outranked repo truth and a master_ci operation classified as branch_ci. derive_op_type now refreshes remote-tracking refs (bounded by GIT_TIMEOUT, fail-open) and containment checks all refs. Generic fix; no SHA special-casing.
495 lines
21 KiB
Python
495 lines
21 KiB
Python
"""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:
|
|
"""SHA is an ancestor of master (local or remote-tracking refs).
|
|
|
|
M10 correction (2026-09-19): --all so a freshly fetched master
|
|
(refs/remotes/gitea/master) counts; local branches alone go stale
|
|
between fetches. derive_op_type refreshes refs before calling this.
|
|
"""
|
|
out = _git("branch", "--all", "--contains", sha)
|
|
for l in out.splitlines():
|
|
s = l.strip().removeprefix("remotes/").removesuffix(" *")
|
|
if s == "master" or s.endswith("/master"):
|
|
return True
|
|
return False
|
|
|
|
|
|
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.
|
|
"""
|
|
# 0. M10 correction (2026-09-19): refresh remote-tracking refs before
|
|
# inference. A just-merged master commit is invisible to
|
|
# branch --contains until refs are fetched; stale local git state
|
|
# must never outrank repo truth. Bounded by GIT_TIMEOUT, fail-open.
|
|
_git("fetch", "--quiet", "gitea", "+refs/heads/*:refs/remotes/gitea/*")
|
|
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,
|
|
task_id: str = "") -> 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}
|
|
task_id: REAL_ASYNC_BRIDGE §9 — when the caller knows its task
|
|
binding, pending-operation resolution is scoped to operations
|
|
registered with that task_id; stale global operations are
|
|
never consulted when a binding exists.
|
|
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
|
|
|
|
# 0. task binding scope (§9): task-bound operations beat global state.
|
|
# Stored operation wins over ANY model hint (sha/op_type/deadline).
|
|
if task_id and not op_id:
|
|
bound = [r for r in pending_operations()
|
|
if str(r.get("ext", {}).get("task_id", "")) == task_id]
|
|
if len(bound) == 1:
|
|
rec = bound[0]
|
|
norm = _canonical_from_record(rec, model_deadline_ignored,
|
|
model_op_type_ignored,
|
|
"task_bound",
|
|
user_deadline, contract_deadline)
|
|
norm["sha_for_wait"] = _rec_sha(rec)
|
|
norm["task_id"] = task_id
|
|
return norm
|
|
if len(bound) > 1:
|
|
return {"normalized": "AMBIGUOUS",
|
|
"candidates": [_rec_operation_id(c) for c in bound],
|
|
"reason": f"multiple pending operations bound to task {task_id}"}
|
|
return {"normalized": False,
|
|
"reason": f"no pending operation bound to task {task_id}"}
|
|
|
|
# 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:
|
|
# REAL_ASYNC_BRIDGE §10: a FRESH operation has no wait record
|
|
# yet — registration happens at the park step (ci_event_wait).
|
|
# For an explicit, well-formed, EXISTING commit id, return the
|
|
# provisional canonical continuation (policy deadline, own-runs
|
|
# authority) so the model-facing tool can park on first call.
|
|
# A typo'd/ghost sha is refused here (A-patch lesson): the
|
|
# commit must exist in the repository.
|
|
import re as _re
|
|
m = _re.fullmatch(r"ci-([0-9a-f]{40})", op_id)
|
|
if m:
|
|
sha40 = m.group(1)
|
|
probe = subprocess.run(
|
|
["git", "-C", str(REPO), "cat-file", "-e",
|
|
f"{sha40}^{{commit}}"],
|
|
capture_output=True, timeout=15)
|
|
if probe.returncode == 0:
|
|
return {
|
|
"normalized": True,
|
|
"action": "ContinuePendingOperation",
|
|
"operation_id": op_id,
|
|
"sha_for_wait": sha40,
|
|
"op_type": "branch_ci",
|
|
"authority": "ci_runs",
|
|
"deadline_secs": POLICY_DEADLINES["branch_ci"],
|
|
"deadline_provenance": "GOVERNED_OPERATION_POLICY",
|
|
"model_deadline_ignored": model_deadline_ignored,
|
|
"model_op_type_ignored": model_op_type_ignored,
|
|
"evidence": "explicit_id_fresh_registration",
|
|
}
|
|
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.
|
|
# H-PATCH H2/H3: the model sha is a HINT. If it doesn't resolve but
|
|
# unambiguous pending operations exist, the STORED target wins — a
|
|
# mistyped model sha must never create a ghost operation (the A-patch
|
|
# live failure: a typo'd sha derived branch_ci for a nonexistent commit
|
|
# and parked a wait on a ghost).
|
|
# 2. no explicit operation_id: resolve by sha / pending operations.
|
|
# H-PATCH: when the model provides NO sha (operation-id-first mode),
|
|
# resolve from unambiguous pending operations — the model should not
|
|
# need to reconstruct a 40-char sha to express "wait for it".
|
|
if not sha:
|
|
all_pending = pending_operations()
|
|
if len(all_pending) == 1:
|
|
rec = all_pending[0]
|
|
norm = _canonical_from_record(rec, model_deadline_ignored,
|
|
model_op_type_ignored,
|
|
"auto_resolved_no_sha",
|
|
user_deadline, contract_deadline)
|
|
norm["sha_for_wait"] = _rec_sha(rec)
|
|
return norm
|
|
if len(all_pending) > 1:
|
|
return {"normalized": "AMBIGUOUS",
|
|
"candidates": [_rec_operation_id(c) for c in all_pending],
|
|
"reason": "multiple pending operations, no sha hint"}
|
|
return {"normalized": False,
|
|
"reason": "no pending operations and no sha provided"}
|
|
|
|
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:
|
|
rec = candidates[0]
|
|
stored_sha = _rec_sha(rec)
|
|
norm = _canonical_from_record(rec, model_deadline_ignored,
|
|
model_op_type_ignored, "pending_match",
|
|
user_deadline, contract_deadline)
|
|
if sha and stored_sha and sha != stored_sha:
|
|
norm["model_sha_mismatch"] = {
|
|
"model_sha": sha, "stored_sha": stored_sha,
|
|
"resolution": "stored_target_used"}
|
|
return norm
|
|
# model sha unresolved: any unambiguous pending op at all? then the
|
|
# model probably mistyped the sha — use the stored target, log it.
|
|
all_pending = pending_operations()
|
|
if len(all_pending) == 1 and sha:
|
|
rec = all_pending[0]
|
|
stored_sha = _rec_sha(rec)
|
|
norm = _canonical_from_record(rec, model_deadline_ignored,
|
|
model_op_type_ignored,
|
|
"unresolved_sha_single_pending",
|
|
user_deadline, contract_deadline)
|
|
norm["model_sha_mismatch"] = {
|
|
"model_sha": sha, "stored_sha": stored_sha,
|
|
"resolution": "stored_target_used"}
|
|
return norm
|
|
|
|
# 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)
|
|
norm["sha_for_wait"] = _rec_sha(rec)
|
|
elif sha:
|
|
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, "sha_for_wait": sha,
|
|
}
|
|
else:
|
|
return None
|
|
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
|