Build an agent

import { pathToFileURL } from "node:url";

type Json = Record<string, unknown>;
type Fetch = typeof globalThis.fetch;
type Sleep = (seconds: number) => Promise<void>;
const object = (value: unknown): Json =>
  value !== null && typeof value === "object" && !Array.isArray(value) ? value as Json : {};

/** One obtain authority: serve.downloadReady plus a READY verdict field. */
export function downloadPlanObtainReady(plan: Json): boolean {
  const serve = object(plan.serve);
  if (serve.downloadReady !== true) return false;
  const explanation = object(plan.serveExplanation);
  const verdict = plan.serveVerdict ?? explanation.serveVerdict ?? explanation.downloadStatus;
  return verdict === "READY";
}
const httpsUrl = (value: unknown) => {
  try { return typeof value === "string" && new URL(value).protocol === "https:"; }
  catch { return false; }
};

export class ApiError extends Error {
  status: number;
  code: string;
  requestId: string | null;
  retryAfter: string | null;
  constructor(status: number, body: unknown, retryAfter: string | null) {
    const data = object(body);
    const error = object(data.error);
    super(String(data.message ?? error.message ?? `HTTP ${status}`));
    this.status = status;
    this.code = String(data.code ?? error.code ?? (typeof data.error === "string" ? data.error : "http_error"));
    this.requestId = typeof (data.request_id ?? data.requestId) === "string" ? String(data.request_id ?? data.requestId) : null;
    this.retryAfter = retryAfter;
  }
}

export function retrySeconds(value: string | null, now = Date.now()): number | null {
  if (value === null) return 1;
  const seconds = /^\d+(?:\.\d+)?$/.test(value) ? Number(value) : (Date.parse(value) - now) / 1000;
  return Number.isFinite(seconds) && seconds >= 0 && seconds <= 30 ? seconds : null;
}

export function createReader(baseUrl = "https://huggingbay.xyz", fetchImpl: Fetch = fetch,
  sleep: Sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000))) {
  return async (path: string): Promise<Json> => {
    for (let attempt = 0; attempt < 3; attempt++) {
      const response = await fetchImpl(new URL(path, baseUrl), {
        method: "GET", headers: { Accept: "application/json" }, signal: AbortSignal.timeout(20_000),
      });
      const text = await response.text();
      let body: unknown;
      try { body = JSON.parse(text); } catch { body = null; }
      if (!response.ok) {
        const retryAfter = response.headers.get("Retry-After");
        const delay = retrySeconds(retryAfter);
        if ([429, 503].includes(response.status) && attempt < 2 && delay !== null) {
          await sleep(delay);
          continue;
        }
        throw new ApiError(response.status, body, retryAfter);
      }
      if (body === null || typeof body !== "object" || Array.isArray(body)) {
        throw new Error("Expected a JSON object from the API.");
      }
      return object(body);
    }
    throw new Error("Retry budget exhausted.");
  };
}

export async function buildAgent(query = "Qwen2.5 0.5B GGUF", read = createReader()) {
  const search = await read(`/api/v1/search?${new URLSearchParams({ q: query, limit: "3", sort: "best" })}`);
  if (!Array.isArray(search.rows)) throw new Error("Search response is missing rows.");
  if (search.rows.length === 0) return { status: "abstained", reason: "no_matches", files: [] };
  for (const row of search.rows.slice(0, 3)) {
    const candidate = object(row);
    const id = candidate.artifactId ?? candidate.id;
    if (typeof id !== "string" || !id) continue;
    let plan: Json;
    try { plan = await read(`/api/v1/artifacts/${encodeURIComponent(id)}/download-plan`); }
    catch (error) { if (error instanceof ApiError && error.status === 404) continue; throw error; }
    if (!downloadPlanObtainReady(plan)) continue;
    const files = (Array.isArray(plan.files) ? plan.files : []).filter((item) => {
      const file = object(item);
      return typeof file.path === "string" && file.path.length > 0
        && typeof file.sha256 === "string" && /^[a-f0-9]{64}$/i.test(file.sha256)
        && Number.isSafeInteger(file.sizeBytes) && Number(file.sizeBytes) >= 0
        && httpsUrl(file.downloadUrl);
    }).map((item) => {
      const file = object(item);
      return { path: file.path, sha256: file.sha256, sizeBytes: file.sizeBytes, downloadUrl: file.downloadUrl };
    });
    if (files.length) return {
      status: "files_available", artifactId: id, license: object(plan.artifact).license ?? null, files,
      note: "These are hosted files. This does not establish a complete runnable model or permission to use it.",
    };
  }
  return { status: "abstained", reason: "no_hosted_files_in_first_three_results", files: [] };
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
  try {
    const read = createReader(process.env.HUGGING_BAY_BASE_URL || "https://huggingbay.xyz");
    console.log(JSON.stringify(await buildAgent(process.argv[2], read), null, 2));
  } catch (error) {
    const details = error instanceof ApiError ? {
      httpStatus: error.status, code: error.code, requestId: error.requestId, retryAfter: error.retryAfter,
    } : {};
    console.error(JSON.stringify({ status: "error", message: error instanceof Error ? error.message : String(error), ...details }));
    process.exitCode = 1;
  }
}

