perf: cut slow pages and APIs found in Sentry traces

.env.dev.example

@@ -115,3 +115,16 @@ VISUALIZER_RESULTS_DIR=./results
 #SIMPLEAUDIT_OTLP_MAX_BODY_BYTES=8388608
 #SIMPLEAUDIT_OTLP_HOLD_SECONDS=600
 #SIMPLEAUDIT_OTLP_AUTH_CACHE_SECONDS=30
+
+# --- Live streams, event retention, Open WebUI sync (defaults shown) -----------
+# Open live audit-run streams per user, and in total per web process
+# (default: half of GUNICORN_THREADS). Over the cap the stream answers 429 and
+# the browser retries.
+# SIMPLEAUDIT_SSE_MAX_STREAMS_PER_USER=3
+# SIMPLEAUDIT_SSE_MAX_STREAMS=8
+# Days after a run finishes before its per-turn progress events are deleted
+# (state and terminal events are kept). 0 keeps them forever.
+# SIMPLEAUDIT_EVENT_RETENTION_DAYS=30
+# Seconds between the Agent page's automatic Open WebUI knowledge/tool syncs
+# per user and workspace (the Sync button always runs). 0 syncs on every load.
+# SIMPLEAUDIT_OPENWEBUI_SYNC_TTL_S=60

.env.example

@@ -171,3 +171,16 @@ WORKOS_API_KEY=
 # GUNICORN_TIMEOUT=60
 # GUNICORN_GRACEFUL_TIMEOUT=30
 # GUNICORN_KEEPALIVE=5
+
+# --- Live streams, event retention, Open WebUI sync (defaults shown) -----------
+# Open live audit-run streams per user, and in total per web process
+# (default: half of GUNICORN_THREADS). Over the cap the stream answers 429 and
+# the browser retries.
+# SIMPLEAUDIT_SSE_MAX_STREAMS_PER_USER=3
+# SIMPLEAUDIT_SSE_MAX_STREAMS=8
+# Days after a run finishes before its per-turn progress events are deleted
+# (state and terminal events are kept). 0 keeps them forever.
+# SIMPLEAUDIT_EVENT_RETENTION_DAYS=30
+# Seconds between the Agent page's automatic Open WebUI knowledge/tool syncs
+# per user and workspace (the Sync button always runs). 0 syncs on every load.
+# SIMPLEAUDIT_OPENWEBUI_SYNC_TTL_S=60

accounts/services.py

@@ -129,6 +129,31 @@ def grant_default_project(user) -> Project | None:
     return project
 
 
+def request_roles(request) -> dict:
+    """``{project_id: role}`` for the request's user, loaded once per request.
+
+    Middleware, context processors and write checks all need the user's
+    memberships; one query here replaces several per page.
+    """
+    cached = getattr(request, "_studio_roles", None)
+    if cached is None:
+        user = getattr(request, "user", None)
+        cached = (
+            dict(ProjectMembership.objects.filter(user=user).values_list("project_id", "role"))
+            if user is not None and user.is_authenticated else {}
+        )
+        request._studio_roles = cached
+    return cached
+
+
+def request_has_access(request, project) -> bool:
+    """``ensure_project_access`` for the request's user, from ``request_roles``."""
+    user = getattr(request, "user", None)
+    if not user or not user.is_authenticated:
+        return False
+    return project.slug == DEFAULT_PROJECT_SLUG or project.pk in request_roles(request)
+
+
 def ensure_project_access(user, project) -> bool:
     """Return True if the user may view this project's content.
 

audits/comparison.py

@@ -121,23 +121,30 @@ def compare_runs(project, run_ids: list[int]) -> dict:
     from scenarios.models import ScenarioSetVersionItem
 
     run_results: dict[int, dict[str, dict]] = {}
+    item_maps: dict[int, dict[str, str]] = {}   # runs often share a pinned version
     for run in runs:
         # Map version_item_id (str) -> scenario key for this run's pinned version.
         item_map: dict[str, str] = {}
         if run.scenario_set_version_id:
-            items = ScenarioSetVersionItem.objects.filter(
-                version_id=run.scenario_set_version_id
-            ).select_related("scenario")
-            for item in items:
-                item_map[str(item.id)] = item.scenario.key
+            if run.scenario_set_version_id not in item_maps:
+                item_maps[run.scenario_set_version_id] = {
+                    str(item_id): key
+                    for item_id, key in ScenarioSetVersionItem.objects.filter(
+                        version_id=run.scenario_set_version_id
+                    ).values_list("id", "scenario__key")
+                }
+            item_map = item_maps[run.scenario_set_version_id]
 
-        rows = ScenarioResult.objects.filter(run_id=str(run.id))
+        # The small list_summary projection carries every field used here.
+        rows = ScenarioResult.objects.filter(run_id=run.id).only(
+            "version_item_id", "status", "attempts", "list_summary"
+        )
         m = {}
         for row in rows:
             key = item_map.get(row.version_item_id, f"item_{row.version_item_id}")
-            r_data = row.result or {}
+            r_data = row.list_summary or {}
             # For repeated results, use aggregated severity
-            if "reps" in r_data and isinstance(r_data.get("reps"), list):
+            if r_data.get("has_reps"):
                 sev = r_data.get("aggregated_severity")
                 agree = r_data.get("agreement_rate")
                 summary = f"{sev} ({int(agree*100)}% agreement)" if agree is not None else sev

audits/events.py

@@ -23,9 +23,9 @@
 class AuditEvent(models.Model):
     """Append-only durable progress event for an audit run."""
 
-    run_id = models.PositiveBigIntegerField(db_index=True)
-    version_item_id = models.CharField(max_length=64, db_index=True)
-    kind = models.CharField(max_length=64, db_index=True)
+    run_id = models.PositiveBigIntegerField()
+    version_item_id = models.CharField(max_length=64)
+    kind = models.CharField(max_length=64)
     payload = models.JSONField(default=dict, blank=True)
     created_at = models.DateTimeField(auto_now_add=True)
 
@@ -34,12 +34,68 @@ class Meta:
         ordering = ["id"]
         indexes = [
             models.Index(fields=["run_id", "id"], name="audit_event_run_id_idx"),
+            # Per-scenario state/attempt lookups and the progress snapshot group-by.
+            models.Index(fields=["run_id", "kind", "version_item_id"], name="audit_event_run_kind_item_idx"),
         ]
 
     def __str__(self) -> str:
         return f"AuditEvent(run={self.run_id} {self.kind} vi={self.version_item_id})"
 
 
+def trial_severities(result: dict) -> list:
+    """Every judged trial in one stored result (one per repetition)."""
+    reps = result.get("reps")
+    if isinstance(reps, list):
+        return [(r or {}).get("severity") for r in reps]
+    return [result.get("severity")]
+
+
+def trial_counts(status: str, result: dict | None) -> dict[str, int]:
+    """Per-row trial tallies kept beside the (large) ``result`` JSON.
+
+    Repetitions count as separate trials. ERROR severities and non-completed
+    scenarios are excluded from ``n_judged`` and counted in ``n_errors``;
+    "ungraded" trials (a judgment without a verdict) are counted apart.
+    """
+    out = {"n_pass": 0, "n_judged": 0, "n_errors": 0, "n_ungraded": 0}
+    if status != "completed":
+        out["n_errors"] = 1
+        return out
+    for sev in trial_severities(result or {}):
+        if not sev or sev == "ERROR":
+            out["n_errors"] += 1
+        elif sev == "ungraded":
+            out["n_ungraded"] += 1
+        else:
+            out["n_judged"] += 1
+            if sev == "pass":
+                out["n_pass"] += 1
+    return out
+
+
+def result_projection(result: dict | None) -> dict:
+    """The few fields the run's Results list shows, lifted out of the big ``result``.
+
+    Transcripts dominate a stored result; the list only needs verdicts and short
+    summaries, so ``ScenarioResult.list_summary`` keeps just those.
+    """
+    r = result if isinstance(result, dict) else {}
+    reps = r.get("reps")
+    return {
+        "severity": r.get("severity"),
+        "aggregated_severity": r.get("aggregated_severity"),
+        "summary": r.get("summary", ""),
+        "agreement_rate": r.get("agreement_rate"),
+        "severity_distribution": r.get("severity_distribution"),
+        "has_reps": isinstance(reps, list),
+        "reps": (
+            [{"severity": x.get("severity"), "summary": x.get("summary", "")} for x in reps if isinstance(x, dict)]
+            if isinstance(reps, list) else []
+        ),
+        "agentic_status": (r.get("agentic_evaluation") or {}).get("status", ""),
+    }
+
+
 class ScenarioResult(models.Model):
     """Idempotent final result for one scenario in one run."""
 
@@ -51,8 +107,29 @@ class ScenarioResult(models.Model):
     # judgment, ...). Large raw transcripts may instead be offloaded to object
     # storage; this column holds the structured result record.
     result = models.JSONField(default=dict, blank=True)
+    # Trial tallies derived from ``status`` + ``result`` (see ``trial_counts``).
+    # They let run totals be a SUM over small columns instead of loading every
+    # transcript; ``save()`` keeps them in step with the JSON.
+    n_pass = models.PositiveIntegerField(default=0)
+    n_judged = models.PositiveIntegerField(default=0)
+    n_errors = models.PositiveIntegerField(default=0)
+    n_ungraded = models.PositiveIntegerField(default=0)
+    # Small projection of ``result`` for list pages (see ``result_projection``).
+    list_summary = models.JSONField(default=dict, blank=True)
     updated_at = models.DateTimeField(auto_now=True)
 
+    COUNT_FIELDS = ("n_pass", "n_judged", "n_errors", "n_ungraded", "list_summary")
+
+    def save(self, *args, **kwargs):
+        update_fields = kwargs.get("update_fields")
+        if update_fields is None or {"status", "result"} & set(update_fields):
+            for name, value in trial_counts(self.status, self.result).items():
+                setattr(self, name, value)
+            self.list_summary = result_projection(self.result)
+            if update_fields is not None:
+                kwargs["update_fields"] = [*update_fields, *self.COUNT_FIELDS]
+        super().save(*args, **kwargs)
+
     class Meta:
         db_table = "core_scenario_result"
         constraints = [
@@ -61,6 +138,9 @@ class Meta:
                 name="unique_scenario_result_per_run_item",
             ),
         ]
+        indexes = [
+            models.Index(fields=["run_id", "status"], name="scenario_result_run_status_idx"),
+        ]
 
     def __str__(self) -> str:
         return f"ScenarioResult(run={self.run_id} vi={self.version_item_id} {self.status})"
@@ -78,6 +158,50 @@ def append_event(run_id: int | str, version_item_id: str, kind: str, payload: di
     return event.id
 
 
+@retry_if_locked
+def append_events(run_id: int | str, version_item_id: str, items: list[tuple[str, dict | None]]) -> None:
+    """Append several events in one INSERT, keeping their order (ids ascend)."""
+    if not items:
+        return
+    AuditEvent.objects.bulk_create([
+        AuditEvent(run_id=int(run_id), version_item_id=version_item_id, kind=kind, payload=payload or {})
+        for kind, payload in items
+    ])
+
+
+# Per-turn progress events: by far the most numerous rows, and only useful
+# while a run is live (the progress view and "Latest Activity"). State and
+# terminal events are kept for good.
+PRUNABLE_EVENT_KINDS = ("scenario_turn", "scenario_rep_started")
+
+
+def prune_finished_run_events(older_than_days: int, batch_size: int = 5000) -> int:
+    """Delete per-turn events of runs that finished more than ``older_than_days`` ago.
+
+    Deletes in id batches so a large backlog never holds one long transaction.
+    Returns the number of rows deleted.
+    """
+    from datetime import timedelta
+
+    from django.utils import timezone
+
+    from audits.models import AuditRun
+
+    if older_than_days <= 0:
+        return 0
+    cutoff = timezone.now() - timedelta(days=older_than_days)
+    finished = AuditRun.objects.filter(
+        status__in=AuditRun.TERMINAL_STATUSES, finished_at__lt=cutoff
+    ).values("id")
+    qs = AuditEvent.objects.filter(run_id__in=finished, kind__in=PRUNABLE_EVENT_KINDS)
+    deleted = 0
+    while True:
+        ids = list(qs.order_by().values_list("id", flat=True)[:batch_size])
+        if not ids:
+            return deleted
+        deleted += AuditEvent.objects.filter(id__in=ids).delete()[0]
+
+
 def _event_dict(e: AuditEvent) -> dict:
     return {
         "id": e.id,

audits/experiments.py

@@ -647,7 +647,12 @@ def prefetch_experiment_runs():
     """Prefetch for Experiment querysets whose pages show factors and summaries."""
     from django.db.models import Prefetch
 
-    return Prefetch("runs", queryset=AuditRun.objects.select_related(*RUN_FACTOR_RELATED))
+    # Factors and summaries read the model/judge snapshots and generation
+    # parameters; the runtime and trace blobs are never shown on these pages.
+    return Prefetch(
+        "runs",
+        queryset=AuditRun.objects.select_related(*RUN_FACTOR_RELATED).defer("runtime_metadata", "trace_config"),
+    )
 
 
 class MembershipError(ValueError):

audits/migrations/0011_scenario_result_trial_counts.py

@@ -0,0 +1,99 @@
+# Generated by Django 5.2.4 on 2026-11-11 16:38
+
+from django.db import migrations, models
+
+
+def _trial_counts(status, result):
+    """Frozen copy of ``audits.events.trial_counts`` (migrations must not import app code)."""
+    out = {"n_pass": 0, "n_judged": 0, "n_errors": 0, "n_ungraded": 0}
+    if status != "completed":
+        out["n_errors"] = 1
+        return out
+    result = result or {}
+    reps = result.get("reps")
+    severities = (
+        [(r or {}).get("severity") for r in reps]
+        if isinstance(reps, list)
+        else [result.get("severity")]
+    )
+    for sev in severities:
+        if not sev or sev == "ERROR":
+            out["n_errors"] += 1
+        elif sev == "ungraded":
+            out["n_ungraded"] += 1
+        else:
+            out["n_judged"] += 1
+            if sev == "pass":
+                out["n_pass"] += 1
+    return out
+
+
+def _projection(result):
+    """Frozen copy of ``audits.events.result_projection``."""
+    r = result if isinstance(result, dict) else {}
+    reps = r.get("reps")
+    return {
+        "severity": r.get("severity"),
+        "aggregated_severity": r.get("aggregated_severity"),
+        "summary": r.get("summary", ""),
+        "agreement_rate": r.get("agreement_rate"),
+        "severity_distribution": r.get("severity_distribution"),
+        "has_reps": isinstance(reps, list),
+        "reps": (
+            [{"severity": x.get("severity"), "summary": x.get("summary", "")} for x in reps if isinstance(x, dict)]
+            if isinstance(reps, list) else []
+        ),
+        "agentic_status": (r.get("agentic_evaluation") or {}).get("status", ""),
+    }
+
+
+def backfill_trial_counts(apps, schema_editor):
+    """Fill the new columns for existing rows, streaming so big transcripts never pile up."""
+    ScenarioResult = apps.get_model("audits", "ScenarioResult")
+    fields = ["n_pass", "n_judged", "n_errors", "n_ungraded", "list_summary"]
+    batch = []
+    for row in ScenarioResult.objects.only("id", "status", "result").iterator(chunk_size=200):
+        for name, value in _trial_counts(row.status, row.result).items():
+            setattr(row, name, value)
+        row.list_summary = _projection(row.result)
+        batch.append(row)
+        if len(batch) >= 500:
+            ScenarioResult.objects.bulk_update(batch, fields)
+            batch = []
+    if batch:
+        ScenarioResult.objects.bulk_update(batch, fields)
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("audits", "0010_experiment_run_membership"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="scenarioresult",
+            name="n_errors",
+            field=models.PositiveIntegerField(default=0),
+        ),
+        migrations.AddField(
+            model_name="scenarioresult",
+            name="n_judged",
+            field=models.PositiveIntegerField(default=0),
+        ),
+        migrations.AddField(
+            model_name="scenarioresult",
+            name="n_pass",
+            field=models.PositiveIntegerField(default=0),
+        ),
+        migrations.AddField(
+            model_name="scenarioresult",
+            name="n_ungraded",
+            field=models.PositiveIntegerField(default=0),
+        ),
+        migrations.AddField(
+            model_name="scenarioresult",
+            name="list_summary",
+            field=models.JSONField(blank=True, default=dict),
+        ),
+        migrations.RunPython(backfill_trial_counts, migrations.RunPython.noop),
+    ]