feat(aggregator): audit queue gets a consumer — drain by priority, not by threshold

CHANGELOG.md

@@ -8,6 +8,21 @@ All notable changes to AgentDescent are documented here. The format follows
 
 ### Added
 
+- **`AggregatorConfig.audit_drain_per_step` — the L-value consumer, the audit
+  queue's missing reader.** `AuditScheduler.submit()` computes
+  `priority = blast_radius × uncertainty / trust` for every merge, but with the
+  default `collect=False` nothing queues, and with `collect=True` nothing pops
+  the heap — the audit that actually runs is the inline `force_oracle` threshold
+  gate, which reads trust but not the ranking. `concepts.md` calls this *"the
+  priority queue has no consumer"* and says to treat the ranking as a priority
+  *model*, not work in flight.
+
+  `_drain_audit_queue(max_per_step)` pops the highest-priority queued audits
+  after each `finish_step`, runs `full_eval` on the stored `(base_state,
+  candidate)` pair, and updates trust from the oracle's verdict. The scheduler's
+  `collect` flag is forced on when `audit_drain_per_step > 0`. `EvolutionResult.
+  audit_drained` reports how many ran. Default `0` = the old behaviour exactly.
+
 - **`ReplaySampler` + `ProposalContext.rejected` — make the settled-evidence pool a consumer.** The aggregator's settled pool (stale / oversized / CAS-conflict evidence cards) was diagnostic-only: nothing read it back. Two consumers now close that loop.
 
   **`ReplaySampler`** (``agentdescent/sampling.py``) is a ``TaskSampler`` that up-weights tasks whose recent proposals were discarded — a signal that is different from pass rate (a task whose proposals keep going stale is not "easy", it is being out-competed by the parallel scheduler). Additive replay bonus, saturating at ``capped_at`` (default 10), scaled by ``temperature`` (default 0.0 = ``DifficultyWeighted`` identically). Wired via ``aggregator.set_settled_consumer(sampler.settle)`` on both the sync and async paths.

agentdescent/aggregator.py

@@ -138,6 +138,14 @@ class AggregatorConfig:
     #: two writers backing off by the same amount collide again on the same
     #: schedule.
     cas_backoff: float = 0.05
+    #: How many highest-priority queued audits the L-value consumer drains per
+    #: merge step. ``0`` (the default) keeps the old behaviour: the audit scheduler
+    #: computes priorities, ``force_oracle`` spends budget on a threshold gate, and
+    #: nothing spends it on the ranking. A positive value pops the heap after each
+    #: ``finish_step`` and runs ``full_eval`` on the candidates, updating trust --
+    #: the out-of-band audit process the design (section 5.3) calls for. The
+    #: scheduler's ``collect`` flag is forced on when this is set.
+    audit_drain_per_step: int = 0
     #: Rank candidates against their fusion on the cheap layer before putting one
     #: forward. **Off**, because the ranking is paid every round while the only
     #: decision it changes from the gate's is recoverable -- see
