gitea-saas-core#39: eval farm HTTP surface — register_eval_routes
Five routes symmetric with the factory front-end: POST /api/eval/batches (funded 201 / unfunded 202 queued) GET /api/eval/batches/<id> status + per-task state GET /api/eval/batches/<id>/results the results table POST /api/eval/batches/<id>/cancel stop draining + release the hold GET /api/eval/health liveness, no auth Auth delegates to gitea_saas.auth (issue #24 single path — no sixth implementation): missing/blank key -> canonical 401; no key configured at all -> 503 SERVICE_UNAVAILABLE (issue #38's assembly acceptance). Unfunded submits now queue in pending_eval_batches (no Gitea issue is ever created); flush_pending ships them after funding, exactly like a fresh funded submit. Funded submits reserve eval_batch_hold_minor (SAAS_EVAL_BATCH_HOLD_MINOR, default = debit floor) via the billing core's reserve_hold; cancel releases it so the balance returns to its pre-submit value exactly, and marks the batch cancelled (poll/emit become no-ops). Rolled-back intake releases the hold too. Schema: pending_eval_batches + eval_batch_cancels tables (idempotent CREATE IF NOT EXISTS), hold_id column with a PRAGMA-guarded ALTER for pre-existing DB files.
This commit is contained in:
Claude
parent
71cc2634ce
commit
8620d4d36e
@ -10,12 +10,25 @@ 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.
|
||||
debit floor. Unfunded/unknown clients get a 402-style structured refusal,
|
||||
a locally queued row (``pending_eval_batches``) and NO Gitea issues are
|
||||
ever created for them; once funding arrives, `flush_pending` ships the
|
||||
queued batch exactly like a fresh funded submit (issue #39).
|
||||
|
||||
Funds discipline (issue #39): a funded submit reserves a batch hold
|
||||
(configurable ``eval_batch_hold_minor`` / ``SAAS_EVAL_BATCH_HOLD_MINOR``,
|
||||
default = the debit floor) so the client's available balance visibly backs
|
||||
the in-flight batch. ``cancel_batch`` releases that hold and stops draining
|
||||
(a cancel marker; the batch no longer polls or emits). Rolled-back intake
|
||||
releases the hold too, so a failed submit never leaves money locked.
|
||||
|
||||
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.
|
||||
``eval_batches``, ``eval_tasks``, ``pending_eval_batches`` and
|
||||
``eval_batch_cancels`` are bootstrapped idempotently against the same
|
||||
SQLite file as the billing core, exactly like the factory does. The
|
||||
``hold_id`` column is added to pre-existing ``eval_batches`` tables via a
|
||||
guarded ``ALTER TABLE`` (SQLite cannot amend a CHECK constraint, but
|
||||
adding a column is idempotent and safe).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@ -30,10 +43,21 @@ from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
# Single auth path (issue #24): the header name, the constant-time comparison
|
||||
# and the 401 body all live in ``gitea_saas.auth``; this module delegates and
|
||||
# never implements its own key check (issue #39 "do not add a sixth").
|
||||
from .auth import API_KEY_HEADER, key_is_valid, unauthorized_response
|
||||
|
||||
# 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 per-batch hold reserved at submit (issue #39): the client's
|
||||
# available balance visibly backs the in-flight batch until it emits or is
|
||||
# cancelled. Config attr ``eval_batch_hold_minor`` / env
|
||||
# ``SAAS_EVAL_BATCH_HOLD_MINOR`` override; 0 disables the hold entirely.
|
||||
DEFAULT_BATCH_HOLD_MINOR = DEFAULT_DEBIT_FLOOR_MINOR
|
||||
|
||||
DEFAULT_EVAL_REPO = "eval-runs"
|
||||
EVAL_LABEL = "eval-task"
|
||||
|
||||
@ -45,6 +69,12 @@ _STATUS_FAILED = "failed"
|
||||
_TASK_PENDING = "pending"
|
||||
_TASK_DONE = "done"
|
||||
|
||||
# pending_eval_batches lifecycle (issue #39, mirroring the factory's
|
||||
# pending_issues): queued while the client is unfunded, shipped once
|
||||
# ``flush_pending`` creates the real batch + task issues.
|
||||
_STATUS_QUEUED = "queued"
|
||||
_STATUS_SHIPPED = "shipped"
|
||||
|
||||
_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).
|
||||
@ -55,6 +85,7 @@ CREATE TABLE IF NOT EXISTS eval_batches (
|
||||
status TEXT NOT NULL DEFAULT 'open'
|
||||
CHECK (status IN ('open', 'done', 'emitted', 'failed')),
|
||||
callback_url TEXT,
|
||||
hold_id INTEGER,
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
|
||||
);
|
||||
|
||||
@ -73,6 +104,26 @@ CREATE TABLE IF NOT EXISTS eval_tasks (
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_eval_tasks_batch ON eval_tasks (batch_id, status);
|
||||
|
||||
-- Unfunded submissions queue here (issue #39, factory mirror): no Gitea
|
||||
-- issue is ever created for them; flush_pending ships them post-funding.
|
||||
CREATE TABLE IF NOT EXISTS pending_eval_batches (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
client_id INTEGER NOT NULL,
|
||||
tasks_json TEXT NOT NULL,
|
||||
callback_url TEXT,
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
|
||||
status TEXT NOT NULL DEFAULT 'queued'
|
||||
CHECK (status IN ('queued', 'shipped'))
|
||||
);
|
||||
|
||||
-- Cancel markers (issue #39): "stop draining" without touching the batch
|
||||
-- status CHECK constraint on pre-existing tables. A marker row means the
|
||||
-- batch no longer polls or emits and its hold was released.
|
||||
CREATE TABLE IF NOT EXISTS eval_batch_cancels (
|
||||
batch_id INTEGER PRIMARY KEY REFERENCES eval_batches(id),
|
||||
cancelled_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
|
||||
);
|
||||
"""
|
||||
|
||||
|
||||
@ -167,6 +218,15 @@ class EvalIntake:
|
||||
self._conn.execute("PRAGMA journal_mode=WAL")
|
||||
self._conn.execute("PRAGMA foreign_keys=ON")
|
||||
self._conn.executescript(_EVAL_DDL)
|
||||
# Idempotent column migration for DB files created before #39's
|
||||
# hold_id landed: SQLite cannot amend a CHECK, but ADD COLUMN is
|
||||
# safe and re-runnable (PRAGMA-guarded).
|
||||
cols = {row[1] for row in self._conn.execute(
|
||||
"PRAGMA table_info(eval_batches)").fetchall()}
|
||||
if "hold_id" not in cols:
|
||||
self._conn.execute(
|
||||
"ALTER TABLE eval_batches ADD COLUMN hold_id INTEGER")
|
||||
self._conn.commit()
|
||||
self._lock = threading.RLock()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
@ -202,6 +262,22 @@ class EvalIntake:
|
||||
def per_batch_repos(self) -> bool:
|
||||
return bool(getattr(self._config, "per_batch_repos", False))
|
||||
|
||||
@property
|
||||
def batch_hold_minor(self) -> int:
|
||||
"""Per-batch hold reserved at submit (issue #39).
|
||||
|
||||
Config attr ``eval_batch_hold_minor`` > env
|
||||
``SAAS_EVAL_BATCH_HOLD_MINOR`` > ``DEFAULT_BATCH_HOLD_MINOR``.
|
||||
0 disables the hold (cancel then has nothing to release).
|
||||
"""
|
||||
value = getattr(self._config, "eval_batch_hold_minor", None)
|
||||
if value is None:
|
||||
raw = os.environ.get("SAAS_EVAL_BATCH_HOLD_MINOR", "").strip()
|
||||
value = int(raw) if raw else DEFAULT_BATCH_HOLD_MINOR
|
||||
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
||||
raise EvalIntakeError(f"invalid batch hold: {value!r}")
|
||||
return value
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Billing helpers (fail-closed, factory semantics)
|
||||
# ------------------------------------------------------------------
|
||||
@ -254,16 +330,45 @@ class EvalIntake:
|
||||
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.
|
||||
Funded clients get ``{ok: True, batch_id, repo, issues,
|
||||
hold_id}``; the batch hold is reserved from the client's balance
|
||||
(issue #39). Unfunded/unknown clients (fail-closed) get
|
||||
``{ok: False, reason: 'unfunded', status: 402, queued: True,
|
||||
pending_id}``, a local ``pending_eval_batches`` row and NO Gitea
|
||||
issues — ``flush_pending`` ships it once funding arrives.
|
||||
"""
|
||||
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
|
||||
return self._queue_batch(client_id, tasks, callback_url)
|
||||
return self._ship_batch(client_id, tasks, callback_url)
|
||||
|
||||
def _queue_batch(self, client_id: int, tasks: List[dict],
|
||||
callback_url: Optional[str]) -> Dict[str, Any]:
|
||||
"""Persist an unfunded submission locally (no Gitea mutations)."""
|
||||
callback_url = (callback_url or "").strip() or None
|
||||
with self._lock:
|
||||
cur = self._conn.execute(
|
||||
"INSERT INTO pending_eval_batches"
|
||||
" (client_id, tasks_json, callback_url, created_at, status)"
|
||||
" VALUES (?, ?, ?, ?, ?)",
|
||||
(client_id, json.dumps(tasks), callback_url, _now_iso(),
|
||||
_STATUS_QUEUED))
|
||||
self._conn.commit()
|
||||
return {"ok": False, "reason": "unfunded", "status": 402,
|
||||
"queued": True, "pending_id": int(cur.lastrowid)}
|
||||
|
||||
def _ship_batch(self, client_id: int, tasks: List[dict],
|
||||
callback_url: Optional[str],
|
||||
tasks_json: Optional[str] = None) -> Dict[str, Any]:
|
||||
"""Create batch row + eval-task issues; reserve the batch hold.
|
||||
|
||||
The hold is placed BEFORE any Gitea mutation and released (with the
|
||||
batch rolled back) if intake fails, so a failed submit never leaves
|
||||
money locked or a partial batch behind. ``tasks_json`` is only
|
||||
passed by ``flush_pending`` (the payload was already persisted in
|
||||
the pending row).
|
||||
"""
|
||||
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)"
|
||||
@ -278,6 +383,27 @@ class EvalIntake:
|
||||
repo = self.eval_repo
|
||||
if not _REPO_NAME_RE.match(repo):
|
||||
repo = DEFAULT_EVAL_REPO
|
||||
|
||||
# Money first (issue #39): reserve the batch hold; a client
|
||||
# whose balance cannot back the batch is refused, not shipped.
|
||||
hold_id = None
|
||||
hold_amount = self.batch_hold_minor
|
||||
if hold_amount > 0:
|
||||
ok, details = self._core.reserve_hold(
|
||||
client_id, hold_amount, f"eval-batch-{batch_id}",
|
||||
ref_type="eval", ref_id=str(batch_id))
|
||||
if not ok:
|
||||
self._conn.execute(
|
||||
"DELETE FROM eval_batches WHERE id = ?", (batch_id,))
|
||||
self._conn.commit()
|
||||
error = details if isinstance(details, dict) else {"raw": details}
|
||||
return {"ok": False, "reason": "intake_failed",
|
||||
"error": error, "status": 500}
|
||||
hold_id = int(details["hold_id"])
|
||||
self._conn.execute(
|
||||
"UPDATE eval_batches SET hold_id = ? WHERE id = ?",
|
||||
(hold_id, batch_id))
|
||||
|
||||
try:
|
||||
if not self._gitea.repo_exists(repo):
|
||||
self._gitea.create_org_repo(repo)
|
||||
@ -297,7 +423,8 @@ class EvalIntake:
|
||||
" task_id, payload_json, status, created_at)"
|
||||
" VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||||
(batch_id, repo, issue_number, task_id,
|
||||
json.dumps({"title": title, "body": body}),
|
||||
tasks_json if tasks_json is not None
|
||||
else json.dumps({"title": title, "body": body}),
|
||||
_TASK_PENDING, _now_iso()))
|
||||
issues.append({"task_id": task_id,
|
||||
"issue_number": issue_number})
|
||||
@ -307,12 +434,116 @@ class EvalIntake:
|
||||
"DELETE FROM eval_tasks WHERE batch_id = ?", (batch_id,))
|
||||
self._conn.execute(
|
||||
"DELETE FROM eval_batches WHERE id = ?", (batch_id,))
|
||||
if hold_id is not None:
|
||||
try:
|
||||
self._core.release_hold(
|
||||
hold_id, f"eval-rollback-{batch_id}")
|
||||
except Exception:
|
||||
pass # best effort; the hold auto-expires in 30 min
|
||||
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}
|
||||
"issues": issues, "hold_id": hold_id}
|
||||
|
||||
def flush_pending(self, client_id: int) -> Dict[str, Any]:
|
||||
"""Ship a client's queued (unfunded) batches once they are funded.
|
||||
|
||||
Factory mirror (issue #39): unfunded work never becomes a Gitea
|
||||
issue, so this refuses to flush while the client is still
|
||||
unfunded. Each shipped batch is created exactly like a fresh
|
||||
funded submit (hold reserved, task issues created, batch row
|
||||
written) and its pending row is marked ``shipped``.
|
||||
"""
|
||||
with self._lock:
|
||||
rows = self._conn.execute(
|
||||
"SELECT * FROM pending_eval_batches WHERE client_id = ?"
|
||||
" AND status = ? ORDER BY id",
|
||||
(client_id, _STATUS_QUEUED)).fetchall()
|
||||
if not rows:
|
||||
return {"ok": True, "flushed": 0, "batch_ids": [], "issues": []}
|
||||
if self.client_unfunded(client_id):
|
||||
return {"ok": False, "reason": "unfunded",
|
||||
"skipped": len(rows), "flushed": 0}
|
||||
batch_ids: List[int] = []
|
||||
issues: List[dict] = []
|
||||
for row in rows:
|
||||
try:
|
||||
tasks = json.loads(row["tasks_json"] or "[]")
|
||||
except ValueError:
|
||||
tasks = []
|
||||
result = self._ship_batch(
|
||||
client_id, tasks, row["callback_url"],
|
||||
tasks_json=row["tasks_json"])
|
||||
if not result.get("ok"):
|
||||
# A single failure stops the flush (fail-closed, mirroring
|
||||
# intake's no-partial-batches rule); earlier shipped rows
|
||||
# stay shipped.
|
||||
return {"ok": False, "reason": "flush_failed",
|
||||
"error": result.get("error"),
|
||||
"flushed": len(batch_ids), "batch_ids": batch_ids,
|
||||
"issues": issues}
|
||||
with self._lock:
|
||||
self._conn.execute(
|
||||
"UPDATE pending_eval_batches SET status = ? WHERE id = ?",
|
||||
(_STATUS_SHIPPED, row["id"]))
|
||||
self._conn.commit()
|
||||
batch_ids.append(result["batch_id"])
|
||||
issues.extend(result["issues"])
|
||||
return {"ok": True, "flushed": len(batch_ids),
|
||||
"batch_ids": batch_ids, "issues": issues}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Cancelling a batch (issue #39)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def cancel_batch(self, batch_id: int) -> Dict[str, Any]:
|
||||
"""Stop draining a batch and release its held funds.
|
||||
|
||||
Writes a cancel marker (``eval_batch_cancels``) so polling and
|
||||
emitting become no-ops, and releases the batch hold — the client's
|
||||
available balance returns to its pre-submit value exactly.
|
||||
Idempotent: a second cancel returns ``already_cancelled`` and
|
||||
never releases twice. A hold the core already auto-expired
|
||||
(``HOLD_ALREADY_RELEASED``/``HOLD_NOT_FOUND``) is treated as
|
||||
already gone; the batch is still marked cancelled.
|
||||
"""
|
||||
batch = self._batch_row(batch_id)
|
||||
if batch is None:
|
||||
return {"ok": False, "error": "unknown_batch"}
|
||||
with self._lock:
|
||||
marker = self._conn.execute(
|
||||
"SELECT 1 FROM eval_batch_cancels WHERE batch_id = ?",
|
||||
(batch_id,)).fetchone()
|
||||
if marker is not None:
|
||||
return {"ok": True, "batch_id": batch_id,
|
||||
"cancelled": True, "already_cancelled": True,
|
||||
"hold_released": False}
|
||||
hold_id = batch["hold_id"]
|
||||
hold_released = False
|
||||
if hold_id is not None:
|
||||
ok, details = self._core.release_hold(
|
||||
int(hold_id), f"eval-cancel-{batch_id}")
|
||||
if not ok:
|
||||
code = ((details or {}).get("error") or {}).get(
|
||||
"code") if isinstance(details, dict) else None
|
||||
if code not in ("HOLD_ALREADY_RELEASED", "HOLD_NOT_FOUND"):
|
||||
return {"ok": False, "reason": "cancel_failed",
|
||||
"error": details}
|
||||
else:
|
||||
hold_released = True
|
||||
self._conn.execute(
|
||||
"INSERT INTO eval_batch_cancels (batch_id, cancelled_at)"
|
||||
" VALUES (?, ?)", (batch_id, _now_iso()))
|
||||
self._conn.commit()
|
||||
return {"ok": True, "batch_id": batch_id, "cancelled": True,
|
||||
"hold_released": hold_released}
|
||||
|
||||
def _is_cancelled(self, batch_id: int) -> bool:
|
||||
return self._conn.execute(
|
||||
"SELECT 1 FROM eval_batch_cancels WHERE batch_id = ?",
|
||||
(batch_id,)).fetchone() is not None
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Polling completed tasks
|
||||
@ -332,11 +563,14 @@ class EvalIntake:
|
||||
|
||||
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.
|
||||
report) plus the issue's labels. A cancelled batch (issue #39)
|
||||
stops draining: ``{ok: False, error: 'cancelled'}``.
|
||||
"""
|
||||
batch = self._batch_row(batch_id)
|
||||
if batch is None:
|
||||
return {"ok": False, "error": "unknown_batch"}
|
||||
if self._is_cancelled(batch_id):
|
||||
return {"ok": False, "error": "cancelled"}
|
||||
done = 0
|
||||
pending = 0
|
||||
for row in self._task_rows(batch_id):
|
||||
@ -411,10 +645,14 @@ class EvalIntake:
|
||||
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).
|
||||
A cancelled batch (issue #39) refuses to emit: ``{ok: False,
|
||||
reason: 'cancelled'}``.
|
||||
"""
|
||||
batch = self._batch_row(batch_id)
|
||||
if batch is None:
|
||||
return {"ok": False, "error": "unknown_batch"}
|
||||
if self._is_cancelled(batch_id):
|
||||
return {"ok": False, "reason": "cancelled"}
|
||||
pending = self._conn.execute(
|
||||
"SELECT COUNT(*) FROM eval_tasks WHERE batch_id = ?"
|
||||
" AND status = 'pending'", (batch_id,)).fetchone()[0]
|
||||
@ -465,4 +703,159 @@ class EvalIntake:
|
||||
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}
|
||||
marker = self._conn.execute(
|
||||
"SELECT cancelled_at FROM eval_batch_cancels WHERE batch_id = ?",
|
||||
(batch_id,)).fetchone()
|
||||
data = dict(batch)
|
||||
if marker is not None:
|
||||
data["status"] = "cancelled"
|
||||
data["cancelled_at"] = marker["cancelled_at"]
|
||||
return {"ok": True, "batch": data, "tasks": tasks,
|
||||
"cancelled": marker is not None}
|
||||
|
||||
def get_results(self, batch_id: int) -> Dict[str, Any]:
|
||||
"""The results table for a batch (issue #39 HTTP surface).
|
||||
|
||||
Read-only view: one row per task with its current status, title
|
||||
and collected summary — regardless of whether the batch is still
|
||||
draining, emitted or cancelled. Unknown batches are
|
||||
``{ok: False, error: 'unknown_batch'}``.
|
||||
"""
|
||||
batch = self._batch_row(batch_id)
|
||||
if batch is None:
|
||||
return {"ok": False, "error": "unknown_batch"}
|
||||
report = self._report(batch)
|
||||
if self._is_cancelled(batch_id):
|
||||
report["status"] = "cancelled"
|
||||
report["ok"] = True
|
||||
return report
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HTTP surface (issue #39; mounted by the assembly issue #38's create_app)
|
||||
# ---------------------------------------------------------------------------
|
||||
#
|
||||
# Symmetric with register_factory_routes: the same five-route shape, the
|
||||
# same X-API-Key discipline (gitea_saas.auth only — issue #24; no sixth
|
||||
# implementation), and the same local-queue money rule. The one deliberate
|
||||
# difference: a route reached while NO key is configured returns 503
|
||||
# SERVICE_UNAVAILABLE (fail-closed "cannot serve"), as #38's assembly
|
||||
# acceptance demands app-wide. /api/eval/health is unauthenticated like
|
||||
# the factory's /api/factory/health (its handler never calls _require_auth).
|
||||
|
||||
|
||||
def register_eval_routes(app: Any, core: Any, gitea: Any, config: Any) -> None:
|
||||
"""Mount the eval-farm endpoints on a Flask app (idempotent).
|
||||
|
||||
Routes:
|
||||
POST /api/eval/batches {"client_id": int,
|
||||
"tasks": [{task_id, title,
|
||||
body?}],
|
||||
"callback_url"?: str}
|
||||
GET /api/eval/batches/<id> status + per-task state
|
||||
GET /api/eval/batches/<id>/results the results table
|
||||
POST /api/eval/batches/<id>/cancel stop draining + release hold
|
||||
GET /api/eval/health liveness (no auth)
|
||||
|
||||
Each authed route builds an :class:`EvalIntake` per request (cheap
|
||||
SQLite connection, WAL + per-instance lock — same pattern as the
|
||||
factory). Auth failures use the canonical 401 body; unconfigured key
|
||||
(empty expected) is 503 SERVICE_UNAVAILABLE.
|
||||
"""
|
||||
from flask import jsonify, request
|
||||
|
||||
api_key = app.config.get("API_KEY") or getattr(config, "api_key", None) or ""
|
||||
|
||||
def _eval() -> EvalIntake:
|
||||
return EvalIntake(core, gitea, config)
|
||||
|
||||
def _require_auth():
|
||||
# Fail-closed on a misconfigured deployment: no key configured
|
||||
# means the service cannot authenticate anything -> 503 (issue
|
||||
# #38's assembly acceptance demands this app-wide).
|
||||
if not api_key:
|
||||
resp = jsonify({"error": {
|
||||
"code": "SERVICE_UNAVAILABLE",
|
||||
"message": "API key not configured"}})
|
||||
resp.status_code = 503
|
||||
return resp
|
||||
sent = request.headers.get(API_KEY_HEADER, "")
|
||||
if not key_is_valid(sent, api_key):
|
||||
return unauthorized_response()
|
||||
return None
|
||||
|
||||
@app.get("/api/eval/health")
|
||||
def eval_health():
|
||||
return jsonify({"ok": True})
|
||||
|
||||
@app.post("/api/eval/batches")
|
||||
def eval_submit_batch():
|
||||
if (denied := _require_auth()) is not None:
|
||||
return denied
|
||||
body = request.get_json(silent=True) or {}
|
||||
raw_id = body.get("client_id")
|
||||
if isinstance(raw_id, bool) or not isinstance(raw_id, int) or raw_id <= 0:
|
||||
return _json_error(400, "invalid_request",
|
||||
"client_id must be a positive integer")
|
||||
callback_url = body.get("callback_url")
|
||||
try:
|
||||
result = _eval().submit_batch(raw_id, body.get("tasks") or [],
|
||||
callback_url)
|
||||
except Exception as exc:
|
||||
return _map_eval_error(exc)
|
||||
resp = jsonify(result)
|
||||
# Funded: 201 created. Unfunded/queued: 202 accepted locally
|
||||
# (factory mirror: the row flushes once funding arrives).
|
||||
resp.status_code = 201 if result.get("ok") else 202
|
||||
return resp
|
||||
|
||||
@app.get("/api/eval/batches/<int:batch_id>")
|
||||
def eval_get_batch(batch_id: int):
|
||||
if (denied := _require_auth()) is not None:
|
||||
return denied
|
||||
try:
|
||||
result = _eval().get_batch(batch_id)
|
||||
except Exception as exc:
|
||||
return _map_eval_error(exc)
|
||||
if not result.get("ok"):
|
||||
return _json_error(404, "unknown_batch", "no such batch")
|
||||
return jsonify(result)
|
||||
|
||||
@app.get("/api/eval/batches/<int:batch_id>/results")
|
||||
def eval_get_results(batch_id: int):
|
||||
if (denied := _require_auth()) is not None:
|
||||
return denied
|
||||
try:
|
||||
result = _eval().get_results(batch_id)
|
||||
except Exception as exc:
|
||||
return _map_eval_error(exc)
|
||||
if not result.get("ok"):
|
||||
return _json_error(404, "unknown_batch", "no such batch")
|
||||
return jsonify(result)
|
||||
|
||||
@app.post("/api/eval/batches/<int:batch_id>/cancel")
|
||||
def eval_cancel_batch(batch_id: int):
|
||||
if (denied := _require_auth()) is not None:
|
||||
return denied
|
||||
try:
|
||||
result = _eval().cancel_batch(batch_id)
|
||||
except Exception as exc:
|
||||
return _map_eval_error(exc)
|
||||
if not result.get("ok"):
|
||||
return _json_error(404, "unknown_batch", "no such batch")
|
||||
return jsonify(result)
|
||||
|
||||
|
||||
def _json_error(status: int, code: str, message: str):
|
||||
from flask import jsonify
|
||||
|
||||
resp = jsonify({"error": message, "code": code})
|
||||
resp.status_code = status
|
||||
return resp
|
||||
|
||||
|
||||
def _map_eval_error(exc: Exception):
|
||||
"""Map eval domain errors to HTTP responses (fail-closed on unknown)."""
|
||||
if isinstance(exc, EvalIntakeError):
|
||||
return _json_error(400, "invalid_request", str(exc))
|
||||
return _json_error(500, "eval_error", f"{type(exc).__name__}: {exc}")
|
||||
|
||||
@ -8,6 +8,7 @@ DB (funded vs unfunded gating is real). No network, no sleeps (a noop
|
||||
Run: python3 -m pytest tests/test_intake_eval.py -q
|
||||
"""
|
||||
import json
|
||||
import sqlite3
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
@ -15,8 +16,14 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import pytest
|
||||
|
||||
from gitea_saas.auth import UNAUTHORIZED_BODY
|
||||
from gitea_saas.billing_core import BillingCore
|
||||
from gitea_saas.intake_eval import EvalIntake, EvalIntakeError, EVAL_LABEL
|
||||
from gitea_saas.intake_eval import (
|
||||
EvalIntake,
|
||||
EvalIntakeError,
|
||||
EVAL_LABEL,
|
||||
register_eval_routes,
|
||||
)
|
||||
|
||||
|
||||
class Config:
|
||||
@ -420,3 +427,402 @@ def test_real_client_list_issue_comments_builds_correct_request(monkeypatch):
|
||||
"issues/42/comments")
|
||||
assert req.headers.get("Authorization") == "token tok123"
|
||||
assert req.headers.get("Accept") == "application/json"
|
||||
|
||||
|
||||
# -------------------------------------------------------------------------- #
|
||||
# Issue #39: HTTP surface (register_eval_routes) #
|
||||
# -------------------------------------------------------------------------- #
|
||||
|
||||
@pytest.fixture
|
||||
def app_env(tmp_path):
|
||||
"""core + MockGitea + api-keyed Config for mounting eval routes."""
|
||||
core = BillingCore(tmp_path / "billing.db")
|
||||
gitea = MockGitea()
|
||||
cfg = Config(tmp_path / "saas.db", api_key="test-api-key",
|
||||
debit_floor_minor=500)
|
||||
return core, gitea, cfg
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def eval_app(app_env):
|
||||
from flask import Flask
|
||||
|
||||
core, gitea, cfg = app_env
|
||||
app = Flask(__name__)
|
||||
app.config["API_KEY"] = cfg.api_key
|
||||
app.config["TESTING"] = True
|
||||
register_eval_routes(app, core, gitea, cfg)
|
||||
return app.test_client()
|
||||
|
||||
|
||||
def _balance(core, client_id):
|
||||
ok, details = core.get_client(client_id, include_balance=True)
|
||||
assert ok
|
||||
return details["balance"]
|
||||
|
||||
|
||||
def _queued_batch_rows(db_path, client_id):
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
try:
|
||||
conn.row_factory = sqlite3.Row
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM pending_eval_batches WHERE client_id = ?"
|
||||
" ORDER BY id", (client_id,)).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
class TestEvalRoutes:
|
||||
"""Acceptance 1: each route — success, missing/blank key 401,
|
||||
unconfigured key 503; health stays public."""
|
||||
|
||||
AUTHD_PATHS = [
|
||||
("post", "/api/eval/batches", {"client_id": 1, "tasks": TASKS}),
|
||||
("get", "/api/eval/batches/1", None),
|
||||
("get", "/api/eval/batches/1/results", None),
|
||||
("post", "/api/eval/batches/1/cancel", None),
|
||||
]
|
||||
|
||||
KEY = "test-api-key"
|
||||
|
||||
@staticmethod
|
||||
def _post(client, path, payload=None):
|
||||
return client.post(path, json=payload or {},
|
||||
headers={"X-API-Key": TestEvalRoutes.KEY})
|
||||
|
||||
@staticmethod
|
||||
def _get(client, path):
|
||||
return client.get(path,
|
||||
headers={"X-API-Key": TestEvalRoutes.KEY})
|
||||
|
||||
def test_eval_health_is_public(self, eval_app):
|
||||
resp = eval_app.get("/api/eval/health")
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_json() == {"ok": True}
|
||||
|
||||
def test_eval_health_public_without_any_key(self, app_env):
|
||||
core, gitea, cfg = app_env
|
||||
from flask import Flask
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config["TESTING"] = True
|
||||
register_eval_routes(app, core, gitea, cfg) # API_KEY unset
|
||||
resp = app.test_client().get("/api/eval/health")
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_missing_key_401_on_every_route(self, eval_app):
|
||||
for method, path, _payload in self.AUTHD_PATHS:
|
||||
resp = getattr(eval_app, method)(path, json={})
|
||||
assert resp.status_code == 401, (method, path)
|
||||
# canonical #24 body, byte-identical with every front-end
|
||||
assert resp.get_json() == UNAUTHORIZED_BODY
|
||||
|
||||
def test_blank_key_401_on_every_route(self, eval_app):
|
||||
for method, path, _payload in self.AUTHD_PATHS:
|
||||
resp = getattr(eval_app, method)(
|
||||
path, json={}, headers={"X-API-Key": ""})
|
||||
assert resp.status_code == 401, (method, path)
|
||||
assert resp.get_json() == UNAUTHORIZED_BODY
|
||||
|
||||
def test_wrong_key_401(self, eval_app):
|
||||
resp = eval_app.post(
|
||||
"/api/eval/batches", json={"client_id": 1, "tasks": TASKS},
|
||||
headers={"X-API-Key": "wrong-key"})
|
||||
assert resp.status_code == 401
|
||||
assert resp.get_json() == UNAUTHORIZED_BODY
|
||||
|
||||
def test_unconfigured_key_503_on_every_route(self, app_env):
|
||||
"""No API key configured -> authed routes fail closed 503
|
||||
(issue #38's assembly acceptance demands this app-wide); the
|
||||
liveness route stays reachable."""
|
||||
core, gitea, cfg = app_env
|
||||
cfg_no_key = Config(cfg.db_path) # no api_key attr at all
|
||||
from flask import Flask
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config["TESTING"] = True
|
||||
register_eval_routes(app, core, gitea, cfg_no_key)
|
||||
client = app.test_client()
|
||||
for method, path, payload in self.AUTHD_PATHS:
|
||||
resp = getattr(client, method)(path, json=payload or {})
|
||||
assert resp.status_code == 503, (method, path)
|
||||
assert resp.get_json()["error"]["code"] == "SERVICE_UNAVAILABLE"
|
||||
assert client.get("/api/eval/health").status_code == 200
|
||||
|
||||
def test_submit_funded_route_201(self, app_env, eval_app):
|
||||
core, gitea, _cfg = app_env
|
||||
ok, c = core.create_client("route-client")
|
||||
assert ok
|
||||
core.credit(c["id"], 10_000, "fund-route")
|
||||
resp = self._post(eval_app, "/api/eval/batches", {
|
||||
"client_id": c["id"], "tasks": TASKS})
|
||||
assert resp.status_code == 201
|
||||
body = resp.get_json()
|
||||
assert body["ok"] is True and body["batch_id"] == 1
|
||||
assert len(body["issues"]) == 2
|
||||
# the batch hold is visible in the client's available balance
|
||||
assert _balance(core, c["id"]) == 10_000 - 500
|
||||
|
||||
def test_submit_unfunded_route_202_queued(self, app_env, eval_app):
|
||||
core, gitea, _cfg = app_env
|
||||
ok, c = core.create_client("poor-route")
|
||||
assert ok
|
||||
resp = self._post(eval_app, "/api/eval/batches", {
|
||||
"client_id": c["id"], "tasks": TASKS})
|
||||
assert resp.status_code == 202
|
||||
body = resp.get_json()
|
||||
assert body["ok"] is False and body["reason"] == "unfunded"
|
||||
assert body["queued"] is True and body["pending_id"] == 1
|
||||
assert gitea.created_issues == [] # NO Gitea issues
|
||||
assert gitea.created_repos == []
|
||||
rows = _queued_batch_rows(_cfg.db_path, c["id"])
|
||||
assert len(rows) == 1 and rows[0]["status"] == "queued"
|
||||
|
||||
def test_route_validation_400_and_404(self, eval_app, app_env):
|
||||
core, gitea, _cfg = app_env
|
||||
ok, c = core.create_client("val-client")
|
||||
assert ok
|
||||
core.credit(c["id"], 10_000, "fund-val")
|
||||
bad = self._post(eval_app, "/api/eval/batches", {
|
||||
"client_id": -3, "tasks": TASKS})
|
||||
assert bad.status_code == 400 and bad.get_json()["code"] == "invalid_request"
|
||||
bad_tasks = self._post(eval_app, "/api/eval/batches", {
|
||||
"client_id": c["id"], "tasks": [{"task_id": "x"}]}) # no title
|
||||
assert bad_tasks.status_code == 400
|
||||
assert self._get(eval_app, "/api/eval/batches/99").status_code == 404
|
||||
assert self._get(eval_app, "/api/eval/batches/99/results").status_code == 404
|
||||
assert self._post(eval_app, "/api/eval/batches/99/cancel").status_code == 404
|
||||
|
||||
def test_get_batch_and_results_routes(self, app_env, eval_app):
|
||||
core, gitea, _cfg = app_env
|
||||
ok, c = core.create_client("view-client")
|
||||
assert ok
|
||||
core.credit(c["id"], 10_000, "fund-view")
|
||||
r = self._post(eval_app, "/api/eval/batches", {
|
||||
"client_id": c["id"], "tasks": TASKS}).get_json()
|
||||
batch_id = r["batch_id"]
|
||||
# status + per-task state
|
||||
resp = self._get(eval_app, f"/api/eval/batches/{batch_id}")
|
||||
assert resp.status_code == 200
|
||||
body = resp.get_json()
|
||||
assert body["ok"] and body["batch"]["status"] == "open"
|
||||
assert [t["task_id"] for t in body["tasks"]] == ["t1", "t2"]
|
||||
assert all(t["status"] == "pending" for t in body["tasks"])
|
||||
# the results table reflects the drain state: close t1 in gitea,
|
||||
# then run the drain step (poll) through a direct intake on the
|
||||
# same DB — the results route renders the persisted table
|
||||
gitea.add_comment("eval-runs", r["issues"][0]["issue_number"],
|
||||
"lane report: green")
|
||||
gitea.close_issue("eval-runs", r["issues"][0]["issue_number"])
|
||||
intake = EvalIntake(core, gitea, _cfg, sleep_fn=noop_sleep)
|
||||
assert intake.poll_completed(batch_id)["done"] == 1
|
||||
results = self._get(
|
||||
eval_app, f"/api/eval/batches/{batch_id}/results").get_json()
|
||||
assert results["ok"] is True
|
||||
by_task = {t["task_id"]: t for t in results["results"]}
|
||||
assert by_task["t1"]["status"] == "done"
|
||||
assert by_task["t1"]["summary"] == "lane report: green"
|
||||
# t2 was never drained, so the results table reports it pending.
|
||||
assert by_task["t2"]["status"] == "pending"
|
||||
|
||||
def test_cancel_route_releases_hold_exactly(self, app_env, eval_app):
|
||||
"""Acceptance 3 (route half): balance returns to its pre-submit
|
||||
value exactly after cancel."""
|
||||
core, gitea, _cfg = app_env
|
||||
ok, c = core.create_client("cancel-client")
|
||||
assert ok
|
||||
core.credit(c["id"], 10_000, "fund-cancel")
|
||||
pre = _balance(core, c["id"])
|
||||
r = self._post(eval_app, "/api/eval/batches", {
|
||||
"client_id": c["id"], "tasks": TASKS}).get_json()
|
||||
assert _balance(core, c["id"]) == pre - 500 # hold placed
|
||||
resp = self._post(eval_app, f"/api/eval/batches/{r['batch_id']}/cancel")
|
||||
assert resp.status_code == 200
|
||||
body = resp.get_json()
|
||||
assert body["ok"] is True and body["cancelled"] is True
|
||||
assert body["hold_released"] is True
|
||||
assert _balance(core, c["id"]) == pre # exact return
|
||||
# the batch now reads as cancelled
|
||||
status = self._get(
|
||||
eval_app, f"/api/eval/batches/{r['batch_id']}").get_json()
|
||||
assert status["cancelled"] is True
|
||||
assert status["batch"]["status"] == "cancelled"
|
||||
|
||||
|
||||
# -------------------------------------------------------------------------- #
|
||||
# Issue #39: queue + flush (acceptance 2) and hold/cancel (acceptance 3) #
|
||||
# -------------------------------------------------------------------------- #
|
||||
|
||||
class TestQueueAndFlush:
|
||||
def test_unfunded_submit_queues_no_gitea(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 and r["reason"] == "unfunded"
|
||||
assert r["queued"] is True and r["pending_id"] == 1
|
||||
# half 1: NO Gitea issue, NO repo
|
||||
assert gitea.created_issues == []
|
||||
assert gitea.created_repos == []
|
||||
# half 1: a queued row exists locally
|
||||
rows = _queued_batch_rows(tmp_path / "saas.db", client_id)
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["status"] == "queued"
|
||||
assert json.loads(rows[0]["tasks_json"]) == TASKS
|
||||
|
||||
def test_flush_ships_queued_batch_after_funding(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["pending_id"] == 1
|
||||
|
||||
# funding arrives; flush ships it
|
||||
core.credit(client_id, 10_000, "fund-flush")
|
||||
out = intake.flush_pending(client_id)
|
||||
assert out["ok"] is True and out["flushed"] == 1
|
||||
assert out["batch_ids"] == [1]
|
||||
assert len(out["issues"]) == 2
|
||||
# half 2: issues now exist, one per task, eval-task label
|
||||
assert len(gitea.created_issues) == 2
|
||||
for _repo, _title, body, labels in gitea.created_issues:
|
||||
assert labels == [EVAL_LABEL]
|
||||
assert "batch: 1" in body
|
||||
# the queued row is marked shipped; a real batch row exists
|
||||
rows = _queued_batch_rows(tmp_path / "saas.db", client_id)
|
||||
assert len(rows) == 1 and rows[0]["status"] == "shipped"
|
||||
batch = intake.get_batch(1)
|
||||
assert batch["ok"] and batch["batch"]["client_id"] == client_id
|
||||
assert len(batch["tasks"]) == 2
|
||||
# the shipped batch placed its hold like any fresh submit
|
||||
assert _balance(core, client_id) == 10_000 - 500
|
||||
|
||||
def test_flush_while_still_unfunded_skips(self, unfunded, tmp_path):
|
||||
core, client_id = unfunded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea)
|
||||
intake.submit_batch(client_id, TASKS)
|
||||
out = intake.flush_pending(client_id)
|
||||
assert out["ok"] is False and out["reason"] == "unfunded"
|
||||
assert out["skipped"] == 1 and out["flushed"] == 0
|
||||
assert gitea.created_issues == []
|
||||
rows = _queued_batch_rows(tmp_path / "saas.db", client_id)
|
||||
assert rows[0]["status"] == "queued"
|
||||
|
||||
def test_flush_nothing_queued_is_noop(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
intake = make_intake(core, tmp_path)
|
||||
out = intake.flush_pending(client_id)
|
||||
assert out == {"ok": True, "flushed": 0, "batch_ids": [], "issues": []}
|
||||
|
||||
|
||||
class TestHoldAndCancel:
|
||||
def test_cancel_releases_hold_balance_exact(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
intake = make_intake(core, tmp_path)
|
||||
pre = _balance(core, client_id)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
assert r["hold_id"] is not None
|
||||
assert _balance(core, client_id) == pre - 500 # hold visible
|
||||
out = intake.cancel_batch(r["batch_id"])
|
||||
assert out["ok"] is True and out["cancelled"] is True
|
||||
assert out["hold_released"] is True
|
||||
assert _balance(core, client_id) == pre # EXACT return
|
||||
|
||||
def test_cancel_idempotent_no_double_release(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
intake = make_intake(core, tmp_path)
|
||||
pre = _balance(core, client_id)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
assert intake.cancel_batch(r["batch_id"])["ok"] is True
|
||||
assert _balance(core, client_id) == pre
|
||||
again = intake.cancel_batch(r["batch_id"])
|
||||
assert again["ok"] is True and again["already_cancelled"] is True
|
||||
assert again["hold_released"] is False
|
||||
assert _balance(core, client_id) == pre # no second release
|
||||
|
||||
def test_cancel_unknown_batch(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
intake = make_intake(core, tmp_path)
|
||||
out = intake.cancel_batch(999)
|
||||
assert out == {"ok": False, "error": "unknown_batch"}
|
||||
|
||||
def test_cancel_stops_draining(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea, emitter=MockEmitter())
|
||||
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"])
|
||||
assert intake.cancel_batch(r["batch_id"])["ok"] is True
|
||||
# polling and emitting both refuse a cancelled batch
|
||||
assert intake.poll_completed(r["batch_id"]) == {
|
||||
"ok": False, "error": "cancelled"}
|
||||
assert intake.emit(r["batch_id"]) == {
|
||||
"ok": False, "reason": "cancelled"}
|
||||
# and the batch reads as cancelled
|
||||
status = intake.get_batch(r["batch_id"])
|
||||
assert status["cancelled"] is True
|
||||
assert status["batch"]["status"] == "cancelled"
|
||||
assert "cancelled_at" in status["batch"]
|
||||
|
||||
def test_zero_hold_disables_hold(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
intake = make_intake(core, tmp_path, eval_batch_hold_minor=0)
|
||||
pre = _balance(core, client_id)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
assert r.get("hold_id") is None
|
||||
assert _balance(core, client_id) == pre # nothing held
|
||||
out = intake.cancel_batch(r["batch_id"])
|
||||
assert out["ok"] is True and out["hold_released"] is False
|
||||
assert _balance(core, client_id) == pre
|
||||
|
||||
def test_intake_failure_releases_hold(self, funded, tmp_path):
|
||||
"""A rolled-back submit must not leave money locked."""
|
||||
core, client_id = funded
|
||||
|
||||
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())
|
||||
pre = _balance(core, client_id)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
assert r["ok"] is False and r["reason"] == "intake_failed"
|
||||
assert intake.get_batch(1)["ok"] is False # no partial batch
|
||||
assert _balance(core, client_id) == pre # hold released
|
||||
|
||||
def test_hold_too_large_refuses_batch(self, funded, tmp_path):
|
||||
"""Reserving beyond the available balance fails closed — the
|
||||
batch is refused and no Gitea issue is created."""
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea,
|
||||
eval_batch_hold_minor=50_000) # > 10_000
|
||||
pre = _balance(core, client_id)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
assert r["ok"] is False and r["reason"] == "intake_failed"
|
||||
assert gitea.created_issues == []
|
||||
assert _balance(core, client_id) == pre
|
||||
|
||||
def test_get_results_after_partial_drain(self, funded, tmp_path):
|
||||
core, client_id = funded
|
||||
gitea = MockGitea()
|
||||
intake = make_intake(core, tmp_path, gitea)
|
||||
r = intake.submit_batch(client_id, TASKS)
|
||||
gitea.add_comment("eval-runs", r["issues"][0]["issue_number"],
|
||||
"ok")
|
||||
gitea.close_issue("eval-runs", r["issues"][0]["issue_number"])
|
||||
intake.poll_completed(r["batch_id"])
|
||||
results = intake.get_results(r["batch_id"])
|
||||
assert results["ok"] is True
|
||||
assert results["batch_id"] == r["batch_id"]
|
||||
by_task = {t["task_id"]: t for t in results["results"]}
|
||||
assert by_task["t1"]["status"] == "done"
|
||||
assert by_task["t1"]["summary"] == "ok"
|
||||
assert by_task["t2"]["status"] == "pending"
|
||||
assert intake.get_results(999)["error"] == "unknown_batch"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user