feat: trackio alerts

README.md

@@ -40,6 +40,25 @@ Optional config accepts strings `participant`, `submission_id`, `run_id`, `model
 Supply it on the first batch; it cannot be changed after a run exists. Repeating
 the same config on subsequent events is supported.
 
+The same endpoint accepts an `alerts` list, alone or alongside `logs`:
+
+```json
+{"alerts":[{"project":"burtenshaw","run":"arena-run-id","run_id":"arena-run-id","alert_id":"sandbox-cleanup-39","title":"Sandbox cleanup timed out","text":"Training ended after 39 updates. Evaluation will use the last checkpoint.","level":"error","step":39}]}
+```
+
+An alert has a required stable `alert_id`, a nonempty `title` of up to 200 characters,
+optional plain `text` of up to 4,000 characters, and `level` of `info`, `warn`, or
+`error`. Its optional `step` is a nonnegative integer or null. Identity and config
+follow the same rules as logs. The combined batch limit is 128 events; the entire
+batch must describe one run. Retry with the same `alert_id`; retries cannot replace
+an existing alert. Alert-only failures register a run without inventing numeric metrics.
+
+Alerts are public. Send bounded messages describing the failure and next action;
+never include credentials, private trajectories, or raw exception payloads. They
+appear in Trackio's run-filtered Reports page at
+`/reports?project=<participant>&runs=<full-run-id>`. Trackio 0.40.0 does not expose
+individual-alert anchors or a URL parameter for the severity filter.
+
 The existing `openenvarena/training-dashboard` bucket stays mounted at `/data`, with
 `TRACKIO_DIR=/data/trackio`. The app uses this mounted storage directly and does not
 provision Spaces, buckets or credentials. Run one server process for Trackio's

ingest.py

@@ -1,4 +1,4 @@
-"""Append-only numeric metric ingestion for trusted Arena workers."""
+"""Append-only numeric metric and public alert ingestion for trusted Arena workers."""
 
 import json
 import math
@@ -26,6 +26,17 @@ STRING_CONFIG = {
 }
 NUMBER_CONFIG = {"num_updates", "train_tasks"}
 LOG_FIELDS = {"project", "run", "run_id", "step", "log_id", "metrics", "config"}
+ALERT_FIELDS = {
+    "project",
+    "run",
+    "run_id",
+    "step",
+    "alert_id",
+    "title",
+    "text",
+    "level",
+    "config",
+}
 # Trackio's bucket-backed SQLite storage supports one server process. Serialize
 # the configuration check with ingestion so two first batches cannot replace it.
 _write_lock = Lock()
@@ -59,17 +70,28 @@ def _config(value):
 
 
 def validate(payload):
-    if not isinstance(payload, dict) or set(payload) != {"logs"}:
-        raise ValueError("expected an object containing only logs")
-    logs = payload["logs"]
-    if not isinstance(logs, list) or not 1 <= len(logs) <= MAX_LOGS:
-        raise ValueError(f"logs must contain 1 to {MAX_LOGS} entries")
+    if (
+        not isinstance(payload, dict)
+        or not payload
+        or payload.keys() - {"logs", "alerts"}
+    ):
+        raise ValueError("expected an object containing logs or alerts")
+    logs, alerts = payload.get("logs", []), payload.get("alerts", [])
+    if (
+        not isinstance(logs, list)
+        or not isinstance(alerts, list)
+        or not 1 <= len(logs) + len(alerts) <= MAX_LOGS
+    ):
+        raise ValueError(f"batch must contain 1 to {MAX_LOGS} entries")
     identity = None
     config = None
-    for entry in logs:
-        if not isinstance(entry, dict) or entry.keys() - LOG_FIELDS:
-            raise ValueError("log contains unsupported fields")
-        for key in ("project", "run", "log_id"):
+    for entry, is_alert in [(item, False) for item in logs] + [
+        (item, True) for item in alerts
+    ]:
+        fields = ALERT_FIELDS if is_alert else LOG_FIELDS
+        if not isinstance(entry, dict) or entry.keys() - fields:
+            raise ValueError("entry contains unsupported fields")
+        for key in ("project", "run", "alert_id" if is_alert else "log_id"):
             if not _identifier(entry.get(key)):
                 raise ValueError(f"{key} must contain 1 to 128 letters, digits, _ or -")
         if entry.get("run_id", entry["run"]) != entry["run"]:
@@ -79,26 +101,45 @@ def validate(payload):
             raise ValueError("a batch must contain one project and run")
         identity = current
         step = entry.get("step")
-        if type(step) is not int or not 0 <= step <= 2**63 - 1:
+        if not (is_alert and step is None) and (
+            type(step) is not int or not 0 <= step <= 2**63 - 1
+        ):
             raise ValueError("step must be a nonnegative 64-bit integer")
-        metrics = entry.get("metrics")
-        if not isinstance(metrics, dict) or not 1 <= len(metrics) <= MAX_METRICS:
-            raise ValueError(f"metrics must contain 1 to {MAX_METRICS} numeric values")
-        for key, value in metrics.items():
-            if not isinstance(key, str) or not 1 <= len(key) <= 128:
-                raise ValueError("metric names must contain 1 to 128 characters")
-            key.encode("utf-8")
-            if not _number(value):
-                raise ValueError("metrics must contain only finite numbers")
+        if is_alert:
+            if entry.get("level") not in ("info", "warn", "error"):
+                raise ValueError("alert level must be info, warn or error")
+            for key, limit in (("title", 200), ("text", 4000)):
+                value = entry.get(key)
+                if key == "text" and value is None:
+                    continue
+                if (
+                    not isinstance(value, str)
+                    or len(value) > limit
+                    or (key == "title" and not value.strip())
+                ):
+                    raise ValueError("alert text must be bounded plain text")
+                value.encode("utf-8")
+        else:
+            metrics = entry.get("metrics")
+            if not isinstance(metrics, dict) or not 1 <= len(metrics) <= MAX_METRICS:
+                raise ValueError(
+                    f"metrics must contain 1 to {MAX_METRICS} numeric values"
+                )
+            for key, value in metrics.items():
+                if not isinstance(key, str) or not 1 <= len(key) <= 128:
+                    raise ValueError("metric names must contain 1 to 128 characters")
+                key.encode("utf-8")
+                if not _number(value):
+                    raise ValueError("metrics must contain only finite numbers")
         entry_config = _config(entry.get("config"))
         if entry_config is not None:
             if config is not None and entry_config != config:
                 raise ValueError("config must be consistent within a batch")
             config = entry_config
-    return logs, identity, config
+    return logs, alerts, identity, config
 
 
-def append(logs, identity, config):
+def append(logs, alerts, identity, config):
     project, run = identity
     with _write_lock:
         existing = next(
@@ -116,16 +157,40 @@ def append(logs, identity, config):
                     409, "configuration of an existing run is immutable"
                 )
             config = None
-        SQLiteStorage.bulk_log(
-            project=project,
-            run=run,
-            run_id=run,
-            metrics_list=[entry["metrics"] for entry in logs],
-            steps=[entry["step"] for entry in logs],
-            # Scope event IDs to the run; retries cannot replace existing rows.
-            log_ids=[f"{run}:{entry['log_id']}" for entry in logs],
-            config=config,
-        )
+        if logs:
+            SQLiteStorage.bulk_log(
+                project=project,
+                run=run,
+                run_id=run,
+                metrics_list=[entry["metrics"] for entry in logs],
+                steps=[entry["step"] for entry in logs],
+                # Scope event IDs to the run; retries cannot replace existing rows.
+                log_ids=[f"{run}:{entry['log_id']}" for entry in logs],
+                config=config,
+            )
+        elif existing is None:
+            # Trackio discovers runs from metrics. Register early failures without
+            # inventing a metric so the run-filtered Reports link can find them.
+            SQLiteStorage.bulk_log(
+                project=project,
+                run=run,
+                run_id=run,
+                metrics_list=[{}],
+                steps=[alerts[0].get("step") or 0],
+                log_ids=[f"{run}:alert:init"],
+                config=config,
+            )
+        if alerts:
+            SQLiteStorage.bulk_alert(
+                project=project,
+                run=run,
+                run_id=run,
+                titles=[entry["title"] for entry in alerts],
+                texts=[entry.get("text") for entry in alerts],
+                levels=[entry["level"] for entry in alerts],
+                steps=[entry.get("step") for entry in alerts],
+                alert_ids=[f"{run}:{entry['alert_id']}" for entry in alerts],
+            )
 
 
 async def ingest(request):
@@ -143,8 +208,8 @@ async def ingest(request):
             return JSONResponse({"error": "metric batch is too large"}, status_code=413)
         body.extend(chunk)
     try:
-        logs, identity, config = validate(json.loads(body))
+        logs, alerts, identity, config = validate(json.loads(body))
     except (ValueError, OverflowError, RecursionError):
         return JSONResponse({"error": "invalid metric batch"}, status_code=400)
-    await run_in_threadpool(append, logs, identity, config)
-    return JSONResponse({"accepted": len(logs)})
+    await run_in_threadpool(append, logs, alerts, identity, config)
+    return JSONResponse({"accepted": len(logs) + len(alerts)})

test_ingest.py

@@ -43,6 +43,22 @@ def batch(**updates):
     return {"logs": [entry]}
 
 
+def alert_batch(**updates):
+    entry = {
+        "project": "burtenshaw",
+        "run": "arena-run-123",
+        "run_id": "arena-run-123",
+        "step": 39,
+        "alert_id": "failure-39",
+        "title": "Sandbox cleanup timed out",
+        "text": "Training ended after 39 updates. Evaluation will use the last checkpoint.",
+        "level": "error",
+        "config": batch()["logs"][0]["config"],
+    }
+    entry.update(updates)
+    return {"alerts": [entry]}
+
+
 def test_missing_or_invalid_auth_cannot_write(client, monkeypatch):
     for authorization in ("", "Bearer wrong", "test-ingest-token"):
         assert (
@@ -182,6 +198,7 @@ def test_batches_cannot_mix_runs_or_mutate_partial_data(client):
 
 def test_dashboard_still_serves_without_ingest_credentials(client, monkeypatch):
     import huggingface_hub
+
     from app import create_app
 
     def unexpected_mutation(*args, **kwargs):
@@ -254,3 +271,149 @@ def test_dashboard_does_not_accept_ingest_token_for_delete_or_upload(
             },
         )
         assert response.status_code >= 400 or response.json().get("error")
+
+
+def test_alert_retry_is_idempotent_and_does_not_replace_the_original(client):
+    from trackio.sqlite_storage import SQLiteStorage
+
+    assert client.post("/ingest", json=alert_batch()).json() == {"accepted": 1}
+    assert client.post("/ingest", json=alert_batch()).status_code == 200
+    assert (
+        client.post("/ingest", json=alert_batch(text="replacement")).status_code == 200
+    )
+    assert (
+        client.post("/ingest", json=alert_batch(alert_id="failure-40")).status_code
+        == 200
+    )
+    alerts = SQLiteStorage.get_alerts("burtenshaw", run_id="arena-run-123")
+    assert len(alerts) == 2
+    assert all(alert["text"] == alert_batch()["alerts"][0]["text"] for alert in alerts)
+    assert all(alert["level"] == "error" and alert["step"] == 39 for alert in alerts)
+
+
+def test_early_alert_registers_a_visible_run_without_fabricating_metrics(client):
+    from trackio.sqlite_storage import SQLiteStorage
+
+    from app import create_app
+
+    assert client.post("/ingest", json=alert_batch(step=None)).status_code == 200
+    records = SQLiteStorage.get_run_records("burtenshaw")
+    assert [record["id"] for record in records] == ["arena-run-123"]
+    with SQLiteStorage._get_connection(
+        SQLiteStorage.get_project_db_path("burtenshaw")
+    ) as conn:
+        metrics = conn.execute("SELECT metrics FROM metrics").fetchall()
+    assert [json.loads(row[0]) for row in metrics] == [{}]
+    with TestClient(create_app()) as dashboard:
+        assert (
+            dashboard.get("/reports?project=burtenshaw&runs=arena-run-123").status_code
+            == 200
+        )
+        response = dashboard.post(
+            "/api/get_alerts",
+            json={
+                "kwargs": {
+                    "project": "burtenshaw",
+                    "run": "arena-run-123",
+                    "level": "error",
+                }
+            },
+        )
+    assert response.status_code == 200
+    alerts = response.json()["data"]
+    assert len(alerts) == 1
+    assert alerts[0]["title"] == "Sandbox cleanup timed out"
+    assert alerts[0]["step"] is None
+    # The next real metric can use the same immutable configuration.
+    assert client.post("/ingest", json=batch()).status_code == 200
+
+
+def test_metrics_and_alerts_share_run_identity_and_immutable_config(client):
+    from trackio.sqlite_storage import SQLiteStorage
+
+    payload = {**batch(), **alert_batch()}
+    assert client.post("/ingest", json=payload).json() == {"accepted": 2}
+    assert (
+        client.post(
+            "/ingest", json=alert_batch(config={"model": "changed"})
+        ).status_code
+        == 409
+    )
+    assert len(SQLiteStorage.get_alerts("burtenshaw", run_id="arena-run-123")) == 1
+    another = alert_batch(run="another", run_id="another")
+    assert client.post("/ingest", json=another).status_code == 200
+    assert len(SQLiteStorage.get_alerts("burtenshaw", run_id="another")) == 1
+
+
+@pytest.mark.parametrize(
+    "updates",
+    [
+        {"alert_id": ""},
+        {"title": ""},
+        {"title": " "},
+        {"title": "x" * 201},
+        {"title": {"html": "not plain text"}},
+        {"text": "x" * 4001},
+        {"text": 5},
+        {"level": "ERROR"},
+        {"level": ["error"]},
+        {"step": -1},
+        {"step": True},
+        {"step": 2**64},
+        {"run_id": "another"},
+        {"HF_TOKEN": "forbidden"},
+    ],
+)
+def test_invalid_alert_rejected_before_any_metric_write(client, updates):
+    from trackio.sqlite_storage import SQLiteStorage
+
+    payload = {**batch(), **alert_batch(**updates)}
+    assert client.post("/ingest", json=payload).status_code == 400
+    assert SQLiteStorage.get_run_records("burtenshaw") == []
+
+
+def test_alerts_have_the_same_auth_batch_and_unicode_limits(client):
+    for authorization in ("", "Bearer wrong"):
+        assert (
+            client.post(
+                "/ingest", json=alert_batch(), headers={"Authorization": authorization}
+            ).status_code
+            == 401
+        )
+    payload = {**batch(), "alerts": alert_batch()["alerts"] * 128}
+    assert client.post("/ingest", json=payload).status_code == 400
+    for payload in (
+        {},
+        {"alerts": []},
+        {"logs": [], "alerts": []},
+        {"alerts": {}},
+        {**batch(), **alert_batch(run="another", run_id="another")},
+    ):
+        assert client.post("/ingest", json=payload).status_code == 400
+    for field in ("title", "text"):
+        assert (
+            client.post(
+                "/ingest", content=json.dumps(alert_batch(**{field: "\ud800"}))
+            ).status_code
+            == 400
+        )
+
+
+def test_alerts_survive_fresh_process_with_same_storage(client, tmp_path):
+    assert client.post("/ingest", json=alert_batch()).status_code == 200
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            (
+                "from trackio.sqlite_storage import SQLiteStorage; "
+                "alerts = SQLiteStorage.get_alerts('burtenshaw', run_id='arena-run-123'); "
+                "assert len(alerts) == 1 and alerts[0]['step'] == 39"
+            ),
+        ],
+        env={**os.environ, "TRACKIO_DIR": str(tmp_path)},
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr