diff --git a/arborist/qa/runner.py b/arborist/qa/runner.py index af2eed3..961d420 100644 --- a/arborist/qa/runner.py +++ b/arborist/qa/runner.py @@ -303,6 +303,90 @@ DEFAULT_POLICY = { } +#: Mapping from verdict audit_mode → Δ5F utility signal for the +#: advisory controller step. Same shape the dry-run simulator uses +#: (``bench/scripts/prometheus_sigma_sweep_dryrun.py``) so the +#: live-runtime advisory log and the offline sweep dry-run agree on +#: cost-class economics. +_AUDIT_MODE_DELTA_5F = { + "STRICT": 0.05, + "CANONICAL_PROJECTION": 0.10, + "HYBRID": 0.00, + "UNGROUNDED": -0.10, +} +_AUDIT_MODE_CAPITAL_COST = { + "STRICT": 1.0, + "CANONICAL_PROJECTION": 0.05, + "HYBRID": 0.8, + "UNGROUNDED": 0.4, +} + + +def _emit_qa_controller_advisory( + conn: sqlite3.Connection, cache_key: str, verdict: dict +) -> None: + """#000037 Phase 2 — write an advisory controller_decision row. + + Synthesizes a single-branch ControllerInput from the verdict, runs + the Phase 1 controller (pure function, no LLM call, no I/O), and + persists the resulting decision + difficulty + allocation rows to + the sibling ``controller_events`` table (does NOT enter + ``audit_events.event_hash`` preimage — chain integrity unchanged). + + Idempotent under retry (``UNIQUE(event_kind, body_hash)``). Pure + advisory: every QA cycle emits one of these regardless of cache + hit/miss state, but downstream consumers can prune or aggregate + rows without affecting the QA result. + + Lazy import keeps the QA hot path free of substrate-module load + cost on calls that never reach this helper (cache hits + early + returns). + """ + from arborist.substrate.prometheus import ( + BatteryDeltas, + ControllerBranch, + ControllerInput, + controller_decide, + safe_weights, + ) + from arborist.substrate.prometheus_audit import emit_controller_events + + audit_mode = (verdict.get("audit_mode") or "UNGROUNDED").upper() + n_quotes = int(verdict.get("n_quotes") or 0) + unverified = verdict.get("unverified_quotes") or [] + if isinstance(unverified, str): + try: + unverified = json.loads(unverified) + except (TypeError, ValueError): + unverified = [] + n_unverified = len(unverified) if isinstance(unverified, list) else 0 + witness_divergence = (n_unverified / n_quotes) if n_quotes > 0 else 0.0 + + branch = ControllerBranch( + branch_id=f"qa:{cache_key[:16]}", + deltas=BatteryDeltas( + delta_5s=0.0, delta_5t=0.0, + delta_5f=_AUDIT_MODE_DELTA_5F.get(audit_mode, 0.0), + delta_5r=0.0, + ), + witness_divergence=witness_divergence, + capital_cost=_AUDIT_MODE_CAPITAL_COST.get(audit_mode, 1.0), + payoff_b=1.0, + ) + decision = controller_decide( + ControllerInput( + organism_root=f"qa:{cache_key}", + branches=(branch,), + budget=1, + hermes_utilization=0, + weights=safe_weights(), + difficulty=1.0, + divergence_rate=witness_divergence, + ) + ) + emit_controller_events(conn, decision, organism_root=f"qa:{cache_key}") + + def _ms_since(t: float) -> float: return round((time.monotonic() - t) * 1000, 1) @@ -1080,6 +1164,14 @@ def ask( ), ) + # #000037 Phase 2 — emit advisory controller_events for this QA + # cycle. Single-branch decision, byte-cheap. Wrapped so any + # controller-emit error never blocks the QA return. Audit-only. + try: + _emit_qa_controller_advisory(conn, ckey, verdict) + except Exception: # pragma: no cover — advisory must not break QA + pass + from arborist.qa.dag import localize_failure as _localize failure_stage = _localize( audit_mode=verdict["audit_mode"], diff --git a/tests/test_prometheus_audit.py b/tests/test_prometheus_audit.py index 1b7e824..39e9915 100644 --- a/tests/test_prometheus_audit.py +++ b/tests/test_prometheus_audit.py @@ -386,3 +386,92 @@ def test_query_by_kind_returns_rows(shard): "WHERE event_kind='controller_budget_allocation'" ).fetchone()[0] assert n_alloc == 2 + + +# --------------------------------------------------------------------- +# Phase 2 integration: arborist.qa.runner._emit_qa_controller_advisory +# wires real QA verdicts → controller_events rows +# --------------------------------------------------------------------- + + +def test_qa_runner_advisory_writes_controller_events(shard): + """The QA runner's `_emit_qa_controller_advisory` synthesizes a + single-branch decision from a verdict dict and writes three rows + (decision + difficulty + budget_allocation) to controller_events. + Wired into the cache-miss path of `arborist.qa.runner.ask` so + every live QA cycle produces a row by default.""" + from arborist.qa.runner import _emit_qa_controller_advisory + + conn, _ = shard + verdict = { + "audit_mode": "UNGROUNDED", + "n_quotes": 4, + "n_verified": 1, + "unverified_quotes": '["q1", "q2", "q3"]', + "verifier_method": "quote", + } + with transaction(conn): + _emit_qa_controller_advisory( + conn, "abc123def456ghi7" * 4, verdict + ) + + kinds = [ + r["event_kind"] + for r in conn.execute( + "SELECT event_kind FROM controller_events ORDER BY event_id" + ).fetchall() + ] + assert kinds.count("controller_decision") == 1 + assert kinds.count("controller_difficulty") == 1 + # single-branch chunk → 1 allocation row. + assert kinds.count("controller_budget_allocation") >= 1 + + +def test_qa_runner_advisory_is_non_blocking_on_bad_verdict(shard): + """Helper tolerates malformed verdict dicts — never raises into + the QA hot path. The wire-up in ask() also wraps in try/except, + so this test just exercises the helper's defensive parsing.""" + from arborist.qa.runner import _emit_qa_controller_advisory + + conn, _ = shard + # Bad shape: missing n_quotes, garbage unverified_quotes JSON. + bad_verdict = { + "audit_mode": "HYBRID", + "unverified_quotes": "not-valid-json[", + "verifier_method": "paraphrase", + } + with transaction(conn): + _emit_qa_controller_advisory(conn, "bad-cache-key", bad_verdict) + # Made it through; rows still written. + n = conn.execute( + "SELECT COUNT(*) FROM controller_events" + ).fetchone()[0] + assert n >= 1 + + +def test_qa_runner_advisory_idempotent_on_same_verdict(shard): + """Calling the helper twice with the same verdict + cache_key + is a no-op on the second call (the underlying UNIQUE constraint + on controller_events catches it).""" + from arborist.qa.runner import _emit_qa_controller_advisory + + conn, _ = shard + verdict = { + "audit_mode": "STRICT", + "n_quotes": 3, + "n_verified": 3, + "unverified_quotes": "[]", + "verifier_method": "quote", + } + cache_key = "stable-cache-key-stable-cache" + with transaction(conn): + _emit_qa_controller_advisory(conn, cache_key, verdict) + first_count = conn.execute( + "SELECT COUNT(*) FROM controller_events" + ).fetchone()[0] + with transaction(conn): + _emit_qa_controller_advisory(conn, cache_key, verdict) + second_count = conn.execute( + "SELECT COUNT(*) FROM controller_events" + ).fetchone()[0] + assert first_count == second_count