diff --git a/gitea_saas/intake_eval.py b/gitea_saas/intake_eval.py index c211b24..ea34689 100644 --- a/gitea_saas/intake_eval.py +++ b/gitea_saas/intake_eval.py @@ -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/.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/ status + per-task state + GET /api/eval/batches//results the results table + POST /api/eval/batches//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/") + 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//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//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}") diff --git a/tests/test_intake_eval.py b/tests/test_intake_eval.py index ac2a292..4eb17ab 100644 --- a/tests/test_intake_eval.py +++ b/tests/test_intake_eval.py @@ -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"