feat: add Phase 2 observability and audit retention

.env.example

@@ -24,6 +24,10 @@ LLMBENCHLAB_WORKER_HEARTBEAT_SECONDS=10
 LLMBENCHLAB_WORKER_POLL_SECONDS=1
 LLMBENCHLAB_WORKER_MAX_ATTEMPTS=3
 LLMBENCHLAB_WORKER_SHUTDOWN_GRACE_SECONDS=30
+LLMBENCHLAB_WORKER_PROGRESS_FLUSH_SECONDS=5
+LLMBENCHLAB_WORKER_PROGRESS_STALE_SECONDS=60
+LLMBENCHLAB_WORKER_EXPECTED_PROCESSES=1
+LLMBENCHLAB_WORKER_RECOVERY_ALERT_SECONDS=60
 LLMBENCHLAB_REDIS_PUBLISH_TIMEOUT_SECONDS=1
 LLMBENCHLAB_REDIS_OPERATION_TIMEOUT_SECONDS=2
 

backend/alembic/versions/20260828_0005_worker_progress_retention.py

@@ -0,0 +1,94 @@
+"""Add Worker progress facts and bounded audit scan indexes.
+
+Revision ID: 20260828_0005
+Revises: 20260827_0004
+Create Date: 2026-08-28 08:00:00 UTC
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "20260828_0005"
+down_revision: str | None = "20260827_0004"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    table = "worker_processes"
+    op.create_table(
+        table,
+        sa.Column("generation_id", sa.String(length=36), nullable=False),
+        sa.Column("worker_id", sa.String(length=128), nullable=False),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("last_scan_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_claim_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_progress_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_lease_heartbeat_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("stopped_at", sa.DateTime(timezone=True), nullable=True),
+        sa.CheckConstraint(
+            "length(worker_id) >= 1 AND length(worker_id) <= 128",
+            name=op.f("ck_worker_processes_worker_id_length"),
+        ),
+        sa.CheckConstraint(
+            "last_seen_at >= started_at",
+            name=op.f("ck_worker_processes_seen_after_start"),
+        ),
+        *(
+            sa.CheckConstraint(
+                f"{column} IS NULL OR ({column} >= started_at AND {column} <= last_seen_at)",
+                name=op.f(f"ck_worker_processes_{column}_within_seen"),
+            )
+            for column in (
+                "last_scan_at",
+                "last_claim_at",
+                "last_progress_at",
+                "last_lease_heartbeat_at",
+            )
+        ),
+        sa.CheckConstraint(
+            "stopped_at IS NULL OR stopped_at >= last_seen_at",
+            name=op.f("ck_worker_processes_stopped_after_seen"),
+        ),
+        sa.PrimaryKeyConstraint("generation_id", name=op.f("pk_worker_processes")),
+        sa.UniqueConstraint("worker_id", name=op.f("uq_worker_processes_worker_id")),
+    )
+    op.create_index(
+        "ix_worker_processes_stopped_seen_generation",
+        table,
+        ["stopped_at", "last_seen_at", "generation_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_audit_events_expires_id",
+        "audit_events",
+        ["expires_at", "id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_audit_events_occurred_id",
+        "audit_events",
+        ["occurred_at", "id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    connection = op.get_bind()
+    count = connection.execute(sa.text("SELECT COUNT(*) FROM worker_processes")).scalar_one()
+    if count:
+        raise RuntimeError(
+            "Cannot downgrade Worker progress schema while process facts exist; "
+            "stop Workers, preserve required facts, and explicitly clear the table first"
+        )
+    op.drop_index("ix_audit_events_occurred_id", table_name="audit_events")
+    op.drop_index("ix_audit_events_expires_id", table_name="audit_events")
+    op.drop_index(
+        "ix_worker_processes_stopped_seen_generation",
+        table_name="worker_processes",
+    )
+    op.drop_table("worker_processes")

backend/alembic/versions/20260829_0006_repair_governance_indexes.py

@@ -0,0 +1,125 @@
+"""Repair three governance indexes missing from an early revision 0004 schema.
+
+Revision ID: 20260829_0006
+Revises: 20260828_0005
+Create Date: 2026-08-29 00:00:00 UTC
+"""
+
+from collections.abc import Sequence
+from typing import Any
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "20260829_0006"
+down_revision: str | None = "20260828_0005"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _normalized_predicate(value: object) -> str:
+    predicate = " ".join(str(value).strip().lower().split())
+    while predicate.startswith("(") and predicate.endswith(")"):
+        predicate = predicate[1:-1].strip()
+    return predicate
+
+
+def _ensure_index(
+    table_name: str,
+    index_name: str,
+    columns: tuple[str, ...],
+    *,
+    unique: bool,
+    sqlite_where: sa.TextClause | None = None,
+    postgresql_where: sa.TextClause | None = None,
+) -> None:
+    connection = op.get_bind()
+    existing_indexes: dict[str, dict[str, Any]] = {
+        str(index["name"]): index for index in sa.inspect(connection).get_indexes(table_name)
+    }
+    existing = existing_indexes.get(index_name)
+    if existing is None:
+        op.create_index(
+            index_name,
+            table_name,
+            list(columns),
+            unique=unique,
+            sqlite_where=sqlite_where,
+            postgresql_where=postgresql_where,
+        )
+        return
+
+    actual_columns = tuple(str(column) for column in existing.get("column_names", ()))
+    if actual_columns != columns or bool(existing.get("unique")) is not unique:
+        raise RuntimeError(
+            f"Existing index {index_name} does not match the repair migration definition"
+        )
+
+    expected_predicate: sa.TextClause | None = None
+    reflected_predicate: object | None = None
+    dialect_options = existing.get("dialect_options", {})
+    if connection.dialect.name == "sqlite":
+        expected_predicate = sqlite_where
+        reflected_predicate = dialect_options.get("sqlite_where")
+    elif connection.dialect.name == "postgresql":
+        expected_predicate = postgresql_where
+        reflected_predicate = dialect_options.get("postgresql_where")
+    predicate_mismatch = (expected_predicate is None) != (reflected_predicate is None)
+    if expected_predicate is not None and reflected_predicate is not None:
+        predicate_mismatch = predicate_mismatch or (
+            _normalized_predicate(reflected_predicate) != _normalized_predicate(expected_predicate)
+        )
+    if predicate_mismatch:
+        raise RuntimeError(
+            f"Existing index {index_name} does not match the repair migration predicate"
+        )
+
+
+def _ensure_active_policy_is_unique() -> None:
+    policies = sa.table(
+        "governance_policies",
+        sa.column("is_active", sa.Boolean()),
+    )
+    active_policy_count = (
+        op.get_bind()
+        .execute(
+            sa.select(sa.func.count()).select_from(policies).where(policies.c.is_active.is_(True))
+        )
+        .scalar_one()
+    )
+    if active_policy_count > 1:
+        raise RuntimeError("Cannot repair governance indexes while multiple active policies exist")
+
+
+def upgrade() -> None:
+    # This data guard must run before the first repair DDL. In particular, a
+    # direct Alembic invocation must not leave the two non-unique indexes behind
+    # when restoring the partial unique policy index would be lossy.
+    _ensure_active_policy_is_unique()
+    _ensure_index(
+        "evaluation_runs",
+        "ix_evaluation_runs_started_at_id",
+        ("started_at", "id"),
+        unique=False,
+    )
+    _ensure_index(
+        "evaluation_runs",
+        "ix_evaluation_runs_finished_at_id",
+        ("finished_at", "id"),
+        unique=False,
+    )
+    _ensure_index(
+        "governance_policies",
+        "uq_governance_policies_single_active",
+        ("is_active",),
+        unique=True,
+        sqlite_where=sa.text("is_active = 1"),
+        postgresql_where=sa.text("is_active IS TRUE"),
+    )
+
+
+def downgrade() -> None:
+    """Retain indexes that are already part of the canonical revision 0005 schema."""
+
+    return None

backend/alembic/versions/20260830_0007_repair_observational_overdraw.py

@@ -0,0 +1,145 @@
+"""Repair overdraw flags created from observational input-token estimates.
+
+Revision ID: 20260830_0007
+Revises: 20260829_0006
+Create Date: 2026-08-30 00:00:00 UTC
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "20260830_0007"
+down_revision: str | None = "20260829_0006"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_ACTIVE_STATES = ("reserved", "send_started")
+_SETTLED_STATES = ("settled_actual", "settled_conservative")
+
+
+def _tables() -> tuple[sa.TableClause, sa.TableClause, sa.TableClause]:
+    scopes = sa.table(
+        "governance_scopes",
+        sa.column("id", sa.String()),
+        sa.column("overdrawn", sa.Boolean()),
+    )
+    reservations = sa.table(
+        "provider_call_reservations",
+        sa.column("state", sa.String()),
+        sa.column("run_id", sa.String()),
+        sa.column("global_scope_id", sa.String()),
+        sa.column("provider_scope_id", sa.String()),
+        sa.column("model_scope_id", sa.String()),
+        sa.column("run_scope_id", sa.String()),
+        sa.column("reserved_input_tokens", sa.BigInteger()),
+        sa.column("reserved_output_tokens", sa.BigInteger()),
+        sa.column("reserved_cost_usd", sa.Numeric(20, 8)),
+        sa.column("actual_input_tokens", sa.BigInteger()),
+        sa.column("actual_output_tokens", sa.BigInteger()),
+        sa.column("actual_cost_usd", sa.Numeric(20, 8)),
+    )
+    runs = sa.table(
+        "evaluation_runs",
+        sa.column("id", sa.String()),
+        sa.column("input_token_reservation", sa.BigInteger()),
+    )
+    return scopes, reservations, runs
+
+
+def _lock_and_require_no_active_reservations(reservations: sa.TableClause) -> None:
+    connection = op.get_bind()
+    if connection.dialect.name == "sqlite":
+        connection.exec_driver_sql("BEGIN IMMEDIATE")
+    elif connection.dialect.name == "postgresql":
+        connection.exec_driver_sql(
+            "LOCK TABLE governance_scopes, provider_call_reservations, evaluation_runs "
+            "IN ACCESS EXCLUSIVE MODE"
+        )
+    active_count = connection.scalar(
+        sa.select(sa.func.count())
+        .select_from(reservations)
+        .where(reservations.c.state.in_(_ACTIVE_STATES))
+    )
+    if int(active_count or 0) != 0:
+        raise RuntimeError(
+            "Cannot repair governance overdraw semantics while Provider reservations are active"
+        )
+
+
+def _overdraw_expression(
+    scopes: sa.TableClause,
+    reservations: sa.TableClause,
+    runs: sa.TableClause,
+    *,
+    require_explicit_input: bool,
+) -> sa.Exists:
+    scope_matches = sa.or_(
+        reservations.c.global_scope_id == scopes.c.id,
+        reservations.c.provider_scope_id == scopes.c.id,
+        reservations.c.model_scope_id == scopes.c.id,
+        reservations.c.run_scope_id == scopes.c.id,
+    )
+    explicit_input = sa.or_(
+        reservations.c.run_id.is_(None),
+        sa.exists(
+            sa.select(1)
+            .select_from(runs)
+            .where(
+                runs.c.id == reservations.c.run_id,
+                runs.c.input_token_reservation.is_not(None),
+            )
+        ),
+    )
+    input_overdraw = sa.and_(
+        reservations.c.reserved_input_tokens.is_not(None),
+        reservations.c.actual_input_tokens.is_not(None),
+        reservations.c.actual_input_tokens > reservations.c.reserved_input_tokens,
+    )
+    cost_overdraw = sa.and_(
+        reservations.c.reserved_cost_usd.is_not(None),
+        reservations.c.actual_cost_usd.is_not(None),
+        reservations.c.actual_cost_usd > reservations.c.reserved_cost_usd,
+    )
+    if require_explicit_input:
+        input_overdraw = sa.and_(explicit_input, input_overdraw)
+        cost_overdraw = sa.and_(explicit_input, cost_overdraw)
+    output_overdraw = sa.and_(
+        reservations.c.reserved_output_tokens.is_not(None),
+        reservations.c.actual_output_tokens.is_not(None),
+        reservations.c.actual_output_tokens > reservations.c.reserved_output_tokens,
+    )
+    return sa.exists(
+        sa.select(1)
+        .select_from(reservations)
+        .where(
+            reservations.c.state.in_(_SETTLED_STATES),
+            scope_matches,
+            sa.or_(input_overdraw, output_overdraw, cost_overdraw),
+        )
+    )
+
+
+def _recompute_overdrawn(*, require_explicit_input: bool) -> None:
+    scopes, reservations, runs = _tables()
+    _lock_and_require_no_active_reservations(reservations)
+    op.get_bind().execute(
+        scopes.update().values(
+            overdrawn=_overdraw_expression(
+                scopes,
+                reservations,
+                runs,
+                require_explicit_input=require_explicit_input,
+            )
+        )
+    )
+
+
+def upgrade() -> None:
+    _recompute_overdrawn(require_explicit_input=True)
+
+
+def downgrade() -> None:
+    _recompute_overdrawn(require_explicit_input=False)