FIX: incremental run listing off the event loop, model uploads refused, runs in flight capped per workstation

RUN_PROGRESS.md

@@ -105,6 +105,13 @@ delivered, or when the run directory disappears.
 events with `"kind": "log"` (§4e). **Without it there are none**: a client
 released before log lines existed would draw one as progress.
 
+After 15 seconds with nothing to send, the stream sends an SSE comment,
+`: ping`. A tool can be silent for minutes -- a registration, a model loading --
+and a client's read timeout is shorter than that; without the comment it
+dropped the stream and reconnected, replaying every event from the start. A
+line starting with `:` is never an event, to the Slicer client or to any SSE
+parser.
+
 The last header is not decoration: nginx buffers a proxied response by default,
 which for a stream means the client sees nothing until the run ends - the exact
 failure the endpoint exists to prevent. `curl` needs `-N` for the same reason.
@@ -369,6 +376,13 @@ Every append and every read stamps the directory, so `RUN_TTL_SECONDS` is an
 progress for six hours is never reaped under itself. The same `_reaper_loop`
 that sweeps transfers sweeps these.
 
+A run a previous server process left `pending` or `running` -- `TEMP_DIR`
+outlives a restart -- is ended `failed` when the next one starts, with the
+message "The server restarted during this run.": nobody else would ever end
+it, and its client waits on exactly that terminal event. Its idle clock is
+kept as it was. A `paused` run is left alone, since a resume picks it up
+whichever process serves it. This assumes the single process `serve.py` runs.
+
 The reaper is a safety net rather than the normal path. **A run is removed with
 the request that owned it**, on success (as a background task, so the directory
 survives until the response has finished streaming and a watcher has the whole

docker-compose.yml

@@ -5,6 +5,17 @@
 # Pinned, the stack keeps one identity wherever it is cloned.
 name: visor-serve
 
+# The log every server service keeps: five files of 20 MB, the oldest dropped.
+# Docker's default json-file driver keeps everything a container ever wrote for
+# as long as it exists, and a server up for months -- an access line per panel
+# poll, every two seconds -- fills the disk TEMP_DIR lives on, where a run then
+# fails to stage its upload for a reason nobody looks for in a log.
+x-bounded-log: &bounded-log
+  driver: json-file
+  options:
+    max-size: "20m"
+    max-file: "5"
+
 # Shared by the two server services below, which differ only in whether they
 # get the GPU. Two services rather than one, because a compose device
 # reservation is ALL-OR-NOTHING: a service asking for an nvidia device cannot
@@ -20,6 +31,7 @@ x-inference-base: &inference-base
   image: ${VISOR_SERVER_IMAGE:-ghcr.io/dcbia-ortholab/visor-serve:main}
   shm_size: 8gb
   working_dir: /workspace/server
+  logging: *bounded-log
   volumes:
     # See `inference-venvs`: a public certificate, when TLS_CERT_FILE names one.
     - ./tls:/etc/visor-tls:ro
@@ -282,6 +294,7 @@ services:
   # block as `inference`.
   inference-venvs:
     profiles: ["venvs"]
+    logging: *bounded-log
     build:
       context: .
       dockerfile: docker/Dockerfile

server/.env.example

@@ -324,3 +324,14 @@ RUN_CANCEL_POLL_SECONDS=1.0
 # and the rest are dropped, terminal events always excepted so a watcher's
 # stream can still end.
 MAX_RUN_EVENTS=2000
+
+# --- one workstation's share of the queue -----------------------------------
+
+# Runs one workstation -- one client address, as the server sees it -- may have
+# in flight at once: registered and not yet ended, waiting for the card
+# included. A run stopped at a checkpoint does not count; it waits for a
+# person. Past it POST /run answers 429 before reading a byte. Every
+# workstation holds the same API token, so without it one client could fill
+# the queue the others wait in. 0 means no cap. Behind a proxy that is not in
+# FORWARDED_ALLOW_IPS every workstation IS the proxy, and they share one cap.
+MAX_RUNS_PER_CLIENT=10

server/benchmark_client.py

@@ -55,6 +55,11 @@
 from typing import Optional
 
 RUN_ID_HEADER = "X-Run-Id"
+# This battery's key, which the server minted for it and checks on every
+# POST /run so a battery is not refused as one workstation sending too many
+# runs (benchmark_jobs.KEY_HEADER, which this script cannot import).
+BENCHMARK_KEY_HEADER = "X-Benchmark-Key"
+BENCHMARK_KEY_ENV = "SADT_BENCHMARK_KEY"
 # Long: a cohort segmentation legitimately takes an hour, and a benchmark that
 # times out before the tool does would record a failure that did not happen.
 REQUEST_TIMEOUT = 3 * 3600
@@ -138,6 +143,14 @@ def _request(url: str, token: str, data=None, method=None, timeout=POLL_TIMEOUT,
         return response.status, response.read(), dict(response.headers)
 
 
+def _run_headers(run_id: str) -> dict:
+    headers = {RUN_ID_HEADER: run_id}
+    key = os.environ.get(BENCHMARK_KEY_ENV)
+    if key:
+        headers[BENCHMARK_KEY_HEADER] = key
+    return headers
+
+
 def _send(url: str, token: str, body: bytes, method: str, content_type: str,
           headers=None):
     request = urllib.request.Request(url, data=body, method=method)
@@ -332,7 +345,7 @@ def _one_run(base: str, token: str, item: dict, client: int, origin: float,
         status, body, headers = _request(
             f"{base}/run/{urllib.parse.quote(item['tool'])}",
             token, data=form, method="POST",
-            timeout=REQUEST_TIMEOUT, headers={RUN_ID_HEADER: run_id})
+            timeout=REQUEST_TIMEOUT, headers=_run_headers(run_id))
         _record(LEDGER_END, run_id)
         record["seconds"] = round(time.time() - posted, 2)
         record["invocation"]["produced"] = {"bytes": len(body)}

server/benchmark_jobs.py

@@ -35,9 +35,11 @@
 
 from __future__ import annotations
 
+import hmac
 import json
 import logging
 import os
+import secrets
 import shutil
 import signal
 import subprocess
@@ -63,6 +65,15 @@
 LATE_SWEEP_SECONDS = 10.0
 _LATE_SWEEP_POLL = 0.25
 
+# How a battery's runs say they are the battery's: a key minted per battery,
+# handed to its process alone and sent beside each POST /run, so the cap on
+# runs per workstation (settings.MAX_RUNS_PER_CLIENT) does not refuse the very
+# concurrency a battery exists to measure. Never in this process's own
+# environment, so no tool inherits it, and never in `current()`. Repeated in
+# benchmark_client.py, which runs as a script and imports nothing from here.
+KEY_HEADER = "X-Benchmark-Key"
+KEY_ENV = "SADT_BENCHMARK_KEY"
+
 _lock = threading.Lock()
 _current: Optional[dict] = None
 # Serialises settling a battery's runs: a stop and the watcher both settle,
@@ -102,6 +113,17 @@ def _finished(job: dict) -> bool:
     return process is None or process.poll() is not None
 
 
+def is_battery(presented) -> bool:
+    """Whether `presented` is the key of the battery running now."""
+    if not presented:
+        return False
+    with _lock:
+        job = _current
+    if job is None or not job.get("_key") or _finished(job):
+        return False
+    return hmac.compare_digest(str(presented), job["_key"])
+
+
 def start(plan: dict, summary_dir: str, base_url: str, token: str,
           scratch_dir: str) -> dict:
     """Spawn the battery. Raises `BatteryError` if one is already running."""
@@ -131,8 +153,10 @@ def start(plan: dict, summary_dir: str, base_url: str, token: str,
         with open(plan_path, "w", encoding="utf-8") as handle:
             json.dump(plan, handle)
 
+        key = secrets.token_urlsafe(32)
         environment = dict(os.environ)
         environment["API_TOKEN"] = token
+        environment[KEY_ENV] = key
         environment["SADT_TLS_CAFILE"] = tls.certificate_paths()[0]
         client = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                               "benchmark_client.py")
@@ -155,6 +179,7 @@ def start(plan: dict, summary_dir: str, base_url: str, token: str,
             "exit_code": None,
             "cancelled_runs": None,
             "_process": process,
+            "_key": key,
             "_plan_path": plan_path,
             "_runs_path": runs_path,
         }

server/config.py

@@ -341,6 +341,21 @@ def tool_roots(self) -> tuple:
     # fill TEMP_DIR; past the cap one "not reported" event is delivered and the
     # rest are dropped, terminal events always excepted.
     MAX_RUN_EVENTS: int = 2000
+    # Runs one workstation may have in flight at once -- registered and not yet
+    # ended, waiting for the card included -- past which POST /run answers 429
+    # before it reads a byte. Every workstation holds the same API token, so
+    # without this one client sending sixty runs filled the queue every other
+    # one waits in. A PAUSED run does not count: it waits for a person and
+    # holds no machine. 0 means no cap.
+    #
+    # 10 is well above what the Slicer client sends: one run per panel at a
+    # time (CONCURRENT_RUNS = 1), and on a workstation an operator set to
+    # parallel batches as many as the budget runs side by side (`max_parallel`
+    # of GET /clients/me, CPUs / CPUs per job). Counted per address as the
+    # server sees it: behind a proxy that is not in FORWARDED_ALLOW_IPS every
+    # workstation IS the proxy, and they share one cap. A benchmark battery an
+    # operator starts is not counted.
+    MAX_RUNS_PER_CLIENT: int = 10
 
 
     @field_validator("SADT_DISPATCH_MODE")

server/deployment.toml

@@ -204,8 +204,8 @@ width_from = "structures"
 # AMASSS has one model and resolves it itself when `model` is empty, from
 # DATA/AMASSS/models/AMASSS_Models. The convention would fill the argument with
 # the models FOLDER, which is not a bundle, and would put a picker with a
-# single answer in front of a clinician. So it is neither filled nor asked;
-# a caller that names a bundle is still obeyed.
+# single answer in front of a clinician. So it is neither filled nor asked --
+# nor uploaded: a model never is (main.py `_reject_upload_for_model`).
 server_selectable = { model = "none" }
 hidden = ["model"]
 
@@ -229,8 +229,8 @@ dispatch = { "CBCT" = "AREG_CBCT", "IOS" = "AREG_IOS", "IOS to CBCT" = "AREG_IOS
 # convention that reads `*_model` as "pick one from DATA/ASO/models/" would fill
 # it with ASO's own model folder and pass that on as ALI's bundle. Measured in a
 # real run on 2026-09-11: ALI received `DATA/ASO/models` and predicted with the
-# wrong weights, silently. "none" opts it out; a caller may still name a bundle
-# to pin which weights ran.
+# wrong weights, silently. "none" opts it out, and the panel no longer asks for
+# it: a model is never uploaded (main.py `_reject_upload_for_model`).
 server_selectable = { landmark_model = "none" }
 
 [tools.ALI]