"""A bounded, read-only catalog agent. Python 3.10+, standard library only."""
import json
import os
import re
import sys
import time
from datetime import timezone
from email.utils import parsedate_to_datetime
from urllib.error import HTTPError
from urllib.parse import quote, urlencode, urlsplit
from urllib.request import Request, urlopen


def https_url(value):
    try:
        url = urlsplit(value) if isinstance(value, str) else None
        return bool(url and url.scheme == "https" and url.hostname)
    except ValueError:
        return False


class ApiError(Exception):
    def __init__(self, status, body, retry_after):
        data = body if isinstance(body, dict) else {}
        error = data.get("error") if isinstance(data.get("error"), dict) else {}
        super().__init__(str(data.get("message", error.get("message", f"HTTP {status}"))))
        self.status = status
        self.code = data.get("code", error.get("code", data.get("error", "http_error")))
        self.request_id = data.get("request_id", data.get("requestId"))
        self.retry_after = retry_after


def retry_seconds(value, now=None):
    if value is None:
        return 1
    try:
        if re.fullmatch(r"\d+(?:\.\d+)?", value):
            seconds = float(value)
        else:
            date = parsedate_to_datetime(value)
            if date.tzinfo is None:
                date = date.replace(tzinfo=timezone.utc)
            seconds = date.timestamp() - (time.time() if now is None else now)
        return seconds if 0 <= seconds <= 30 else None
    except (ValueError, TypeError, OverflowError):
        return None


def create_reader(base_url="https://huggingbay.xyz", opener=urlopen, sleep=time.sleep):
    def read(path):
        for attempt in range(3):
            request = Request(base_url.rstrip("/") + path, headers={"Accept": "application/json"}, method="GET")
            try:
                with opener(request, timeout=20) as response:
                    body = json.load(response)
                if not isinstance(body, dict):
                    raise ValueError("Expected a JSON object from the API.")
                return body
            except HTTPError as error:
                retry_after = error.headers.get("Retry-After")
                with error:
                    try:
                        body = json.load(error)
                    except (ValueError, UnicodeDecodeError):
                        body = None
                delay = retry_seconds(retry_after)
                if error.code in (429, 503) and attempt < 2 and delay is not None:
                    sleep(delay)
                    continue
                raise ApiError(error.code, body, retry_after) from error
        raise RuntimeError("Retry budget exhausted.")
    return read


def download_plan_obtain_ready(plan):
    if not isinstance(plan, dict):
        return False
    serve = plan.get("serve") or {}
    if not isinstance(serve, dict) or serve.get("downloadReady") is not True:
        return False
    explanation = plan.get("serveExplanation") or {}
    verdict = plan.get("serveVerdict")
    if verdict is None and isinstance(explanation, dict):
        verdict = explanation.get("serveVerdict")
        if verdict is None:
            verdict = explanation.get("downloadStatus")
    return verdict == "READY"


def build_agent(query="Qwen2.5 0.5B GGUF", read=None):
    read = read or create_reader()
    search = read("/api/v1/search?" + urlencode({"q": query, "limit": 3, "sort": "best"}))
    rows = search.get("rows")
    if not isinstance(rows, list):
        raise ValueError("Search response is missing rows.")
    if not rows:
        return {"status": "abstained", "reason": "no_matches", "files": []}
    for row in rows[:3]:
        artifact_id = row.get("artifactId", row.get("id")) if isinstance(row, dict) else None
        if not isinstance(artifact_id, str) or not artifact_id:
            continue
        try:
            plan = read(f"/api/v1/artifacts/{quote(artifact_id, safe='')}/download-plan")
        except ApiError as error:
            if error.status == 404:
                continue
            raise
        if not download_plan_obtain_ready(plan):
            continue
        files = []
        for item in plan.get("files", []) if isinstance(plan.get("files"), list) else []:
            if (isinstance(item, dict) and isinstance(item.get("path"), str) and item["path"]
                    and isinstance(item.get("sha256"), str) and re.fullmatch(r"[a-fA-F0-9]{64}", item["sha256"])
                    and type(item.get("sizeBytes")) is int and 0 <= item["sizeBytes"] <= 2**53 - 1
                    and https_url(item.get("downloadUrl"))):
                files.append({key: item[key] for key in ("path", "sha256", "sizeBytes", "downloadUrl")})
        if files:
            artifact = plan.get("artifact") or {}
            return {"status": "files_available", "artifactId": artifact_id,
                    "license": artifact.get("license") if isinstance(artifact, dict) else None, "files": files,
                    "note": "These are hosted files. This does not establish a complete runnable model or permission to use it."}
    return {"status": "abstained", "reason": "no_hosted_files_in_first_three_results", "files": []}


if __name__ == "__main__":
    try:
        reader = create_reader(os.environ.get("HUGGING_BAY_BASE_URL", "https://huggingbay.xyz"))
        query = sys.argv[1] if len(sys.argv) > 1 else "Qwen2.5 0.5B GGUF"
        print(json.dumps(build_agent(query, reader), indent=2))
    except Exception as error:
        details = ({"httpStatus": error.status, "code": error.code,
                    "requestId": error.request_id, "retryAfter": error.retry_after}
                   if isinstance(error, ApiError) else {})
        print(json.dumps({"status": "error", "message": str(error), **details}), file=sys.stderr)
        sys.exit(1)