Add persistent run store with agent API and redesigned Runs panel

.gitignore

@@ -68,3 +68,4 @@ data/graph_library/assets.json
 /docs/locales/**/*.mo
 .diff-worktrees/
 data/runtime/
+data/runs/

comfy_research/api/runs.py

@@ -0,0 +1,176 @@
+# comfy_research/api/runs.py
+"""Run store API: async submit, query, metrics, abort, delete. Files are truth."""
+from __future__ import annotations
+
+import shutil
+
+from fastapi import APIRouter, Header, HTTPException, Request
+
+from comfy_research.engine.runs import run_index, run_store
+from comfy_research.engine.runs.run_index import RunQuery
+from comfy_research.engine.runs.run_worker import get_worker_pool
+from comfy_research.schemas.run_record import TERMINAL_STATUSES
+from comfy_research.schemas.train_request import TrainRequest
+
+router = APIRouter(prefix="/api/runs", tags=["runs"])
+
+_HYPERPARAM_PREFIX = "hyperparam."
+
+
+def _err(status: int, code: str, detail: str) -> HTTPException:
+    return HTTPException(status_code=status, detail={"code": code, "detail": detail})
+
+
+def _validate_run_id(run_id: str) -> None:
+    """Reject a malformed ``run_id`` path param (e.g. path-traversal-shaped) with a
+    structured 400 instead of letting ``run_store.run_dir``'s ``ValueError`` surface
+    as an unstructured 500 from deeper in the call stack."""
+    try:
+        run_store.run_dir(run_id)
+    except ValueError:
+        raise _err(400, "invalid_run_id", f"{run_id!r} is not a valid run ID.")
+
+
+def _parse_float_param(p, name: str) -> float | None:
+    raw = p.get(name)
+    if not raw:
+        return None
+    try:
+        return float(raw)
+    except ValueError:
+        raise _err(400, "invalid_query_param", f"{name}={raw!r} is not a valid number.")
+
+
+def _parse_int_param(p, name: str, default: int) -> int:
+    raw = p.get(name)
+    if not raw:
+        return default
+    try:
+        return int(raw)
+    except ValueError:
+        raise _err(400, "invalid_query_param", f"{name}={raw!r} is not a valid integer.")
+
+
+def _query_from_request(request: Request) -> RunQuery:
+    p = request.query_params
+    hyper = {k[len(_HYPERPARAM_PREFIX):]: v for k, v in p.items()
+             if k.startswith(_HYPERPARAM_PREFIX)}
+    ids = [s for s in (p.get("ids") or "").split(",") if s] or None
+    return RunQuery(
+        status=p.get("status"), origin=p.get("origin"), group_id=p.get("group_id"),
+        since_ms=_parse_float_param(p, "since"),
+        ids=ids, hyperparams=hyper,
+        order_by=p.get("order_by") or "-created_at",
+        limit=_parse_int_param(p, "limit", 100), cursor=p.get("cursor"),
+    )
+
+
+@router.post("", status_code=202)
+def submit_run(body: TrainRequest,
+               idempotency_key: str | None = Header(default=None)) -> dict:
+    rec = get_worker_pool().submit(body, idempotency_key=idempotency_key)
+    return {"run_id": rec.run_id, "status": rec.status}
+
+
+@router.get("")
+def list_runs(request: Request) -> dict:
+    q = _query_from_request(request)
+    try:
+        rows, next_cursor = run_index.query_runs(q)
+    except (ValueError, IndexError):
+        raise _err(400, "invalid_cursor", f"cursor {q.cursor!r} could not be parsed.")
+    return {"runs": rows, "next_cursor": next_cursor}
+
+
+@router.get("/groups")
+def list_groups() -> dict:
+    return {"groups": run_index.group_summary()}
+
+
+@router.get("/{run_id}")
+def get_run(run_id: str) -> dict:
+    _validate_run_id(run_id)
+    rec = run_store.read_run_record(run_id)
+    if rec is None:
+        if (run_store.run_dir(run_id) / "run.json").exists():
+            # file present but unparseable: surface it, don't 404 (spec: unreadable)
+            return {"run_id": run_id, "status": "unreadable", "summary": None}
+        raise _err(404, "run_not_found", f"No run {run_id!r}.")
+    _, series = run_store.load_series(run_id)
+    return {**rec.model_dump(mode="json"), "summary": run_store.summarize(series)}
+
+
+@router.get("/{run_id}/metrics")
+def get_run_metrics(run_id: str, downsample: int | None = None) -> dict:
+    _validate_run_id(run_id)
+    if run_store.read_run_record(run_id) is None:
+        raise _err(404, "run_not_found", f"No run {run_id!r}.")
+    source, data = run_store.load_series(run_id)
+    downsampled = False
+    if downsample and downsample > 0:
+        out = {}
+        for key, series in data.items():
+            if isinstance(series, list) and len(series) > downsample:
+                stride = -(-len(series) // downsample)
+                out[key] = series[::stride]
+                downsampled = True
+            else:
+                out[key] = series
+        data = out
+    return {"source": source, "data": data, "downsampled": downsampled}
+
+
+@router.post("/{run_id}/abort")
+def abort_run(run_id: str) -> dict:
+    _validate_run_id(run_id)
+    ok = get_worker_pool().abort(run_id)
+    if not ok:
+        rec = run_store.read_run_record(run_id)
+        if rec is None:
+            raise _err(404, "run_not_found", f"No run {run_id!r}.")
+        return {"ok": False}
+    return {"ok": True}
+
+
+def _delete_run_files(run_id: str) -> None:
+    shutil.rmtree(run_store.run_dir(run_id), ignore_errors=True)
+    run_index.delete_rows([run_id])
+
+
+_DELETABLE_STATUSES = TERMINAL_STATUSES | {"unreadable"}
+
+
+@router.delete("/{run_id}")
+def delete_run(run_id: str) -> dict:
+    _validate_run_id(run_id)
+    rec = run_store.read_run_record(run_id)
+    if rec is None:
+        if (run_store.run_dir(run_id) / "run.json").exists():
+            _delete_run_files(run_id)  # unreadable: deletable, that's the point
+            return {"ok": True}
+        raise _err(404, "run_not_found", f"No run {run_id!r}.")
+    if rec.status not in _DELETABLE_STATUSES:
+        raise _err(409, "run_active", "Abort the run before deleting it.")
+    _delete_run_files(run_id)
+    return {"ok": True}
+
+
+@router.delete("")
+def bulk_delete(request: Request) -> dict:
+    q = _query_from_request(request)
+    if not any([q.status, q.origin, q.group_id, q.since_ms, q.ids, q.hyperparams]):
+        raise _err(400, "filter_required",
+                   "Bulk delete requires at least one filter (status/origin/group_id/since/ids).")
+    # Collect ALL matching ids first (paginate to the end), then delete the deletable
+    # ones — deleting while paginating would shift the cursor and skip matches.
+    q.limit = 500
+    to_delete: list[str] = []
+    while True:
+        rows, cursor = run_index.query_runs(q)
+        to_delete.extend(r["run_id"] for r in rows if r["status"] in _DELETABLE_STATUSES)
+        if cursor is None:
+            break
+        q.cursor = cursor
+    for run_id in to_delete:
+        _delete_run_files(run_id)
+    return {"deleted": len(to_delete)}

comfy_research/api/train.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
 import json
+import logging
 from dataclasses import asdict
 
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field
 
+logger = logging.getLogger(__name__)
+
 from comfy_research.engine.runs.sweep_control import request_sweep_abort
 from comfy_research.engine.runs.train_coordinate_descent import (
     TrainCoordinateDescentRequest,
@@ -19,6 +22,7 @@
 from comfy_research.engine.crl.crl_run import iter_crl_events_from_context, prepare_crl_run
 from comfy_research.engine.runs.cuda_devices import list_local_cuda_devices
 from comfy_research.engine.runs.trainer_run import iter_trainer_events_from_context, prepare_trainer_run
+from comfy_research.engine.runs.run_writer import RunWriter, build_run_record, capture_events
 from comfy_research.schemas.graph import NodeKind
 from comfy_research.config.remote_train_config import (
     RemoteTrainConfig,
@@ -412,7 +416,28 @@ def generate_remote():
                 set_last_validation_result(False, str(exc))
                 yield _ndjson_encode({"type": "error", "detail": str(exc)})
                 return
-            yield from iter_remote_train_stdout_lines(body, config=cfg_holder["cfg"])
+            writer = RunWriter(build_run_record(body))
+            yield _ndjson_encode({"type": "run_registered", "run_id": writer.record.run_id})
+            writer.mark_running()
+            logged_capture_error = False
+            try:
+                for raw in iter_remote_train_stdout_lines(body, config=cfg_holder["cfg"]):
+                    try:
+                        writer.on_event(json.loads(raw.decode("utf-8")))
+                    except Exception:
+                        # Unparseable remote line: forward it to the client but don't
+                        # capture it into the run store. Log once per stream (not per
+                        # line) so a chatty remote process can't spam the log.
+                        if not logged_capture_error:
+                            logger.warning(
+                                "run %s: dropping unparseable remote NDJSON line(s) "
+                                "from capture (forwarding to client unchanged)",
+                                writer.record.run_id, exc_info=True,
+                            )
+                            logged_capture_error = True
+                    yield raw
+            finally:
+                writer.finalize_disconnect()
 
         return StreamingResponse(
             generate_remote(),
@@ -452,8 +477,11 @@ def generate_crl():
         hessian_oversized_policy=body.hessian_oversized_policy,
     )
 
+    writer = RunWriter(build_run_record(body))
+
     def generate():
-        for event in iter_trainer_events_from_context(ctx):
+        yield _ndjson_encode({"type": "run_registered", "run_id": writer.record.run_id})
+        for event in capture_events(iter_trainer_events_from_context(ctx), writer):
             yield _ndjson_encode(event)
 
     return StreamingResponse(

comfy_research/engine/runs/run_gc.py

@@ -0,0 +1,69 @@
+"""Single-owner retention GC for agent- and sweep-origin runs. Runs only in the API server process."""
+from __future__ import annotations
+
+import json
+import logging
+import shutil
+
+from comfy_research.engine.runs import run_index, run_store
+from comfy_research.engine.runs.run_index import RunQuery
+from comfy_research.schemas.run_record import TERMINAL_STATUSES, now_ms
+
+logger = logging.getLogger(__name__)
+
+_GRACE_MS = 10 * 60 * 1000.0
+_DEFAULTS = {
+    "max_runs_agent": 2000, "max_age_days_agent": None,
+    "max_runs_sweep": 2000,
+    "worker_slots": 2,
+}
+# (origin, cap-config-key): sweep is the highest-volume producer (one run per
+# evaluated point in a sweep or coordinate-descent session), so it gets the same
+# cap/grace/terminal treatment as agent-origin runs.
+_CAPPED_ORIGINS = (("agent", "max_runs_agent"), ("sweep", "max_runs_sweep"))
+
+
+def load_gc_config() -> dict:
+    path = run_store.runs_root() / "config.json"
+    try:
+        raw = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        raw = {}
+    return {**_DEFAULTS, **{k: raw[k] for k in _DEFAULTS if k in raw}}
+
+
+def run_gc_once() -> list[str]:
+    cfg = load_gc_config()
+    cutoff_ms = now_ms() - _GRACE_MS
+    max_age = cfg["max_age_days_agent"]
+    age_cutoff = now_ms() - max_age * 86_400_000.0 if max_age else None
+
+    def prunable(r: dict) -> bool:
+        return (r["status"] in TERMINAL_STATUSES
+                and r["finished_at"] is not None and r["finished_at"] < cutoff_ms)
+
+    pruned: list[str] = []
+    for origin, cap_key in _CAPPED_ORIGINS:
+        rows: list[dict] = []
+        q = RunQuery(origin=origin, order_by="-created_at", limit=500)
+        while True:
+            page, cursor = run_index.query_runs(q)
+            rows.extend(page)
+            if cursor is None:
+                break
+            q.cursor = cursor
+        cap = cfg[cap_key]
+        over_cap = rows[cap:]  # rows are newest-first
+        for r in over_cap:
+            if prunable(r):
+                pruned.append(r["run_id"])
+        if origin == "agent" and age_cutoff is not None:
+            for r in rows[:cap]:
+                if prunable(r) and r["finished_at"] < age_cutoff:
+                    pruned.append(r["run_id"])
+    for run_id in pruned:
+        shutil.rmtree(run_store.run_dir(run_id), ignore_errors=True)
+    run_index.delete_rows(pruned)
+    for run_id in pruned:
+        logger.info("run GC pruned %s", run_id)
+    return pruned