fix: recover AntSeed routing from unhealthy sellers (#103)

* fix: recover AntSeed routing from unhealthy sellers

* fix: preserve AntSeed reachability evidence

Author: jmlago

.github/workflows/ci.yml

@@ -97,6 +97,17 @@ jobs:
       - name: AntSeed CLI resolves its package dependencies inside the image
         run: docker run --rm --entrypoint antseed unhardcoded-antseed:ci --help >/dev/null
 
+      # CLI >=0.1.153 moved the default buyer router into a separately installed
+      # plugin. Loading it here (with runtime updates disabled by the image) proves
+      # a fresh pod does not need npm access before it can join the P2P network.
+      - name: Vendored AntSeed router plugin loads without runtime npm
+        run: >-
+          docker run --rm --network none --entrypoint node unhardcoded-antseed:ci
+          --input-type=module --eval
+          "const {loadRouterPlugin}=await import('/usr/local/lib/node_modules/@antseed/cli/dist/plugins/loader.js');
+          const plugin=await loadRouterPlugin('local');
+          if (plugin.type !== 'router') throw new Error('local router plugin did not load');"
+
       # --network host so the container reaches the runner's postgres service on
       # localhost; a bridged container cannot. No `|| true` on the run: a
       # container that fails to start must fail the job, not fall through to a

Dockerfile.antseed

@@ -11,11 +11,19 @@ FROM node:22-slim AS antseed-dependencies
 RUN apt-get update \
     && apt-get install -y --no-install-recommends python3 make g++ \
     && for install_attempt in 1 2 3; do \
-         if npm install -g @antseed/cli@0.1.128 pg@8.16.3; then break; fi; \
+         if npm install -g @antseed/cli@0.1.153 pg@8.16.3; then break; fi; \
          rm -rf /usr/local/lib/node_modules/@antseed \
            /usr/local/lib/node_modules/pg /usr/local/bin/antseed; \
          if [ "${install_attempt}" = 3 ]; then exit 1; fi; \
        done \
+    && mkdir -p /opt/antseed-plugins \
+    && cd /opt/antseed-plugins \
+    && npm init -y \
+    && for install_attempt in 1 2 3; do \
+         if npm install --ignore-scripts --save-exact @antseed/router-local@0.1.45; then break; fi; \
+         rm -rf node_modules package-lock.json; \
+         if [ "${install_attempt}" = 3 ]; then exit 1; fi; \
+       done \
     && rm -rf /var/lib/apt/lists/*
 
 FROM node:22-slim
@@ -28,10 +36,15 @@ RUN apt-get update \
 # `pg` is the market writer's Postgres client (write-market.js upserts peer_offers
 # into the shared host store). NODE_PATH lets the scripts require these globals.
 COPY --from=antseed-dependencies /usr/local/lib/node_modules /usr/local/lib/node_modules
+# Since CLI 0.1.153 the default buyer router is a separately installed plugin.
+# Vendor the compatible release in the exact directory the CLI loads from so a
+# fresh container never reaches npm before it can join the P2P network.
+COPY --from=antseed-dependencies /opt/antseed-plugins /root/.antseed/plugins
 # Docker COPY dereferences the npm-created bin symlink. Recreate it explicitly
 # so Node resolves package imports from @antseed/cli instead of /usr/local/bin.
 RUN ln -s ../lib/node_modules/@antseed/cli/dist/cli/index.js /usr/local/bin/antseed
-ENV NODE_PATH=/usr/local/lib/node_modules
+ENV NODE_PATH=/usr/local/lib/node_modules \
+    ANTSEED_SKIP_PLUGIN_UPDATE_CHECK=1
 
 # Every non-test file under antseed/ — an explicit list silently ships a module
 # whose `require('./x.js')` has no target, and the control server then dies at

antseed/broadcast.js

@@ -26,7 +26,7 @@
 // ---------------------------------------------------------------------------
 // WHAT CANNOT BE PROVED FROM CLI OUTPUT — read this before widening the rule.
 //
-// `@antseed/cli@0.1.128`'s `buyer deposit` runs SIX RPC calls inside one ora
+// `@antseed/cli@0.1.153`'s `buyer deposit --onchain` runs several RPC calls inside one ora
 // spinner, and TWO of them are broadcasts (an unconditional ERC-20 `approve`,
 // then the deposit itself), each followed by a `wait()` receipt poll:
 //

antseed/cli-args.js

@@ -0,0 +1,14 @@
+'use strict';
+
+// Keep the dashboard wallet API insulated from CLI syntax drift. Since
+// @antseed/cli 0.1.137, `buyer deposit` is the QR/watch flow; the legacy direct
+// on-chain operation moved to `buyer deposit --onchain <amount>`. The control
+// endpoint already receives funded-wallet amounts and must retain that exact
+// transaction semantics.
+function walletCommandArgs(verb, amount) {
+  if (verb === 'deposit') return ['buyer', 'deposit', '--onchain', amount];
+  if (verb === 'withdraw') return ['buyer', 'withdraw', amount];
+  throw new Error(`unsupported wallet command: ${verb}`);
+}
+
+module.exports = { walletCommandArgs };

antseed/cli-args.test.js

@@ -0,0 +1,19 @@
+'use strict';
+
+const test = require('node:test');
+const assert = require('node:assert/strict');
+const { walletCommandArgs } = require('./cli-args.js');
+
+test('direct deposits use the post-0.1.137 --onchain syntax', () => {
+  assert.deepEqual(walletCommandArgs('deposit', '1.25'),
+    ['buyer', 'deposit', '--onchain', '1.25']);
+});
+
+test('withdraw syntax remains positional', () => {
+  assert.deepEqual(walletCommandArgs('withdraw', '2'),
+    ['buyer', 'withdraw', '2']);
+});
+
+test('unknown wallet verbs fail closed', () => {
+  assert.throws(() => walletCommandArgs('sweep', '1'), /unsupported wallet command/);
+});

antseed/control.js

@@ -25,6 +25,7 @@ const { createQueue } = require('./queue.js');
 // reasoning about what is and is not provable from CLI stdio is long enough to
 // deserve a file. See antseed/broadcast.js.
 const { classifyCliFailure } = require('./broadcast.js');
+const { walletCommandArgs } = require('./cli-args.js');
 
 const path = require('path');
 
@@ -197,7 +198,7 @@ const server = http.createServer(async (req, res) => {
       return refuse(res, 400, 'amount must be a positive USDC value (<=6 decimals, <=' + MAX_AMOUNT_USDC + ')');
     }
     return serialize(async () => {
-      const r = await run(['buyer', verb, amount], DEPOSIT_TIMEOUT_MS);
+      const r = await run(walletCommandArgs(verb, amount), DEPOSIT_TIMEOUT_MS);
       if (r.code !== 0) {
         const why = (r.stderr || r.stdout || 'cli failed').slice(0, 600);
         // A CLI we KILLED on the timeout may already have broadcast the

antseed/write-market.js

@@ -47,27 +47,34 @@ for (const peer of fresh.peers) {
   const maxc = posIntOrNull(peer.maxConcurrency);
   const rep = numOrNull(peer.onChainReputationScore);
   const lastSeen = numOrNull(peer.lastSeen);
+  // `lastSeen` is only a DHT advertisement sighting. `lastReachedAt` is the
+  // buyer's stronger signal that it actually connected to the peer; retain both
+  // so host admission never mistakes a repeatedly re-announced dead seller for
+  // an inference-ready one.
+  const lastReachedAt = numOrNull(peer.lastReachedAt);
   for (const pricing of Object.values(peer.providerPricing || {})) {
     for (const [service, sp] of Object.entries((pricing || {}).services || {})) {
       rows.push([
         peer.peerId, service,
         numOr0(sp.inputUsdPerMillion), numOr0(sp.outputUsdPerMillion),
         numOrNull(sp.cachedInputUsdPerMillion),
-        maxc, rep, lastSeen, now, now, now,
+        maxc, rep, lastSeen, lastReachedAt, now, now, now,
       ]);
     }
   }
 }
 
 const UPSERT = `INSERT INTO peer_offers
   (peer_id, service, price_in, price_out, price_cached_in, max_concurrency,
-   reputation, last_seen, observed_at, first_seen, fetched_at)
-  VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
+   reputation, last_seen, last_reached_at, observed_at, first_seen, fetched_at)
+  VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
   ON CONFLICT (peer_id, service) DO UPDATE SET
     price_in=EXCLUDED.price_in, price_out=EXCLUDED.price_out,
     price_cached_in=EXCLUDED.price_cached_in,
     max_concurrency=EXCLUDED.max_concurrency, reputation=EXCLUDED.reputation,
-    last_seen=EXCLUDED.last_seen, observed_at=EXCLUDED.observed_at,
+    last_seen=EXCLUDED.last_seen,
+    last_reached_at=COALESCE(EXCLUDED.last_reached_at, peer_offers.last_reached_at),
+    observed_at=EXCLUDED.observed_at,
     fetched_at=EXCLUDED.fetched_at`;  // first_seen preserved across conflicts
 
 (async () => {

host_store.py

@@ -132,13 +132,17 @@ def _retention_days() -> int:
         served_by          TEXT,
         ok                 BOOLEAN NOT NULL,
         latency_ms         DOUBLE PRECISION,
+        error_kind         TEXT,
+        http_status        INTEGER,
         tools_requested    BOOLEAN,
         tool_calls_emitted BOOLEAN
     )""",
     "CREATE INDEX IF NOT EXISTS idx_route_obs_ts ON route_observations(ts)",
     "CREATE INDEX IF NOT EXISTS idx_route_obs_route"
     " ON route_observations(provider_id, model_family, served_by, ts)",
     # #4c: learned tool capability is derived from these per-attempt signals.
+    "ALTER TABLE route_observations ADD COLUMN IF NOT EXISTS error_kind TEXT",
+    "ALTER TABLE route_observations ADD COLUMN IF NOT EXISTS http_status INTEGER",
     "ALTER TABLE route_observations ADD COLUMN IF NOT EXISTS tools_requested BOOLEAN",
     "ALTER TABLE route_observations ADD COLUMN IF NOT EXISTS tool_calls_emitted BOOLEAN",
     """CREATE TABLE IF NOT EXISTS settings_overrides (
@@ -224,11 +228,13 @@ def _retention_days() -> int:
         max_concurrency INTEGER,
         reputation      DOUBLE PRECISION,
         last_seen       BIGINT,
+        last_reached_at BIGINT,
         observed_at     BIGINT NOT NULL,
         first_seen      BIGINT,
         fetched_at      BIGINT,
         PRIMARY KEY (peer_id, service)
     )""",
+    "ALTER TABLE peer_offers ADD COLUMN IF NOT EXISTS last_reached_at BIGINT",
     "CREATE INDEX IF NOT EXISTS idx_peer_offers_observed ON peer_offers(observed_at)",
     # The antseed buyer's status (escrow + session pin + wallet), one row per
     # buyer pid. WRITTEN by the antseed sidecar (write-status.js on the poll loop
@@ -901,12 +907,14 @@ def _insert_route_observation(row: dict[str, Any]) -> None:
             conn.execute(
                 "INSERT INTO route_observations"
                 " (ts, provider_id, model_family, served_by, ok, latency_ms,"
-                " tools_requested, tool_calls_emitted)"
-                " VALUES (%s,%s,%s,%s,%s,%s,%s,%s)",
+                " error_kind, http_status, tools_requested, tool_calls_emitted)"
+                " VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
                 (int(row.get("ts") or time.time() * 1000),
                  row.get("provider_id"), row.get("model_family"), row.get("served_by"),
                  bool(row.get("ok")),
                  float(row["latency_ms"]) if row.get("latency_ms") is not None else None,
+                 row.get("error_kind"),
+                 int(row["http_status"]) if row.get("http_status") is not None else None,
                  bool(row.get("tools_requested")), bool(row.get("tool_calls_emitted"))))
     except Exception as exc:  # noqa: BLE001 — the fold must never break a request
         _log.warning("host_store route observation insert failed: %s", exc)
@@ -942,6 +950,107 @@ def route_stats(window_ms: int = 900_000) -> dict[str, dict[str, Any]]:
         return {}
 
 
+# Failures in these classes describe the request rather than the route. They
+# remain in route_stats() for backwards-compatible measured reliability, but do
+# not put a marketplace seller into a durable cooldown.
+_ROUTE_HEALTH_NEUTRAL_ERRORS = frozenset({
+    "bad_request", "content_filter", "context_overflow", "payment_required",
+})
+
+# A failure in one service can prove the whole peer unhealthy only for transport,
+# capacity and server faults. A model_unavailable/404 is deliberately absent: it
+# quarantines that peer+family route without hiding the peer's other models.
+_PEER_HEALTH_FAILURE_ERRORS = frozenset({
+    "rate_limit", "timeout", "server_error", "network_error", "auth_error",
+    "bad_response", "unknown",
+})
+
+
+def _fold_health_rows(rows: list[tuple], provider_id: str, *, peer: bool) -> dict:
+    """Fold newest-first observation rows into consecutive attributable failures.
+
+    The SQL caps each identity to a bounded recent sample. A success ends the
+    current failure streak; client/request faults are ignored. Old observations
+    have no error_kind, so they remain route evidence (the safe migration
+    direction) but are not promoted to peer-wide blame.
+    """
+    out: dict[str, dict[str, Any]] = {}
+    for family, served_by, ts, ok, error_kind, http_status in rows:
+        key = served_by if peer else f"{provider_id}|{family}|{served_by}"
+        state = out.setdefault(key, {
+            "consecutive_failures": 0,
+            "last_failure_at": None,
+            "last_success_at": None,
+            "latest_error_kind": None,
+            "latest_http_status": None,
+            "sample_count": 0,
+            "_ended": False,
+        })
+        state["sample_count"] += 1
+        if state["_ended"]:
+            continue
+        if ok:
+            state["last_success_at"] = int(ts)
+            state["_ended"] = True
+            continue
+        kind = str(error_kind) if error_kind else None
+        attributable = (
+            kind in _PEER_HEALTH_FAILURE_ERRORS if peer
+            else kind not in _ROUTE_HEALTH_NEUTRAL_ERRORS
+        )
+        if not attributable:
+            continue
+        state["consecutive_failures"] += 1
+        if state["last_failure_at"] is None:
+            state["last_failure_at"] = int(ts)
+            state["latest_error_kind"] = kind
+            state["latest_http_status"] = (
+                int(http_status) if http_status is not None else None)
+    for key in list(out):
+        state = out[key]
+        state.pop("_ended", None)
+        # A group containing only neutral failures carries no health evidence.
+        if not state["consecutive_failures"] and state["last_success_at"] is None:
+            out.pop(key)
+    return out
+
+
+def marketplace_route_health(provider_id: str, window_ms: int = 86_400_000,
+                             sample_limit: int = 64) -> dict[str, dict]:
+    """Bounded durable health for one marketplace provider.
+
+    Returns ``{"routes": {provider|family|peer: state}, "peers": {peer: state}}``.
+    Route state drives service-specific cooldowns; peer state is restricted to
+    failures that can safely be attributed across services. Both are derived
+    from the shared Postgres ledger, so replicas and restarts agree.
+    """
+    try:
+        cutoff = int(time.time() * 1000) - max(0, window_ms)
+        limit = max(1, min(int(sample_limit), 512))
+
+        def read(partition: str, order_prefix: str) -> list[tuple]:
+            with _get_pool().connection() as conn:
+                cur = conn.execute(
+                    "SELECT model_family,served_by,ts,ok,error_kind,http_status"
+                    " FROM (SELECT id,model_family,served_by,ts,ok,error_kind,http_status,"
+                    f" row_number() OVER (PARTITION BY {partition} ORDER BY ts DESC,id DESC) AS rn"
+                    " FROM route_observations WHERE provider_id=%s AND ts >= %s) recent"
+                    " WHERE rn <= %s"
+                    f" ORDER BY {order_prefix},ts DESC,id DESC",
+                    (provider_id, cutoff, limit))
+                return list(cur.fetchall())
+
+        route_rows = read("model_family,served_by", "served_by,model_family")
+        peer_rows = read("served_by", "served_by")
+        return {
+            "routes": _fold_health_rows(route_rows, provider_id, peer=False),
+            "peers": _fold_health_rows(peer_rows, provider_id, peer=True),
+        }
+    except Exception as exc:  # noqa: BLE001 — admission degrades to legacy behavior
+        _log.warning("host_store marketplace_route_health failed: %s", exc)
+        return {"routes": {}, "peers": {}}
+
+
 def provider_attempt_counts(provider_id: str, window_ms: int = 3_600_000) -> dict[str, int]:
     """{ok, failed, total} attempts for one provider over the last `window_ms`,
     across every family and peer. The wallet keeper's "is this provider fully
@@ -1752,7 +1861,7 @@ def recent_logins(limit: int = 100) -> list[dict[str, Any]]:
 # window/housekeeping columns (observed_at/first_seen/fetched_at) stay internal.
 _PEER_OFFER_FIELDS = ("peer_id", "service", "price_in", "price_out",
                       "price_cached_in", "max_concurrency", "reputation",
-                      "last_seen")
+                      "last_seen", "last_reached_at")
 
 
 def peer_offers(window_ms: int = 900_000) -> list[dict[str, Any]]:

llm_router_host.py

@@ -805,6 +805,8 @@ def _fold_route_outcome(request: dict, result: dict,
     host_store.observe_route_call_async({
         "ts": int(time.time() * 1000), "provider_id": pid, "model_family": fam,
         "served_by": peer_id or pid, "ok": ok, "latency_ms": result.get("latency_ms"),
+        "error_kind": None if ok else result.get("error_kind"),
+        "http_status": None if ok else result.get("http_status"),
         "tools_requested": bool(request.get("tools")),
         "tool_calls_emitted": bool((result.get("response") or {}).get("tool_calls"))})
     # Cache affinity + the per-session meter are DERIVED on the fly from `calls`

providers.py

@@ -160,7 +160,8 @@ def _present(provider_id):
             "offers_top_n": {
                 "type": "int", "default": env_int("ANTSEED_OFFERS_TOP_N", 3),
                 "min": 1, "max": 10, "label": "Offers per family (top-N peers)",
-                "help": "Cheapest distinct seller peers surfaced per family to rotate between on failure."},
+                "help": "Best viable distinct sellers per family after route health, "
+                        "reachability and reputation admission; price ranks inside that set."},
             "reputation_min": {
                 "type": "float", "default": env_float("ANTSEED_REPUTATION_MIN", 0),
                 "min": 0, "max": 100, "label": "Min peer on-chain reputation",