@@ -540,6 +548,11 @@ def __init__(
         #: better than one that counts a fraction of the run.
         self.meter = meter
         self.config = config or AggregatorConfig()
+        # Force the audit scheduler to collect when the L-value consumer is on:
+        # without this, submit() returns the priority without queueing, and
+        # _drain_audit_queue pops an empty heap.
+        if self.config.audit_drain_per_step > 0:
+            self.audit._collect = True
         # The decisions, as objects. Swapping one is the point; the defaults are
         # the code that used to be inline here, moved rather than rewritten.
         cfg = self.config
@@ -561,6 +574,9 @@ def __init__(
             install_policy(policy, verifier, cfg)
         self.buffer = EvidenceBuffer()
         self._posteriors: Dict[str, BetaPosterior] = defaultdict(BetaPosterior)
+        #: How many queued audits the L-value consumer drained and ran against
+        #: the oracle. Zero when ``audit_drain_per_step`` is 0 (the default).
+        self.audit_drained = 0
         # dev-branch survival counter for EMA-style promotion.
         # Artifacts this aggregator has seen, so `finalize` knows what to publish.
         # The survival *counter* belongs to the promotion policy; this is only the
@@ -742,6 +758,7 @@ def finish_step(self, items: List[Union["_Candidate", MergeReport]]) -> List[Mer
             # itself against a number nothing else in the run agrees with.
             self._observe_trust_region(report.category)
         self._age_and_promote(reports)
+        self._drain_audit_queue(self.config.audit_drain_per_step)
         return reports
 
     # -- dual-branch EMA promotion (section 4.5) -----------------------------
@@ -755,6 +772,95 @@ def _age_and_promote(self, reports: List[MergeReport]) -> None:
         for promotion in self.promotion_policy.observe(reports):
             self._promote(promotion.artifact_id)
 
+    def _drain_audit_queue(self, max_per_step: int = 0) -> int:
+        """Pop the highest-priority queued audits and run them against the oracle.
+
+        This is the L-value consumer the design (section 5.3) calls for and the
+        docstring of :class:`~agentdescent.scheduler.AuditScheduler` says does not
+        exist: ``submit`` computes a priority for every merge, ``force_oracle``
+        spends budget on a threshold gate, and nothing spends it on the *ranking*
+        the priority model produces. This does: it pops the highest-priority items
+        the queue holds, runs ``full_eval`` on each candidate, and updates trust.
+
+        ``max_per_step=0`` (the default) drains nothing, which is the old
+        behaviour exactly -- the queue is built (when ``collect=True``) and never
+        read. A caller sets it through :class:`AggregatorConfig`.
+
+        Returns how many audits it ran. The budget is the verifier's own
+        ``oracle_calls_remaining``; when it is spent the drain stops early.
+
+        Budget is checked **before** pop: popping an item and then discovering
+        the budget is gone discards the highest-priority audit without running
+        it. A pair needs two ``full_eval`` calls (base + candidate); checking
+        ``can_spend(2)`` prevents the half-measured case where the first call
+        consumes the last budget and the second falls back to the cheap layer,
+        making trust compare measurements from different task sets.
+        """
+        if max_per_step <= 0 or not getattr(self.audit, "_collect", False):
+            return 0
+        n = 0
+        for _ in range(max_per_step):
+            if not self._verifier_can_spend(2):
+                break
+            item = self.audit.pop()
+            if item is None:
+                break
+            payload = item.payload
+            if not isinstance(payload, tuple) or len(payload) != 3:
+                continue                   # malformed: will never get better
+            artifact_id, base_state, candidate = payload
+            try:
+                cand_full = self.verifier.full_eval(candidate)
+                base_full = self.verifier.full_eval(base_state)
+            except Exception as exc:      # noqa: BLE001 - transient oracle failure
+                # Re-queue: a dead oracle is often a dead *endpoint* (a rate
+                # limit, a timeout), and dropping the highest-priority audit
+                # for a transient fault loses work the ranking already paid
+                # for. Only keep trying while the drain has budget to retry.
+                self._requeue(item)
+                if self.meter is not None:
+                    self.meter.add("audit_drain_oracle_errors")
+                continue
+            self.audit.update_trust(
+                artifact_id, (cand_full > base_full) == (
+                    self.verifier.learned_eval(candidate)[0]
+                    > self.verifier.learned_eval(base_state)[0]))
+            n += 1
+        self.audit_drained += n
+        return n
+
+    def _requeue(self, item) -> None:
+        """Put a popped audit item back with its original priority.
+
+        Uses :meth:`AuditScheduler.requeue` rather than ``submit`` to preserve
+        the priority the ranking already paid for. ``submit`` recomputes
+        priority from ``blast_radius * uncertainty / trust``, and trust may
+        have moved since the item was first submitted."""
+        try:
+            self.audit.requeue(item)
+        except Exception:  # noqa: BLE001 - never let bookkeeping kill the drain
+            pass
+
+    def _verifier_can_spend(self, n: int = 1) -> bool:
+        """Whether the verifier has at least ``n`` oracle calls left.
+
+        Read through ``getattr`` rather than ``hasattr`` + direct access: the
+        shipped :class:`~agentdescent.verifier.ThreeLayerVerifier` exposes
+        ``budget.can_spend()``, but :class:`~agentdescent.policies.
+        VerifierProtocol` does not declare ``budget``, and a custom verifier
+        that satisfies the protocol has no such attribute. ``getattr`` with a
+        default keeps the protocol intact -- a verifier with no budget is one
+        that can always spend, which is the behaviour ``force_oracle`` had
+        before this method existed.
+        """
+        budget = getattr(self.verifier, "budget", None)
+        if budget is None:
+            return True
+        remaining = getattr(budget, "oracle_calls_remaining", None)
+        if remaining is None:
+            return True
+        return remaining >= n
+
     def _promote(self, artifact_id: str) -> None:
         """Copy dev onto stable, skipping the git work when they already agree.
 
@@ -920,7 +1026,7 @@ def _audit(self, artifact, artifact_id, best_state, best_diff,
 """
         _, uncertainty = self.verifier.learned_eval(best_state)
         self.audit.submit(best_diff.diff_id, artifact_id, artifact.blast_radius,
-                          uncertainty, payload=best_diff)
+                          uncertainty, payload=(artifact_id, artifact, best_state))
         # Trust is "how often does the cheap layer agree with the full held-out
         # set", and it has to be measurable WITHOUT spending oracle budget --
         # otherwise it is circular. It was: `force_oracle` fires on

agentdescent/async_evolve.py

@@ -1194,4 +1194,5 @@ def _merger() -> None:
         gate_pool.shutdown(wait=False)
     eng.cleanup()          # do not hold a scratch git repo for the whole process
     result.replay = _replay_stats(sampler)
+    result.audit_drained = getattr(eng.aggregator, "audit_drained", 0)
     return result

agentdescent/evolution.py

@@ -1280,6 +1280,9 @@ class EvolutionResult:
     #: `error` stays `None` while throughput quietly drops -- check this to tell a
     #: fast run from a lucky one.
     retired_workers: int = 0
+    #: How many queued audits the L-value consumer drained and ran against the
+    #: oracle. Zero when ``audit_drain_per_step`` is 0 (the default).
+    audit_drained: int = 0
     #: Replay-sampler statistics, when one was installed. Reports how many cards
     #: the settled pool pushed to the sampler, how many picks were affected by the
     #: replay bonus, and how many proposals carried rejection evidence:
@@ -3307,4 +3310,5 @@ def _bounded(u=None):
                              **_cost_fields(eng.meter))
     eng.cleanup()
     result.replay = _replay_stats(sampler)
+    result.audit_drained = getattr(aggregator, "audit_drained", 0)
     return result

agentdescent/metrics.py

@@ -78,6 +78,7 @@ class MeterSnapshot:
     redispatched: int = 0
     duplicates_dropped: int = 0
     cas_conflicts: int = 0
+    audit_drain_oracle_errors: int = 0
     sandbox_wait_s: float = 0.0
     sandbox_setup_s: float = 0.0
     sandboxes_created: int = 0
@@ -96,6 +97,7 @@ class MeterSnapshot:
     "cache_hits", "cache_misses", "cache_inflight_joins",
     "stale_considered", "stale_discarded",
     "redispatched", "duplicates_dropped", "cas_conflicts",
+    "audit_drain_oracle_errors",
     "sandbox_wait_s", "sandbox_setup_s", "sandboxes_created",
     "sandboxes_reused", "sandbox_failures", "env_mismatch",
 })
