Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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');"
Comment thread
coderabbitai[bot] marked this conversation as resolved.

# --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
Expand Down
17 changes: 15 additions & 2 deletions Dockerfile.antseed
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
2 changes: 1 addition & 1 deletion antseed/broadcast.js
Original file line number Diff line number Diff line change
Expand Up @@ -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:
//
Expand Down
14 changes: 14 additions & 0 deletions antseed/cli-args.js
Original file line number Diff line number Diff line change
@@ -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 };
19 changes: 19 additions & 0 deletions antseed/cli-args.test.js
Original file line number Diff line number Diff line change
@@ -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/);
});
3 changes: 2 additions & 1 deletion antseed/control.js
Original file line number Diff line number Diff line change
Expand Up @@ -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');

Expand Down Expand Up @@ -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
Expand Down
15 changes: 11 additions & 4 deletions antseed/write-market.js
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Comment thread
coderabbitai[bot] marked this conversation as resolved.
fetched_at=EXCLUDED.fetched_at`; // first_seen preserved across conflicts

(async () => {
Expand Down
115 changes: 112 additions & 3 deletions host_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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]]:
Expand Down
2 changes: 2 additions & 0 deletions llm_router_host.py
Original file line number Diff line number Diff line change
Expand Up @@ -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`
Expand Down
3 changes: 2 additions & 1 deletion providers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
Loading
Loading