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/<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}")

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"