gitea-saas-core#8: eval farm intake — batch -> lane drain -> results emitter (402 funded gate, eval-task issues, poll, emit w/ retry + file drop)
This commit is contained in:
Claude
parent
59c3d035b1
commit
2f2f53f3f7
132
gitea_saas/gitea_client.py
Normal file
132
gitea_saas/gitea_client.py
Normal file
@ -0,0 +1,132 @@
|
||||
"""Gitea API client (mockable seam, env token).
|
||||
|
||||
Shared surface for the SaaS front-ends: the factory front-end (issue #4),
|
||||
the review gate (issue #5) and the eval farm (issue #8) all drive Gitea
|
||||
through this seam. Real HTTP only when a token is available; the token is
|
||||
read from ``GITEA_TOKEN`` (or passed in) and is never printed or placed on
|
||||
a command line. Tests use an in-memory fake, so this module is never
|
||||
exercised over the network in CI.
|
||||
|
||||
``owner`` defaults to the ``GITEA_OWNER`` environment variable and then to
|
||||
``pbwraith`` (the self-hosted instance's user). Repo-scoped calls build
|
||||
``/api/v1/repos/{owner}/{repo}/...``; org-repo creation uses
|
||||
``/api/v1/orgs/{owner}/repos`` and falls back to the authenticated user's
|
||||
repos when ``owner`` is a user account (orgs endpoint 404s).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from typing import Any, List, Optional
|
||||
|
||||
|
||||
class GiteaClient:
|
||||
"""Thin urllib client for the subset of the Gitea API the SaaS needs."""
|
||||
|
||||
def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None,
|
||||
owner: Optional[str] = None):
|
||||
self.base_url = (base_url or os.environ.get(
|
||||
"GITEA_BASE_URL", "https://gitea.185.214.135.234.nip.io")).rstrip("/")
|
||||
self.token = token if token is not None else os.environ.get("GITEA_TOKEN", "")
|
||||
self.owner = (owner or os.environ.get("GITEA_OWNER") or "pbwraith").strip() or "pbwraith"
|
||||
|
||||
# -- low-level ---------------------------------------------------------
|
||||
|
||||
def _api(self, method: str, api_path: str, body: Optional[dict] = None) -> Any:
|
||||
"""Perform an API call under ``/api/v1/{api_path}`` and parse JSON."""
|
||||
data = json.dumps(body).encode() if body is not None else None
|
||||
url = f"{self.base_url}/api/v1/{api_path}"
|
||||
req = urllib.request.Request(
|
||||
url, data=data, method=method,
|
||||
headers={"Authorization": "token " + self.token,
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json"})
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
raw = resp.read()
|
||||
return json.loads(raw) if raw else None
|
||||
|
||||
def _repo_api(self, method: str, repo: str, path: str = "",
|
||||
body: Optional[dict] = None) -> Any:
|
||||
return self._api(method, f"repos/{self.owner}/{repo}{path}", body)
|
||||
|
||||
# -- repos -------------------------------------------------------------
|
||||
|
||||
def repo_exists(self, name: str) -> bool:
|
||||
"""True when ``owner/<name>`` exists (404 -> False)."""
|
||||
try:
|
||||
self._repo_api("GET", name)
|
||||
return True
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 404:
|
||||
return False
|
||||
raise
|
||||
|
||||
def create_org_repo(self, name: str, private: bool = True) -> dict:
|
||||
"""Create ``name`` under the owner (org endpoint, user fallback)."""
|
||||
payload = {"name": name, "private": private,
|
||||
"description": "internal SaaS eval/task repo",
|
||||
"auto_init": True}
|
||||
try:
|
||||
created = self._api("POST", f"orgs/{self.owner}/repos", payload)
|
||||
except urllib.error.HTTPError:
|
||||
created = self._api("POST", "user/repos", payload)
|
||||
return created or {"name": name}
|
||||
|
||||
# -- issues ------------------------------------------------------------
|
||||
|
||||
def _label_ids(self, repo: str, labels: List[str]) -> List[int]:
|
||||
"""Resolve label names to ids, creating missing labels first."""
|
||||
if not labels:
|
||||
return []
|
||||
try:
|
||||
existing = self._repo_api("GET", repo, "/labels?limit=50") or []
|
||||
except urllib.error.HTTPError:
|
||||
existing = []
|
||||
by_name = {l.get("name"): l.get("id") for l in existing if isinstance(l, dict)}
|
||||
ids: List[int] = []
|
||||
for name in labels:
|
||||
label_id = by_name.get(name)
|
||||
if label_id is None:
|
||||
try:
|
||||
created = self._repo_api(
|
||||
"POST", repo, "/labels",
|
||||
{"name": name, "color": "#1d76db",
|
||||
"description": "SaaS worker label"})
|
||||
label_id = created.get("id")
|
||||
except urllib.error.HTTPError:
|
||||
label_id = None
|
||||
if label_id is not None:
|
||||
ids.append(int(label_id))
|
||||
return ids
|
||||
|
||||
def create_issue(self, repo: str, title: str, body: str,
|
||||
labels: Optional[List[str]] = None) -> dict:
|
||||
"""Create an issue; returns the created issue record."""
|
||||
payload: dict = {"title": title, "body": body or ""}
|
||||
label_ids = self._label_ids(repo, list(labels or []))
|
||||
if label_ids:
|
||||
payload["labels"] = label_ids
|
||||
return self._repo_api("POST", repo, "/issues", payload)
|
||||
|
||||
def get_issue(self, repo: str, number: int) -> dict:
|
||||
"""Fetch an issue: {title, body, state, labels: [name, ...]}."""
|
||||
issue = self._repo_api("GET", repo, f"/issues/{number}") or {}
|
||||
return {
|
||||
"title": issue.get("title"),
|
||||
"body": issue.get("body"),
|
||||
"state": issue.get("state"),
|
||||
"labels": [l.get("name") for l in issue.get("labels", [])
|
||||
if isinstance(l, dict)],
|
||||
}
|
||||
|
||||
def list_issue_comments(self, repo: str, number: int) -> List[dict]:
|
||||
"""All comments on an issue conversation: [{body, ...}, ...]."""
|
||||
return self._repo_api("GET", repo, f"/issues/{number}/comments") or []
|
||||
|
||||
def post_comment(self, repo: str, number: int, body: str) -> dict:
|
||||
"""Post a comment on an issue or PR conversation."""
|
||||
return self._repo_api("POST", repo, f"/issues/{number}/comments",
|
||||
{"body": body})
|
||||
466
gitea_saas/intake_eval.py
Normal file
466
gitea_saas/intake_eval.py
Normal file
@ -0,0 +1,466 @@
|
||||
"""Eval farm front-end (issue #8): batch intake -> lane drain -> results emitter.
|
||||
|
||||
Turns the lane machine into a paid eval/QA grind (option 2 from
|
||||
ARCHITECTURE.md). An eval batch is a set of tasks; each task becomes one
|
||||
Gitea issue (labelled ``eval-task``, carrying a ``batch: <id>`` marker
|
||||
in its body) so the lane fleet picks it up like any funded work. Results
|
||||
are collected by polling the task issues for closure, then a JSON report
|
||||
is emitted to a callback webhook (retry x3 with backoff) or dropped as a
|
||||
file into ``SAAS_EVAL_OUTPUT_DIR``.
|
||||
|
||||
Money rule (fail-closed, mirroring issue #4's factory semantics): a batch
|
||||
is only accepted when the client's available balance is at or above the
|
||||
debit floor. Unfunded/unknown clients get a 402-style structured refusal
|
||||
and NO Gitea issues are ever created for them.
|
||||
|
||||
Schema additions are owned here (``schema.sql`` belongs to issue #1):
|
||||
``eval_batches`` and ``eval_tasks`` are bootstrapped idempotently against
|
||||
the same SQLite file as the billing core, exactly like the factory does.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
# Default funded threshold: a client must hold at least 5.00 USD before an
|
||||
# eval batch is accepted (same default as the factory front-end, issue #4).
|
||||
DEFAULT_DEBIT_FLOOR_MINOR = 500
|
||||
|
||||
DEFAULT_EVAL_REPO = "eval-runs"
|
||||
EVAL_LABEL = "eval-task"
|
||||
|
||||
_STATUS_OPEN = "open"
|
||||
_STATUS_DONE = "done"
|
||||
_STATUS_EMITTED = "emitted"
|
||||
_STATUS_FAILED = "failed"
|
||||
|
||||
_TASK_PENDING = "pending"
|
||||
_TASK_DONE = "done"
|
||||
|
||||
_REPO_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$")
|
||||
|
||||
# Idempotent by construction (same convention as the billing schema).
|
||||
_EVAL_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS eval_batches (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
client_id INTEGER NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'open'
|
||||
CHECK (status IN ('open', 'done', 'emitted', 'failed')),
|
||||
callback_url TEXT,
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS eval_tasks (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
batch_id INTEGER NOT NULL REFERENCES eval_batches(id),
|
||||
repo TEXT NOT NULL,
|
||||
issue_number INTEGER NOT NULL,
|
||||
task_id TEXT NOT NULL,
|
||||
payload_json TEXT NOT NULL DEFAULT '{}',
|
||||
status TEXT NOT NULL DEFAULT 'pending'
|
||||
CHECK (status IN ('pending', 'done')),
|
||||
result_json TEXT,
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
|
||||
UNIQUE (batch_id, task_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_eval_tasks_batch ON eval_tasks (batch_id, status);
|
||||
"""
|
||||
|
||||
|
||||
class EvalIntakeError(ValueError):
|
||||
"""Base class for eval intake domain errors (malformed input only)."""
|
||||
|
||||
|
||||
def _now_iso() -> str:
|
||||
"""Current UTC time as an ISO-8601 string (matches billing core format)."""
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
class HttpEmitter:
|
||||
"""Webhook emitter: POST a JSON report; retry x3 with backoff.
|
||||
|
||||
Returns ``True`` only on a 2xx response. ``sleep_fn`` is injectable so
|
||||
tests stay hermetic (no real backoff sleeps).
|
||||
"""
|
||||
|
||||
def __init__(self, timeout: int = 20, max_attempts: int = 3,
|
||||
sleep_fn=None):
|
||||
self.timeout = timeout
|
||||
self.max_attempts = max_attempts
|
||||
self._sleep = sleep_fn or time.sleep
|
||||
|
||||
def push(self, report: dict, url: str) -> bool:
|
||||
import urllib.request
|
||||
|
||||
attempts = 0
|
||||
while attempts < self.max_attempts:
|
||||
attempts += 1
|
||||
req = urllib.request.Request(
|
||||
url, data=json.dumps(report).encode(), method="POST",
|
||||
headers={"Content-Type": "application/json",
|
||||
"Accept": "application/json"})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
|
||||
if 200 <= resp.status < 300:
|
||||
return True
|
||||
except Exception:
|
||||
pass # retry below
|
||||
if attempts < self.max_attempts:
|
||||
self._sleep(2 ** attempts) # 2s, 4s backoff
|
||||
return False
|
||||
|
||||
|
||||
class EvalIntake:
|
||||
"""Batch eval intake: submit -> lane drain -> poll -> emit.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
core:
|
||||
The billing core (issue #1 contract): ``get_client(client_id,
|
||||
include_balance=True) -> (ok, details)``. Duck-typed; never
|
||||
imported directly.
|
||||
gitea:
|
||||
A :class:`gitea_saas.gitea_client.GiteaClient` (or in-memory mock):
|
||||
``repo_exists``, ``create_org_repo``, ``create_issue``,
|
||||
``get_issue``, ``list_issue_comments``.
|
||||
config:
|
||||
Configuration object exposing ``db_path`` (a ``pathlib.Path`` to
|
||||
the billing SQLite file). Optional attributes: ``eval_repo``
|
||||
(shared eval repo name), ``output_dir`` (file-drop destination),
|
||||
``debit_floor_minor`` (funded threshold), ``per_batch_repos``.
|
||||
Env fallbacks: ``SAAS_DB_PATH``, ``SAAS_EVAL_REPO``,
|
||||
``SAAS_EVAL_OUTPUT_DIR``, ``SAAS_DEBIT_FLOOR_MINOR``.
|
||||
emitter:
|
||||
Optional duck-typed emitter used for callback delivery:
|
||||
``push(report, url) -> bool``. Defaults to :class:`HttpEmitter`.
|
||||
sleep_fn:
|
||||
Backoff sleep override for hermetic tests (default ``time.sleep``).
|
||||
"""
|
||||
|
||||
def __init__(self, core: Any, gitea: Any, config: Any,
|
||||
emitter: Optional[Any] = None, sleep_fn=None):
|
||||
db_path = getattr(config, "db_path", None)
|
||||
if db_path is None:
|
||||
raw = os.environ.get("SAAS_DB_PATH", "").strip()
|
||||
db_path = Path(raw).expanduser() if raw else None
|
||||
if db_path is None:
|
||||
raise EvalIntakeError(
|
||||
"EvalIntake requires config.db_path (or SAAS_DB_PATH): "
|
||||
"the batch/task queue persists in the billing SQLite file")
|
||||
self._core = core
|
||||
self._gitea = gitea
|
||||
self._config = config
|
||||
self._emitter = emitter if emitter is not None else HttpEmitter(
|
||||
sleep_fn=sleep_fn)
|
||||
self._sleep = sleep_fn or time.sleep
|
||||
self._conn = sqlite3.connect(str(Path(db_path)), check_same_thread=False)
|
||||
self._conn.row_factory = sqlite3.Row
|
||||
self._conn.execute("PRAGMA journal_mode=WAL")
|
||||
self._conn.execute("PRAGMA foreign_keys=ON")
|
||||
self._conn.executescript(_EVAL_DDL)
|
||||
self._lock = threading.RLock()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Config
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@property
|
||||
def debit_floor_minor(self) -> int:
|
||||
value = getattr(self._config, "debit_floor_minor", None)
|
||||
if value is None:
|
||||
raw = os.environ.get("SAAS_DEBIT_FLOOR_MINOR", "").strip()
|
||||
value = int(raw) if raw else DEFAULT_DEBIT_FLOOR_MINOR
|
||||
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
||||
raise EvalIntakeError(f"invalid debit floor: {value!r}")
|
||||
return value
|
||||
|
||||
@property
|
||||
def eval_repo(self) -> str:
|
||||
value = getattr(self._config, "eval_repo", None)
|
||||
if value is None:
|
||||
value = os.environ.get("SAAS_EVAL_REPO", "").strip() or DEFAULT_EVAL_REPO
|
||||
return str(value).strip() or DEFAULT_EVAL_REPO
|
||||
|
||||
@property
|
||||
def output_dir(self) -> Optional[Path]:
|
||||
value = getattr(self._config, "output_dir", None)
|
||||
if value is None:
|
||||
raw = os.environ.get("SAAS_EVAL_OUTPUT_DIR", "").strip()
|
||||
value = Path(raw).expanduser() if raw else None
|
||||
return Path(value) if value else None
|
||||
|
||||
@property
|
||||
def per_batch_repos(self) -> bool:
|
||||
return bool(getattr(self._config, "per_batch_repos", False))
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Billing helpers (fail-closed, factory semantics)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _client_balance(self, client_id: int) -> Optional[int]:
|
||||
"""Available balance, or None when funding cannot be confirmed."""
|
||||
try:
|
||||
ok, details = self._core.get_client(client_id, include_balance=True)
|
||||
except Exception:
|
||||
return None
|
||||
if not ok or not isinstance(details, dict):
|
||||
return None
|
||||
balance = details.get("balance")
|
||||
return balance if isinstance(balance, int) else None
|
||||
|
||||
def client_unfunded(self, client_id: int) -> bool:
|
||||
"""True when the client's balance is below the debit floor.
|
||||
|
||||
Fail-closed: an unknown client or a core error also reports True.
|
||||
"""
|
||||
balance = self._client_balance(client_id)
|
||||
if balance is None:
|
||||
return True
|
||||
return balance < self.debit_floor_minor
|
||||
|
||||
def _require_valid_inputs(self, client_id: int, tasks: List[dict]) -> None:
|
||||
if isinstance(client_id, bool) or not isinstance(client_id, int) \
|
||||
or client_id <= 0:
|
||||
raise EvalIntakeError("client_id must be a positive integer")
|
||||
if not isinstance(tasks, list) or not tasks:
|
||||
raise EvalIntakeError("tasks must be a non-empty list")
|
||||
seen: set = set()
|
||||
for t in tasks:
|
||||
if not isinstance(t, dict):
|
||||
raise EvalIntakeError("each task must be an object")
|
||||
task_id = str(t.get("task_id") or "").strip()
|
||||
if not task_id:
|
||||
raise EvalIntakeError("each task needs a non-empty task_id")
|
||||
if task_id in seen:
|
||||
raise EvalIntakeError(f"duplicate task_id {task_id!r}")
|
||||
seen.add(task_id)
|
||||
if not str(t.get("title") or "").strip():
|
||||
raise EvalIntakeError("each task needs a title")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Submitting a batch
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def submit_batch(self, client_id: int, tasks: List[dict],
|
||||
callback_url: Optional[str] = None) -> Dict[str, Any]:
|
||||
"""Accept a funded batch; one eval-task issue per task.
|
||||
|
||||
Returns ``{ok: True, batch_id, repo, issues: [{task_id,
|
||||
issue_number}]}``. Unfunded/unknown clients get a 402-style
|
||||
``{ok: False, reason: 'unfunded', status: 402}`` and NO Gitea
|
||||
issues are created.
|
||||
"""
|
||||
self._require_valid_inputs(client_id, tasks)
|
||||
if self.client_unfunded(client_id):
|
||||
return {"ok": False, "reason": "unfunded", "status": 402}
|
||||
callback_url = (callback_url or "").strip() or None
|
||||
|
||||
with self._lock:
|
||||
cur = self._conn.execute(
|
||||
"INSERT INTO eval_batches (client_id, status, callback_url, created_at)"
|
||||
" VALUES (?, ?, ?, ?)",
|
||||
(client_id, _STATUS_OPEN, callback_url, _now_iso()))
|
||||
batch_id = int(cur.lastrowid)
|
||||
|
||||
# Shared eval repo (SAAS_EVAL_REPO) or one repo per batch.
|
||||
if self.per_batch_repos:
|
||||
repo = f"eval-{client_id}-{batch_id}"
|
||||
else:
|
||||
repo = self.eval_repo
|
||||
if not _REPO_NAME_RE.match(repo):
|
||||
repo = DEFAULT_EVAL_REPO
|
||||
try:
|
||||
if not self._gitea.repo_exists(repo):
|
||||
self._gitea.create_org_repo(repo)
|
||||
issues = []
|
||||
for t in tasks:
|
||||
task_id = str(t["task_id"]).strip()
|
||||
title = str(t["title"]).strip()
|
||||
body = str(t.get("body") or "").strip()
|
||||
body = f"{body}\n\n---\nbatch: {batch_id}".strip()
|
||||
issue = self._gitea.create_issue(
|
||||
repo, title, body, labels=[EVAL_LABEL])
|
||||
issue_number = int(issue["number"])
|
||||
self._conn.execute(
|
||||
"INSERT INTO eval_tasks (batch_id, repo, issue_number,"
|
||||
" task_id, payload_json, status, created_at)"
|
||||
" VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||||
(batch_id, repo, issue_number, task_id,
|
||||
json.dumps({"title": title, "body": body}),
|
||||
_TASK_PENDING, _now_iso()))
|
||||
issues.append({"task_id": task_id,
|
||||
"issue_number": issue_number})
|
||||
self._conn.commit()
|
||||
except Exception as exc: # fail-closed: no partial batches
|
||||
self._conn.execute(
|
||||
"DELETE FROM eval_tasks WHERE batch_id = ?", (batch_id,))
|
||||
self._conn.execute(
|
||||
"DELETE FROM eval_batches WHERE id = ?", (batch_id,))
|
||||
self._conn.commit()
|
||||
return {"ok": False, "reason": "intake_failed",
|
||||
"error": str(exc), "status": 500}
|
||||
|
||||
return {"ok": True, "batch_id": batch_id, "repo": repo,
|
||||
"issues": issues}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Polling completed tasks
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _batch_row(self, batch_id: int) -> Optional[sqlite3.Row]:
|
||||
return self._conn.execute(
|
||||
"SELECT * FROM eval_batches WHERE id = ?", (batch_id,)).fetchone()
|
||||
|
||||
def _task_rows(self, batch_id: int) -> List[sqlite3.Row]:
|
||||
return self._conn.execute(
|
||||
"SELECT * FROM eval_tasks WHERE batch_id = ? ORDER BY id",
|
||||
(batch_id,)).fetchall()
|
||||
|
||||
def poll_completed(self, batch_id: int) -> Dict[str, Any]:
|
||||
"""Collect results from closed task issues; skip open ones.
|
||||
|
||||
A task is done when its issue is closed (the lane merged); the
|
||||
result record captures the final conversation comment (the lane's
|
||||
report) plus the issue's labels.
|
||||
"""
|
||||
batch = self._batch_row(batch_id)
|
||||
if batch is None:
|
||||
return {"ok": False, "error": "unknown_batch"}
|
||||
done = 0
|
||||
pending = 0
|
||||
for row in self._task_rows(batch_id):
|
||||
if row["status"] == _TASK_DONE:
|
||||
done += 1
|
||||
continue
|
||||
issue = self._gitea.get_issue(row["repo"], row["issue_number"])
|
||||
if not issue or issue.get("state") != "closed":
|
||||
pending += 1
|
||||
continue
|
||||
comments = self._gitea.list_issue_comments(
|
||||
row["repo"], row["issue_number"]) or []
|
||||
summary = ""
|
||||
if comments:
|
||||
last = comments[-1]
|
||||
summary = last.get("body") if isinstance(last, dict) else str(last)
|
||||
result = {
|
||||
"status": "done",
|
||||
"repo": row["repo"],
|
||||
"issue_number": row["issue_number"],
|
||||
"labels": list(issue.get("labels") or []),
|
||||
"summary": (summary or "").strip(),
|
||||
}
|
||||
with self._lock:
|
||||
self._conn.execute(
|
||||
"UPDATE eval_tasks SET status = ?, result_json = ?"
|
||||
" WHERE id = ?",
|
||||
(_TASK_DONE, json.dumps(result), row["id"]))
|
||||
remaining = self._conn.execute(
|
||||
"SELECT COUNT(*) FROM eval_tasks WHERE batch_id = ?"
|
||||
" AND status = 'pending'", (batch_id,)).fetchone()[0]
|
||||
if remaining == 0 and batch["status"] == _STATUS_OPEN:
|
||||
self._conn.execute(
|
||||
"UPDATE eval_batches SET status = ? WHERE id = ?",
|
||||
(_STATUS_DONE, batch_id))
|
||||
self._conn.commit()
|
||||
done += 1
|
||||
return {"ok": True, "batch_id": batch_id, "done": done,
|
||||
"pending": pending}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Emitting results
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _report(self, batch: sqlite3.Row) -> dict:
|
||||
results = []
|
||||
for row in self._task_rows(batch["id"]):
|
||||
try:
|
||||
payload = json.loads(row["payload_json"] or "{}")
|
||||
except ValueError:
|
||||
payload = {}
|
||||
try:
|
||||
result = json.loads(row["result_json"] or "{}")
|
||||
except ValueError:
|
||||
result = {}
|
||||
results.append({
|
||||
"task_id": row["task_id"],
|
||||
"status": "done" if row["status"] == _TASK_DONE else row["status"],
|
||||
"summary": result.get("summary") or "",
|
||||
"title": payload.get("title") or "",
|
||||
})
|
||||
return {
|
||||
"batch_id": batch["id"],
|
||||
"client_id": batch["client_id"],
|
||||
"status": batch["status"],
|
||||
"results": results,
|
||||
}
|
||||
|
||||
def emit(self, batch_id: int) -> Dict[str, Any]:
|
||||
"""Emit the full batch report once every task is done.
|
||||
|
||||
Callback mode: POST to ``callback_url`` (retry x3, backoff) via the
|
||||
emitter seam. File mode: write ``SAAS_EVAL_OUTPUT_DIR/<batch_id>.json``.
|
||||
A batch already emitted re-emits idempotently (no second POST/write).
|
||||
"""
|
||||
batch = self._batch_row(batch_id)
|
||||
if batch is None:
|
||||
return {"ok": False, "error": "unknown_batch"}
|
||||
pending = self._conn.execute(
|
||||
"SELECT COUNT(*) FROM eval_tasks WHERE batch_id = ?"
|
||||
" AND status = 'pending'", (batch_id,)).fetchone()[0]
|
||||
if pending > 0:
|
||||
return {"ok": False, "reason": "tasks_pending",
|
||||
"pending": int(pending)}
|
||||
if batch["status"] == _STATUS_EMITTED:
|
||||
return {"ok": True, "idempotent": True, "batch_id": batch_id,
|
||||
"report": self._report(batch)}
|
||||
report = self._report(batch)
|
||||
|
||||
if batch["callback_url"]:
|
||||
delivered = False
|
||||
attempts = 0
|
||||
while attempts < 3 and not delivered:
|
||||
attempts += 1
|
||||
delivered = bool(self._emitter.push(report, batch["callback_url"]))
|
||||
if not delivered and attempts < 3:
|
||||
self._sleep(2 ** attempts)
|
||||
if not delivered:
|
||||
return {"ok": False, "reason": "emit_failed",
|
||||
"attempts": attempts}
|
||||
else:
|
||||
out_dir = self.output_dir
|
||||
if out_dir is None:
|
||||
return {"ok": False, "reason": "no_callback_no_output_dir"}
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
(out_dir / f"{batch_id}.json").write_text(
|
||||
json.dumps(report, indent=2), encoding="utf-8")
|
||||
|
||||
with self._lock:
|
||||
self._conn.execute(
|
||||
"UPDATE eval_batches SET status = ? WHERE id = ?",
|
||||
(_STATUS_EMITTED, batch_id))
|
||||
self._conn.commit()
|
||||
return {"ok": True, "batch_id": batch_id, "emitted": True,
|
||||
"report": report}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Reads (for server/UI wiring — issue #2's file may expose these)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_batch(self, batch_id: int) -> Dict[str, Any]:
|
||||
batch = self._batch_row(batch_id)
|
||||
if batch is None:
|
||||
return {"ok": False, "error": "unknown_batch"}
|
||||
tasks = [dict(r) for r in self._task_rows(batch_id)]
|
||||
for t in tasks:
|
||||
t["payload_json"] = json.loads(t["payload_json"] or "{}")
|
||||
t["result_json"] = json.loads(t["result_json"] or "null")
|
||||
return {"ok": True, "batch": dict(batch), "tasks": tasks}
|
||||
364
tests/test_intake_eval.py
Normal file
364
tests/test_intake_eval.py
Normal file
@ -0,0 +1,364 @@
|
||||
"""Hermetic tests for the eval farm intake (issue #8 acceptance).
|
||||
|
||||
MockGitea (in-memory repos/issues/comments/labels), a scripted emitter
|
||||
(retry/backoff verified via call counts), and a real BillingCore on a tmp
|
||||
DB (funded vs unfunded gating is real). No network, no sleeps (a noop
|
||||
``sleep_fn`` is injected everywhere).
|
||||
|
||||
Run: python3 -m pytest tests/test_intake_eval.py -q
|
||||
"""
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import pytest
|
||||
|
||||
from gitea_saas.billing_core import BillingCore
|
||||
from gitea_saas.intake_eval import EvalIntake, EvalIntakeError, EVAL_LABEL
|
||||
|
||||
|
||||
class Config:
|
||||
"""Tiny config object for the intake (db_path + optional overrides)."""
|
||||
|
||||
def __init__(self, db_path, **kwargs):
|
||||
self.db_path = Path(db_path)
|
||||
for k, v in kwargs.items():
|
||||
setattr(self, k, v)
|
||||
|
||||
|
||||
class MockGitea:
|
||||
"""In-memory Gitea seam recording every mutation."""
|
||||
|
||||
def __init__(self):
|
||||
self.repos = {} # repo -> {"issues": {n: dict}, "comments": {n: [str]}}
|
||||
self.next_number = 0 # global issue numbering
|
||||
self.created_repos = [] # repo names created
|
||||
self.created_issues = [] # (repo, title, body, labels)
|
||||
|
||||
def _repo(self, name):
|
||||
if name not in self.repos:
|
||||
self.repos[name] = {"issues": {}, "comments": {}}
|
||||
return self.repos[name]
|
||||
|
||||
def repo_exists(self, name):
|
||||
return name in self.repos
|
||||
|
||||
def create_org_repo(self, name):
|
||||
self._repo(name)
|
||||
self.created_repos.append(name)
|
||||
|
||||
def create_issue(self, repo, title, body, labels=None):
|
||||
r = self._repo(repo)
|
||||
self.next_number += 1
|
||||
n = self.next_number
|
||||
r["issues"][n] = {"title": title, "body": body,
|
||||
"state": "open", "labels": list(labels or [])}
|
||||
r["comments"][n] = []
|
||||
self.created_issues.append((repo, title, body, list(labels or [])))
|
||||
return {"number": n, "title": title, "body": body, "state": "open"}
|
||||
|
||||
def get_issue(self, repo, number):
|
||||
r = self._repo(repo)
|
||||
issue = r["issues"].get(number)
|
||||
if issue is None:
|
||||
return None
|
||||
return dict(issue)
|
||||
|
||||
def list_issue_comments(self, repo, number):
|
||||
return [{"body": b} for b in self._repo(repo)["comments"].get(number, [])]
|
||||
|
||||
# -- test helpers: simulate the lane draining a task -------------------
|
||||
def add_comment(self, repo, number, body):
|
||||
self._repo(repo)["comments"].get(number, []).append(body)
|
||||
|
||||
def close_issue(self, repo, number):
|
||||
self._repo(repo)["issues"][number]["state"] = "closed"
|
||||
|
||||
|
||||
class MockEmitter:
|
||||
"""Scripted emitter: fails the first ``fail_first`` pushes."""
|
||||
|
||||
def __init__(self, fail_first=0):
|
||||
self.fail_first = fail_first
|
||||
self.push_count = 0
|
||||
self.push_payloads = [] # (report, url)
|
||||
|
||||
def push(self, report, url):
|
||||
self.push_count += 1
|
||||
if self.push_count <= self.fail_first:
|
||||
return False
|
||||
self.push_payloads.append((report, url))
|
||||
return True
|
||||
|
||||
|
||||
def noop_sleep(_seconds):
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def funded(tmp_path):
|
||||
core = BillingCore(tmp_path / "billing.db")
|
||||
ok, c = core.create_client("eval-client")
|
||||
assert ok
|
||||
ok, _ = core.credit(c["client_id"], 10_000, "fund-1")
|
||||
assert ok
|
||||
return core, c["client_id"]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def unfunded(tmp_path):
|
||||
core = BillingCore(tmp_path / "billing.db")
|
||||
ok, c = core.create_client("poor-client")
|
||||
assert ok
|
||||
return core, c["client_id"]
|
||||
|
||||
|
||||
def make_intake(core, tmp_path, gitea=None, emitter=None, **config_kwargs):
|
||||
gitea = gitea or MockGitea()
|
||||
cfg = Config(tmp_path / "saas.db", **config_kwargs)
|
||||
return EvalIntake(core, gitea, cfg, emitter=emitter, sleep_fn=noop_sleep)
|
||||
|
||||
|
||||
TASKS = [
|
||||
{"task_id": "t1", "title": "Eval task one",
|
||||
"body": "## Spec\nrun X\n## Acceptance\n- [ ] a\n- [ ] b"},
|
||||
{"task_id": "t2", "title": "Eval task two", "body": "## Spec\nrun Y"},
|
||||
]
|
||||
|
||||
|
||||
# -------------------------------------------------------------------------- #
|
||||
# Submission #
|
||||
# -------------------------------------------------------------------------- #
|
||||
|
||||
class TestSubmitBatch:
|
||||
def test_funded_batch_creates_one_issue_per_task(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
assert r["ok"] is True
|
||||
assert r["batch_id"] > 0
|
||||
assert r["repo"] == "eval-runs"
|
||||
assert [i["task_id"] for i in r["issues"]] == ["t1", "t2"]
|
||||
# shared repo exists; two issues; each carries the eval-task label
|
||||
assert "eval-runs" in gitea.created_repos
|
||||
assert len(gitea.created_issues) == 2
|
||||
for repo, title, body, labels in gitea.created_issues:
|
||||
assert repo == "eval-runs"
|
||||
assert labels == [EVAL_LABEL]
|
||||
assert f"batch: {r['batch_id']}" in body
|
||||
# both tasks persisted as pending
|
||||
batch = intake.get_batch(r["batch_id"])
|
||||
assert batch["ok"] and len(batch["tasks"]) == 2
|
||||
assert all(t["status"] == "pending" for t in batch["tasks"])
|
||||
assert batch["batch"]["status"] == "open"
|
||||
|
||||
def test_unfunded_batch_402_and_no_gitea_mutations(self, unfunded, tmp_path):
|
||||
core, client_id = unfunded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
assert r["ok"] is False
|
||||
assert r["reason"] == "unfunded"
|
||||
assert r["status"] == 402
|
||||
assert gitea.created_issues == [] # no issues created
|
||||
assert gitea.created_repos == [] # no repo created either
|
||||
assert gitea.next_number == 0
|
||||
|
||||
def test_unknown_client_is_fail_closed_unfunded(self, tmp_path):
|
||||
core = BillingCore(tmp_path / "billing.db")
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea)
|
||||
r = intake.submit_batch(9999, TASKS)
|
||||
assert r["ok"] is False and r["reason"] == "unfunded"
|
||||
assert r["status"] == 402
|
||||
assert gitea.created_issues == []
|
||||
|
||||
def test_duplicate_task_id_rejected(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
intake = make_intake(core, tmp_path)
|
||||
with pytest.raises(EvalIntakeError):
|
||||
intake.submit_batch(client_id, [TASKS[0], dict(TASKS[0])])
|
||||
|
||||
def test_intake_failure_creates_no_partial_batch(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
|
||||
class FlakyGitea(MockGitea):
|
||||
def create_issue(self, *a, **kw):
|
||||
if self.next_number >= 1: # second issue fails
|
||||
raise RuntimeError("gitea hiccup")
|
||||
return super().create_issue(*a, **kw)
|
||||
|
||||
intake = make_intake(core, tmp_path, FlakyGitea())
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
assert r["ok"] is False and r["reason"] == "intake_failed"
|
||||
# no batch row left behind
|
||||
assert intake.get_batch(1)["ok"] is False
|
||||
|
||||
|
||||
# -------------------------------------------------------------------------- #
|
||||
# Polling #
|
||||
# -------------------------------------------------------------------------- #
|
||||
|
||||
class TestPollCompleted:
|
||||
def test_poll_stores_closed_results_and_skips_open(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
n1 = r["issues"][0]["issue_number"]
|
||||
gitea.add_comment("eval-runs", n1, "lane report: all checks green")
|
||||
gitea.close_issue("eval-runs", n1) # task 1 drained; task 2 still open
|
||||
out = intake.poll_completed(r["batch_id"])
|
||||
assert out["done"] == 1 and out["pending"] == 1
|
||||
batch = intake.get_batch(r["batch_id"])
|
||||
done_task = [t for t in batch["tasks"] if t["task_id"] == "t1"][0]
|
||||
pending_task = [t for t in batch["tasks"] if t["task_id"] == "t2"][0]
|
||||
assert done_task["status"] == "done"
|
||||
result = done_task["result_json"]
|
||||
assert result["summary"] == "lane report: all checks green"
|
||||
assert result["labels"] == [EVAL_LABEL]
|
||||
assert pending_task["status"] == "pending"
|
||||
assert batch["batch"]["status"] == "open" # not all done yet
|
||||
|
||||
def test_poll_marks_batch_done_when_all_closed(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
for issue in r["issues"]:
|
||||
gitea.add_comment("eval-runs", issue["issue_number"], "ok")
|
||||
gitea.close_issue("eval-runs", issue["issue_number"])
|
||||
out = intake.poll_completed(r["batch_id"])
|
||||
assert out["done"] == 2 and out["pending"] == 0
|
||||
assert intake.get_batch(r["batch_id"])["batch"]["status"] == "done"
|
||||
|
||||
def test_poll_unknown_batch(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
intake = make_intake(core, tmp_path)
|
||||
out = intake.poll_completed(123)
|
||||
assert out["ok"] is False and out["error"] == "unknown_batch"
|
||||
|
||||
|
||||
# -------------------------------------------------------------------------- #
|
||||
# Emitting #
|
||||
# -------------------------------------------------------------------------- #
|
||||
|
||||
class TestEmit:
|
||||
def test_emit_posts_report_to_callback_with_retry(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
# fails twice (500s), succeeds on the third attempt
|
||||
emitter = MockEmitter(fail_first=2)
|
||||
intake = make_intake(core, tmp_path, gitea, emitter=emitter)
|
||||
r = intake.submit_batch(client_id, TASKS,
|
||||
callback_url="https://client.example/hook")
|
||||
for issue in r["issues"]:
|
||||
gitea.add_comment("eval-runs", issue["issue_number"],
|
||||
"done: all checks passed")
|
||||
gitea.close_issue("eval-runs", issue["issue_number"])
|
||||
intake.poll_completed(r["batch_id"])
|
||||
|
||||
out = intake.emit(r["batch_id"])
|
||||
assert out["ok"] is True and out["emitted"] is True
|
||||
assert emitter.push_count == 3 # retry x3 total
|
||||
assert len(emitter.push_payloads) == 1 # delivered once
|
||||
report, url = emitter.push_payloads[0]
|
||||
assert url == "https://client.example/hook"
|
||||
assert report["batch_id"] == r["batch_id"]
|
||||
assert report["client_id"] == client_id
|
||||
assert [res["task_id"] for res in report["results"]] == ["t1", "t2"]
|
||||
assert all(res["status"] == "done" for res in report["results"])
|
||||
assert all("summary" in res for res in report["results"])
|
||||
assert intake.get_batch(r["batch_id"])["batch"]["status"] == "emitted"
|
||||
|
||||
def test_emit_file_mode_writes_json_drop(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea,
|
||||
output_dir=tmp_path / "drops")
|
||||
r = intake.submit_batch(client_id, TASKS) # no callback_url
|
||||
for issue in r["issues"]:
|
||||
gitea.close_issue("eval-runs", issue["issue_number"])
|
||||
intake.poll_completed(r["batch_id"])
|
||||
out = intake.emit(r["batch_id"])
|
||||
assert out["ok"] is True and out["emitted"] is True
|
||||
drop = tmp_path / "drops" / f"{r['batch_id']}.json"
|
||||
assert drop.exists()
|
||||
payload = json.loads(drop.read_text(encoding="utf-8"))
|
||||
assert payload["batch_id"] == r["batch_id"]
|
||||
assert len(payload["results"]) == 2
|
||||
|
||||
def test_emit_refuses_before_all_tasks_done(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
intake = make_intake(core, tmp_path)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
out = intake.emit(r["batch_id"])
|
||||
assert out["ok"] is False and out["reason"] == "tasks_pending"
|
||||
assert intake.get_batch(r["batch_id"])["batch"]["status"] == "open"
|
||||
|
||||
def test_reemit_is_idempotent_no_double_post(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
emitter = MockEmitter()
|
||||
intake = make_intake(core, tmp_path, gitea, emitter=emitter)
|
||||
r = intake.submit_batch(client_id, TASKS,
|
||||
callback_url="https://cb.example/x")
|
||||
for issue in r["issues"]:
|
||||
gitea.close_issue("eval-runs", issue["issue_number"])
|
||||
intake.poll_completed(r["batch_id"])
|
||||
assert intake.emit(r["batch_id"])["ok"] is True
|
||||
pushes_after_first = emitter.push_count
|
||||
again = intake.emit(r["batch_id"])
|
||||
assert again["ok"] is True and again.get("idempotent") is True
|
||||
assert emitter.push_count == pushes_after_first # no second POST
|
||||
|
||||
def test_callback_failure_keeps_batch_emittable(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
emitter = MockEmitter(fail_first=99) # always fails
|
||||
intake = make_intake(core, tmp_path, gitea, emitter=emitter)
|
||||
r = intake.submit_batch(client_id, TASKS,
|
||||
callback_url="https://cb.example/x")
|
||||
for issue in r["issues"]:
|
||||
gitea.close_issue("eval-runs", issue["issue_number"])
|
||||
intake.poll_completed(r["batch_id"])
|
||||
out = intake.emit(r["batch_id"])
|
||||
assert out["ok"] is False and out["reason"] == "emit_failed"
|
||||
assert out["attempts"] == 3
|
||||
assert intake.get_batch(r["batch_id"])["batch"]["status"] == "done"
|
||||
|
||||
|
||||
# -------------------------------------------------------------------------- #
|
||||
# Persistence + config #
|
||||
# -------------------------------------------------------------------------- #
|
||||
|
||||
class TestPersistence:
|
||||
def test_batch_survives_intake_reopen(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
intake2 = make_intake(core, tmp_path, gitea) # fresh connection
|
||||
batch = intake2.get_batch(r["batch_id"])
|
||||
assert batch["ok"] and batch["batch"]["client_id"] == client_id
|
||||
assert len(batch["tasks"]) == 2
|
||||
|
||||
def test_shared_repo_name_from_config(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea, eval_repo="qa-runs")
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
assert r["repo"] == "qa-runs"
|
||||
assert "qa-runs" in gitea.created_repos
|
||||
|
||||
def test_per_batch_repo_mode(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea, per_batch_repos=True)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
assert r["repo"] == f"eval-{client_id}-{r['batch_id']}"
|
||||
assert r["repo"] in gitea.created_repos
|
||||
Loading…
Reference in New Issue
Block a user