diff --git a/control-plane/README.md b/control-plane/README.md index 95fe5a3..5b69b72 100644 --- a/control-plane/README.md +++ b/control-plane/README.md @@ -190,7 +190,7 @@ instance would need Postgres `LISTEN`/`NOTIFY` or a shared queue, which is out o **Configuration** (`mergit_cp/config.py`). Required: `DATABASE_URL`, `APP_DOMAIN`, `SESSION_SECRET` (32+ characters), `GATEWAY_URL`, `STORE_URL`, `AGENTS_FILE`, `PLANNER_MODEL`, `EVENTS_TOKEN`, `CHAIN_EVENTS_TOKEN`, `WORKER_TOKEN`, plus the model provider key. Defaults: `CHAIN_ID` (84532), `PLANS_PER_HOUR` (10), -`PLANNER_MAX_TASKS` (8), `MAX_ATTEMPTS` (3), `REPO_HOSTS` (github.com), `CORS_ORIGINS`, `SSE_POLL_S` (1), +`SUGGESTIONS_PER_HOUR` (60, exploring suggestions only), `PLANNER_MAX_TASKS` (8), `MAX_ATTEMPTS` (3), `REPO_HOSTS` (github.com), `CORS_ORIGINS`, `SSE_POLL_S` (1), `OUTCOME_POLL_S` (1). `WORKER_TOKEN` is this deployment's own choice, like `CHAIN_EVENTS_TOKEN`: spec/events-v0.md does not name a token for this leg either, since the whole endpoint is new in O7. The planner model must be priced in diff --git a/control-plane/mergit_cp/config.py b/control-plane/mergit_cp/config.py index f7a5c52..3bc0c8a 100644 --- a/control-plane/mergit_cp/config.py +++ b/control-plane/mergit_cp/config.py @@ -66,10 +66,39 @@ class Settings(BaseSettings): """How often finished jobs are checked for a retry, an escalation or plan completion (plan O6).""" reputation_poll_s: float = Field(default=5.0, gt=0) """How often finished tasks' costs are read from the gateway and the posteriors refreshed.""" + reputation_cost_batch: int = Field(default=200, ge=1) + """Pending costs read from the gateway in one pass of ``record_missing_costs``; the rest wait for the + next poll. Bounds one pass to a known number of sequential gateway requests, however large the backlog + (an outage, or heavy volume) grows.""" + reputation_cost_budget_s: float = Field(default=10.0, gt=0) + """A pass of ``record_missing_costs`` stops asking for more costs once it has run this long, even inside + the batch above: a gateway that answers but is slow must not make one pass run arbitrarily long.""" routing_epsilon: float = Field(default=0.1, ge=0) """Slack on the task budget when a suggestion skips an agent whose cost would not fit (paper §9.3).""" routing_min_observations: int = Field(default=5, ge=1) """Counted outcomes below which a suggested agent is called unproven.""" + routing_probation_s: int = Field(default=604800, gt=0) + """How long, in seconds, an operator's agents are left out of suggestions after a proven falsification on + any of their tasks (an invalid verdict or an upheld challenge, whatever the plan's funding). Paper §7.5 + iii sets ``p = 1`` for a probationary period without fixing its length, and audits rather than excludes; + exclusion and seven days are our parameterisation, not the paper's.""" + routing_operator_concurrency_cap: int = Field(default=5, ge=1) + """Unresolved dispatches (no release or return yet) an operator's agents may hold between them before + its agents are left out of suggestions. Paper §7.5 ii caps concurrent exposure per operator and sizes + the audit rate for the cap; nothing here sizes an audit rate yet. The value 5 (one column of the paper's + table) and counting from dispatch rather than from execution are our parameterisation, not the paper's.""" + routing_min_feasible_probability: float = Field(default=0.01, gt=0, le=0.25) + """The smallest chance that one Thompson draw of an agent's cost fits a task's budget (with + ``routing_epsilon``) for an exploring suggestion to draw that agent at all; below it the agent is listed + as over budget for that task. The chance is computed from the agent's cost posterior (the upper + regularized incomplete gamma), once per agent and task, not per draw, so a lucky draw cannot bring such an + agent back. Capped at 0.25 because an agent whose mean cost fits always has at least 1/e, so the cap + keeps exploring from leaving out anything greedy would suggest. Paper §9.3 screens only each draw on + budget (so an agent whose cost is merely uncertain is not shut out for good) and has no such floor; the + floor and its value are our parameterisation, not the paper's.""" + suggestions_per_hour: int = Field(default=60, ge=1) + """Exploring suggestions per requester per hour. Each one appends a replay record per task to + ``routing_decisions``, which is append-only; greedy suggestions write nothing and are not limited.""" reputation_cost_prior_micros: int = Field(default=500_000, ge=1) """The cost prior's mean for an agent with no history, in USDC micros. Configuration, not a measurement: it only matters until an agent has a few validated outcomes.""" diff --git a/control-plane/mergit_cp/eligibility.py b/control-plane/mergit_cp/eligibility.py new file mode 100644 index 0000000..c5ae3f6 --- /dev/null +++ b/control-plane/mergit_cp/eligibility.py @@ -0,0 +1,122 @@ +"""Which roster agents a suggestion must leave out, and why (paper §7.5 ii and iii). Read-only. + +Two rules, both per operator rather than per agent, since an operator running several agents can defect +across all of them at once: + +* **Probation** (``IN_PROBATION``). An operator is in probation while any task dispatched to any of its + agents has a proven falsification less than ``routing_probation_s`` old. A falsification is a + ``money_events`` row with ``kind = 'verdict.recorded' AND verdict = 'invalid'`` or + ``kind = 'challenge.resolved' AND upheld``, timed by its ``received_at``. A task is falsified once: the + window runs from the first such row for it, so a second ruling on the same false receipt does not restart + it. The plan's funding does not matter. Invariant 6 keeps free-credit work out of the posteriors because + it must not *earn* reputation; a proven false receipt earns nothing, it is evidence of fraud, and + ignoring it on a credits plan would let an operator forge where the stakes are lowest and stay suggested + where they are highest. A validated failure (``verdict = 'fail'``) is not a falsification: it is a + capability signal, and the posteriors already carry it. +* **The concurrency cap** (``AT_CAP``). An operator is at cap while its agents hold, between them, at least + ``routing_operator_concurrency_cap`` unresolved wallet-funded dispatches. Only wallet-funded ones count, + unlike probation: invariant 6 says free-credit work must not influence routing, and here it could do so + cheaply and on purpose, since anyone with a credits grant could send a few plans to a rival operator's + agents and have its whole roster shown as at cap to everyone. A paid dispatch costs its requester the + escrow, so holding an operator's capacity is not free. Unresolved means a ``task_dispatches`` row with no + ``task.released`` or ``task.returned`` row yet, except: + + - a dispatch whose ``jobs`` row has ``status = 'error' AND receipt_root IS NULL``: nothing was sealed for + that attempt, so there is nothing to forge and nothing in flight, though its escrow only comes back + on the chain's own deadline; + - a dispatch with no ``jobs`` row at all that is older than ``dispatch.DISPATCH_DEADLINE_S``: it never + escrowed and never will (the gateway rejected it, or the outbox gave up), and since nothing will ever + arrive to resolve it, it would otherwise count against its operator forever. Reusing the dispatch's own + deadline means it stops counting exactly when the dispatch itself gives up. One still within its + deadline is a real in-flight commitment and counts. + +Probation is checked first, so an operator that is both reports ``IN_PROBATION``. The cap is not relaxed +during probation (paper §7.5 iii); here that is moot, since either one already leaves the agent out. + +Which operator an agent belongs to comes from the roster passed in, the same one the suggestions are for: +a dispatch or falsification by an agent no longer on the roster counts toward no operator. + +Ported from the reviewed ``in_probation`` and ``_UNRESOLVED_BY_AGENT`` on ``archive/auto-dispatch-routing``. +That branch read probation from its own ``reputation_observations`` log; this one has none, so it reads +``money_events`` through ``task_dispatches`` directly, with the same rule. +""" + +from datetime import datetime, timedelta + +from psycopg import AsyncConnection +from psycopg.rows import TupleRow + +from .agents import Agent +from .config import Settings +from .dispatch import DISPATCH_DEADLINE_S + +IN_PROBATION = "operator in probation" +AT_CAP = "operator at concurrency cap" +"""The reasons ``ineligible`` gives. Both are transient: each clears on its own as time passes or as +dispatches resolve.""" + +# The agents among `agent_ids` with a task whose first falsification was received after `since`. +_FALSIFIED_AGENTS = """ +SELECT DISTINCT d.agent_id +FROM task_dispatches d +WHERE d.agent_id = ANY(%(agent_ids)s) + AND ( + SELECT min(m.received_at) FROM money_events m + WHERE m.task_id = d.task_id + AND ((m.kind = 'verdict.recorded' AND m.verdict = 'invalid') + OR (m.kind = 'challenge.resolved' AND m.upheld)) + ) > %(since)s +""" + +# Unresolved wallet-funded dispatches per agent; `ineligible` sums them per operator. `jobs.task_id` is its +# primary key, so the join adds at most one row per dispatch, and `j.status IS NULL` is only true when no +# `jobs` row exists at all (a dispatch that did escrow always has one, whatever its later status). +_UNRESOLVED_BY_AGENT = """ +SELECT d.agent_id, count(*) +FROM task_dispatches d +JOIN plans p ON p.plan_uuid = d.plan_uuid AND p.funding = 'wallet' +LEFT JOIN jobs j ON j.task_id = d.task_id +WHERE d.agent_id = ANY(%(agent_ids)s) + AND NOT EXISTS ( + SELECT 1 FROM money_events m + WHERE m.task_id = d.task_id AND m.kind IN ('task.released', 'task.returned') + ) + AND NOT COALESCE(j.status = 'error' AND j.receipt_root IS NULL, false) + AND NOT (j.status IS NULL AND d.dispatched_at < %(stale_before)s) +GROUP BY d.agent_id +""" + + +async def ineligible( + conn: AsyncConnection[TupleRow], agents: list[Agent], settings: Settings, now: datetime +) -> dict[str, str]: + """Every agent in ``agents`` that must be left out of a suggestion at ``now``, mapped to the reason. + An agent that may be suggested, as far as these two rules go, is absent: presence means excluded.""" + if not agents: + return {} + agent_ids = [a.agent_id for a in agents] + + cur = await conn.execute( + _FALSIFIED_AGENTS, + {"agent_ids": agent_ids, "since": now - timedelta(seconds=settings.routing_probation_s)}, + ) + falsified = {row[0] for row in await cur.fetchall()} + in_probation = {a.operator for a in agents if a.agent_id in falsified} + + cur = await conn.execute( + _UNRESOLVED_BY_AGENT, + {"agent_ids": agent_ids, "stale_before": now - timedelta(seconds=DISPATCH_DEADLINE_S)}, + ) + unresolved_by_agent: dict[str, int] = {agent_id: count for agent_id, count in await cur.fetchall()} + unresolved_by_operator: dict[str, int] = {} + for a in agents: + count = unresolved_by_agent.get(a.agent_id, 0) + unresolved_by_operator[a.operator] = unresolved_by_operator.get(a.operator, 0) + count + + out: dict[str, str] = {} + for a in agents: + if a.operator in in_probation: + out[a.agent_id] = IN_PROBATION + elif unresolved_by_operator[a.operator] >= settings.routing_operator_concurrency_cap: + out[a.agent_id] = AT_CAP + return out diff --git a/control-plane/mergit_cp/reputation.py b/control-plane/mergit_cp/reputation.py index a40f4c8..73f9760 100644 --- a/control-plane/mergit_cp/reputation.py +++ b/control-plane/mergit_cp/reputation.py @@ -2,8 +2,9 @@ Three steps, none of which writes anything the others do not own: -1. ``task_outcomes`` (a view, migration 0015) says which tasks are validated outcomes. Only those count - (invariant 4); a release nobody audited, or a deadline return, is not evidence. +1. ``task_outcomes`` (a view, migrations 0015 and 0020) says which tasks are validated outcomes: audited, or + challenged and adjudicated. Only those count (invariant 4); a release nobody audited or challenged, or a + deadline return, is not evidence. 2. ``record_missing_costs`` copies each such task's metered cost from the gateway's on-chain commitment into ``task_costs``, once. A cost that cannot be read yet leaves the outcome pending; it is never taken as zero. @@ -22,6 +23,7 @@ import asyncio import logging +import time from contextlib import suppress from dataclasses import dataclass from typing import Any, cast @@ -41,6 +43,17 @@ class AgentCategory: category: str +class GatewayUnavailable(Exception): + """The gateway cannot be asked right now: unreachable, failing, refusing us, or not answering with a list + of commitments. Every task would get the same answer, so the pass stops and tries again next time.""" + + +class BadCommitment(ValueError): + """The gateway's answer about this one task cannot be used (another task's commitment, a cost that is not + a non-negative integer, a row that is not an object, a request refused for this task id). Other tasks are + not affected, so the pass goes on to them.""" + + class GatewayCosts: """Reads a task's committed cost from the gateway (``GET /v0/commitments?task_id=``).""" @@ -54,47 +67,84 @@ async def aclose(self) -> None: await self._http.aclose() async def cost_micros(self, task_id: str) -> int | None: - """The cost in USDC micros, or None when the gateway has not seen the commitment yet. Anything else - (an unreachable gateway, an answer that is not a commitment) raises.""" - response = await self._http.get(self._url, params={"task_id": task_id, "limit": 1}) - response.raise_for_status() - rows = cast("list[dict[str, Any]]", response.json()) - if not isinstance(rows, list): # pyright: ignore[reportUnnecessaryIsInstance] - raise ValueError("the gateway's answer is not a list") + """The cost in USDC micros, or None when the gateway has not seen the commitment yet. Raises + ``GatewayUnavailable`` when the gateway as a whole cannot answer, ``BadCommitment`` when its answer + about this task is unusable.""" + try: + response = await self._http.get(self._url, params={"task_id": task_id, "limit": 1}) + except httpx.HTTPError as problem: + raise GatewayUnavailable(f"{type(problem).__name__}: {problem}") from problem + if response.status_code in (400, 422): # the gateway refused this task id, not every request + raise BadCommitment(f"HTTP {response.status_code}") + if not response.is_success: + raise GatewayUnavailable(f"HTTP {response.status_code}") + try: + rows: object = response.json() + except ValueError as problem: + raise GatewayUnavailable("the gateway's answer is not JSON") from problem + if not isinstance(rows, list): + raise GatewayUnavailable("the gateway's answer is not a list") if not rows: return None - row = rows[0] - if row.get("task_id") != task_id: - raise ValueError("the gateway answered with another task's commitment") - cost = int(row["cost_micros"]) + row = cast("list[object]", rows)[0] + if not isinstance(row, dict): + raise BadCommitment("the gateway's commitment is not an object") + fields = cast("dict[str, Any]", row) + if fields.get("task_id") != task_id: + raise BadCommitment("the gateway answered with another task's commitment") + raw = fields.get("cost_micros") + if not isinstance(raw, str | int) or isinstance(raw, bool): + raise BadCommitment("the commitment has no cost") + try: + cost = int(raw) + except ValueError as problem: + raise BadCommitment(f"the cost {raw!r} is not an integer") from problem if cost < 0: - raise ValueError("a negative cost") + raise BadCommitment("a negative cost") return cost -async def record_missing_costs(pool: Pool, costs: GatewayCosts, limit: int = 50) -> int: - """Writes ``task_costs`` for counted outcomes that lack one. Returns how many were written. A task whose - commitment is not there yet, or a gateway that cannot be asked, leaves it for the next pass.""" +async def record_missing_costs( + pool: Pool, costs: GatewayCosts, *, batch: int = 200, budget_s: float = 10.0 +) -> int: + """Writes ``task_costs`` for up to ``batch`` counted outcomes that lack one, stopping early past + ``budget_s``. Returns how many were written. + + Pending outcomes are tried oldest first. A task whose commitment is not there yet, or whose answer is + unusable, is left for the next pass and does not hold back the tasks after it (a commitment that never + arrives must not block every newer outcome). A gateway that cannot answer at all ends the pass early, + since every remaining task would fail the same way. ``batch`` and ``budget_s`` bound one pass to a known + number of sequential gateway requests and a known wall-clock time, however large the backlog (an outage, + or heavy volume) grows, or however slow a gateway that does answer is; the rest wait for the next poll, + which also serves ``GET /v0/agents/reputation`` from the same table.""" async with pool.connection() as conn: cur = await conn.execute( - "SELECT task_id FROM task_outcomes WHERE cost_micros IS NULL ORDER BY position LIMIT %s", (limit,) + "SELECT task_id FROM task_outcomes WHERE cost_micros IS NULL ORDER BY position LIMIT %s", + (batch,), ) pending = [row[0] for row in await cur.fetchall()] written = 0 + started = time.monotonic() for task_id in pending: + if time.monotonic() - started > budget_s: + log.warning("costs not read this pass (budget exceeded, %d left pending)", len(pending) - written) + break try: cost = await costs.cost_micros(task_id) - except (httpx.HTTPError, ValueError, KeyError, TypeError) as problem: - log.warning("cost of %s not read: %s: %s", task_id, type(problem).__name__, problem) - break # the gateway is unwell or odd; do not hammer it for each remaining task + except GatewayUnavailable as problem: + log.warning("costs not read this pass (%d pending): %s", len(pending), problem) + break # the gateway is unwell; do not hammer it for each remaining task + except BadCommitment as problem: + log.warning("cost of %s not read: %s", task_id, problem) + continue if cost is None: continue async with pool.connection() as conn: - await conn.execute( + cur = await conn.execute( "INSERT INTO task_costs (task_id, cost_micros) VALUES (%s, %s) ON CONFLICT DO NOTHING", (task_id, cost), ) - written += 1 + written += cur.rowcount # 0 when another instance recorded it first return written @@ -103,6 +153,9 @@ async def refresh_posteriors(pool: Pool, settings: Settings) -> int: prior = Posterior.prior(Params(tau0=float(settings.reputation_cost_prior_micros))) async with pool.connection() as conn, conn.transaction(): await conn.execute("SELECT pg_advisory_xact_lock(hashtext('reputation'))") + # An outcome whose cost is still pending waits (it is never taken as zero). A recorded cost of 0 is a + # true copy of the commitment: the outcome's success or failure counts like any other, but the cost + # model needs c > 0 (paper §8.3), so it adds nothing to the cost posterior rather than a made-up cost. cur = await conn.execute( "SELECT agent_id, category, success, cost_micros FROM task_outcomes" " WHERE funding = 'wallet' AND cost_micros IS NOT NULL ORDER BY position, task_id" @@ -110,7 +163,7 @@ async def refresh_posteriors(pool: Pool, settings: Settings) -> int: folded: dict[tuple[str, str], Posterior] = {} for agent_id, category, success, cost in await cur.fetchall(): key = (agent_id, category) - folded[key] = folded.get(key, prior).observe(bool(success), float(cost)) + folded[key] = folded.get(key, prior).observe(bool(success), float(cost) if cost > 0 else None) await conn.execute("DELETE FROM agent_posteriors") for (agent_id, category), p in sorted(folded.items()): await conn.execute( @@ -131,7 +184,12 @@ def __init__(self, pool: Pool, settings: Settings, costs: GatewayCosts) -> None: self._costs = costs async def update(self) -> None: - await record_missing_costs(self._pool, self._costs) + await record_missing_costs( + self._pool, + self._costs, + batch=self._settings.reputation_cost_batch, + budget_s=self._settings.reputation_cost_budget_s, + ) await refresh_posteriors(self._pool, self._settings) async def run(self, poll_s: float, stop: asyncio.Event) -> None: diff --git a/control-plane/mergit_cp/reputation_model.py b/control-plane/mergit_cp/reputation_model.py index 0524e71..637d7b5 100644 --- a/control-plane/mergit_cp/reputation_model.py +++ b/control-plane/mergit_cp/reputation_model.py @@ -47,18 +47,20 @@ def prior(cls, params: Params) -> "Posterior": raise ValueError("every prior parameter must be positive") return cls(params.alpha0, params.beta0, params.kappa0, params.tau0) - def observe(self, success: bool, cost: float, gamma: float = GAMMA) -> "Posterior": - """One validated outcome. Every parameter decays, then the outcome adds to one Beta parameter, and the - cost (of a failure too) adds to the Gamma pair.""" - if not math.isfinite(cost) or cost < 0: - raise ValueError("cost must be a finite, non-negative number") + def observe(self, success: bool, cost: float | None, gamma: float = GAMMA) -> "Posterior": + """One validated outcome. The Beta pair decays, then the outcome adds 1 to one of them (paper §8.2); + every validated outcome does this. The Gamma pair decays and gains the cost (of a failure too) only + when there is a cost to learn from. ``None`` means the outcome carries no usable cost (its receipt + metered 0, which the cost model's ``c > 0``, §8.3, cannot take) and leaves kappa and tau alone.""" + if cost is not None and (not math.isfinite(cost) or cost <= 0): + raise ValueError("cost must be a finite, positive number, or None for no cost evidence") if not 0 < gamma <= 1: raise ValueError("gamma must be in (0, 1]") return Posterior( alpha=gamma * self.alpha + (1.0 if success else 0.0), beta=gamma * self.beta + (0.0 if success else 1.0), - kappa=gamma * self.kappa + 1.0, - tau=gamma * self.tau + cost, + kappa=self.kappa if cost is None else gamma * self.kappa + 1.0, + tau=self.tau if cost is None else gamma * self.tau + cost, n=self.n + 1, ) diff --git a/control-plane/mergit_cp/routes_routing.py b/control-plane/mergit_cp/routes_routing.py index a91af6f..45c255e 100644 --- a/control-plane/mergit_cp/routes_routing.py +++ b/control-plane/mergit_cp/routes_routing.py @@ -3,23 +3,33 @@ Advice only. Approval is unchanged: the requester still submits ``agents`` to ``/approve``, and may ignore every suggestion. Free-credit work never shapes a suggestion, because it is not in the posteriors it ranks from (invariant 6). See ``routing.py`` for the two modes. + +Every roster agent not in a task's ``picks`` is in its ``ineligible`` list with one reason, the most +fundamental that applies: probation or the operator cap (``eligibility``, checked first, so such an agent is +never ranked), then the budget (``routing.OVER_BUDGET``, or ``OVER_BUDGET_IN_THIS_DRAW`` when exploring), then +the ``max_attempts`` cut. When ``picks`` is empty, ``reason`` says only what is true of every agent left out: +whether one merely missed this draw (``NO_FIT_THIS_DRAW``), and otherwise whether the budget call rests on +evidence or only on the configured cost prior (``NO_FIT`` or ``NO_FIT_ON_PRIOR_ALONE``): ``_empty_reason``. """ import json import random import secrets +from datetime import UTC, datetime from typing import Annotated, Any from uuid import UUID from fastapi import APIRouter, Depends, HTTPException, Query +from psycopg import AsyncConnection +from psycopg.rows import TupleRow from psycopg.types.json import Jsonb -from . import plans +from . import eligibility, plans from .agents import Agent from .auth import PoolDep, PrincipalDep, SettingsDep from .reputation_model import Params, Posterior from .routes_plans import get_agents -from .routing import Pick, rank +from .routing import BELOW_MAX_ATTEMPTS, OVER_BUDGET_IN_THIS_DRAW, Pick, rank router = APIRouter(prefix="/v0/plans", tags=["routing"]) @@ -30,8 +40,36 @@ "cover the cost of an unproven agent that fails, so choosing one can cost you." ) +# A task's ``reason`` when it has no picks (``_empty_reason`` chooses). ``NO_FIT`` is the text this field +# always had; it now stands only when every agent ranked was over budget on its posterior (not just in one +# draw) and at least one of them has a recorded cost behind that call. +NO_FIT = "no agent is expected to fit this budget; choose one yourself or raise the budget" +NO_FIT_ON_PRIOR_ALONE = ( + "no agent is expected to fit this budget, but that expectation is only the configured starting cost " + "estimate, not evidence: none of the agents left out on budget has a recorded cost in this category yet. " + "Choose one yourself or raise the budget" +) +NO_FIT_THIS_DRAW = ( + "no agent's drawn cost fit the budget this time, but at least one could have (each is listed under " + "ineligible with the reason): this is one unlucky draw, not evidence that none can do it. Ask again, use " + "the default ranking (without explore), or choose one yourself" +) +ALL_LEFT_OUT = ( + "every agent is left out of suggestions for now (each is listed under ineligible with the reason); " + "you can still choose one yourself" +) +PARTLY_LEFT_OUT = ( + "some agents are left out for probation or the operator cap, whatever their cost (see ineligible); " + "of the others, " +) +"""Put before any of the budget reasons when probation or the cap also left agents out, so the answer never +implies the budget is the only obstacle: an agent out for probation may well fit the budget.""" + -def _pick_view(pick: Pick, name: str) -> dict[str, Any]: +def _pick_view(pick: Pick, name: str, cost_is_prior: bool) -> dict[str, Any]: + """``cost_is_prior`` (additive) is ``_cost_is_the_prior`` for this agent: no cost recorded in this + category yet, even if it has counted outcomes (all metered 0). A client then must not show + ``assumed_cost_micros`` as a measured cost; it is only the configured starting estimate.""" return { "agent_id": pick.agent_id, "name": name, @@ -40,6 +78,7 @@ def _pick_view(pick: Pick, name: str) -> dict[str, Any]: "assumed_cost_micros": pick.cost, "counted_outcomes": pick.counted_outcomes, "unproven": pick.unproven, + "cost_is_prior": cost_is_prior, } @@ -52,8 +91,11 @@ async def suggestions( agents: Annotated[list[Agent], Depends(get_agents)], explore: Annotated[bool, Query()] = False, ) -> dict[str, Any]: - """Per task: the agents expected to fit its budget, best first (at most ``max_attempts``). With - ``explore=true`` the ranking is a seeded Thompson draw, and the draw is recorded so it can be replayed.""" + """Per task: the agents expected to fit its budget, best first (at most ``max_attempts``), and every + other roster agent under ``ineligible`` with why (the agents cut by ``max_attempts`` first, in rank order, + then the rest by agent id). With ``explore=true`` the ranking is a seeded Thompson draw over the agents + probation and the cap leave in, and the draw is recorded so it can be replayed; since that writes, + exploring calls are limited to ``suggestions_per_hour`` per requester (then 429).""" async with pool.connection() as conn: plan = await plans.load(conn, plan_uuid) if plan is None: @@ -71,8 +113,11 @@ async def suggestions( (plan.category,), ) known = {r[0]: Posterior(r[1], r[2], r[3], r[4], r[5]) for r in await cur.fetchall()} + excluded = await eligibility.ineligible(conn, agents, settings, datetime.now(UTC)) prior = Posterior.prior(Params(tau0=float(settings.reputation_cost_prior_micros))) - candidates = {a.agent_id: known.get(a.agent_id, prior) for a in agents} + # Probation and the operator cap come first: an agent they exclude is never ranked, so it is listed with + # that reason even when it would also be over budget. + candidates = {a.agent_id: known.get(a.agent_id, prior) for a in agents if a.agent_id not in excluded} names = {a.agent_id: a.name for a in agents} seed = secrets.randbits(62) if explore else None @@ -81,28 +126,46 @@ async def suggestions( for task in plan.tasks: budget = task.draft.budget_usdc_units rng = random.Random(f"{seed}/{task.draft.index}") # noqa: S311 (a replayable draw, not a secret) - picks = rank( + ranked, left_out = rank( candidates, float(budget), explore=explore, epsilon=settings.routing_epsilon, min_observations=settings.routing_min_observations, + min_feasible_probability=settings.routing_min_feasible_probability, rng=rng, - )[: settings.max_attempts] + ) + picks, cut = ranked[: settings.max_attempts], ranked[settings.max_attempts :] decisions.append((task.draft.index, budget, picks)) view: dict[str, Any] = { "index": task.draft.index, "budget_usdc_units": str(budget), - "picks": [_pick_view(p, names[p.agent_id]) for p in picks], + "picks": [ + _pick_view(p, names[p.agent_id], _cost_is_the_prior(candidates[p.agent_id], prior)) + for p in picks + ], + "ineligible": [ + {"agent_id": p.agent_id, "name": names[p.agent_id], "reason": BELOW_MAX_ATTEMPTS} for p in cut + ] + + [ + {"agent_id": agent_id, "name": names[agent_id], "reason": reason} + for agent_id, reason in sorted((excluded | left_out).items()) + ], } if not picks: - view["reason"] = ( - "no agent is expected to fit this budget; choose one yourself or raise the budget" - ) + view["reason"] = _empty_reason(left_out, excluded, candidates, prior) tasks.append(view) if seed is not None: + address = principal.address async with pool.connection() as conn, conn.transaction(): + # One lock per requester, so two concurrent calls cannot both pass the limit (as for plans). + await conn.execute("SELECT pg_advisory_xact_lock(hashtext(%s))", (f"suggestions:{address}",)) + if await _recent_explorations(conn, address) >= settings.suggestions_per_hour: + raise HTTPException( + status_code=429, detail="too many exploring suggestions this hour; try again later" + ) + await conn.execute("INSERT INTO suggestion_attempts (requester) VALUES (%s)", (address,)) for index, budget, picks in decisions: await conn.execute( "INSERT INTO routing_decisions (plan_uuid, task_index, category, seed, budget_micros," @@ -126,6 +189,50 @@ async def suggestions( } +def _empty_reason( + left_out: dict[str, str], excluded: dict[str, str], candidates: dict[str, Posterior], prior: Posterior +) -> str: + """Why a task has no picks. With none, every candidate ``rank`` saw is in ``left_out`` on budget; the + answer says only what is true of all of them: + + * If any of them only missed this draw (``OVER_BUDGET_IN_THIS_DRAW``), that agent could fit, so "no agent + is expected to fit" is false: ``NO_FIT_THIS_DRAW``, whatever the others' reasons. + * Otherwise every one is over budget on its posterior. If none of those has a recorded cost, that call is + only the configured prior (``NO_FIT_ON_PRIOR_ALONE``); else it rests on evidence (``NO_FIT``). + * With nobody ranked at all, probation and the cap left out the whole roster (``ALL_LEFT_OUT``). + + When probation or the cap also left agents out, a budget reason is prefixed with ``PARTLY_LEFT_OUT``, + since those agents' cost was never the question.""" + if not left_out: + return ALL_LEFT_OUT if excluded else NO_FIT + if OVER_BUDGET_IN_THIS_DRAW in left_out.values(): + budget = NO_FIT_THIS_DRAW + elif all(_cost_is_the_prior(candidates[agent_id], prior) for agent_id in left_out): + budget = NO_FIT_ON_PRIOR_ALONE + else: + budget = NO_FIT + return PARTLY_LEFT_OUT + budget if excluded else budget + + +def _cost_is_the_prior(posterior: Posterior, prior: Posterior) -> bool: + """No cost has been recorded for this agent in this category, so its cost posterior is still the prior. + Kappa alone tells: ``Posterior.observe`` leaves kappa and tau alone for an outcome without a cost (one + that metered 0), and moves kappa to ``0.95 kappa + 1`` with one, which from the prior's 1 only grows and + never returns. So this also covers an agent with no counted outcome (it is the prior) and one whose + counted outcomes all metered 0, which is why it does not test ``n``. It relies on the prior's kappa being + below 20, the fixed point of that update; the route's prior has 1.""" + return posterior.kappa == prior.kappa + + +async def _recent_explorations(conn: AsyncConnection[TupleRow], requester: str) -> int: + cur = await conn.execute( + "SELECT count(*) FROM suggestion_attempts WHERE requester = %s AND at > now() - interval '1 hour'", + (requester,), + ) + row = await cur.fetchone() + return int(row[0]) if row else 0 + + def _snapshot(candidates: dict[str, Posterior]) -> list[dict[str, Any]]: return [ {"agent_id": a, "alpha": p.alpha, "beta": p.beta, "kappa": p.kappa, "tau": p.tau, "n": p.n} diff --git a/control-plane/mergit_cp/routing.py b/control-plane/mergit_cp/routing.py index ce23295..52a9771 100644 --- a/control-plane/mergit_cp/routing.py +++ b/control-plane/mergit_cp/routing.py @@ -4,21 +4,40 @@ * **greedy** (the default): each candidate is scored by its posterior-mean value density, mean success over mean cost. Nothing is random. This is the paper's exploration suppression (§9.4), and it is what a requester - gets unless they ask to explore. -* **explore**: Thompson sampling. Each candidate draws a success probability from its Beta and a cost from - its Gamma. A candidate whose drawn cost exceeds the task budget (with ``epsilon`` of slack) is skipped, and - the rest rank by drawn success over drawn cost. A draw is reproducible from the seed. + gets unless they ask to explore. A candidate whose mean cost exceeds the task budget (with ``epsilon`` of + slack) is left out as ``OVER_BUDGET``. +* **explore**: Thompson sampling. A candidate whose chance of drawing a cost within the budget + (``feasible_probability``) is below ``min_feasible_probability`` is left out as ``OVER_BUDGET`` before the + draw, and takes no draw. Each other candidate draws a success probability from its Beta and a cost from its + Gamma; one whose drawn cost exceeds the budget is left out of this draw only (``OVER_BUDGET_IN_THIS_DRAW``), + and the rest rank by drawn success over drawn cost. A draw is reproducible from the seed. + +Every candidate ``rank`` does not return is reported with its reason; nothing is dropped silently. What else +leaves an agent out (probation, the operator cap: ``eligibility``) is decided before ``rank`` is called, and +the ``max_attempts`` cut after; ``rank`` knows about neither. A candidate with fewer than ``min_observations`` counted outcomes is marked unproven. The platform does not yet absorb the cost of a failed exploratory draw (paper §9.5: the subsidy ledger is not built), so an unproven pick can cost the requester, and the caller must say so. + +``budget_limit``, ``mean_feasible``, ``_upper_regularized_gamma`` and ``feasible_probability`` are ported from +the reviewed ``router.py`` on ``archive/auto-dispatch-routing``. """ +import math import random from dataclasses import dataclass from .reputation_model import Posterior, draw +OVER_BUDGET = "over budget for this task" +"""The posterior says this agent will not fit the budget: its mean cost is over it (greedy), or its chance of +drawing a cost within it is below ``min_feasible_probability`` (explore).""" +OVER_BUDGET_IN_THIS_DRAW = "over budget in this draw" +"""Explore only: this agent could fit, but the cost it drew this time did not. Another draw may include it.""" +BELOW_MAX_ATTEMPTS = "ranked below the first max_attempts" +"""Fits the budget, but ranked after the ``max_attempts`` agents a task can list (applied by the caller).""" + @dataclass(frozen=True, slots=True) class Pick: @@ -33,6 +52,77 @@ class Pick: unproven: bool +def budget_limit(budget_micros: float, epsilon: float) -> float: + """The most a cost may be and still fit ``budget_micros``: the one budget rule ``rank``, + ``mean_feasible`` and ``feasible_probability`` all share.""" + return budget_micros * (1.0 + epsilon) + + +def mean_feasible(posterior: Posterior, budget_micros: float, epsilon: float) -> bool: + """Whether the posterior *mean* cost fits the budget: the test greedy ranking applies.""" + return posterior.mean_cost <= budget_limit(budget_micros, epsilon) + + +def _upper_regularized_gamma(a: float, x: float) -> float: + """Q(a, x) = Gamma(a, x) / Gamma(a), the upper regularized incomplete gamma function, for a > 0. + + The standard pair of evaluations (Numerical Recipes 6.2): the power series for P = 1 - Q when + x < a + 1, where it converges fast, and the modified-Lentz continued fraction for Q itself otherwise. + Both run to double precision; the result is clamped into [0, 1] against rounding.""" + if x <= 0: + return 1.0 + log_prefactor = -x + a * math.log(x) - math.lgamma(a) + if x < a + 1: + term = 1.0 / a + total = term + n = a + for _ in range(10_000): + n += 1 + term *= x / n + total += term + if abs(term) < abs(total) * 1e-16: + break + return min(1.0, max(0.0, 1.0 - total * math.exp(log_prefactor))) + tiny = 1e-300 + b = x + 1 - a + c = 1 / tiny + d = 1 / b + h = d + for i in range(1, 10_000): + an = -i * (i - a) + b += 2 + d = an * d + b + if abs(d) < tiny: + d = tiny + c = b + an / c + if abs(c) < tiny: + c = tiny + d = 1 / d + delta = d * c + h *= delta + if abs(delta - 1) < 1e-16: + break + return min(1.0, max(0.0, math.exp(log_prefactor) * h)) + + +def feasible_probability(posterior: Posterior, budget_micros: float, epsilon: float) -> float: + """The probability that one Thompson draw of this agent's cost fits the budget: P(cost <= limit). + + The drawn cost is ``1 / r`` with ``r ~ Gamma(shape=kappa, rate=tau)`` (``reputation_model.draw``), so + ``cost <= limit`` is ``r >= 1 / limit``, whose probability is the upper regularized incomplete gamma + function ``Q(kappa, tau / limit)``. Closed form for the cold prior (``kappa = 1``): ``exp(-tau / limit)``. + + An agent whose mean cost fits has ``tau / limit <= kappa``, so this is at least ``Q(kappa, kappa)``: 1/e + at ``kappa = 1``, rising toward 1/2. Every posterior here has ``kappa >= 1`` (the prior's is 1, and an + observed cost moves it to ``0.95 kappa + 1``), which is why ``routing_min_feasible_probability`` is capped + at 0.25: exploring never leaves out an agent that greedy ranking would suggest. + """ + limit = budget_limit(budget_micros, epsilon) + if limit <= 0: + return 0.0 + return _upper_regularized_gamma(posterior.kappa, posterior.tau / limit) + + def rank( candidates: dict[str, Posterior], budget_micros: float, @@ -40,17 +130,31 @@ def rank( explore: bool, epsilon: float, min_observations: int, + min_feasible_probability: float, rng: random.Random, -) -> list[Pick]: - """The feasible candidates, best first (ties by agent id, so the order never depends on dict order). - Empty when none is expected to fit the budget. ``rng`` is only used when exploring, in agent-id order.""" - limit = budget_micros * (1.0 + epsilon) +) -> tuple[list[Pick], dict[str, str]]: + """The feasible candidates, best first (ties by agent id, so the order never depends on dict order), and + every other candidate mapped to why it is left out (``OVER_BUDGET`` or ``OVER_BUDGET_IN_THIS_DRAW``). + Each candidate is in exactly one of the two. ``rng`` is only used when exploring, in agent-id order, and + only by candidates not already left out by ``min_feasible_probability``.""" + limit = budget_limit(budget_micros, epsilon) picks: list[Pick] = [] + left_out: dict[str, str] = {} for agent_id in sorted(candidates): posterior = candidates[agent_id] - success, cost = draw(posterior, rng) if explore else (posterior.mean_success, posterior.mean_cost) - if not cost <= limit: # also skips an infinite or undefined cost - continue + if explore: + if feasible_probability(posterior, budget_micros, epsilon) < min_feasible_probability: + left_out[agent_id] = OVER_BUDGET + continue + success, cost = draw(posterior, rng) + if not cost <= limit: # also an infinite or undefined drawn cost + left_out[agent_id] = OVER_BUDGET_IN_THIS_DRAW + continue + else: + if not mean_feasible(posterior, budget_micros, epsilon): + left_out[agent_id] = OVER_BUDGET + continue + success, cost = posterior.mean_success, posterior.mean_cost picks.append( Pick( agent_id=agent_id, @@ -62,4 +166,4 @@ def rank( ) ) picks.sort(key=lambda p: (-p.score, p.agent_id)) - return picks + return picks, left_out diff --git a/control-plane/migrations/0019_money_events_task.sql b/control-plane/migrations/0019_money_events_task.sql new file mode 100644 index 0000000..6adbf4c --- /dev/null +++ b/control-plane/migrations/0019_money_events_task.sql @@ -0,0 +1,6 @@ +-- A task's ledger rows, by kind, oldest first. `task_outcomes` (0015) looks up each dispatch's first event of a +-- kind (`WHERE task_id = ? AND kind = ? ORDER BY id LIMIT 1`), and nothing else served that lookup: the only +-- index with `task_id` in it, `money_events_on_chain`, is partial (`tx IS NOT NULL`), so each lookup walked the +-- whole ledger and the view cost tasks x events. That view runs on every reputation pass and on every +-- `GET /v0/agents/reputation`, which needs no sign-in. +CREATE INDEX money_events_task ON money_events (task_id, kind, id); diff --git a/control-plane/migrations/0020_task_outcomes_adjudicated.sql b/control-plane/migrations/0020_task_outcomes_adjudicated.sql new file mode 100644 index 0000000..b86d362 --- /dev/null +++ b/control-plane/migrations/0020_task_outcomes_adjudicated.sql @@ -0,0 +1,51 @@ +-- `task_outcomes` (0015) counted only audited outcomes. The paper (§8.2) also counts an outcome that "was +-- challenged and adjudicated", and dropping those lost evidence: an unaudited task overturned by a challenge, or +-- one whose challenge failed and which was then paid, never reached a posterior. Same columns, same meaning; the +-- audited cases keep exactly the row they had. +-- failure: the validator's verdict was fail or invalid (at the first such verdict); or a challenge against the +-- task was upheld (the adjudicated outcome replaces the audit's). A passed and released task keeps +-- the release's position; otherwise the upheld resolution is what made it count, so it sits there +-- success: no fail/invalid verdict, no upheld challenge, the task was released, and it was either audited +-- (a pass verdict: at the release) or challenged and the challenge rejected (at the later of the +-- release and the rejection, the event that completed the evidence) +-- Still not evidence: a release nobody audited or challenged, a deadline return, a bare receipt, a pass verdict +-- with no release, and a rejected challenge with no release yet. +CREATE OR REPLACE VIEW task_outcomes AS +SELECT d.task_id, + d.agent_id, + p.category, + p.funding, + (bad.id IS NULL AND upheld.id IS NULL) AS success, + CASE + WHEN bad.id IS NOT NULL THEN bad.id + WHEN passed.id IS NOT NULL AND rel.id IS NOT NULL THEN rel.id + WHEN upheld.id IS NOT NULL THEN upheld.id + ELSE GREATEST(rel.id, rejected.id) + END AS position, + c.cost_micros +FROM task_dispatches d +JOIN plans p ON p.plan_uuid = d.plan_uuid +LEFT JOIN LATERAL ( + SELECT id FROM money_events + WHERE task_id = d.task_id AND kind = 'verdict.recorded' AND verdict IN ('fail', 'invalid') + ORDER BY id LIMIT 1) bad ON true +LEFT JOIN LATERAL ( + SELECT id FROM money_events + WHERE task_id = d.task_id AND kind = 'verdict.recorded' AND verdict = 'pass' + ORDER BY id LIMIT 1) passed ON true +LEFT JOIN LATERAL ( + SELECT id FROM money_events + WHERE task_id = d.task_id AND kind = 'task.released' + ORDER BY id LIMIT 1) rel ON true +LEFT JOIN LATERAL ( + SELECT id FROM money_events + WHERE task_id = d.task_id AND kind = 'challenge.resolved' AND upheld + ORDER BY id LIMIT 1) upheld ON true +LEFT JOIN LATERAL ( + SELECT id FROM money_events + WHERE task_id = d.task_id AND kind = 'challenge.resolved' AND NOT upheld + ORDER BY id LIMIT 1) rejected ON true +LEFT JOIN task_costs c ON c.task_id = d.task_id +WHERE bad.id IS NOT NULL + OR upheld.id IS NOT NULL + OR (rel.id IS NOT NULL AND (passed.id IS NOT NULL OR rejected.id IS NOT NULL)); diff --git a/control-plane/migrations/0021_suggestion_attempts.sql b/control-plane/migrations/0021_suggestion_attempts.sql new file mode 100644 index 0000000..15a3d3e --- /dev/null +++ b/control-plane/migrations/0021_suggestion_attempts.sql @@ -0,0 +1,10 @@ +-- One row per exploring suggestion (GET /v0/plans/{uuid}/suggestions?explore=true), for the per-requester rate +-- limit. Each one appends a replay record per task to `routing_decisions` (0016), which is append-only, so +-- without a limit anyone with a draft plan could grow it without bound. A greedy suggestion writes nothing and is +-- not counted. The same shape as `planning_attempts` (0003), which limits the planner the same way. +CREATE TABLE suggestion_attempts ( + requester text NOT NULL, + at timestamptz NOT NULL DEFAULT now() +); + +CREATE INDEX suggestion_attempts_requester ON suggestion_attempts (requester, at DESC); diff --git a/control-plane/migrations/0022_routing_decisions_outlive_drafts.sql b/control-plane/migrations/0022_routing_decisions_outlive_drafts.sql new file mode 100644 index 0000000..1eff7c3 --- /dev/null +++ b/control-plane/migrations/0022_routing_decisions_outlive_drafts.sql @@ -0,0 +1,7 @@ +-- `routing_decisions` (0016) is a replay log of exploring draws, and draws are only made for draft plans. A +-- draft can be discarded (DELETE /v0/plans/{uuid}), which really deletes the plan row, so the foreign key made +-- any draft that ever had a draw impossible to discard. The log is an audit record, not a live reference: it +-- should outlive the draft it was drawn for, so `plan_uuid` keeps its value (still NOT NULL) and stops +-- referencing `plans`. Cascading the delete would not work anyway, and would erase the record: the append-only +-- trigger, unchanged here, still refuses every UPDATE and DELETE on the log. +ALTER TABLE routing_decisions DROP CONSTRAINT IF EXISTS routing_decisions_plan_uuid_fkey; diff --git a/control-plane/tests/test_eligibility.py b/control-plane/tests/test_eligibility.py new file mode 100644 index 0000000..d9ac0e4 --- /dev/null +++ b/control-plane/tests/test_eligibility.py @@ -0,0 +1,354 @@ +"""Which roster agents a suggestion leaves out, and why: probation after a proven falsification, and the +operator concurrency cap. Both are derived from ``money_events``, ``task_dispatches`` and ``jobs``. + +Run with `pytest -m postgres`. Every test uses its own fresh agent ids, so the shared database's other rows +never count toward a test's operator. +""" + +from __future__ import annotations + +import secrets +import uuid +from datetime import UTC, datetime, timedelta + +import mergit_receipt +import pytest +from psycopg.types.json import Jsonb + +from mergit_cp.agents import Agent +from mergit_cp.config import Settings +from mergit_cp.db import Pool +from mergit_cp.dispatch import DISPATCH_DEADLINE_S +from mergit_cp.eligibility import AT_CAP, IN_PROBATION, ineligible +from tests.test_money import run + +pytestmark = pytest.mark.postgres + +NOW = datetime(2026, 11, 2, 12, 0, tzinfo=UTC) +REQUESTER = "0x" + "1" * 40 +OPERATOR_A = "0x" + "a" * 40 +OPERATOR_B = "0x" + "b" * 40 + + +def agent(operator: str) -> Agent: + """A roster agent with an id no other test uses.""" + agent_id = str(secrets.randbelow(10**15) + 10**15) + return Agent( + agent_id=agent_id, + name=f"agent-{agent_id}", + operator=operator, + runtime_key="0x" + "2" * 40, + model="claude-sonnet-5", + system_prompt="You fix failing tests.", + ) + + +async def plan(pool: Pool, funding: str = "wallet") -> str: + plan_uuid = str(uuid.uuid4()) + async with pool.connection() as conn: + await conn.execute( + "INSERT INTO plans (plan_uuid, plan_id, requester, goal, repo_url, ref, category," + " budget_usdc_units, state, funding, total_budget_usdc_units, planner_model, approved_at)" + " VALUES (%s, %s, %s, 'fix it', 'https://github.com/acme/widgets', 'main', 'test-repair'," + " 1000000, 'running', %s, 1000000, 'claude-opus-5', now())", + (plan_uuid, mergit_receipt.plan_id(plan_uuid), REQUESTER, funding), + ) + return plan_uuid + + +async def dispatch( + pool: Pool, + a: Agent, + *, + funding: str = "wallet", + at: datetime = NOW, + job: str | None = "running", + receipt: bool = False, +) -> tuple[str, str]: + """One dispatch of ``a`` on a plan of its own, ``at`` this time; with a ``jobs`` row in status ``job`` + unless ``job`` is None (the escrow never confirmed). Returns (plan_uuid, task_id).""" + plan_uuid = await plan(pool, funding) + task_id = "0x" + secrets.token_hex(32) + async with pool.connection() as conn: + await conn.execute( + 'INSERT INTO task_dispatches (task_id, plan_uuid, "index", attempt, agent_id, dispatched_at)' + " VALUES (%s, %s, 0, 1, %s, %s)", + (task_id, plan_uuid, a.agent_id, at), + ) + if job is not None: + await conn.execute( + 'INSERT INTO jobs (task_id, plan_uuid, "index", attempt, agent_id, status, receipt_root,' + " created_at) VALUES (%s, %s, 0, 1, %s, %s, %s, %s)", + (task_id, plan_uuid, a.agent_id, job, ("0x" + "e" * 64) if receipt else None, at), + ) + return plan_uuid, task_id + + +async def ledger(pool: Pool, plan_uuid: str, task_id: str, kind: str, at: datetime, **cols: object) -> None: + """A ``money_events`` row received at ``at``. Written directly rather than through + ``chain_events.receive`` because the ledger is append-only and ``received_at`` is otherwise always now.""" + event_id = str(uuid.uuid4()) + names = ["event_id", "kind", "plan_uuid", "task_id", "tx", "received_at", *cols] + values = [event_id, kind, plan_uuid, task_id, "0x" + secrets.token_hex(32), at, *cols.values()] + async with pool.connection() as conn, conn.transaction(): + await conn.execute( + "INSERT INTO chain_events_seen (event_id, type, payload, received_at) VALUES (%s, %s, %s, %s)", + (event_id, kind, Jsonb({"task_id": task_id}), at), + ) + await conn.execute( + f"INSERT INTO money_events ({', '.join(names)}) VALUES ({', '.join(['%s'] * len(names))})", + values, + ) + + +async def invalid_verdict(pool: Pool, plan_uuid: str, task_id: str, at: datetime) -> None: + await ledger(pool, plan_uuid, task_id, "verdict.recorded", at, verdict="invalid") + + +async def challenge_resolved(pool: Pool, plan_uuid: str, task_id: str, at: datetime, upheld: bool) -> None: + await ledger(pool, plan_uuid, task_id, "challenge.resolved", at, challenge_id=7, upheld=upheld) + + +async def released(pool: Pool, plan_uuid: str, task_id: str, at: datetime = NOW) -> None: + await ledger(pool, plan_uuid, task_id, "task.released", at, payee="0x" + "c" * 40, amount_usdc_units=1) + + +async def returned(pool: Pool, plan_uuid: str, task_id: str, at: datetime = NOW) -> None: + await ledger(pool, plan_uuid, task_id, "task.returned", at, amount_usdc_units=1, reason="deadline") + + +async def excluded( + pool: Pool, agents: list[Agent], settings: Settings, now: datetime = NOW +) -> dict[str, str]: + async with pool.connection() as conn: + return await ineligible(conn, agents, settings, now) + + +# Probation. + + +def test_wallet_falsification_puts_operator_in_probation(database_url: str, settings: Settings): + async def body(pool: Pool) -> None: + a = agent(OPERATOR_A) + plan_uuid, task_id = await dispatch(pool, a, at=NOW - timedelta(days=1), receipt=True) + await released(pool, plan_uuid, task_id, NOW - timedelta(days=1)) + await invalid_verdict(pool, plan_uuid, task_id, NOW - timedelta(hours=1)) + assert await excluded(pool, [a], settings) == {a.agent_id: IN_PROBATION} + + run(database_url, body) + + +def test_credits_falsification_puts_operator_in_probation(database_url: str, settings: Settings): + """Invariant 6 keeps free-credit work out of the posteriors, not out of probation: a proven false + receipt is evidence of fraud, so forging where the stakes are lowest still costs wallet-funded work.""" + + async def body(pool: Pool) -> None: + a = agent(OPERATOR_A) + plan_uuid, task_id = await dispatch(pool, a, funding="credits", at=NOW - timedelta(days=1)) + await invalid_verdict(pool, plan_uuid, task_id, NOW - timedelta(hours=1)) + assert await excluded(pool, [a], settings) == {a.agent_id: IN_PROBATION} + + run(database_url, body) + + +def test_upheld_challenge_puts_operator_in_probation(database_url: str, settings: Settings): + async def body(pool: Pool) -> None: + a, b = agent(OPERATOR_A), agent(OPERATOR_B) + plan_a, task_a = await dispatch(pool, a, at=NOW - timedelta(days=1)) + plan_b, task_b = await dispatch(pool, b, at=NOW - timedelta(days=1)) + for plan_uuid, task_id, upheld in ((plan_a, task_a, True), (plan_b, task_b, False)): + await challenge_resolved(pool, plan_uuid, task_id, NOW - timedelta(hours=1), upheld) + # A rejected challenge proves nothing against b. + assert await excluded(pool, [a, b], settings) == {a.agent_id: IN_PROBATION} + + run(database_url, body) + + +def test_a_failing_verdict_is_not_a_falsification(database_url: str, settings: Settings): + """A validated failure is a capability signal for the posteriors (paper §7.5 iii), not fraud.""" + + async def body(pool: Pool) -> None: + a = agent(OPERATOR_A) + plan_uuid, task_id = await dispatch(pool, a, at=NOW - timedelta(days=1)) + await ledger(pool, plan_uuid, task_id, "verdict.recorded", NOW - timedelta(hours=1), verdict="fail") + await returned(pool, plan_uuid, task_id, NOW - timedelta(hours=1)) + assert await excluded(pool, [a], settings) == {} + + run(database_url, body) + + +def test_probation_expires_after_window(database_url: str, settings: Settings): + async def body(pool: Pool) -> None: + a = agent(OPERATOR_A) + plan_uuid, task_id = await dispatch(pool, a, at=NOW - timedelta(days=30)) + await returned(pool, plan_uuid, task_id, NOW - timedelta(days=30)) + proven = NOW - timedelta(days=10) + await invalid_verdict(pool, plan_uuid, task_id, proven) + window = timedelta(seconds=settings.routing_probation_s) + assert await excluded(pool, [a], settings, now=proven + window - timedelta(seconds=1)) == { + a.agent_id: IN_PROBATION + } + assert await excluded(pool, [a], settings, now=proven + window) == {} + assert await excluded(pool, [a], settings) == {} # NOW is ten days on, past the seven-day default + + run(database_url, body) + + +def test_a_second_falsification_of_the_same_task_does_not_restart_probation( + database_url: str, settings: Settings +): + """One false receipt is one falsification, timed when it was first proven: a later upheld challenge on a + task already ruled invalid is the same fraud, not a new one.""" + + async def body(pool: Pool) -> None: + a = agent(OPERATOR_A) + plan_uuid, task_id = await dispatch(pool, a, at=NOW - timedelta(days=30)) + await invalid_verdict(pool, plan_uuid, task_id, NOW - timedelta(days=10)) + await challenge_resolved(pool, plan_uuid, task_id, NOW - timedelta(hours=1), upheld=True) + await returned(pool, plan_uuid, task_id, NOW - timedelta(days=10)) + assert await excluded(pool, [a], settings) == {} + + run(database_url, body) + + +def test_probation_covers_every_agent_of_the_operator(database_url: str, settings: Settings): + async def body(pool: Pool) -> None: + forger, sibling, other = agent(OPERATOR_A), agent(OPERATOR_A), agent(OPERATOR_B) + plan_uuid, task_id = await dispatch(pool, forger, at=NOW - timedelta(days=1)) + await invalid_verdict(pool, plan_uuid, task_id, NOW - timedelta(hours=1)) + # The sibling has never been dispatched at all; the other operator's agent is untouched. + assert await excluded(pool, [forger, sibling, other], settings) == { + forger.agent_id: IN_PROBATION, + sibling.agent_id: IN_PROBATION, + } + + run(database_url, body) + + +# The operator concurrency cap. + + +def test_operator_at_cap_is_ineligible(database_url: str, settings: Settings): + """The cap is operator-wide (paper §7.5 ii): two agents' dispatches add up, and the bound is inclusive.""" + + async def body(pool: Pool) -> None: + cap = settings.routing_operator_concurrency_cap + busy, idle, other = agent(OPERATOR_A), agent(OPERATOR_A), agent(OPERATOR_B) + for _ in range(cap - 1): + await dispatch(pool, busy) + await dispatch(pool, other) + assert await excluded(pool, [busy, idle, other], settings) == {} # one below the cap + await dispatch(pool, idle, job=None) # mid-escrow, within its deadline: a real commitment + assert await excluded(pool, [busy, idle, other], settings) == { + busy.agent_id: AT_CAP, + idle.agent_id: AT_CAP, + } + + run(database_url, body) + + +def test_free_credit_dispatches_do_not_count_toward_the_cap(database_url: str, settings: Settings): + """Invariant 6: free-credit work must not influence routing. Otherwise anyone with a credits grant could + send a few cheap plans to a rival operator's agents and have its whole roster shown as "at cap" to + everyone. Only wallet-funded dispatches, where someone has paid to hold the operator's capacity, count.""" + + async def body(pool: Pool) -> None: + s = settings.model_copy(update={"routing_operator_concurrency_cap": 2}) + a = agent(OPERATOR_A) + for _ in range(5): + await dispatch(pool, a, funding="credits") + assert await excluded(pool, [a], s) == {} + await dispatch(pool, a, funding="wallet") + assert await excluded(pool, [a], s) == {} # one paid dispatch, still under the cap of 2 + await dispatch(pool, a, funding="wallet") + assert await excluded(pool, [a], s) == {a.agent_id: AT_CAP} + + run(database_url, body) + + +def test_released_and_returned_dispatches_are_resolved(database_url: str, settings: Settings): + async def body(pool: Pool) -> None: + s = settings.model_copy(update={"routing_operator_concurrency_cap": 2}) + a = agent(OPERATOR_A) + first = await dispatch(pool, a, job="done", receipt=True) + second = await dispatch(pool, a, job="done", receipt=True) + assert await excluded(pool, [a], s) == {a.agent_id: AT_CAP} + await released(pool, *first) + assert await excluded(pool, [a], s) == {} + await dispatch(pool, a) + assert await excluded(pool, [a], s) == {a.agent_id: AT_CAP} + await returned(pool, *second) + assert await excluded(pool, [a], s) == {} + + run(database_url, body) + + +def test_stale_dispatch_without_job_does_not_count(database_url: str, settings: Settings): + async def body(pool: Pool) -> None: + s = settings.model_copy(update={"routing_operator_concurrency_cap": 1}) + a = agent(OPERATOR_A) + await dispatch(pool, a, job=None, at=NOW - timedelta(seconds=DISPATCH_DEADLINE_S + 1)) + assert await excluded(pool, [a], s) == {} # never escrowed, and never will + b = agent(OPERATOR_B) + await dispatch(pool, b, job=None, at=NOW - timedelta(seconds=DISPATCH_DEADLINE_S - 1)) + assert await excluded(pool, [b], s) == {b.agent_id: AT_CAP} # still within its deadline + c = agent("0x" + "c" * 40) + await dispatch(pool, c, job="running", at=NOW - timedelta(seconds=DISPATCH_DEADLINE_S + 1)) + assert await excluded(pool, [c], s) == {c.agent_id: AT_CAP} # escrowed: age alone never resolves it + + run(database_url, body) + + +def test_an_errored_job_with_no_receipt_does_not_count(database_url: str, settings: Settings): + """Nothing was sealed for that attempt, so there is nothing to forge and nothing in flight; its escrow + comes back on the chain's own deadline, which may be long after the worker gave up.""" + + async def body(pool: Pool) -> None: + s = settings.model_copy(update={"routing_operator_concurrency_cap": 1}) + a, b = agent(OPERATOR_A), agent(OPERATOR_B) + await dispatch(pool, a, job="error") + await dispatch(pool, b, job="error", receipt=True) + assert await excluded(pool, [a, b], s) == {b.agent_id: AT_CAP} + + run(database_url, body) + + +def test_probation_reason_wins_over_cap(database_url: str, settings: Settings): + async def body(pool: Pool) -> None: + s = settings.model_copy(update={"routing_operator_concurrency_cap": 1}) + a = agent(OPERATOR_A) + plan_uuid, task_id = await dispatch(pool, a, at=NOW - timedelta(minutes=5)) # unresolved + await invalid_verdict(pool, plan_uuid, task_id, NOW - timedelta(minutes=1)) + assert await excluded(pool, [a], s) == {a.agent_id: IN_PROBATION} + + run(database_url, body) + + +# Everyone else. + + +def test_an_ordinary_agent_is_not_in_the_result(database_url: str, settings: Settings): + """Present means excluded; an agent that may be suggested is absent, never mapped to an empty reason.""" + + async def body(pool: Pool) -> None: + a, fresh = agent(OPERATOR_A), agent(OPERATOR_B) + done = await dispatch(pool, a, at=NOW - timedelta(days=1), job="done", receipt=True) + await ledger(pool, *done, "verdict.recorded", NOW - timedelta(days=1), verdict="pass") + await released(pool, *done, NOW - timedelta(days=1)) + await dispatch(pool, a) # one in flight, well under the cap + assert await excluded(pool, [a, fresh], settings) == {} + + run(database_url, body) + + +def test_an_empty_roster_excludes_nobody(database_url: str, settings: Settings): + async def body(pool: Pool) -> None: + assert await excluded(pool, [], settings) == {} + + run(database_url, body) + + +def test_the_default_settings(): + fields = Settings.model_fields + assert fields["routing_probation_s"].default == 7 * 24 * 3600 + assert fields["routing_operator_concurrency_cap"].default == 5 diff --git a/control-plane/tests/test_money.py b/control-plane/tests/test_money.py index 69d3704..a955818 100644 --- a/control-plane/tests/test_money.py +++ b/control-plane/tests/test_money.py @@ -334,8 +334,14 @@ def test_the_migration_backfills_history_from_the_raw_log(database_url: str, tmp old.mkdir() # Every migration except the one under test, not "< 0006": a later migration added after this # test was written (0009, from #15) must not also show up as newly applied below. 0015 reads the - # ledger (a view over money_events), so it cannot exist without 0006 and is applied with it. - held_back = {"0006_money_events", "0015_reputation"} + # ledger (a view over money_events), so it cannot exist without 0006 and is applied with it; so do + # 0019, an index on money_events, and 0020, which replaces 0015's view. + held_back = { + "0006_money_events", + "0015_reputation", + "0019_money_events_task", + "0020_task_outcomes_adjudicated", + } for path in sorted(MIGRATIONS.glob("*.sql")): if path.stem not in held_back: (old / path.name).write_text(path.read_text()) @@ -417,7 +423,7 @@ async def go() -> list[dict[str, Any]]: "task.released", {"task_id": task_id, "to": ADDR, "amount_usdc_units": "9" * 78, "tx": tx("6")}, ) - assert await migrate(conn) == ["0006_money_events", "0015_reputation"] + assert await migrate(conn) == sorted(held_back) cur = conn.cursor(row_factory=dict_row) await cur.execute("SELECT * FROM money_events ORDER BY id") return await cur.fetchall() diff --git a/control-plane/tests/test_reputation.py b/control-plane/tests/test_reputation.py index f05c34a..fd7437c 100644 --- a/control-plane/tests/test_reputation.py +++ b/control-plane/tests/test_reputation.py @@ -10,12 +10,13 @@ from typing import Any import httpx +import psycopg import pytest from psycopg.rows import dict_row from mergit_cp.config import Settings from mergit_cp.db import Pool, migrate, open_pool -from mergit_cp.reputation import GatewayCosts, record_missing_costs, refresh_posteriors +from mergit_cp.reputation import GatewayCosts, ReputationUpdater, record_missing_costs, refresh_posteriors from mergit_cp.reputation_model import Params, Posterior from tests.test_api import seed_plan from tests.test_money import receive, tx @@ -219,3 +220,250 @@ async def snapshot() -> list[Any]: assert await snapshot() == before run(database_url, body) + + +def scripted(answer: Callable[[str], httpx.Response], asked: list[str]) -> GatewayCosts: + """A gateway whose answer for each task id is ``answer(task_id)``; every task asked about is appended to + ``asked``. ``answer`` may raise an ``httpx`` error to play an unreachable gateway.""" + + def handle(request: httpx.Request) -> httpx.Response: + task_id = request.url.params["task_id"] + asked.append(task_id) + return answer(task_id) + + return GatewayCosts( + "http://gateway.test", client=httpx.AsyncClient(transport=httpx.MockTransport(handle)) + ) + + +def commitment(task_id: str, cost: str) -> httpx.Response: + return httpx.Response(200, json=[{"task_id": task_id, "cost_micros": cost}]) + + +async def fail_each(pool: Pool, settings: Settings, tasks: list[str]) -> None: + """A validated failure for each task, in order.""" + for i, task in enumerate(tasks): + await receive( + pool, settings, "verdict.recorded", {"task_id": task, "verdict": "fail", "tx": f"0x{i + 1:064x}"} + ) + + +async def recorded(pool: Pool) -> set[str]: + async with pool.connection() as conn: + cur = await conn.execute("SELECT task_id FROM task_costs") + return {r[0] for r in await cur.fetchall()} + + +def test_outcomes_whose_cost_never_arrives_do_not_hold_back_newer_ones(database_url: str, settings: Settings): + """The review's case: 51 validated failures, and the gateway knows only the newest one's commitment. The + 50 older ones used to fill every slot of every pass, so the 51st was never asked about.""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester=REQUESTER, n_tasks=51) + await fail_each(pool, settings, tasks) + asked: list[str] = [] + newest = tasks[-1] + costs = scripted( + lambda t: commitment(t, "300000") if t == newest else httpx.Response(200, json=[]), asked + ) + assert await record_missing_costs(pool, costs) == 1 + assert await recorded(pool) == {newest} + assert asked == tasks # every pending outcome was tried, oldest first, in one pass + assert await refresh_posteriors(pool, settings) == 1 + + run(database_url, body) + + +def test_an_unusable_answer_for_one_task_does_not_stop_the_others(database_url: str, settings: Settings): + """Another task's commitment, a negative cost, a row that is not an object (D11), a cost that is not a + number, a 400: each is that task's problem. The task after them still gets its cost.""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester=REQUESTER, n_tasks=6) + await fail_each(pool, settings, tasks) + bad = { + tasks[0]: httpx.Response(200, json=[{"task_id": "0x" + "9" * 64, "cost_micros": "1"}]), + tasks[1]: commitment(tasks[1], "-5"), + tasks[2]: httpx.Response(200, json=["x"]), + tasks[3]: commitment(tasks[3], "lots"), + tasks[4]: httpx.Response(400), + } + asked: list[str] = [] + costs = scripted(lambda t: bad.get(t) or commitment(t, "200000"), asked) + assert await record_missing_costs(pool, costs) == 1 + assert await recorded(pool) == {tasks[5]} + assert asked == tasks + + run(database_url, body) + + +def gateway_cannot_answer( + database_url: str, settings: Settings, answer: Callable[[str], httpx.Response] +) -> None: + """Every task would fail the same way, so the rest are left for the next pass rather than each asked.""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester=REQUESTER, n_tasks=3) + await fail_each(pool, settings, tasks) + asked: list[str] = [] + assert await record_missing_costs(pool, scripted(answer, asked)) == 0 + assert asked == tasks[:1] + assert await recorded(pool) == set() + + run(database_url, body) + + +def _unreachable(task_id: str) -> httpx.Response: + raise httpx.ConnectError(f"refused while asking about {task_id}") + + +@pytest.mark.parametrize( + ("status", "content"), + [(503, b"[]"), (429, b"[]"), (401, b"[]"), (200, b"not json"), (200, b'{"not": "a list"}')], + ids=["503", "429", "401", "not-json", "not-a-list"], +) +def test_a_gateway_that_cannot_answer_ends_the_pass_after_one_request( + database_url: str, settings: Settings, status: int, content: bytes +): + gateway_cannot_answer(database_url, settings, lambda _: httpx.Response(status, content=content)) + + +def test_an_unreachable_gateway_ends_the_pass_after_one_request(database_url: str, settings: Settings): + gateway_cannot_answer(database_url, settings, _unreachable) + + +def test_a_cost_another_instance_recorded_first_is_not_counted_as_written( + database_url: str, settings: Settings +): + """Two API instances read the same pending task; the other one's INSERT lands first, ours does nothing.""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester=REQUESTER) + await verdict(pool, settings, tasks[0], "fail", "1") + + def raced(task_id: str) -> httpx.Response: + with psycopg.connect(database_url) as conn: + conn.execute("INSERT INTO task_costs (task_id, cost_micros) VALUES (%s, 7)", (task_id,)) + return commitment(task_id, "7") + + assert await record_missing_costs(pool, scripted(raced, [])) == 0 + assert await recorded(pool) == {tasks[0]} + + run(database_url, body) + + +def test_a_commitment_that_is_not_an_object_does_not_skip_the_refresh(database_url: str, settings: Settings): + """A gateway row that is not a dict used to raise AttributeError, which escaped the pass before the + posteriors were refreshed.""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester=REQUESTER) + await verdict(pool, settings, tasks[0], "fail", "1") + await record_missing_costs(pool, gateway({tasks[0]: "300000"})) + await verdict(pool, settings, tasks[1], "fail", "2") + await ReputationUpdater(pool, settings, gateway(raw=["x"])).update() + assert (await posteriors(pool))[("1", "test-repair")]["n"] == 1 + + run(database_url, body) + + +def test_a_zero_cost_outcome_still_counts_as_a_success_or_failure_but_not_as_a_cost( + database_url: str, settings: Settings +): + """Paper §8.3's c > 0 constrains only the cost (Gamma) update; the success (Beta) update of §8.2 has no + cost precondition. A validated failure whose receipt metered 0 (say, a run that died before its first + billed call) is still a failure, so it moves beta and n; it says nothing about cost, so kappa and tau skip + it. A positive cost folds as before; free-credit work and outcomes whose cost is pending stay out.""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester=REQUESTER, n_tasks=4) + zero, paid, pending, free_zero = tasks + await fail_each(pool, settings, [zero, paid, pending]) + credits_uuid, _, credit_tasks = await seed_plan(pool, requester=REQUESTER, n_tasks=1) + async with pool.connection() as conn: + await conn.execute("UPDATE plans SET funding = 'credits' WHERE plan_uuid = %s", (credits_uuid,)) + await receive( + pool, + settings, + "verdict.recorded", + {"task_id": credit_tasks[0], "verdict": "fail", "tx": f"0x{99:064x}"}, + ) + del free_zero # never validated: no row at all + costs = gateway({zero: "0", paid: "300000", credit_tasks[0]: "0"}) # `pending` is not known yet + assert await record_missing_costs(pool, costs) == 3 + assert await refresh_posteriors(pool, settings) == 1 + + got = (await posteriors(pool))[("1", "test-repair")] + expected = prior(settings).observe(False, None).observe(False, 300000.0) # in ledger order + assert got["n"] == 2 # the zero-cost failure counts; the pending one and the credits one do not + for name in ("alpha", "beta", "kappa", "tau"): + assert got[name] == pytest.approx(getattr(expected, name)) + # beta saw two failures; kappa and tau saw one cost (and one decay before it, from the prior only) + assert got["beta"] == pytest.approx(0.95 * (0.95 * 1 + 1) + 1) + assert got["kappa"] == pytest.approx(0.95 * 1 + 1) + assert got["tau"] == pytest.approx(0.95 * settings.reputation_cost_prior_micros + 300000) + + run(database_url, body) + + +def test_record_missing_costs_stops_at_the_batch_size(database_url: str, settings: Settings): + """Without a cap, a backlog of pending outcomes (an outage, or heavy volume) makes one pass make one + sequential gateway request per task with no bound; a request is required on every poll on top of + whatever the pass is still working through, so the backlog never shrinks.""" + + def body_with(n: int, batch: int) -> Callable[[Pool], Awaitable[None]]: + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester=REQUESTER, n_tasks=n) + for i, task in enumerate(tasks): + await verdict(pool, settings, task, "pass", f"{i:x}") + await released(pool, settings, task, f"{i + 8:x}") + costs = gateway({t: "100000" for t in tasks}) + assert await record_missing_costs(pool, costs, batch=batch) == min(n, batch) + + return body + + run(database_url, body_with(5, 2)) + + +def test_record_missing_costs_stops_at_the_time_budget(database_url: str, settings: Settings): + """A gateway that answers, but slowly, must not make one pass run arbitrarily long either; the rest + wait for the next poll, same as hitting the batch cap.""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester=REQUESTER, n_tasks=3) + for i, task in enumerate(tasks): + await verdict(pool, settings, task, "pass", f"{i:x}") + await released(pool, settings, task, f"{i + 8:x}") + + def handle(request: httpx.Request) -> httpx.Response: + task_id = request.url.params["task_id"] + return httpx.Response(200, json=[{"task_id": task_id, "cost_micros": "100000"}]) + + costs = GatewayCosts( + "http://gateway.test", client=httpx.AsyncClient(transport=httpx.MockTransport(handle)) + ) + written = await record_missing_costs(pool, costs, batch=100, budget_s=0.0) + assert written == 0 # the budget is spent before the first request is even tried + + run(database_url, body) + + +def test_the_updater_passes_its_settings_batch_and_budget( + database_url: str, settings: Settings, monkeypatch: pytest.MonkeyPatch +): + s = settings.model_copy(update={"reputation_cost_batch": 3, "reputation_cost_budget_s": 7.0}) + seen: dict[str, Any] = {} + + async def fake(pool: Pool, costs: GatewayCosts, *, batch: int, budget_s: float) -> int: + seen["batch"], seen["budget_s"] = batch, budget_s + return 0 + + import mergit_cp.reputation as reputation_module + + monkeypatch.setattr(reputation_module, "record_missing_costs", fake) + + async def body(pool: Pool) -> None: + await ReputationUpdater(pool, s, gateway()).update() + + run(database_url, body) + assert seen == {"batch": 3, "budget_s": 7.0} diff --git a/control-plane/tests/test_reputation_model.py b/control-plane/tests/test_reputation_model.py index 940201f..77aa670 100644 --- a/control-plane/tests/test_reputation_model.py +++ b/control-plane/tests/test_reputation_model.py @@ -73,7 +73,7 @@ def test_old_evidence_fades() -> None: @pytest.mark.parametrize("bad", [-1.0, float("nan"), float("inf")]) -def test_a_cost_that_is_not_a_finite_non_negative_number_is_refused(bad: float) -> None: +def test_a_cost_that_is_not_a_finite_positive_number_is_refused(bad: float) -> None: with pytest.raises(ValueError, match="cost"): Posterior.prior(PARAMS).observe(True, cost=bad) @@ -98,3 +98,19 @@ def test_draws_centre_on_the_posterior() -> None: draws = [draw(p, rng) for _ in range(4000)] assert sum(t for t, _ in draws) / len(draws) == pytest.approx(p.mean_success, abs=0.01) assert all(0.0 <= t <= 1.0 and c > 0.0 for t, c in draws) + + +def test_a_zero_cost_is_refused() -> None: + # Paper §8.3: c > 0. A run of zero costs decays tau toward 0, and density = E[theta] * kappa / tau then + # overflows to inf after ~14,000 of them (which JSON cannot carry: the reputation endpoint would 500). + with pytest.raises(ValueError, match="cost"): + Posterior.prior(PARAMS).observe(True, cost=0.0) + + +def test_an_outcome_without_a_usable_cost_moves_only_the_success_posterior() -> None: + # cost=None: the outcome is validated, so the Beta pair decays and increments as always (paper §8.2), and + # n counts it; the Gamma pair is untouched, neither decayed nor incremented (no cost evidence, §8.3). + p = Posterior.prior(PARAMS).observe(True, 4.0) # alpha 1.95, beta 0.95, kappa 1.95, tau 5.9 + q = p.observe(False, None) + assert (q.alpha, q.beta) == (pytest.approx(0.95 * 1.95), pytest.approx(0.95 * 0.95 + 1)) + assert (q.kappa, q.tau, q.n) == (p.kappa, p.tau, 2) diff --git a/control-plane/tests/test_reputation_outcomes.py b/control-plane/tests/test_reputation_outcomes.py index 332282c..b7b53c7 100644 --- a/control-plane/tests/test_reputation_outcomes.py +++ b/control-plane/tests/test_reputation_outcomes.py @@ -181,3 +181,151 @@ async def body(pool: Pool) -> None: await conn.execute("DELETE FROM task_costs WHERE task_id = %s", (tasks[0],)) run(database_url, body) + + +def _scans(plan: dict[str, Any]) -> list[tuple[str, str | None]]: + """Every (relation, index) a JSON EXPLAIN plan reads, depth first.""" + found: list[tuple[str, str | None]] = [] + if "Relation Name" in plan: + found.append((plan["Relation Name"], plan.get("Index Name"))) + for child in plan.get("Plans", []): + found.extend(_scans(child)) + return found + + +@pytest.mark.parametrize( + "query", + [ + # record_missing_costs, every REPUTATION_POLL_S + "SELECT task_id FROM task_outcomes WHERE cost_micros IS NULL ORDER BY position LIMIT 50", + # the credit-only count behind the unauthenticated GET /v0/agents/reputation + "SELECT agent_id, category, count(*) FROM task_outcomes WHERE funding = 'credits'" + " GROUP BY agent_id, category", + ], +) +def test_the_outcome_view_finds_each_tasks_ledger_rows_through_an_index( + database_url: str, settings: Settings, query: str +): + """Each of the view's per-task lookups into ``money_events`` must be an index lookup on the task, not a + walk of the whole ledger: otherwise the view costs tasks x ledger (12 s at 4k tasks / 12k events).""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester="0x" + "1" * 40) + await verdict(pool, settings, tasks[0], "fail", "1") + async with pool.connection() as conn, conn.transaction(): + await conn.execute("SET LOCAL enable_seqscan = off") # the tables are tiny; ask for the shape + cur = await conn.execute(f"EXPLAIN (FORMAT JSON) {query}") # test-only constant SQL + row = await cur.fetchone() + assert row is not None + scans = [index for relation, index in _scans(row[0][0]["Plan"]) if relation == "money_events"] + assert scans, "the view no longer reads money_events?" + assert set(scans) == {"money_events_task"}, scans + + run(database_url, body) + + +async def returned(pool: Pool, s: Settings, task: str, reason: str, t: str) -> None: + await receive( + pool, + s, + "task.returned", + {"task_id": task, "amount_usdc_units": "500000", "reason": reason, "tx": tx(t)}, + ) + + +def test_an_unaudited_release_that_a_challenge_overturns_is_an_adjudicated_failure( + database_url: str, settings: Settings +): + """Paper §8.2: validated is audited, *or challenged and adjudicated*. Released unaudited, then upheld.""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester="0x" + "1" * 40) + await released(pool, settings, tasks[0], "1") + assert await outcomes(pool, tasks) == {} # unaudited and unchallenged: still not evidence + await challenge(pool, settings, tasks[0], upheld=True, t="2") + got = await outcomes(pool, tasks) + assert list(got) == [tasks[0]] + assert got[tasks[0]]["success"] is False + # Nothing counted until the adjudication, so that is where it sits. + assert got[tasks[0]]["position"] == await ledger_id(pool, tasks[0], "challenge.resolved") + + run(database_url, body) + + +def test_a_challenge_upheld_before_release_is_an_adjudicated_failure(database_url: str, settings: Settings): + """No verdict and no release: the challenge is upheld inside the window and the escrow goes back.""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester="0x" + "1" * 40) + await challenge(pool, settings, tasks[0], upheld=True, t="1") + await returned(pool, settings, tasks[0], "challenge", "2") + got = await outcomes(pool, tasks) + assert list(got) == [tasks[0]] + assert got[tasks[0]]["success"] is False + assert got[tasks[0]]["position"] == await ledger_id(pool, tasks[0], "challenge.resolved") + + run(database_url, body) + + +def test_an_unaudited_release_that_survived_a_challenge_is_an_adjudicated_success( + database_url: str, settings: Settings +): + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester="0x" + "1" * 40) + await challenge(pool, settings, tasks[0], upheld=False, t="1") + assert await outcomes(pool, tasks) == {} # adjudicated, but not paid yet: wait for the release + await released(pool, settings, tasks[0], "2") + got = await outcomes(pool, tasks) + assert list(got) == [tasks[0]] + assert got[tasks[0]]["success"] is True + assert got[tasks[0]]["position"] == await ledger_id(pool, tasks[0], "task.released") + + run(database_url, body) + + +def test_a_challenge_rejected_after_an_unaudited_release_counts_from_the_rejection( + database_url: str, settings: Settings +): + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester="0x" + "1" * 40) + await released(pool, settings, tasks[0], "1") + await challenge(pool, settings, tasks[0], upheld=False, t="2") + got = (await outcomes(pool, tasks))[tasks[0]] + assert got["success"] is True + # The later of the two events is the one that made it count. + assert got["position"] == await ledger_id(pool, tasks[0], "challenge.resolved") + + run(database_url, body) + + +def test_the_audited_cases_keep_their_outcome_and_position(database_url: str, settings: Settings): + """Regression for the adjudication cases above: every audited case reads exactly as before.""" + + async def body(pool: Pool) -> None: + _, _, tasks = await seed_plan(pool, requester="0x" + "1" * 40, n_tasks=7) + fail, invalid, passed, upheld, rejected, fail_upheld, pass_unreleased = tasks + await verdict(pool, settings, fail, "fail", "1") + await verdict(pool, settings, invalid, "invalid", "2") + await verdict(pool, settings, passed, "pass", "3") + await released(pool, settings, passed, "4") + await verdict(pool, settings, upheld, "pass", "5") + await released(pool, settings, upheld, "6") + await challenge(pool, settings, upheld, upheld=True, t="7") + await verdict(pool, settings, rejected, "pass", "8") + await released(pool, settings, rejected, "9") + await challenge(pool, settings, rejected, upheld=False, t="a") + await verdict(pool, settings, fail_upheld, "fail", "e") + await challenge(pool, settings, fail_upheld, upheld=True, t="c") + await verdict(pool, settings, pass_unreleased, "pass", "d") # a verdict with no release + + got = {t: (r["success"], r["position"]) for t, r in (await outcomes(pool, tasks)).items()} + assert got == { + fail: (False, await ledger_id(pool, fail, "verdict.recorded")), + invalid: (False, await ledger_id(pool, invalid, "verdict.recorded")), + passed: (True, await ledger_id(pool, passed, "task.released")), + upheld: (False, await ledger_id(pool, upheld, "task.released")), + rejected: (True, await ledger_id(pool, rejected, "task.released")), + fail_upheld: (False, await ledger_id(pool, fail_upheld, "verdict.recorded")), + } + + run(database_url, body) diff --git a/control-plane/tests/test_routes_routing.py b/control-plane/tests/test_routes_routing.py index 17d7f25..5400dfc 100644 --- a/control-plane/tests/test_routes_routing.py +++ b/control-plane/tests/test_routes_routing.py @@ -7,18 +7,45 @@ # (the imported fixtures are used by name, so their parameters look like redefinitions) from __future__ import annotations +import json import random +from collections.abc import Iterator +from contextlib import contextmanager +from datetime import UTC, datetime, timedelta +from pathlib import Path +from types import SimpleNamespace +from typing import Any import psycopg import pytest from fastapi.testclient import TestClient +from mergit_cp import routes_routing +from mergit_cp.agents import Agent +from mergit_cp.app import create_app from mergit_cp.config import Settings from mergit_cp.db import Pool +from mergit_cp.eligibility import AT_CAP, IN_PROBATION from mergit_cp.reputation import record_missing_costs, refresh_posteriors -from mergit_cp.reputation_model import Posterior -from mergit_cp.routing import rank +from mergit_cp.reputation_model import Params, Posterior +from mergit_cp.routes_routing import ( + ALL_LEFT_OUT, + NO_FIT, + NO_FIT_ON_PRIOR_ALONE, + NO_FIT_THIS_DRAW, + PARTLY_LEFT_OUT, +) +from mergit_cp.routing import ( + BELOW_MAX_ATTEMPTS, + OVER_BUDGET, + OVER_BUDGET_IN_THIS_DRAW, + feasible_probability, + mean_feasible, + rank, +) +from tests.cp_fakes import FakeLLM from tests.test_api import seed_plan +from tests.test_eligibility import dispatch, invalid_verdict from tests.test_plans import app_client, create, llm, settings, sign_in # noqa: F401 (fixtures) from tests.test_reputation import gateway, run from tests.test_reputation_outcomes import verdict @@ -29,9 +56,13 @@ @pytest.fixture(autouse=True) def _no_history(database_url: str) -> None: """The posteriors are global and the database is shared, so outcomes other tests left behind would - otherwise already be in them.""" - with psycopg.connect(database_url) as conn: - conn.execute("TRUNCATE plans, agent_posteriors, task_costs CASCADE") + otherwise already be in them. ``run`` migrates before it empties the tables, like every other ledger + test, so this file also passes when it is the first (or only) one to touch a fresh database.""" + + async def nothing(_: Pool) -> None: + return None + + run(database_url, nothing) def decisions(settings: Settings, plan_uuid: str) -> list[tuple[int, int, list[dict[str, float | str]]]]: @@ -86,15 +117,44 @@ def test_an_exploring_answer_is_recorded_and_can_be_replayed(app_client: TestCli candidates = { c["agent_id"]: Posterior(c["alpha"], c["beta"], c["kappa"], c["tau"], c["n"]) for c in snapshot } # type: ignore[arg-type] - replay = rank( + replay, _ = rank( candidates, 500_000.0, explore=True, epsilon=settings.routing_epsilon, min_observations=settings.routing_min_observations, + min_feasible_probability=settings.routing_min_feasible_probability, rng=random.Random(f"{seed}/{index}"), - )[: settings.max_attempts] - assert [p.agent_id for p in replay] == [p["agent_id"] for p in got["tasks"][index]["picks"]] + ) + picked = [p.agent_id for p in replay[: settings.max_attempts]] + assert picked == [p["agent_id"] for p in got["tasks"][index]["picks"]] + + +def test_a_draft_with_a_recorded_draw_can_still_be_discarded_and_the_draw_outlives_it( + app_client: TestClient, settings: Settings +): + """A draw is the first thing that can point at a still-draft plan, and discarding a draft really deletes + it. The replay record is an audit log, so it stays, and it must not block the discard (migration 0022).""" + sign_in(app_client) + plan = create(app_client) + url = f"/v0/plans/{plan['plan_uuid']}" + drawn = app_client.get(f"{url}/suggestions", params={"explore": "true"}) + assert drawn.status_code == 200, drawn.text + assert len(decisions(settings, plan["plan_uuid"])) == 2 + + assert app_client.delete(url).status_code == 204 + assert app_client.get(url).status_code == 404 + logged = decisions(settings, plan["plan_uuid"]) + assert [(i, seed) for i, seed, _ in logged] == [(0, drawn.json()["seed"]), (1, drawn.json()["seed"])] + # Outliving the plan does not make the record editable: the append-only trigger still holds. + with psycopg.connect(settings.database_url) as conn: + for sql in ( + "DELETE FROM routing_decisions WHERE plan_uuid = %s", + "UPDATE routing_decisions SET seed = 0 WHERE plan_uuid = %s", + ): + with pytest.raises(psycopg.errors.RaiseException, match="append-only"), conn.transaction(): + conn.execute(sql, (plan["plan_uuid"],)) + assert decisions(settings, plan["plan_uuid"]) == logged def test_free_credit_work_does_not_change_a_suggestion( @@ -130,3 +190,384 @@ async def body(pool: Pool) -> None: picks = app_client.get(f"/v0/plans/{plan['plan_uuid']}/suggestions").json()["tasks"][0]["picks"] assert [p["agent_id"] for p in picks] == ["2", "1"] # the untried agent's prior beats a failing record assert {p["agent_id"]: p["counted_outcomes"] for p in picks} == {"1": 2, "2": 0} + + +def test_exploring_suggestions_are_rate_limited_per_requester(settings: Settings, llm: FakeLLM) -> None: + """Each exploring call appends a replay record per task to the append-only routing_decisions, so it is + limited per requester like the planner (plans_per_hour). A greedy call writes nothing and is not.""" + limited = settings.model_copy(update={"suggestions_per_hour": 2}) + with TestClient(create_app(limited, llm=llm, deliver_events=False), base_url="https://testserver") as c: + sign_in(c) + plan = create(c) + url = f"/v0/plans/{plan['plan_uuid']}/suggestions" + for _ in range(2): + assert c.get(url, params={"explore": "true"}).status_code == 200 + refused = c.get(url, params={"explore": "true"}) + assert refused.status_code == 429, refused.text + assert len(decisions(settings, plan["plan_uuid"])) == 2 * 2 # two calls x two tasks; not the third + assert c.get(url).status_code == 200 # greedy is still answered + + c.cookies.clear() + sign_in(c) # someone else has their own allowance + other = create(c) + theirs = f"/v0/plans/{other['plan_uuid']}/suggestions" + assert c.get(theirs, params={"explore": "true"}).status_code == 200 + + +# Why each agent is left out: every roster agent is in a task's picks or in its ineligible list, once. + +OPERATORS = {str(i): "0x" + c * 40 for i, c in zip(range(1, 6), "abcde", strict=True)} +PRICEY = Posterior(8.0, 3.0, 10.0, 20_000_000.0, 10) # mean cost 2M against a 500k task, on evidence +EXCELLENT = Posterior(18.0, 2.0, 15.0, 3_000_000.0, 20) # mean cost 200k, passes nine in ten +PICK_FIELDS = { + "agent_id", + "name", + "score_per_usdc", + "assumed_success", + "assumed_cost_micros", + "counted_outcomes", + "unproven", + "cost_is_prior", +} + + +def roster_agent(agent_id: str) -> Agent: + return Agent( + agent_id=agent_id, + name=f"agent-{agent_id}", + operator=OPERATORS[agent_id], + runtime_key="0x" + "2" * 40, + model="m", + system_prompt="p", + ) + + +@pytest.fixture +def roster(settings: Settings, tmp_path: Path) -> Settings: + """Five agents, each with its own operator, so probation or the cap on one leaves the others alone.""" + chain = tmp_path / "five-agents.json" + chain.write_text( + json.dumps( + { + "agents": [ + {"agent_id": i, "name": f"agent-{i}", "operator": op, "runtime_key": "0x" + "2" * 40} + for i, op in OPERATORS.items() + ] + } + ) + ) + config = tmp_path / "five-agents.toml" + config.write_text("".join(f'[agents.agent-{i}]\nmodel = "m"\nsystem_prompt = "p"\n' for i in OPERATORS)) + return settings.model_copy(update={"agents_file": str(chain), "agents_config": str(config)}) + + +@contextmanager +def suggesting(settings: Settings, llm: FakeLLM, **update: Any) -> Iterator[TestClient]: + """A signed-in requester. The reputation updater is off, since these tests write ``agent_posteriors`` + directly and a refresh would rebuild it from the (empty) ledger.""" + app = create_app( + settings.model_copy(update=update), + llm=llm, + deliver_events=False, + process_outcomes=False, + update_reputation=False, + ) + with TestClient(app, base_url="https://testserver") as c: + sign_in(c) + yield c + with psycopg.connect(settings.database_url) as conn: + conn.execute( + "TRUNCATE plans, outbox, planning_attempts, planner_costs, sessions, auth_nonces CASCADE" + ) + + +def history( + database_url: str, + *, + posteriors: dict[str, Posterior] | None = None, + probation: tuple[str, ...] = (), + in_flight: tuple[str, ...] = (), +) -> None: + """Posteriors in the plan's category; a falsification an hour ago for each agent in ``probation``; one + running dispatch for each agent in ``in_flight`` (at cap when ``routing_operator_concurrency_cap`` is + 1).""" + + async def body(pool: Pool) -> None: + async with pool.connection() as conn: + for agent_id, p in (posteriors or {}).items(): + await conn.execute( + "INSERT INTO agent_posteriors (agent_id, category, alpha, beta, kappa, tau, n)" + " VALUES (%s, 'test-repair', %s, %s, %s, %s, %s)", + (agent_id, p.alpha, p.beta, p.kappa, p.tau, p.n), + ) + now = datetime.now(UTC) + for agent_id in probation: + plan_uuid, task_id = await dispatch(pool, roster_agent(agent_id), at=now - timedelta(days=1)) + await invalid_verdict(pool, plan_uuid, task_id, now - timedelta(hours=1)) + for agent_id in in_flight: + await dispatch(pool, roster_agent(agent_id), at=now) + + run(database_url, body) + + +def suggest(c: TestClient, explore: bool = False) -> dict[str, Any]: + plan = create(c) + got = c.get(f"/v0/plans/{plan['plan_uuid']}/suggestions", params={"explore": str(explore).lower()}) + assert got.status_code == 200, got.text + return got.json() + + +def why(task: dict[str, Any]) -> dict[str, str]: + return {e["agent_id"]: e["reason"] for e in task["ineligible"]} + + +def picked(task: dict[str, Any]) -> list[str]: + return [p["agent_id"] for p in task["picks"]] + + +def test_greedy_lists_over_budget_agents_with_reason(roster: Settings, llm: FakeLLM, database_url: str): + history(database_url, posteriors={"1": PRICEY}) + with suggesting(roster, llm) as c: + body = suggest(c) + for task in body["tasks"]: + assert "1" not in picked(task) + assert {"agent_id": "1", "name": "agent-1", "reason": OVER_BUDGET} in task["ineligible"] + + +def test_probation_or_cap_reason_wins_over_over_budget(roster: Settings, llm: FakeLLM, database_url: str): + """1 and 2 would be over budget anyway, but probation and the cap are the more fundamental reasons. 3 is + the best agent on the roster by its record, and probation still keeps it out of the ranking.""" + history( + database_url, + posteriors={"1": PRICEY, "2": PRICEY, "3": EXCELLENT}, + probation=("1", "3"), + in_flight=("2",), + ) + with suggesting(roster, llm, routing_operator_concurrency_cap=1) as c: + greedy, explored = suggest(c), suggest(c, explore=True) + for task in greedy["tasks"]: + assert picked(task) == ["4", "5"] + assert why(task) == {"1": IN_PROBATION, "2": AT_CAP, "3": IN_PROBATION} + for task in explored["tasks"]: + assert set(picked(task)) <= {"4", "5"} + assert {a: r for a, r in why(task).items() if a in "123"} == { + "1": IN_PROBATION, + "2": AT_CAP, + "3": IN_PROBATION, + } + + +def test_max_attempts_cut_agents_are_listed(roster: Settings, llm: FakeLLM): + with suggesting(roster, llm) as c: + body = suggest(c) + for task in body["tasks"]: # five equal priors, ties by agent id + assert picked(task) == ["1", "2", "3"] + assert task["ineligible"] == [ + {"agent_id": "4", "name": "agent-4", "reason": BELOW_MAX_ATTEMPTS}, + {"agent_id": "5", "name": "agent-5", "reason": BELOW_MAX_ATTEMPTS}, + ] + + +def test_every_roster_agent_is_in_picks_or_ineligible(roster: Settings, llm: FakeLLM, database_url: str): + history(database_url, posteriors={"1": PRICEY, "3": EXCELLENT}, probation=("2",)) + reasons = {IN_PROBATION, AT_CAP, OVER_BUDGET, OVER_BUDGET_IN_THIS_DRAW, BELOW_MAX_ATTEMPTS} + with suggesting(roster, llm) as c: + bodies = [suggest(c)] + [suggest(c, explore=True) for _ in range(6)] + for body in bodies: + for task in body["tasks"]: + ids = picked(task) + [e["agent_id"] for e in task["ineligible"]] + assert sorted(ids) == sorted(OPERATORS) # each exactly once + assert set(why(task).values()) <= reasons + assert {e["name"] for e in task["ineligible"]} == {f"agent-{a}" for a in why(task)} + assert why(task)["2"] == IN_PROBATION and why(task)["1"] == OVER_BUDGET + + +@pytest.mark.parametrize("explore", [False, True]) +def test_existing_response_fields_unchanged(roster: Settings, llm: FakeLLM, explore: bool): + with suggesting(roster, llm) as c: + body = suggest(c, explore=explore) + assert set(body) == {"plan_id", "mode", "seed", "note", "tasks"} + assert isinstance(body["plan_id"], str) and body["mode"] == ("explore" if explore else "greedy") + assert isinstance(body["seed"], int) if explore else body["seed"] is None + assert "not an assignment" in body["note"] + assert [t["index"] for t in body["tasks"]] == [0, 1] + for task in body["tasks"]: + # "ineligible" is the one new field; "reason" is there exactly when "picks" is empty, as before. + reason = set() if task["picks"] else {"reason"} + assert set(task) == {"index", "budget_usdc_units", "picks", "ineligible"} | reason + assert task["budget_usdc_units"] == "500000" + for pick in task["picks"]: + assert set(pick) == PICK_FIELDS + assert isinstance(pick["agent_id"], str) and pick["name"] == f"agent-{pick['agent_id']}" + assert all(isinstance(pick[k], float) for k in ("score_per_usdc", "assumed_success")) + assert isinstance(pick["assumed_cost_micros"], float) + assert (pick["counted_outcomes"], pick["unproven"], pick["cost_is_prior"]) == (0, True, True) + if not explore: # the prior's own means, exactly as before (plus the additive "cost_is_prior") + assert task["picks"][0] == { + "agent_id": "1", + "name": "agent-1", + "score_per_usdc": 1.0, + "assumed_success": 0.5, + "assumed_cost_micros": 500_000.0, + "counted_outcomes": 0, + "unproven": True, + "cost_is_prior": True, + } + + +@pytest.mark.parametrize("explore", [False, True]) +def test_a_pick_says_when_its_cost_is_only_the_prior( + roster: Settings, llm: FakeLLM, database_url: str, explore: bool +): + """Agent 1 has counted outcomes, but every one metered 0, so its cost is still the configured prior + (``Posterior.observe``) and a client must not show it as measured. Agent 2 has recorded costs. Agents + 3-5 have no history at all.""" + prior = Posterior.prior(Params(tau0=float(roster.reputation_cost_prior_micros))) + no_cost = prior.observe(True, None).observe(True, None).observe(False, None) + assert no_cost.n == 3 + history(database_url, posteriors={"1": no_cost, "2": EXCELLENT}) + with suggesting(roster, llm) as c: + body = suggest(c, explore=explore) + for task in body["tasks"]: + for pick in task["picks"]: + assert pick["cost_is_prior"] is (pick["agent_id"] != "2"), pick + if not explore: + by_id = {p["agent_id"]: p for p in task["picks"]} + assert by_id["1"]["counted_outcomes"] == 3 and by_id["1"]["cost_is_prior"] is True + assert by_id["2"]["cost_is_prior"] is False + + +@pytest.mark.parametrize("explore", [False, True]) +def test_an_empty_answer_says_when_it_rests_on_the_prior_alone( + roster: Settings, llm: FakeLLM, database_url: str, explore: bool +): + """No agent has a recorded cost, and the configured prior's mean (5M) is ten times the task's budget: + the answer is still empty, but it says the expectation is an assumption, not what anyone was seen to do. + One agent in probation does not change that, but the answer says it is out for another reason too.""" + history(database_url, probation=("1",)) + with suggesting(roster, llm, reputation_cost_prior_micros=5_000_000) as c: + body = suggest(c, explore=explore) + for task in body["tasks"]: + assert task["picks"] == [] + assert why(task) == {"1": IN_PROBATION} | {a: OVER_BUDGET for a in "2345"} + assert task["reason"] == PARTLY_LEFT_OUT + NO_FIT_ON_PRIOR_ALONE + assert "not evidence" in task["reason"] and "starting" in task["reason"] + + +def test_an_empty_answer_backed_by_evidence_says_so_plainly( + roster: Settings, llm: FakeLLM, database_url: str +): + history(database_url, posteriors=dict.fromkeys(OPERATORS, PRICEY)) + with suggesting(roster, llm) as c: + body = suggest(c) + for task in body["tasks"]: + assert task["picks"] == [] and set(why(task).values()) == {OVER_BUDGET} + assert task["reason"] == NO_FIT + + +def test_an_agent_with_outcomes_but_no_recorded_cost_still_rests_on_the_prior( + roster: Settings, llm: FakeLLM, database_url: str +): + """Outcomes whose receipts metered 0 count toward success but leave the cost posterior at the prior + (``Posterior.observe``), so a budget call on it is still only the assumption.""" + prior = Posterior(1.0, 1.0, 1.0, 5_000_000.0) + no_cost = prior.observe(True, None).observe(False, None).observe(True, None) + history(database_url, posteriors=dict.fromkeys(OPERATORS, no_cost)) + with suggesting(roster, llm, reputation_cost_prior_micros=5_000_000) as c: + body = suggest(c) + for task in body["tasks"]: + assert task["picks"] == [] and task["reason"] == NO_FIT_ON_PRIOR_ALONE + + +def test_an_empty_answer_when_every_agent_is_left_out_for_now( + roster: Settings, llm: FakeLLM, database_url: str +): + history(database_url, probation=tuple(OPERATORS)) + with suggesting(roster, llm) as c: + body = suggest(c) + for task in body["tasks"]: + assert task["picks"] == [] and set(why(task).values()) == {IN_PROBATION} + assert task["reason"] == ALL_LEFT_OUT + + +# Explore: an empty answer because every draw missed is not "no agent is expected to fit". + + +def unlucky_seed(settings: Settings, candidates: dict[str, Posterior], reasons: set[str]) -> int: + """The first seed whose draw for task 0 (as the route seeds it) ranks nobody and leaves the candidates out + for exactly ``reasons``. Found by search over the same pure ``rank`` the route calls, so it is fixed.""" + for seed in range(10_000): + ranked, left_out = rank( + candidates, + 500_000.0, + explore=True, + epsilon=settings.routing_epsilon, + min_observations=settings.routing_min_observations, + min_feasible_probability=settings.routing_min_feasible_probability, + rng=random.Random(f"{seed}/0"), + ) + if not ranked and set(left_out.values()) == reasons: + return seed + raise AssertionError(f"no seed in 10000 gives {reasons}") + + +def seeded(monkeypatch: pytest.MonkeyPatch, seed: int) -> None: + """Make the route's exploring seed ``seed`` instead of a random one.""" + monkeypatch.setattr(routes_routing, "secrets", SimpleNamespace(randbits=lambda _bits: seed)) + + +def test_an_explore_answer_where_every_draw_missed_does_not_claim_no_agent_fits( + roster: Settings, llm: FakeLLM, monkeypatch: pytest.MonkeyPatch +): + """Cold start: five agents on the prior (mean 500k) for a 500k task. Each one fits (greedy suggests + them) and each draw fits about 40% of the time, so all five missing at once happens on about 7% of + exploring calls. That is one unlucky draw, not evidence, and the answer must not say otherwise.""" + prior = Posterior.prior(Params(tau0=float(roster.reputation_cost_prior_micros))) + eps, floor = roster.routing_epsilon, roster.routing_min_feasible_probability + assert mean_feasible(prior, 500_000, eps) and feasible_probability(prior, 500_000, eps) > 0.4 > floor + seed = unlucky_seed(roster, dict.fromkeys(OPERATORS, prior), {OVER_BUDGET_IN_THIS_DRAW}) + seeded(monkeypatch, seed) + with suggesting(roster, llm) as c: + explored, greedy = suggest(c, explore=True), suggest(c) + task = explored["tasks"][0] + assert explored["seed"] == seed and task["picks"] == [] + assert set(why(task).values()) == {OVER_BUDGET_IN_THIS_DRAW} + assert task["reason"] == NO_FIT_THIS_DRAW + assert "expected to fit" not in task["reason"] and "raise the budget" not in task["reason"] + assert picked(greedy["tasks"][0]) == ["1", "2", "3"] # the same posteriors, ranked greedily + + +def test_an_explore_answer_with_some_over_budget_and_the_rest_unlucky_says_it_was_the_draw( + roster: Settings, llm: FakeLLM, database_url: str, monkeypatch: pytest.MonkeyPatch +): + """Agent 1 is over budget on evidence; the other four could fit and only missed this draw. "No agent is + expected to fit" is true of agent 1 only, so it is not what the answer says.""" + history(database_url, posteriors={"1": PRICEY}) + prior = Posterior.prior(Params(tau0=float(roster.reputation_cost_prior_micros))) + candidates = dict.fromkeys(OPERATORS, prior) | {"1": PRICEY} + seed = unlucky_seed(roster, candidates, {OVER_BUDGET, OVER_BUDGET_IN_THIS_DRAW}) + seeded(monkeypatch, seed) + with suggesting(roster, llm) as c: + task = suggest(c, explore=True)["tasks"][0] + assert task["picks"] == [] + assert why(task) == {"1": OVER_BUDGET} | dict.fromkeys("2345", OVER_BUDGET_IN_THIS_DRAW) + assert task["reason"] == NO_FIT_THIS_DRAW + + +def test_an_empty_answer_with_agents_out_for_probation_does_not_blame_the_budget_alone( + roster: Settings, llm: FakeLLM, database_url: str +): + """Agent 3 would fit easily; it is out for probation. The rest are over budget on evidence. The answer + says the budget rules out only the others.""" + history( + database_url, + posteriors=dict.fromkeys(OPERATORS, PRICEY) | {"3": EXCELLENT}, + probation=("3",), + ) + with suggesting(roster, llm) as c: + body = suggest(c) + for task in body["tasks"]: + assert task["picks"] == [] + assert why(task) == {"3": IN_PROBATION} | dict.fromkeys("1245", OVER_BUDGET) + assert task["reason"] == PARTLY_LEFT_OUT + NO_FIT + assert task["reason"].startswith("some agents are left out for probation or the operator cap") diff --git a/control-plane/tests/test_routing.py b/control-plane/tests/test_routing.py index 53464bd..95bb4dc 100644 --- a/control-plane/tests/test_routing.py +++ b/control-plane/tests/test_routing.py @@ -1,13 +1,25 @@ """Ranking agents for a task: greedy by posterior means, or a seeded Thompson draw (paper §9.3 and §9.4).""" +import math import random import pytest - -from mergit_cp.reputation_model import Params, Posterior -from mergit_cp.routing import rank +from annotated_types import Le + +from mergit_cp.config import Settings +from mergit_cp.reputation_model import Params, Posterior, draw +from mergit_cp.routing import ( + OVER_BUDGET, + OVER_BUDGET_IN_THIS_DRAW, + Pick, + budget_limit, + feasible_probability, + mean_feasible, + rank, +) PRIOR = Posterior.prior(Params(tau0=500_000.0)) +THRESHOLD = 0.01 # routing_min_feasible_probability's default def seasoned(successes: int, failures: int, cost: float) -> Posterior: @@ -20,14 +32,25 @@ def seasoned(successes: int, failures: int, cost: float) -> Posterior: return p -def ranked(candidates: dict[str, Posterior], budget: float = 1_000_000, **over: object) -> list[str]: +def ranking( + candidates: dict[str, Posterior], budget: float = 1_000_000, **over: object +) -> tuple[list[Pick], dict[str, str]]: args: dict[str, object] = { "explore": False, "epsilon": 0.1, "min_observations": 5, + "min_feasible_probability": THRESHOLD, "rng": random.Random(0), } | over - return [p.agent_id for p in rank(candidates, budget, **args)] # type: ignore[arg-type] + return rank(candidates, budget, **args) # type: ignore[arg-type] + + +def picks(candidates: dict[str, Posterior], budget: float = 1_000_000, **over: object) -> list[Pick]: + return ranking(candidates, budget, **over)[0] + + +def ranked(candidates: dict[str, Posterior], budget: float = 1_000_000, **over: object) -> list[str]: + return [p.agent_id for p in picks(candidates, budget, **over)] def test_greedy_ranks_by_mean_success_over_mean_cost(): @@ -48,9 +71,11 @@ def test_ties_break_by_agent_id_not_by_insertion_order(): assert ranked({"b": PRIOR, "a": PRIOR}) == ["a", "b"] -def test_an_agent_whose_cost_would_not_fit_the_budget_is_skipped(): +def test_an_agent_whose_cost_would_not_fit_the_budget_is_left_out_and_says_so(): pricey = seasoned(10, 0, 5_000_000) - assert ranked({"cheap": seasoned(5, 5, 200_000), "pricey": pricey}, budget=1_000_000) == ["cheap"] + got, left_out = ranking({"cheap": seasoned(5, 5, 200_000), "pricey": pricey}, budget=1_000_000) + assert [p.agent_id for p in got] == ["cheap"] + assert left_out == {"pricey": OVER_BUDGET} def test_the_budget_slack_lets_a_slightly_dearer_agent_in(): @@ -60,27 +85,19 @@ def test_the_budget_slack_lets_a_slightly_dearer_agent_in(): def test_when_nothing_fits_the_answer_is_empty_not_a_guess(): - assert ranked({"a": seasoned(10, 0, 9_000_000)}, budget=100_000) == [] + assert ranking({"a": seasoned(10, 0, 9_000_000)}, budget=100_000) == ([], {"a": OVER_BUDGET}) def test_a_newcomer_is_ranked_on_its_prior_and_called_unproven(): - picks = rank( - {"new": PRIOR, "old": seasoned(20, 0, 300_000)}, - 1_000_000, - explore=False, - epsilon=0.1, - min_observations=5, - rng=random.Random(0), - ) - by_id = {p.agent_id: p for p in picks} + by_id = {p.agent_id: p for p in picks({"new": PRIOR, "old": seasoned(20, 0, 300_000)})} assert by_id["new"].unproven and by_id["new"].counted_outcomes == 0 assert not by_id["old"].unproven and by_id["old"].counted_outcomes == 20 def test_an_exploring_draw_is_reproducible_from_its_seed(): cands = {"a": seasoned(3, 3, 300_000), "b": seasoned(3, 3, 300_000), "c": PRIOR} - first = rank(cands, 1_000_000, explore=True, epsilon=0.1, min_observations=5, rng=random.Random(5)) - again = rank(cands, 1_000_000, explore=True, epsilon=0.1, min_observations=5, rng=random.Random(5)) + first = ranking(cands, explore=True, rng=random.Random(5)) + again = ranking(cands, explore=True, rng=random.Random(5)) assert first == again @@ -88,29 +105,17 @@ def test_an_exploring_draw_skips_an_agent_whose_drawn_cost_cannot_fit(): # kappa 1, tau 1e12: the drawn cost is around a million times the budget, every time. hopeless = Posterior(alpha=5.0, beta=1.0, kappa=1.0, tau=1e12, n=10) for seed in range(20): - got = rank( - {"h": hopeless, "ok": seasoned(5, 5, 300_000)}, - 1_000_000, - explore=True, - epsilon=0.1, - min_observations=5, - rng=random.Random(seed), + got, left_out = ranking( + {"h": hopeless, "ok": seasoned(5, 5, 300_000)}, explore=True, rng=random.Random(seed) ) assert [p.agent_id for p in got] == ["ok"] + assert left_out == {"h": OVER_BUDGET} def test_exploring_still_favours_the_better_agent_most_of_the_time(): good, poor = seasoned(18, 2, 300_000), seasoned(3, 17, 300_000) wins = sum( - rank( - {"good": good, "poor": poor}, - 1_000_000, - explore=True, - epsilon=0.1, - min_observations=5, - rng=random.Random(seed), - )[0].agent_id - == "good" + picks({"good": good, "poor": poor}, explore=True, rng=random.Random(seed))[0].agent_id == "good" for seed in range(300) ) assert wins > 270 # an experienced, clearly better agent wins nearly every draw @@ -119,14 +124,7 @@ def test_exploring_still_favours_the_better_agent_most_of_the_time(): def test_exploring_gives_an_uncertain_newcomer_some_of_the_draws(): veteran, newcomer = seasoned(14, 6, 300_000), PRIOR # newcomer's prior is wide: worth trying sometimes firsts = [ - rank( - {"vet": veteran, "new": newcomer}, - 1_000_000, - explore=True, - epsilon=0.1, - min_observations=5, - rng=random.Random(seed), - )[0].agent_id + picks({"vet": veteran, "new": newcomer}, explore=True, rng=random.Random(seed))[0].agent_id for seed in range(400) ] assert 0 < firsts.count("new") < 400 @@ -134,14 +132,156 @@ def test_exploring_gives_an_uncertain_newcomer_some_of_the_draws(): @pytest.mark.parametrize("explore", [False, True]) def test_every_pick_says_what_it_was_scored_on(explore: bool): - picks = rank( - {"a": seasoned(4, 1, 300_000)}, - 1_000_000, - explore=explore, - epsilon=0.1, - min_observations=5, - rng=random.Random(3), - ) - p = picks[0] + p = picks({"a": seasoned(4, 1, 300_000)}, explore=explore, rng=random.Random(3))[0] assert p.score == pytest.approx(p.success / p.cost) assert 0.0 <= p.success <= 1.0 and p.cost > 0.0 + + +# Budget feasibility. + + +def at(kappa: float, x: float, budget: float = 1_000_000, epsilon: float = 0.1) -> float: + """``feasible_probability`` for a posterior whose ``tau / limit`` is ``x``.""" + tau = x * budget_limit(budget, epsilon) + return feasible_probability(Posterior(1.0, 1.0, kappa, tau), budget, epsilon) + + +def q_integer(n: int, x: float) -> float: + """Q(n, x) for a whole n: the chance a Poisson(x) count is below n.""" + return math.exp(-x) * sum(x**k / math.factorial(k) for k in range(n)) + + +def q_half(n: int, x: float) -> float: + """Q(n + 1/2, x): erfc(sqrt x) plus a finite sum, from Q(a + 1, x) = Q(a, x) + x^a e^-x / Gamma(a + 1).""" + return math.erfc(math.sqrt(x)) + math.exp(-x) * sum( + x ** (k + 0.5) / math.gamma(k + 1.5) for k in range(n) + ) + + +@pytest.mark.parametrize( + ("kappa", "x", "expected"), + [ + # kappa = 1, the cold prior: exp(-x), as the reference branch's docstring says. Both evaluation + # branches (the series for x < kappa + 1, the continued fraction otherwise). + (1.0, 0.5, math.exp(-0.5)), + (1.0, 3.0, math.exp(-3.0)), + # Whole kappa: a Poisson tail. Series, near the switch, and continued fraction. + (4.0, 2.0, q_integer(4, 2.0)), + (4.0, 4.9, q_integer(4, 4.9)), + (4.0, 11.0, q_integer(4, 11.0)), + (20.0, 20.0, q_integer(20, 20.0)), + # Half-whole kappa: erfc plus a finite sum. + (0.5, 0.2, q_half(0, 0.2)), + (2.5, 0.7, q_half(2, 0.7)), + (2.5, 9.0, q_half(2, 9.0)), + # Neither: values from mpmath 1.3, gammainc(a, x, inf, regularized=True) at 40 digits. + (7.3, 4.1, 0.90203619465415758655), + (7.3, 12.6, 0.040957636455877532386), + (0.4, 3.0, 0.010001123188919289088), + (1.95, 0.95, 0.74047986432453140266), + ], +) +def test_feasible_probability_matches_independent_computation(kappa: float, x: float, expected: float): + assert at(kappa, x) == pytest.approx(expected, rel=1e-12, abs=1e-15) + + +@pytest.mark.parametrize(("budget", "epsilon"), [(1_000_000, 0.1), (250_000, 0.0), (40_000, 0.5)]) +def test_feasible_probability_is_the_chance_one_draw_fits(budget: float, epsilon: float): + """What the closed form claims, checked against the draw itself (``reputation_model.draw``).""" + posterior = Posterior(1.0, 1.0, kappa=3.0, tau=2.4 * budget) + rng = random.Random(11) + limit = budget_limit(budget, epsilon) + fits = sum(draw(posterior, rng)[1] <= limit for _ in range(40_000)) / 40_000 + assert fits == pytest.approx(feasible_probability(posterior, budget, epsilon), abs=0.01) + + +def test_feasible_probability_of_a_budget_of_nothing_is_zero(): + assert feasible_probability(PRIOR, 0, 0.1) == 0.0 + + +def test_mean_feasible_is_the_mean_cost_within_the_limit(): + p = Posterior(1.0, 1.0, kappa=2.0, tau=2_200_000) # mean cost 1.1M + assert mean_feasible(p, 1_000_000, 0.1) + assert not mean_feasible(p, 999_999, 0.1) + assert budget_limit(1_000_000, 0.1) == pytest.approx(1_100_000) + + +@pytest.mark.parametrize("kappa", [1.0, 1.5, 1.95, 3.0, 7.3, 20.0, 400.0]) +def test_mean_feasible_agent_is_never_below_min_feasible_probability(kappa: float): + """If the mean cost fits, ``tau / limit <= kappa``, so the chance a draw fits is at least + ``Q(kappa, kappa)``: 1/e at kappa 1, rising toward 1/2. Every posterior here has kappa >= 1 (the prior's + is 1 and each observed cost moves it to 0.95 kappa + 1). The setting is capped well below that, so the + explore-mode threshold can never leave out an agent greedy mode would suggest.""" + worst = Posterior(1.0, 1.0, kappa=kappa, tau=kappa * budget_limit(300_000, 0.1)) # mean exactly at limit + assert mean_feasible(worst, 300_000, 0.1) + p = feasible_probability(worst, 300_000, 0.1) + assert p >= math.exp(-1) - 1e-12 + + field = Settings.model_fields["routing_min_feasible_probability"] + cap = next(m.le for m in field.metadata if isinstance(m, Le)) + assert p > cap == 0.25 + + +def test_explore_skips_agent_below_threshold_even_if_draw_would_fit(): + """The threshold is applied to the agent up front, never to one draw: an agent whose chance of fitting is + below it is left out of every draw, including those where its own draw would have fit.""" + budget = 1_000_000 + limit = budget_limit(budget, 0.1) + unlikely = Posterior(5.0, 1.0, kappa=1.0, tau=limit * math.log(1 / 0.005), n=12) # fits 0.5% of draws + assert feasible_probability(unlikely, budget, 0.1) == pytest.approx(0.005) + + # Seeds where its own draw (the first one taken from the generator) would fit the budget. + lucky = [s for s in range(5_000) if draw(unlikely, random.Random(s))[1] <= limit] + assert lucky, "no draw of a 0.5% posterior fit in 5000 tries" + for seed in [*lucky, *range(200)]: + got, left_out = ranking({"u": unlikely}, budget, explore=True, rng=random.Random(seed)) + assert got == [] and left_out == {"u": OVER_BUDGET} + + # The threshold is what does it: below the agent's own chance, the lucky draws pick it. + for seed in lucky: + got, _ = ranking( + {"u": unlikely}, budget, explore=True, min_feasible_probability=0.001, rng=random.Random(seed) + ) + assert [p.agent_id for p in got] == ["u"] + + +def test_an_agent_left_out_up_front_takes_no_draw(): + """Leaving an agent out before the draw means it consumes nothing from the generator, so the others' + draws are the ones they would have had without it.""" + budget = 1_000_000 + unlikely = Posterior(5.0, 1.0, kappa=1.0, tau=budget_limit(budget, 0.1) * 10, n=12) # e^-10 + ok = seasoned(5, 5, 300_000) + for seed in range(20): + with_it, _ = ranking({"a": unlikely, "b": ok}, budget, explore=True, rng=random.Random(seed)) + without, _ = ranking({"b": ok}, budget, explore=True, rng=random.Random(seed)) + assert with_it == without + + +def test_a_draw_over_budget_is_reported_as_this_draw_only(): + """An agent above the threshold whose draw happens not to fit is out of this draw, not of the task.""" + budget = 1_000_000 + coin = Posterior(5.0, 1.0, kappa=1.0, tau=budget_limit(budget, 0.1) * math.log(2), n=12) # fits half + left_out = [ranking({"c": coin}, budget, explore=True, rng=random.Random(s))[1] for s in range(60)] + reasons = {r.get("c") for r in left_out} + assert reasons == {None, OVER_BUDGET_IN_THIS_DRAW} + + +@pytest.mark.parametrize("explore", [False, True]) +def test_every_candidate_is_ranked_or_left_out_with_a_reason(explore: bool): + budget = 400_000 + cands = { + "new": PRIOR, # mean 500k: over in greedy, a ~33% draw in explore + "cheap": seasoned(5, 5, 100_000), + "dear": seasoned(9, 1, 2_000_000), + "hopeless": Posterior(1.0, 1.0, kappa=1.0, tau=1e12, n=3), + "edge": seasoned(10, 0, 430_000), + } + for seed in range(50): + got, left_out = ranking(cands, budget, explore=explore, rng=random.Random(seed)) + ids = [p.agent_id for p in got] + assert len(ids) == len(set(ids)) + assert set(ids).isdisjoint(left_out) + assert set(ids) | set(left_out) == set(cands) + assert set(left_out.values()) <= {OVER_BUDGET, OVER_BUDGET_IN_THIS_DRAW} + if not explore: + assert OVER_BUDGET_IN_THIS_DRAW not in left_out.values() diff --git a/docs/reputation-v0.md b/docs/reputation-v0.md index 7aea1c9..26b941f 100644 --- a/docs/reputation-v0.md +++ b/docs/reputation-v0.md @@ -10,26 +10,36 @@ Per agent and task category, from validated outcomes only: - **Success** `θ`: a Beta posterior. Prior (1, 1). On each outcome both parameters decay by γ = 0.95, then a success adds 1 to α or a failure adds 1 to β (paper §8.2). The memory is about twenty outcomes. -- **Cost** `c`: a Gamma posterior over the rate of an exponential cost. On each outcome both parameters decay, - then κ gains 1 and τ gains the task's cost (paper §8.3, every completed task, passed or failed). +- **Cost** `c`: a Gamma posterior over the rate of an exponential cost. On a validated outcome with a real, positive + cost, both parameters decay, then κ gains 1 and τ gains the task's cost. A validated outcome whose metered cost + is zero still updates success/failure (α/β) but leaves κ/τ untouched entirely, neither decayed nor incremented + (paper §8.3 requires cost `> 0`). Only validated outcomes are folded, so the cost of a task nobody audited or + challenged does not move κ or τ. The paper (§8.3) uses cost from *all* completed tasks, and cost is attested + whether or not the task was audited; whether unaudited tasks' costs should count here is an open question, not + a decision. - **Value density** `D = E[θ] / E[c]`, a ratio of expectations (paper §8.4): verified success per unit of cost. -Read them at `GET /v0/agents/reputation` (every agent-category pair with counted outcomes; with -`?category=` every agent on the roster in that category) and `GET /v0/agents/{id}/reputation`. An agent with no -counted outcome shows the prior and says so (`basis: "prior (no counted outcomes)"`). +Read them at `GET /v0/agents/reputation` (every agent-category pair with counted outcomes, and every pair with +only free-credit outcomes, which shows the prior and its `uncounted_credit_outcomes`; with `?category=` every agent +on the roster in that category) and `GET /v0/agents/{id}/reputation`. An agent with no counted outcome shows the +prior and says so (`basis: "prior (no counted outcomes)"`). ## What counts as an outcome (invariant 4) -One row per task, in the view `task_outcomes` (migration 0015), which is the only place the rule lives: +One row per task, in the view `task_outcomes` (migration 0015, replaced by 0020), which is the only place the +rule lives. An outcome counts if it was audited or was challenged and adjudicated (paper §8.2): | Ledger | Counts as | | --- | --- | | validator verdict `fail` or `invalid` | a failure, at that event | | verdict `pass` and `task.released`, no upheld challenge | a success, at the release | | verdict `pass`, released, then a challenge **upheld** | one failure (the adjudicated outcome replaces the audit's), placed where the release was | -| released with no verdict (unaudited) | nothing | +| no verdict, and a challenge **upheld** (before or after a release) | a failure, at the upheld resolution | +| verdict `pass`, no release, and a challenge **upheld** | a failure, at the upheld resolution | +| no verdict, a challenge **rejected**, and released | a success, at whichever of the release and the rejection came later | +| released with no verdict and no challenge resolved (unaudited) | nothing | | returned at the deadline | nothing | -| a receipt, or a verdict with no release | nothing | +| a receipt, a verdict with no release (and no upheld challenge), or a rejected challenge with no release yet | nothing | | Φ results from the worker | nothing, ever | A task that never delivered (deadline) is therefore not penalised. That is the paper's rule (only audited or @@ -45,13 +55,18 @@ The flag is the plan's funding, not the address shape: a wallet user may also fu - **Outcomes:** the money ledger, `money_events`, which is append-only. - **Cost:** the receipt's metered cost, as committed on chain, read from the gateway (`GET /v0/commitments?task_id=`) and copied once into `task_costs` (write-once). A cost that cannot be read yet - leaves the outcome out of the posterior until it can; it is never taken as zero. + leaves the outcome out of the posterior until it can; it is never taken as zero. A commitment that really says + zero is recorded as it is, and the outcome's success or failure counts like any other; but the cost model needs + a positive cost (paper §8.3), so that outcome leaves the cost posterior (κ, τ) unchanged rather than adding an + invented cost. - **Posteriors:** `agent_posteriors` is a cache. It is rebuilt from the two sources above on every pass, so it cannot disagree with a rebuild, and a challenge that is upheld later changes it correctly. Settings: `REPUTATION_POLL_S` (how often, default 5) and `REPUTATION_COST_PRIOR_MICROS` (the prior mean cost, default 500000 USDC micros). The default is configuration, not a measurement; it only matters until an agent -has a few validated outcomes. +has a few validated outcomes. Reading a cost from the gateway is one sequential request per pending task, so +one pass is also bounded: `REPUTATION_COST_BATCH` (default 200 tasks) and `REPUTATION_COST_BUDGET_S` (default +10s wall-clock); either limit stops the pass, and whatever is left over waits for the next poll. ## Suggestions (routing, advice only) @@ -59,26 +74,92 @@ has a few validated outcomes. to fit the task's budget, best first, at most `max_attempts` of them. Approval is unchanged: the requester sends `agents` to `/approve` and may ignore the suggestion. -- **Greedy (default):** rank by posterior-mean value density; nothing random. This is the paper's exploration - suppression (§9.4). -- **`?explore=true`:** Thompson sampling (§9.3). Each agent draws θ from its Beta and a cost from its Gamma; an agent - whose drawn cost exceeds the budget (plus `ROUTING_EPSILON`, default 0.1) is skipped; the rest rank by drawn - θ over drawn cost. The seed and each agent's posterior are stored in `routing_decisions`, so a draw can be replayed. +- **Greedy (default):** rank by posterior-mean value density; nothing random. An agent whose posterior mean cost + exceeds the budget (plus `ROUTING_EPSILON`, default 0.1) is left out as over budget. Greedy is the paper's exploration + suppression (§9.4), which the paper allows only as a per-task escape hatch: "§9.5 explains why it cannot be the + default". Making it the default is this slice's divergence from the paper, not something the paper recommends. + It is defensible while the §9.5 subsidy (the platform, not the requester, bearing a failed exploratory draw) + does not exist, since exploring would otherwise spend the requester's money; it should be revisited when it does. +- **`?explore=true`:** Thompson sampling (§9.3). First, an agent whose chance of drawing a cost within the budget + (plus `ROUTING_EPSILON`) is below `ROUTING_MIN_FEASIBLE_PROBABILITY` (default 0.01) is left out as over budget and + takes no draw. That chance is computed from the agent's cost posterior (cost `1/r`, `r ~ Gamma(κ, rate τ)`, so it is + the upper regularized incomplete gamma `Q(κ, τ / limit)`), once per agent and task, so no lucky draw brings such an + agent back. The paper screens each draw only; this floor and its value are our parameterisation. It is capped at + 0.25 because an agent whose mean cost fits always has at least a 1/e chance, so exploring never leaves out an agent + greedy would suggest. Each remaining agent draws θ from its Beta and a cost from its Gamma; one whose drawn cost + exceeds the budget is left out of this draw only; the rest rank by drawn θ over drawn cost. The seed and the + posterior of every agent passed to the ranking (all those probation and the cap left in, including any the ranking + then left out on budget) are stored in `routing_decisions`, so a draw can be replayed. Agents probation or the cap + left out are not in that record. + Because each exploring call writes that record, a requester may make `SUGGESTIONS_PER_HOUR` (default 60) of them + an hour; past that the answer is 429. Greedy calls write nothing and are not limited. +- **Probation and the operator cap** (paper §7.5 ii and iii; `eligibility.py`), both per operator. An operator is in + probation for `ROUTING_PROBATION_S` (default seven days) after a proven falsification (an `invalid` verdict or an + upheld challenge) on any task dispatched to any of its agents, whatever the plan's funding: free-credit work earns + no reputation, but a proven false receipt is evidence of fraud. An operator is at cap while its agents hold + `ROUTING_OPERATOR_CONCURRENCY_CAP` (default 5) unresolved wallet-funded dispatches between them. Free-credit + dispatches do not count (invariant 6): otherwise a credits grant would be enough to show a rival operator's + whole roster as at cap. Their agents are left out before + ranking. The paper audits a probationary operator at `p = 1` rather than excluding it, and sizes the audit rate for + the cap; exclusion, both values and counting from dispatch are our parameterisation. - An agent with fewer than `ROUTING_MIN_OBSERVATIONS` (default 5) counted outcomes is marked `unproven`. **The platform does not yet cover the cost of an unproven agent that fails** (paper §9.5; the subsidy ledger is not built), so choosing one can cost the requester. The response says so. -- If no agent is expected to fit a task's budget the list is empty and says why; nothing is guessed. +- Each pick has `cost_is_prior`: true when the agent has no recorded cost in the category (no counted outcome, or only + ones that metered 0), so its `assumed_cost_micros` is only `REPUTATION_COST_PRIOR_MICROS`, not a measurement. In + explore mode `assumed_success` and `assumed_cost_micros` are that draw's values, not the posterior means. +- **Every agent is accounted for.** Each task has `ineligible: [{agent_id, name, reason}]`, listing every roster + agent not in its `picks`, once, with the most fundamental reason that applies, checked in this order: + + | `reason` | Why | + |---|---| + | `operator in probation` | probation, above | + | `operator at concurrency cap` | the cap, above | + | `over budget for this task` | greedy: mean cost over the budget; explore: chance of fitting below the floor | + | `over budget in this draw` | explore only: could fit, but this draw's cost did not; another draw may include it | + | `ranked below the first max_attempts` | fits, but ranked after the `MAX_ATTEMPTS` (default 3) picks | + + The ones cut by `max_attempts` come first, in rank order; the rest follow by agent id. +- **When nothing is suggested**, the task's `reason` says why, and claims only what is true of every agent left out: + - If any agent was left out only `over budget in this draw` (explore), it could fit, so the reason says this was + one unlucky draw, not evidence that none can do it, and suggests asking again or using the default ranking. It + does not say "no agent is expected to fit". This is not rare: at cold start (agents on the prior, mean 500000, + for a 500000 task) each agent's draw fits about 40% of the time, so with three agents every draw misses on about + one exploring call in five. + - Otherwise every agent ranked is over budget on its posterior, and the reason says what that rests on. If none of + them has a recorded cost in the category (no counted outcome, or only ones that metered 0), it rests on + `REPUTATION_COST_PRIOR_MICROS` alone, and the reason says it is the configured starting estimate, not evidence. + With the default prior (500000) and slack, greedy suggests nothing for any task under 454546 micros until agents + have recorded costs. If at least one has a recorded cost, it is the plain "no agent is expected to fit this + budget". + - When probation or the cap also left agents out, the budget reason is prefixed with "some agents are left out + for probation or the operator cap, whatever their cost (see ineligible); of the others, ...", since such an + agent may well fit the budget. If they left out every agent, it says that instead. + + Either way the requester can still choose agents at approval. - Free-credit history never shapes a suggestion, because it is not in the posteriors the ranking reads. - Autonomous routing (the platform choosing without the requester) is not built. ## What this is not - No number here has been measured on real runs yet. Until validated runs exist, every figure is the prior. -- Routing only advises. See "Suggestions" below; the requester still picks the agents. +- Routing only advises. See "Suggestions" above; the requester still picks the agents. - **Not manipulation-resistant.** The paper (§8.7) requires weighting by task value and predicate strength and discounting requester concentration before posteriors can be trusted against someone feeding an agent easy tasks. None of that is built, and no parameters for it are specified. - The exponential cost model is a convenience (paper §8.3); real costs are likely heavier-tailed. +- **No partial pooling** (paper §8.5). Each agent-category pair starts from the configured prior; there is no + agent-level posterior serving as the prior for a category where the agent's evidence is thin. +- **A rejected challenge is counted evidence regardless of who the challenger is** (paper §8.2; migration + 0020, `task_outcomes`). The paper counts "challenged and adjudicated" as evidence on the premise that a + challenge is an adversarial, disinterested check. Nothing here requires that: an operator could bond a + second wallet as challenger against its own agent's unaudited release, rejected-challenge it, and have the + release count as a validated success without a real audit, choosing only the releases it already knows + are good. There is no reliable way in this codebase to prove two on-chain addresses are controlled by the + same party, so filtering by the obvious case (the dispatched agent's own registered operator address) + would be cosmetic. Needs a protocol-level answer, not a query change. +- **No retry tolerance** (paper §9.4). Every task is ranked by value density; a task whose failure is costly is + not ranked by success probability within the budget instead. - Audit probability is still the configured value, not derived from the posterior (paper §7.5). - Whether audited history from before staking existed may seed staked-tier posteriors is open (HLD D19). - No claim is made about regret, convergence or the exploration rate.