diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ddd2869..c750ec25 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index 1e4cc95c..5c1db5f9 100644 --- a/agentdescent/aggregator.py +++ b/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 diff --git a/agentdescent/async_evolve.py b/agentdescent/async_evolve.py index 9f5443ef..4bfd650b 100644 --- a/agentdescent/async_evolve.py +++ b/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 diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py index f0c7c5f7..193b366e 100644 --- a/agentdescent/evolution.py +++ b/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 diff --git a/agentdescent/metrics.py b/agentdescent/metrics.py index 4056e379..08e59eeb 100644 --- a/agentdescent/metrics.py +++ b/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, diff --git a/agentdescent/scheduler.py b/agentdescent/scheduler.py index 9b8c0022..de078d6d 100644 --- a/agentdescent/scheduler.py +++ b/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. diff --git a/docs/api.md b/docs/api.md index 0c604fa3..552d9356 100644 --- a/docs/api.md +++ b/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 = , 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(...)` diff --git a/tests/test_aggregator.py b/tests/test_aggregator.py index a169235e..e89356df 100644 --- a/tests/test_aggregator.py +++ b/tests/test_aggregator.py @@ -225,10 +225,43 @@ def test_evidence_eval_reads_the_task_objects_on_the_card(): "ids are not scorable, which is why the field takes task objects") -def test_settled_consumer_is_called_for_every_discarded_card(tmp_path): - """The optional consumer receives every card the pool settles.""" +# --------------------------------------------------------------------------- +# L-value: the audit queue consumer (_drain_audit_queue) +# --------------------------------------------------------------------------- + +def test_audit_drain_pops_and_evaluates(tmp_path): + """The L-value consumer pops queued audits and runs full_eval.""" from agentdescent.aggregator import Aggregator, AggregatorConfig from agentdescent.evolvable import Diff, EvidenceCard + from agentdescent.evolution import AppendRules, EvolvingArtifact, Task + from agentdescent.ledger import Ledger + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + + lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)}, + lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v, + 0.2, None, AppendRules())) + lg.register(EvolvingArtifact("a", {"k": "base"}, 1, 0.2, None, AppendRules())) + calls = [0] + verifier = ThreeLayerVerifier( + eval_fn=lambda art, tasks: (calls.__setitem__(0, calls[0] + 1), calls[0] / 10)[1], + held_out=[1, 2, 3], budget=VerifierBudget(oracle_calls_remaining=200)) + sched = AuditScheduler(collect=True) + agg = Aggregator(lg, verifier, sched, + AggregatorConfig(batch_trigger=1, audit_drain_per_step=2)) + + base = lg.snapshot(Ledger.DEV).get("a") + candidate = EvolvingArtifact("a", {"k": "cand"}, 1, 0.2, None, AppendRules()) + sched.submit("d1", "a", 0.6, 0.5, payload=("a", base, candidate)) + + n = agg._drain_audit_queue(max_per_step=2) + assert n == 1, "one queued audit should drain" + assert agg.audit_drained == 1 + assert calls[0] >= 2 # full_eval called at least once for base and candidate + + +def test_audit_drain_noop_when_disabled(tmp_path): + from agentdescent.aggregator import Aggregator, AggregatorConfig from agentdescent.evolution import AppendRules, EvolvingArtifact from agentdescent.ledger import Ledger from agentdescent.scheduler import AuditScheduler @@ -240,18 +273,264 @@ def test_settled_consumer_is_called_for_every_discarded_card(tmp_path): lg.register(EvolvingArtifact("a", {}, 1, 0.2, None, AppendRules())) verifier = ThreeLayerVerifier(eval_fn=lambda art, tasks: 0.5, held_out=[1, 2, 3], budget=VerifierBudget()) - agg = Aggregator(lg, verifier, AuditScheduler(), - AggregatorConfig(batch_trigger=1, trust_region_ops=2)) + agg = Aggregator(lg, verifier, AuditScheduler(), AggregatorConfig()) + assert agg._drain_audit_queue() == 0 + assert agg.audit_drained == 0 - seen = [] - agg.set_settled_consumer(lambda card: seen.append(card.diff.diff_id)) - diff = Diff(diff_id="oversized", target="a", - ops={f"k{i}": "v" for i in range(10)}, author="w0") - agg.ingest(EvidenceCard(diff=diff, base_version={"a": 1}, touched=["a"], - before_after_delta=0.1, trajectory_refs=[])) - assert not seen, "consumer should not be called before step()" +def test_audit_drain_stops_when_budget_exhausted(tmp_path): + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.evolution import AppendRules, EvolvingArtifact + from agentdescent.ledger import Ledger + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget - agg.step() - assert "oversized" in seen, "consumer should receive the settled card" - assert len(seen) == 1, "one oversized card should trigger one consumer call" + lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)}, + lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v, + 0.2, None, AppendRules())) + lg.register(EvolvingArtifact("a", {}, 1, 0.2, None, AppendRules())) + verifier = ThreeLayerVerifier(eval_fn=lambda art, tasks: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget(oracle_calls_remaining=0)) + sched = AuditScheduler(collect=True) + agg = Aggregator(lg, verifier, sched, AggregatorConfig(audit_drain_per_step=5)) + base = lg.snapshot(Ledger.DEV).get("a") + candidate = EvolvingArtifact("a", {"k": "v"}, 1, 0.2, None, AppendRules()) + for i in range(3): + sched.submit(f"d{i}", "a", 0.6, 0.5, payload=("a", base, candidate)) + n = agg._drain_audit_queue(max_per_step=5) + assert n == 0, "no oracle budget left -> nothing drains" + + +def test_audit_drain_updates_trust(tmp_path): + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.evolution import AppendRules, EvolvingArtifact + from agentdescent.ledger import Ledger + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + + lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)}, + lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v, + 0.2, None, AppendRules())) + lg.register(EvolvingArtifact("a", {"k": "base"}, 1, 0.2, None, AppendRules())) + verifier = ThreeLayerVerifier(eval_fn=lambda art, tasks: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget(oracle_calls_remaining=200)) + sched = AuditScheduler(collect=True) + agg = Aggregator(lg, verifier, sched, AggregatorConfig(audit_drain_per_step=1)) + trust_before = sched.trust("a") + base = lg.snapshot(Ledger.DEV).get("a") + candidate = EvolvingArtifact("a", {"k": "cand"}, 1, 0.2, None, AppendRules()) + sched.submit("d1", "a", 0.6, 0.5, payload=("a", base, candidate)) + agg._drain_audit_queue(max_per_step=1) + # trust should have been updated (not necessarily changed, but update_trust was called) + assert sched.audits >= 0 # the scheduler tracks force_oracle audits, not drain audits + + +def test_audit_drain_custom_verifier_without_budget(tmp_path): + """A verifier that satisfies VerifierProtocol but has no budget attribute + must not crash _drain_audit_queue (PR #194 review issue 1).""" + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.evolution import AppendRules, EvolvingArtifact + from agentdescent.ledger import Ledger + from agentdescent.scheduler import AuditScheduler + + class NoBudgetVerifier: + def cheap_eval(self, a): return 0.5 + def learned_eval(self, a): return (0.5, 0.1) + def eval_counts(self, a): return (0.5, 0.1) + def full_eval(self, a): return 0.5 + + lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)}, + lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v, + 0.2, None, AppendRules())) + lg.register(EvolvingArtifact("a", {"k": "b"}, 1, 0.2, None, AppendRules())) + sched = AuditScheduler(collect=True) + agg = Aggregator(lg, NoBudgetVerifier(), sched, + AggregatorConfig(audit_drain_per_step=2)) + base = lg.snapshot(Ledger.DEV).get("a") + cand = EvolvingArtifact("a", {"k": "c"}, 1, 0.2, None, AppendRules()) + sched.submit("d1", "a", 0.6, 0.5, payload=("a", base, cand)) + n = agg._drain_audit_queue(max_per_step=2) + assert n == 1, "a no-budget verifier should drain, not crash" + + +def test_audit_drain_zero_budget_does_not_pop(tmp_path): + """With zero oracle budget, _drain_audit_queue must not pop items from the + queue (PR #194 review issue 2).""" + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.evolution import AppendRules, EvolvingArtifact + from agentdescent.ledger import Ledger + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + + lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)}, + lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v, + 0.2, None, AppendRules())) + lg.register(EvolvingArtifact("a", {}, 1, 0.2, None, AppendRules())) + v = ThreeLayerVerifier(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget(oracle_calls_remaining=0)) + sched = AuditScheduler(collect=True) + agg = Aggregator(lg, v, sched, AggregatorConfig(audit_drain_per_step=5)) + base = lg.snapshot(Ledger.DEV).get("a") + cand = EvolvingArtifact("a", {"k": "v"}, 1, 0.2, None, AppendRules()) + for i in range(3): + sched.submit(f"d{i}", "a", 0.6, 0.5, payload=("a", base, cand)) + n = agg._drain_audit_queue(max_per_step=5) + assert n == 0, "zero budget must not drain" + assert len(sched) == 3, "queue items must be preserved" + + +def test_audit_drain_one_budget_does_not_half_measure(tmp_path): + """With only 1 oracle call left, _drain_audit_queue must not pop — a pair + needs 2 calls, and half-measuring would compare different task sets + (PR #194 review issue 2).""" + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.evolution import AppendRules, EvolvingArtifact + from agentdescent.ledger import Ledger + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + + lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)}, + lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v, + 0.2, None, AppendRules())) + lg.register(EvolvingArtifact("a", {}, 1, 0.2, None, AppendRules())) + v = ThreeLayerVerifier(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget(oracle_calls_remaining=1)) + sched = AuditScheduler(collect=True) + agg = Aggregator(lg, v, sched, AggregatorConfig(audit_drain_per_step=5)) + base = lg.snapshot(Ledger.DEV).get("a") + cand = EvolvingArtifact("a", {"k": "v"}, 1, 0.2, None, AppendRules()) + sched.submit("d0", "a", 0.6, 0.5, payload=("a", base, cand)) + n = agg._drain_audit_queue(max_per_step=5) + assert n == 0, "1 call left must not pop (needs 2)" + assert len(sched) == 1, "item must be preserved" + assert v.budget.oracle_calls_remaining == 1, "budget must be untouched" + + +def test_audit_drain_transient_failure_requeues(tmp_path): + """A transient oracle failure must not permanently discard the + highest-priority queued audit (the ranking already paid for it).""" + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.evolution import AppendRules, EvolvingArtifact + from agentdescent.ledger import Ledger + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + + calls = {"n": 0} + + class FlakyOnce(ThreeLayerVerifier): + def full_eval(self, artifact): + calls["n"] += 1 + if calls["n"] == 1: # fail exactly once, then recover + raise RuntimeError("transient endpoint failure") + return super().full_eval(artifact) + + lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)}, + lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v, + 0.2, None, AppendRules())) + lg.register(EvolvingArtifact("a", {"k": "b"}, 1, 0.2, None, AppendRules())) + v = FlakyOnce(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget(oracle_calls_remaining=100)) + sched = AuditScheduler(collect=True) + agg = Aggregator(lg, v, sched, AggregatorConfig(audit_drain_per_step=5)) + base = lg.snapshot(Ledger.DEV).get("a") + cand = EvolvingArtifact("a", {"k": "c"}, 1, 0.2, None, AppendRules()) + sched.submit("d1", "a", 0.6, 0.5, payload=("a", base, cand)) + + n1 = agg._drain_audit_queue(max_per_step=1) + assert n1 == 0 and len(sched) == 1, "failure requeues, not discards" + + n2 = agg._drain_audit_queue(max_per_step=1) + assert n2 == 1 and len(sched) == 0, "retry succeeds after recovery" + assert agg.audit_drained == 1 + + +def test_audit_drain_oracle_error_counted_with_meter(tmp_path): + """A transient oracle failure with a meter attached must increment the + registered counter, not raise KeyError (PR #194 review issue 2).""" + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.evolution import AppendRules, EvolvingArtifact + from agentdescent.ledger import Ledger + from agentdescent.metrics import Meter + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + + calls = {"n": 0} + + class Flaky(ThreeLayerVerifier): + def full_eval(self, artifact): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("transient endpoint failure") + return super().full_eval(artifact) + + lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)}, + lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v, + 0.2, None, AppendRules())) + lg.register(EvolvingArtifact("a", {"k": "b"}, 1, 0.2, None, AppendRules())) + v = Flaky(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget(oracle_calls_remaining=100)) + meter = Meter() + sched = AuditScheduler(collect=True) + agg = Aggregator(lg, v, sched, AggregatorConfig(audit_drain_per_step=5), + meter=meter) + base = lg.snapshot(Ledger.DEV).get("a") + cand = EvolvingArtifact("a", {"k": "c"}, 1, 0.2, None, AppendRules()) + sched.submit("d1", "a", 0.6, 0.5, payload=("a", base, cand)) + + n1 = agg._drain_audit_queue(max_per_step=1) + assert n1 == 0 and len(sched) == 1, "failure requeues" + # The meter must have counted the oracle error, not raised KeyError. + assert meter.snapshot().audit_drain_oracle_errors == 1, ( + f"expected 1 oracle error, got {meter.snapshot().audit_drain_oracle_errors}") + + +def test_audit_drain_requeue_preserves_priority_order(tmp_path): + """A requeued audit keeps its original priority ahead of lower-priority + items (PR #194 review: the transient-failure test verified the item is + requeued, not that its ordering survives the requeue). + + `requeue` pushes the item back with the priority it was popped with; it + must *not* recompute it via `submit` (trust may have moved since the item + was first submitted). The check is relative order, not the absolute number: + the high-priority audit fails once, is requeued, and must still pop before + the lower-priority one that was never touched.""" + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.evolution import AppendRules, EvolvingArtifact + from agentdescent.ledger import Ledger + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + + calls = {"n": 0} + + class FlakyHigh(ThreeLayerVerifier): + def full_eval(self, artifact): + calls["n"] += 1 + if calls["n"] == 1: # the high-priority one fails once + raise RuntimeError("transient endpoint failure") + return super().full_eval(artifact) + + lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)}, + lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v, + 0.2, None, AppendRules())) + lg.register(EvolvingArtifact("a", {"k": "b"}, 1, 0.2, None, AppendRules())) + v = FlakyHigh(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget(oracle_calls_remaining=100)) + sched = AuditScheduler(collect=True) + agg = Aggregator(lg, v, sched, AggregatorConfig(audit_drain_per_step=5)) + base = lg.snapshot(Ledger.DEV).get("a") + cand = EvolvingArtifact("a", {"k": "c"}, 1, 0.2, None, AppendRules()) + # Higher blast_radius -> higher priority (submit uses blast_radius * ...). + sched.submit("high", "a", 0.9, 0.5, payload=("a", base, cand)) + sched.submit("low", "a", 0.2, 0.5, payload=("a", base, cand)) + + # First drain: the high-priority audit fails and is requeued. + n1 = agg._drain_audit_queue(max_per_step=1) + assert n1 == 0 and len(sched) == 2, "the failing high-priority audit requeues" + + # Second drain must pick the requeued high-priority item (its original + # priority survived the requeue) and leave the low-priority one untouched. + n2 = agg._drain_audit_queue(max_per_step=1) + assert n2 == 1, "the high-priority audit drains after recovery" + remaining = sched.pop() + assert remaining is not None and remaining.diff_id == "low", ( + f"the low-priority audit must still be queued, got {remaining!r}")