From aea46f6f8c8c00d943d0520b0d52d2bd18576704 Mon Sep 17 00:00:00 2001 From: Sushant Gautam Date: Mon, 9 Nov 2026 10:58:34 +0100 Subject: [PATCH 1/5] Show which role (target, auditor, judge, system) failed a scenario Store a structured failure object (role required; kind, http_status, endpoint, model, retryable, attempt best effort) next to the raw error. Role comes from the engine's error_phase plus the last on_turn phase. A semantic-judge endpoint failure is now a failed scenario, not UNCERTAIN. Per-role counts in summary_metrics, run-level message, role badge, summary and filter in the UI. Results without the structure render as before. Refs #85 --- audits/agentic/semantic_judge.py | 3 + audits/failures.py | 250 +++++++++++++++++++++++ docs/domain-model.md | 17 +- infra/tests/test_failure_roles.py | 253 ++++++++++++++++++++++++ infra/ui.py | 46 ++++- infra/worker.py | 93 ++++++++- templates/partials/failure_badge.html | 1 + templates/partials/run_detail_body.html | 26 +++ templates/partials/run_result_body.html | 10 +- templates/partials/run_results.html | 20 +- 10 files changed, 705 insertions(+), 14 deletions(-) create mode 100644 audits/failures.py create mode 100644 infra/tests/test_failure_roles.py create mode 100644 templates/partials/failure_badge.html diff --git a/audits/agentic/semantic_judge.py b/audits/agentic/semantic_judge.py index a773bf60..1164af65 100644 --- a/audits/agentic/semantic_judge.py +++ b/audits/agentic/semantic_judge.py @@ -407,6 +407,9 @@ def run_semantic_judge( "span_ids": [], } result["rationale"] = UNCERTAIN_REASON_LLM_UNAVAILABLE + # The judge call itself failed (not a verdict): the worker records this + # as a judge failure instead of letting UNCERTAIN pass for an answer. + result["llm_error"] = f"{type(exc).__name__}: {exc}" return result dimensions = parse_semantic_judgement(raw, criteria) diff --git a/audits/failures.py b/audits/failures.py new file mode 100644 index 00000000..3bf23c9e --- /dev/null +++ b/audits/failures.py @@ -0,0 +1,250 @@ +"""Which role failed a scenario, and why (structured, best effort). + +A failed scenario stores a ``failure`` object next to the raw error text:: + + {"role": "target|auditor|judge|system", # required + "kind": "rate_limit|unavailable|auth|timeout|bad_response|config|unknown", + "message": "raw text", "http_status": 429, "endpoint": "...", + "model": "...", "retryable": true, "attempt": 2} + +``role`` is the only required field; the rest is derived from the error text +and the frozen endpoint snapshot, and left out when unknown. Nothing here +assumes a target type or provider. Results stored without ``failure`` (older +runs) have no structure and render as before. +""" + +from __future__ import annotations + +import re +from typing import Any + +ROLES = ("target", "auditor", "judge", "system") +ROLE_LABELS = {"target": "Target", "auditor": "Auditor", "judge": "Judge", "system": "System"} +KINDS = ("rate_limit", "unavailable", "auth", "timeout", "bad_response", "config", "unknown") +KIND_LABELS = { + "rate_limit": "rate limited", + "unavailable": "unavailable", + "auth": "auth rejected", + "timeout": "timeout", + "bad_response": "bad response", + "config": "misconfigured", + "unknown": "", +} +_RETRYABLE = {"rate_limit", "unavailable", "timeout", "bad_response"} + +# Exception class names that imply a status when the message carries no number. +_NAME_STATUS = ( + ("RateLimit", 429), ("AuthenticationError", 401), ("PermissionDenied", 403), + ("ServiceUnavailable", 503), ("InternalServerError", 500), ("BadGateway", 502), +) +_STATUS_RE = re.compile( + r"(?:status|http|code|error)\W{0,4}([45]\d\d)\b" + r"|\b([45]\d\d)\s+(?:too many|internal|bad gateway|service|gateway|unauthorized|forbidden|not found)", + re.IGNORECASE, +) + + +def _http_status(text: str) -> int | None: + match = _STATUS_RE.search(text) + if match: + return int(match.group(1) or match.group(2)) + for name, status in _NAME_STATUS: + if name in text: + return status + return None + + +def classify_kind(text: str, http_status: int | None = None) -> str: + """Provider-agnostic cause from an HTTP status or the error wording.""" + low = (text or "").lower() + if http_status == 429 or "ratelimit" in low or "rate limit" in low or "rate_limit" in low or "too many requests" in low: + return "rate_limit" + if http_status in (401, 403) or any(w in low for w in ("authentication", "unauthorized", "forbidden", "invalid api key", "permission denied")): + return "auth" + if http_status == 408 or http_status == 504 or "timeout" in low or "timed out" in low: + return "timeout" + if "is not set" in low and "environment variable" in low: + return "config" + if http_status is not None and http_status >= 500 or any( + w in low for w in ("unavailable", "connection", "overloaded", "bad gateway", "internal server error", "temporarily", "unreachable", "name resolution") + ): + return "unavailable" + if any(w in low for w in ("could not parse", "no choices", "empty response", "successful (http 200)", "returned an error", "no message", "parse")): + return "bad_response" + if http_status is not None and 400 <= http_status < 500: + return "config" + return "unknown" + + +def build_failure( + role: str, + message: str, + *, + endpoint: str | None = None, + model: str | None = None, + attempt: int | None = None, + kind: str | None = None, + http_status: int | None = None, +) -> dict[str, Any]: + """A failure object; ``role`` is required, everything else only when known.""" + role = role if role in ROLES else "system" + message = str(message or "") + status = http_status if http_status is not None else _http_status(message) + kind = kind if kind in KINDS else classify_kind(message, status) + failure: dict[str, Any] = {"role": role, "kind": kind, "message": message} + if status is not None: + failure["http_status"] = status + if endpoint: + failure["endpoint"] = endpoint + if model: + failure["model"] = model + if kind != "unknown": + failure["retryable"] = kind in _RETRYABLE + if attempt is not None: + failure["attempt"] = attempt + return failure + + +def endpoint_of(snapshot: dict | None) -> dict[str, str]: + """Endpoint/model of a frozen role snapshot, only the parts it records.""" + snap = snapshot if isinstance(snapshot, dict) else {} + out = {} + url = snap.get("base_url") or "" + # Never persist credentials that ride in a URL. + if url: + out["endpoint"] = re.sub(r"//[^/@]*@", "//", str(url)) + model = snap.get("model_id") or "" + if model: + out["model"] = str(model) + return out + + +class PhaseTracker: + """Last phase that completed, per repetition, from the engine's ``on_turn``. + + The engine fires ``on_turn`` after a phase returns and fires nothing for a + phase that fails, so the failed phase is the one after the last that + completed: ``auditor`` -> ``target`` -> next turn's ``auditor``. + """ + + def __init__(self) -> None: + self._last: dict[int, tuple[int, int, str]] = {} + + def note(self, rep: int, turn: int, max_turns: int, role: str) -> None: + self._last[rep] = (turn, max_turns, role) + + def failed_role(self, rep: int, error_phase: str | None, *, probe_is_fixed: bool = False) -> str: + """Role that failed in repetition ``rep``; ``system`` when not attributable.""" + if error_phase == "judge": + return "judge" + last = self._last.get(rep) + if last is None: + # Nothing completed: the first call. A fixed test prompt skips the auditor. + return "target" if probe_is_fixed else "auditor" + turn, max_turns, role = last + if role == "auditor": + return "target" + if role == "target": + return "auditor" if turn < max_turns - 1 else "system" + return "system" + + +def attach_failures( + payload: dict, + tracker: PhaseTracker, + snapshots: dict[str, dict | None], + *, + attempt: int | None = None, + probe_is_fixed: bool = False, + reps: bool = False, +) -> dict | None: + """Stamp ``failure`` onto an engine result (and each ERROR rep); return the first. + + Mutates ``payload``. ``snapshots`` maps role -> frozen endpoint snapshot. + A semantic judge failure recorded by the worker (``failure`` already set) + is left alone. + """ + first = None + + def one(item: dict, rep: int) -> dict | None: + if str(item.get("severity", "")).upper() != "ERROR": + return None + if isinstance(item.get("failure"), dict): + return item["failure"] + judgment = item.get("judgment") if isinstance(item.get("judgment"), dict) else {} + message = str(judgment.get("error") or item.get("error") or item.get("summary") or "") + role = tracker.failed_role(rep, judgment.get("error_phase"), probe_is_fixed=probe_is_fixed) + failure = build_failure(role, message, attempt=attempt, **endpoint_of(snapshots.get(role))) + item["failure"] = failure + return failure + + rep_items = payload.get("reps") + if isinstance(rep_items, list): + for index, rep in enumerate(rep_items): + if isinstance(rep, dict): + got = one(rep, index) + first = first or got + if first is not None: + payload["failure"] = first + else: + first = one(payload, 0) + return first + + +def result_failure(payload: Any) -> dict | None: + """The stored structured failure of a result, or None (old results).""" + payload = payload if isinstance(payload, dict) else {} + failure = payload.get("failure") + if isinstance(failure, dict) and failure.get("role") in ROLES: + return failure + for rep in payload.get("reps") or []: + failure = rep.get("failure") if isinstance(rep, dict) else None + if isinstance(failure, dict) and failure.get("role") in ROLES: + return failure + return None + + +def failure_label(failure: dict | None) -> str: + """``"Target · rate limited"`` style badge text.""" + if not failure: + return "" + label = ROLE_LABELS.get(failure.get("role"), "System") + cause = KIND_LABELS.get(failure.get("kind") or "", "") + return f"{label} · {cause}" if cause else label + + +def failure_where(failure: dict | None) -> str: + """Model if known, else the endpoint; never an assumed model.""" + if not failure: + return "" + return failure.get("model") or failure.get("endpoint") or "" + + +def count_by_role(results: list[dict] | Any) -> dict[str, int]: + """Failures per role over stored result payloads that failed (role-less ones skipped).""" + counts: dict[str, int] = {} + for payload in results: + failure = result_failure(payload) + if failure: + counts[failure["role"]] = counts.get(failure["role"], 0) + 1 + return counts + + +def summary_text(counts: dict[str, int], total_failed: int) -> str: + """``"47 failed: 44 target, 3 judge"``; just the total when roles are unknown.""" + parts = [f"{counts[r]} {r}" for r in ROLES if counts.get(r)] + return f"{total_failed} failed: {', '.join(parts)}" if parts else f"{total_failed} failed" + + +_SECRET_ROLE_RE = re.compile(r"Secret for (target|auditor|judge) endpoint", re.IGNORECASE) + + +def engine_error_failure(message: str, snapshots: dict[str, dict | None], *, attempt: int | None = None) -> dict: + """Failure for an ``EngineError`` raised before/around the conversation. + + These are setup or crash errors that belong to no single call, so ``system``, + unless the message names the endpoint whose configuration is broken. + """ + match = _SECRET_ROLE_RE.search(message or "") + role = match.group(1).lower() if match else "system" + return build_failure(role, message, attempt=attempt, **endpoint_of(snapshots.get(role))) diff --git a/docs/domain-model.md b/docs/domain-model.md index 975d8325..856d9456 100644 --- a/docs/domain-model.md +++ b/docs/domain-model.md @@ -310,7 +310,7 @@ Fields: - `successful_scenarios` - `failed_scenarios` - `retried_scenarios` -- `summary_metrics` — JSON +- `summary_metrics` — JSON (`failures_by_role`: failed scenarios per role, e.g. `{"target": 44, "judge": 3}`) - `error_code` - `error_message` - experiments — many-to-many through `ExperimentRun` (a run can belong to several) @@ -523,3 +523,18 @@ The platform must preserve existing SimpleAudit semantics: - result aggregation Any change to these semantics needs regression tests against existing expected outputs. + + +## Scenario failure attribution + +A failed `ScenarioResult` stores a structured `failure` object beside the raw +error text (top level of the result, and on each failed repetition). `role` +(`target` | `auditor` | `judge` | `system`) is the only required field; `kind` +(`rate_limit`, `unavailable`, `auth`, `timeout`, `bad_response`, `config`, +`unknown`), `message`, `http_status`, `endpoint`, `model`, `retryable` and +`attempt` are best effort and omitted when unknown. The role comes from the +engine's `error_phase` plus the last phase the engine reported through +`on_turn`, so it needs no target cooperation; the model is never assumed (the +endpoint is shown when the model is unknown). A judge endpoint failure is a +failed scenario, not an `UNCERTAIN` verdict. Results stored without `failure` +render as before. Code: `audits/failures.py`. diff --git a/infra/tests/test_failure_roles.py b/infra/tests/test_failure_roles.py new file mode 100644 index 00000000..cf9535d0 --- /dev/null +++ b/infra/tests/test_failure_roles.py @@ -0,0 +1,253 @@ +"""Which role failed a scenario: classification, worker wiring and UI. + +Role is the only required field of the stored ``failure`` object; kind, +http_status, endpoint, model and retryable are best effort. +""" +import json + +import pytest +from django.test import Client, TestCase + +from audits.failures import ( + PhaseTracker, + attach_failures, + build_failure, + count_by_role, + engine_error_failure, + result_failure, + summary_text, +) +from infra.tests.factories import ( + AuditRunFactory, + MembershipFactory, + ProjectFactory, + ScenarioResultFactory, + UserFactory, +) +from infra.tests.test_agentic_semantic_judge_lane import ( + _rep, + _semantic_metadata, + _spans, + _workflow_input, +) +from infra.tests.test_agentic_semantic_judge_lane import ( + patched_worker as _patched_worker, +) +from infra.worker import _scenario_execute_impl + +patched_worker = _patched_worker # re-exported so pytest finds the fixture + +# What each failure looks like as the engine records it: ": ". +ERRORS = { + "429": ("RateLimitError: Error code: 429 - Too Many Requests", "rate_limit", 429), + "5xx": ("APIError: The language model provider is unavailable (status 503)", "unavailable", 503), + "timeout": ("APITimeoutError: Request timed out.", "timeout", None), + "connection": ("APIConnectionError: Connection error.", "unavailable", None), +} +SNAPSHOTS = { + "target": {"base_url": "http://target/v1", "model_id": ""}, + "auditor": {"base_url": "http://auditor/v1", "model_id": "aud-1"}, + "judge": {"base_url": "http://judge/v1", "model_id": "jud-1"}, +} + + +def _error_result(message, phase): + return { + "severity": "ERROR", + "summary": f"Scenario did not complete due to an error: {message}", + "judgment": {"severity": "ERROR", "error": message, "error_phase": phase}, + } + + +def _tracker_for(role): + """A tracker whose engine callbacks stopped right before ``role`` ran.""" + tracker = PhaseTracker() + if role == "target": + tracker.note(0, 0, 3, "auditor") + elif role == "auditor": + tracker.note(0, 0, 3, "target") + return tracker + + +@pytest.mark.parametrize("role", ["target", "auditor", "judge"]) +@pytest.mark.parametrize("name", list(ERRORS)) +def test_role_and_kind_for_each_failure(role, name): + message, kind, status = ERRORS[name] + payload = _error_result(message, "judge" if role == "judge" else "conversation") + failure = attach_failures(payload, _tracker_for(role), SNAPSHOTS, attempt=2) + assert failure["role"] == role + assert failure["kind"] == kind + assert failure.get("http_status") == status + assert failure["message"] == message # raw text kept + assert failure["attempt"] == 2 + assert failure["endpoint"] == SNAPSHOTS[role]["base_url"] + assert payload["failure"] is failure + + +def test_model_only_when_known_never_assumed(): + payload = _error_result(ERRORS["429"][0], "conversation") + failure = attach_failures(payload, _tracker_for("target"), SNAPSHOTS) + assert "model" not in failure and failure["endpoint"] == "http://target/v1" + judge = attach_failures(_error_result("x", "judge"), PhaseTracker(), SNAPSHOTS) + assert judge["model"] == "jud-1" + + +def test_role_is_the_only_required_field(): + failure = build_failure("target", "something odd") + assert failure == {"role": "target", "kind": "unknown", "message": "something odd"} + assert build_failure("nonsense", "x")["role"] == "system" + + +def test_fixed_test_prompt_skips_auditor(): + payload = _error_result("APIConnectionError: Connection error.", "conversation") + failure = attach_failures(payload, PhaseTracker(), SNAPSHOTS, probe_is_fixed=True) + assert failure["role"] == "target" + + +def test_repeated_result_tags_each_error_rep(): + tracker = PhaseTracker() + tracker.note(1, 0, 3, "auditor") + payload = {"reps": [{"severity": "pass"}, _error_result(ERRORS["5xx"][0], "conversation")]} + attach_failures(payload, tracker, SNAPSHOTS) + assert "failure" not in payload["reps"][0] + assert payload["reps"][1]["failure"]["role"] == "target" + assert result_failure(payload)["role"] == "target" + + +@pytest.mark.parametrize("name", list(ERRORS)) +def test_engine_crash_is_system(name): + failure = engine_error_failure(f"Scenario execution crashed: {ERRORS[name][0]}", SNAPSHOTS) + assert failure["role"] == "system" + + +def test_missing_secret_names_the_endpoint_role(): + failure = engine_error_failure( + "Secret for judge endpoint is not set: environment variable 'K' is missing or empty.", SNAPSHOTS + ) + assert (failure["role"], failure["kind"], failure["retryable"]) == ("judge", "config", False) + + +def test_summary_and_counts(): + results = [{"failure": {"role": "target"}}] * 44 + [{"failure": {"role": "judge"}}] * 3 + [{"error": "old"}] + counts = count_by_role(results) + assert counts == {"target": 44, "judge": 3} + assert summary_text(counts, 47) == "47 failed: 44 target, 3 judge" + assert summary_text({}, 5) == "5 failed" + + +# --- worker -------------------------------------------------------------------- + +@pytest.fixture +def engine_patch(monkeypatch, patched_worker): + import infra.engine as engine_module + + def drive(phases, result): + def fake(**kwargs): + for turn, role in phases: + kwargs["on_turn"](turn, 3, role) + return result + monkeypatch.setattr(engine_module, "run_scenario", fake) + return drive + + +@pytest.mark.parametrize( + ("phases", "phase", "role"), + [ + ([(0, "auditor")], "conversation", "target"), + ([(0, "auditor"), (0, "target")], "conversation", "auditor"), + ([(0, "auditor"), (0, "target"), (2, "target")], "judge", "judge"), + ], +) +def test_worker_stores_role_and_counts_it(engine_patch, patched_worker, phases, phase, role): + engine_patch(phases, _error_result(ERRORS["429"][0], phase)) + with pytest.raises(Exception, match="attempt 1 failed"): # provisional: Hatchet retries + _scenario_execute_impl(_workflow_input(), ctx=None) + stored = patched_worker["upserts"][0][1] + assert stored["status"] == "failed" + assert stored["result"]["failure"]["role"] == role + event = next(e for e in patched_worker["events"] if e[0][2] == "scenario_failed") + assert event[0][3]["failure"]["role"] == role + + +def test_judge_endpoint_failure_is_failed_not_uncertain(monkeypatch, patched_worker): + """The semantic judge's endpoint raising must not surface as UNCERTAIN.""" + import infra.worker as worker_module + + def broken(prompt): + raise RuntimeError("RateLimitError: Error code: 429") + + monkeypatch.setattr(worker_module, "_build_semantic_judge_llm", lambda snap: broken) + patched_worker["revision_metadata"] = _semantic_metadata() + patched_worker["rep"] = _rep(trace_evidence={"all_spans": _spans()}) + + with pytest.raises(Exception, match="attempt 1 failed"): + _scenario_execute_impl(_workflow_input(), ctx=None) + + stored = patched_worker["upserts"][0][1] + assert stored["status"] == "failed" + result = stored["result"] + assert result["severity"] == "ERROR" + assert result["failure"]["role"] == "judge" and result["failure"]["kind"] == "rate_limit" + assert result["failure"]["http_status"] == 429 + + +# --- UI ------------------------------------------------------------------------ + +class FailureUiTests(TestCase): + def setUp(self): + user = UserFactory() + user.set_password("pw") + user.save() + self.project = ProjectFactory() + MembershipFactory(user=user, project=self.project, role="admin") + self.client = Client() + self.client.login(username=user.username, password="pw") + self.run = AuditRunFactory(project=self.project, status="completed") + + def _fail(self, role, kind="rate_limit", message="RateLimitError: 429", **extra): + failure = build_failure(role, message, kind=kind, **extra) + return ScenarioResultFactory( + run_id=self.run.id, status="failed", + result={"severity": "ERROR", "summary": message, "failure": failure, + "judgment": {"error": message}}, + ) + + def test_summary_badges_and_filter_buttons(self): + for _ in range(2): + self._fail("target") + self._fail("judge", kind="unavailable", message="APIError: provider is unavailable") + html = self.client.get(f"/runs/{self.run.id}/results-fragment/").content.decode() + assert "3 failed: 2 target, 1 judge" in html + assert html.count('data-testid="failure-badge"') == 3 + assert "Target · rate limited" in html and "Judge · unavailable" in html + assert 'data-role-filter="judge"' in html and 'data-failure-role="target"' in html + + def test_old_result_without_structure_renders_as_before(self): + ScenarioResultFactory( + run_id=self.run.id, status="failed", + result={"severity": "ERROR", "summary": "old failure text", "error": "boom"}, + ) + html = self.client.get(f"/runs/{self.run.id}/results-fragment/").content.decode() + assert "old failure text" in html + assert "failure-badge" not in html and "role-filter" not in html and "Filter:" not in html + + def test_detail_shows_badge_and_raw_message(self): + row = self._fail("auditor", message="APITimeoutError: Request timed out.", kind="timeout", + endpoint="http://auditor/v1") + html = self.client.get(f"/runs/{self.run.id}/results/{row.pk}/").content.decode() + assert 'data-testid="failure-detail"' in html + assert "Auditor · timeout" in html and "http://auditor/v1" in html + assert "APITimeoutError: Request timed out." in html + + def test_old_result_detail_has_no_badge(self): + row = ScenarioResultFactory( + run_id=self.run.id, status="failed", result={"severity": "ERROR", "error": "boom"}, + ) + html = self.client.get(f"/runs/{self.run.id}/results/{row.pk}/").content.decode() + assert "Scenario failed: boom" in html and "failure-detail" not in html + + +def test_result_is_json_serialisable(): + payload = _error_result(ERRORS["429"][0], "judge") + attach_failures(payload, PhaseTracker(), SNAPSHOTS) + json.dumps(payload) diff --git a/infra/ui.py b/infra/ui.py index e0ed6bcb..40f7bb5a 100644 --- a/infra/ui.py +++ b/infra/ui.py @@ -32,6 +32,14 @@ from audits.agentic.trajectory import inside_retrieval from audits.comparison import compare_runs from audits.events import ScenarioResult +from audits.failures import ( + ROLE_LABELS, + ROLES, + failure_label, + failure_where, + result_failure, + summary_text, +) from audits.models import AuditRun from audits.services import ( _repeated_export_entry, @@ -3504,7 +3512,12 @@ def _result_rows(run: AuditRun, items: dict | None = None) -> list[dict]: severity = r.get("severity", sr.status) agreement = None sev_dist = None + failure = result_failure(r) if sr.status == "failed" else None results.append({ + "failure": failure, + "failure_role": failure["role"] if failure else "", + "failure_label": failure_label(failure), + "failure_where": failure_where(failure), "result_id": sr.pk, "version_item_id": sr.version_item_id, "scenario_name": item.scenario.title if item else sr.version_item_id, @@ -3522,6 +3535,25 @@ def _result_rows(run: AuditRun, items: dict | None = None) -> list[dict]: return results +def _results_context(run: AuditRun, items: dict | None = None) -> dict: + results = _result_rows(run, items) + return {"results": results, "failure_summary": _failure_summary(results)} + + +def _failure_summary(results: list[dict]) -> dict: + """Failed-scenario summary for the Results list: ``47 failed: 44 target, 3 judge``.""" + failed = [r for r in results if r["status"] == "failed"] + counts: dict[str, int] = {} + for r in failed: + if r["failure_role"]: + counts[r["failure_role"]] = counts.get(r["failure_role"], 0) + 1 + return { + "total": len(failed), + "text": summary_text(counts, len(failed)) if failed else "", + "roles": [{"role": role, "label": ROLE_LABELS[role], "count": counts[role]} for role in ROLES if counts.get(role)], + } + + class RunResultsFragmentView(ProjectMixin, View): """GET /runs//results-fragment/ — the Results list alone, for live refresh.""" @@ -3532,7 +3564,7 @@ def get(self, request, run_id): AuditRun.objects.select_related("scenario_set_version"), pk=run_id, project=request.project ) html = render_to_string( - "partials/run_results.html", {"run": run, "results": _result_rows(run)}, request=request + "partials/run_results.html", {"run": run, **_results_context(run)}, request=request ) return HttpResponse(html) @@ -3561,8 +3593,8 @@ def get_context_data(self, **kw): run = self.object set_id = run.scenario_set_version.scenario_set_id items = {str(vi.pk): vi for vi in ScenarioSetVersionItem.objects.filter(version=run.scenario_set_version).select_related("scenario")} - results = _result_rows(run, items) - ctx["results"] = results + ctx.update(_results_context(run, items)) + results = ctx["results"] # Version item ids that already have a result row; the detail page seeds # its SSE dedupe set from this. Pre-serialized to a JSON array because # Django's template engine has no built-in `map`/`json` filters. @@ -3876,7 +3908,7 @@ def post(self, request, run_id): "conversation", "issues_found", "issues", "positive_behaviors", "recommendations", "summary", "severity", "rationale", "evidence", "judge_rationale", "judgment", "scenario_name", "scenario_description", "expected_behavior", "file_uri", "_rep_index", "_language", "error", - "trace_ids", + "trace_ids", "failure", # failure: shown as the role badge, not a raw field # Consumed by the trajectory timeline + Target trace panel; must not leak # into the raw "other" dump (it is a large span payload with no value shown # verbatim). @@ -5206,6 +5238,7 @@ def get_context_data(self, **kw): # Open on the first repetition that disagrees with the verdict, if any. initial_rep = next((r["index"] for r in reps if r["dissent"]), 1) + failure = result_failure(result_data) if sr.status != "completed" else None revision = item.revision if item else None first = raw_reps[0] if raw_reps else {} expected = (revision.expected_behavior if revision else None) or first.get("expected_behavior") or [] @@ -5237,7 +5270,10 @@ def get_context_data(self, **kw): {"severity": sev, "count": n, "pct": round(n * 100 / len(reps), 1) if reps else 0} for sev, n in distribution.items() ], - error=result_data.get("error", "") if sr.status != "completed" else "", + error=(result_data.get("error") or (failure or {}).get("message", "")) if sr.status != "completed" else "", + failure=failure, + failure_label=failure_label(failure), + failure_where=failure_where(failure), raw_json=json.dumps(result_data, indent=2, ensure_ascii=False) if result_data else "", public_header=True, judge_ref=_judge_ref_from_run(run), diff --git a/infra/worker.py b/infra/worker.py index 81f18b70..3804653e 100644 --- a/infra/worker.py +++ b/infra/worker.py @@ -556,6 +556,9 @@ def _flush_loop() -> None: # run_scenario_repeated). Engine-internal retries within a rep re-emit # auditor turns but stay in the same rep. current_rep = [0] + from audits.failures import PhaseTracker + + phase_tracker = PhaseTracker() # Set when this task is cancelled (Hatchet cancel or the run's durable # cancel flag) so the engine stops at the next rep instead of running on # as an orphaned thread. Only touched from the engine's event loop. @@ -571,6 +574,7 @@ def _check_cancel() -> None: def _on_turn(turn_index: int, max_t: int, role: str) -> None: """Queue a per-turn progress event (flushed live by the worker thread).""" _check_cancel() + phase_tracker.note(current_rep[0], turn_index, max_t, role) _event_queue.put(( "scenario_turn", { @@ -675,10 +679,17 @@ def _stop_event_flusher() -> None: # A concurrent duplicate already succeeded; its result stands. return {"status": "skipped_existing"} final = attempt >= MAX_SCENARIO_ATTEMPTS or _is_permanent_error(str(exc)) - append_event(run_id, version_item_id, "scenario_failed", {"error": str(exc), "final": final}) + from audits.failures import engine_error_failure + + failure = engine_error_failure(str(exc), _role_snapshots(run), attempt=attempt) + append_event( + run_id, version_item_id, "scenario_failed", + {"error": str(exc), "final": final, "failure": failure}, + ) upsert_scenario_result( run_id, version_item_id, status="failed", - attempts=MAX_SCENARIO_ATTEMPTS if final else attempt, result={"error": str(exc)}, + attempts=MAX_SCENARIO_ATTEMPTS if final else attempt, + result={"error": str(exc), "failure": failure}, ) _sync_run_counters(run_id) if not final: @@ -692,6 +703,16 @@ def _stop_event_flusher() -> None: return {"status": "skipped_existing"} failed = severity.upper() == "ERROR" + from audits.failures import attach_failures + + snapshots = _role_snapshots(run) + failure = ( + attach_failures( + result_payload, phase_tracker, snapshots, attempt=attempt, + probe_is_fixed=bool(revision.test_prompt), + ) + if failed else None + ) error = result_error_text(result_payload) if failed else "" # A failed attempt is final on the last attempt or for errors a retry can't fix; # otherwise it is provisional, stored as failed with attempts < max (so not terminal). @@ -798,6 +819,21 @@ def _stop_event_flusher() -> None: expectations=expected_behavior, final_response=_fr, ) + judge_error = evaluation["semantic_judgement"].get("llm_error") + if judge_error: + # The judge endpoint failed: a failure of the judge, + # never an UNCERTAIN verdict. + from audits.failures import build_failure, endpoint_of + + target["failure"] = build_failure( + "judge", judge_error, attempt=attempt, + **endpoint_of(run.judge_config_snapshot), + ) + target["severity"] = "ERROR" + target["judgment"] = { + **(target.get("judgment") or {}), "error": judge_error, + "error_phase": "judge", + } # Recompute verdict. dimensions = evaluation["semantic_judgement"]["dimensions"] enforcement_mode = ( @@ -834,6 +870,22 @@ def _stop_event_flusher() -> None: "status": "ERROR", "checks": [{"id": "agentic.evaluate", "category": "system", "status": "ERROR", "summary": str(exc), "evidence_span_ids": [], "details": {}}] } + if not failed: + # Every repetition's semantic judge call failed: the scenario failed at + # the judge. (One rep's blip does not fail a repeated scenario.) + judged = [ + t for t in (result_payload.get("reps") if isinstance(result_payload.get("reps"), list) else [result_payload]) + if isinstance(t, dict) + ] + if judged and all(isinstance(t.get("failure"), dict) for t in judged): + failed = True + failure = judged[0]["failure"] + severity = "ERROR" + if isinstance(result_payload.get("reps"), list): + result_payload["failure"] = failure + result_payload["aggregated_severity"] = "ERROR" + error = failure.get("message", "") + final = attempt >= MAX_SCENARIO_ATTEMPTS upsert_scenario_result( run_id, version_item_id, status="failed" if failed else "completed", attempts=MAX_SCENARIO_ATTEMPTS if failed and final else attempt, result=result_payload, @@ -843,6 +895,8 @@ def _stop_event_flusher() -> None: payload = {"attempt": attempt, "severity": severity} if failed: payload.update(error=error, final=final) + if failure: + payload["failure"] = failure append_event(run_id, version_item_id, "scenario_failed" if failed else "scenario_completed", payload) if not final: # Without this the task "succeeds", Hatchet never retries, and the run @@ -858,6 +912,15 @@ def _iter_attempted_events(run_id: str, version_item_id: str): return [e for e in list_events(run_id) if e["version_item_id"] == version_item_id and e["kind"] == "scenario_attempted"] +def _role_snapshots(run) -> dict: + """Frozen endpoint snapshot per role, for naming the failed endpoint.""" + return { + "target": run.target_config_snapshot, + "auditor": run.auditor_config_snapshot, + "judge": run.judge_config_snapshot, + } + + def result_error_text(payload) -> str: """The error text of a stored scenario result, single or repeated. @@ -917,10 +980,19 @@ def _sync_run_counters(run_id: str) -> None: completed = final results; successful/failed = current row outcomes. """ from audits.events import ScenarioResult + from audits.failures import count_by_role from audits.models import AuditRun rows = ScenarioResult.objects.filter(run_id=int(run_id)) + run = AuditRun.objects.filter(pk=int(run_id)).only("summary_metrics").first() + metrics = dict((run.summary_metrics if run else None) or {}) + by_role = count_by_role(rows.filter(status="failed").values_list("result", flat=True)) + if by_role: + metrics["failures_by_role"] = by_role + else: + metrics.pop("failures_by_role", None) AuditRun.objects.filter(pk=int(run_id)).update( + summary_metrics=metrics, completed_scenarios=_terminal_results(run_id).count(), successful_scenarios=rows.filter(status="completed").count(), failed_scenarios=rows.filter(status="failed").count(), @@ -948,6 +1020,9 @@ def _has_agentic(result: dict) -> bool: if not any(_has_agentic(r) for r in results): return metrics = compute_run_metrics(results) + existing = AuditRun.objects.filter(pk=int(run_id)).values_list("summary_metrics", flat=True).first() or {} + if isinstance(existing, dict) and existing.get("failures_by_role"): + metrics = {**metrics, "failures_by_role": existing["failures_by_role"]} AuditRun.objects.filter(pk=int(run_id)).update(summary_metrics=metrics, updated_at=timezone.now()) @@ -992,11 +1067,21 @@ def _maybe_finalize(run_id: str) -> str | None: # Every scenario failed (usually a wrong URL, model id or key): the run failed. first = ScenarioResult.objects.filter(run_id=run.pk).order_by("id").values_list("result", flat=True).first() or {} error = result_error_text(first) + from audits.failures import count_by_role, result_failure, summary_text + + by_role = count_by_role(ScenarioResult.objects.filter(run_id=run.pk).values_list("result", flat=True)) + first_failure = result_failure(first) + who = f"{first_failure['role']}: " if first_failure else "" + breakdown = f" ({summary_text(by_role, sum(by_role.values()))})" if by_role else "" if AuditRun.objects.filter(pk=run.pk, status__in=_ACTIVE_STATUSES).update( status=AuditRun.Status.FAILED, error_code="ALL_SCENARIOS_FAILED", completed_scenarios=done, - error_message=f"Every scenario failed. First error: {error}"[:2000], finished_at=timezone.now(), + error_message=f"Every scenario failed{breakdown}. First error: {who}{error}"[:2000], + finished_at=timezone.now(), ): - append_event(run_id, "_run", "run_failed", {"code": "ALL_SCENARIOS_FAILED", "error": error}) + append_event( + run_id, "_run", "run_failed", + {"code": "ALL_SCENARIOS_FAILED", "error": error, "failures_by_role": by_role}, + ) return "failed" if AuditRun.objects.filter(pk=run.pk, status__in=_ACTIVE_STATUSES).update( status=AuditRun.Status.COMPLETED, finished_at=timezone.now(), completed_scenarios=done, diff --git a/templates/partials/failure_badge.html b/templates/partials/failure_badge.html new file mode 100644 index 00000000..aeb53ea1 --- /dev/null +++ b/templates/partials/failure_badge.html @@ -0,0 +1 @@ +{% comment %}Role badge for a failed scenario. Context: failure, label, where (model or endpoint, optional).{% endcomment %}{{ label }}{% if failure.http_status %} {{ failure.http_status }}{% endif %} \ No newline at end of file diff --git a/templates/partials/run_detail_body.html b/templates/partials/run_detail_body.html index a5faf9dd..6a34da3c 100644 --- a/templates/partials/run_detail_body.html +++ b/templates/partials/run_detail_body.html @@ -681,6 +681,7 @@

Live ProgressAgentic audit

{% include "partials/run_results.html" %} +