@@ -203,6 +205,10 @@ class Meter:
     #: writer, by construction; non-zero is how much contention a multi-writer
     #: configuration is actually producing.
     cas_conflicts: int = 0
+    #: Transient oracle failures the L-value consumer hit while draining the
+    #: audit queue -- rate limits, timeouts. The audit was requeued, not dropped,
+    #: so this is how many retries happened, not how many audits were lost.
+    audit_drain_oracle_errors: int = 0
 
     #: Sandbox accounting. Zero on the default single-workspace path; filled once
     #: the sandbox pool exists. Kept here from the start so the result schema
@@ -291,6 +297,7 @@ def snapshot(self) -> MeterSnapshot:
                 redispatched=self.redispatched,
                 duplicates_dropped=self.duplicates_dropped,
                 cas_conflicts=self.cas_conflicts,
+                audit_drain_oracle_errors=self.audit_drain_oracle_errors,
                 sandbox_wait_s=self.sandbox_wait_s,
                 sandbox_setup_s=self.sandbox_setup_s,
                 sandboxes_created=self.sandboxes_created,

agentdescent/scheduler.py

@@ -213,6 +213,18 @@ def submit(
                 heapq.heapify(self._items)
             return priority
 
+    def requeue(self, item: _AuditItem) -> None:
+        """Put a popped item back with its **original** priority.
+
+        ``submit`` recomputes priority from ``blast_radius * uncertainty /
+        trust``, and trust may have moved since the item was first submitted --
+        so re-submitting with the original ``blast_radius`` and ``uncertainty``
+        would give a different (and potentially lower) priority. This method
+        pushes the item back with the priority it was popped with, so relative
+        order is preserved across a transient failure and retry."""
+        with self._lock:
+            heapq.heappush(self._items, item)
+
     def force_oracle(self, blast_radius: float, artifact_id: str) -> bool:
         """High-impact or low-trust changes are forced through the oracle.
 

docs/api.md

@@ -39,6 +39,7 @@ EvolutionResult(
     forced_refreshes: int = 0,
     stragglers: int = 0,
     retired_workers: int = 0,
+    audit_drained: int = 0,
     replay: Optional[Dict[str, Any]] = None,
     usage: Usage = <factory>,
     wallclock: float = 0.0,
@@ -1061,6 +1062,7 @@ AggregatorConfig(
     accept_samples: int = 4000,
     cas_attempts: int = 3,
     cas_backoff: float = 0.05,
+    audit_drain_per_step: int = 0,
     fusion_tournament: bool = False,
     bounded_gate: bool = False
 ) -> None
@@ -2915,6 +2917,7 @@ Allocates oracle budget by estimated value G-hat (design doc, 5.3).
 | method | what it does |
 |---|---|
 | `force_oracle(blast_radius: float, artifact_id: str) -> bool` | High-impact or low-trust changes are forced through the oracle. |
+| `requeue(item: _AuditItem) -> None` | Put a popped item back with its **original** priority. |
 | `update_trust(artifact_id: str, oracle_agreed: bool) -> None` | Raise trust when cheap eval agreed with the oracle, lower it when not. |
 
 ### `DurationEstimator(...)`